Extend test_fetch_coredumps.py for zstd cases

Signed-off-by: Jose J Palacios-Perez <[email protected]>

Author: perezjosibm

teuthology/task/tests/test_fetch_coredumps.py

@@ -1,4 +1,5 @@
 from teuthology.task.internal import fetch_binaries_for_coredumps
+from teuthology.task.internal import get_backtraces_from_coredumps
 from unittest.mock import patch, Mock
 import gzip
 import os
@@ -19,6 +20,36 @@ def __init__(self, ret):
         def communicate(self, input=None):
             return [TestFetchCoreDumps.MockDecode(self.ret)]
 
+    class MockCompletedProcess(object):
+        def __init__(self, ret):
+            self.ret = ret
+
+        @property
+        def stdout(self):
+            return self.ret
+
+    class MockGdb(object):
+        def __init__(self, ret):
+            self.ret = ret
+
+        def run(self, *args, **kwargs):
+            return TestFetchCoreDumps.MockCompletedProcess(self.ret)
+
+    class TestGetBacktracesFromCoreDumps(object):
+        @patch('teuthology.task.internal.subprocess.run')
+        def test_get_backtraces_from_coredumps(self, mock_run):
+            mock_run.return_value = TestFetchCoreDumps.MockCompletedProcess(
+                "Backtrace line 1\nBacktrace line 2\nBacktrace line 3\n"
+            )
+            backtraces = get_backtraces_from_coredumps(coredump_path="core_dump_path", dump_path="binary_path", 
+                                                       dump_program="ceph_test_rados_api_io", dump="core_dump")
+            expected_backtraces = [
+                "Backtrace line 1",
+                "Backtrace line 2",
+                "Backtrace line 3"
+            ]
+            assert backtraces == expected_backtraces
+
     def setup_method(self):
         self.the_function = fetch_binaries_for_coredumps
         with gzip.open('file.gz', 'wb') as f:
@@ -44,6 +75,20 @@ def setup_method(self):
             " 19:56:56 2022, from Unix, original size modulo 2^32 11"
         )
 
+        # Centos 9 coredumps are zstd compressed:
+        self.zstd_compressed_correct = self.MockPopen(
+            "Zstandard compressed data"\
+            "'correct.format.core', last modified: Wed Jun 29"\
+            " 19:55:29 2022, from Unix, original size modulo 2^32 3167080"
+        )
+
+        self.zstd_compressed_incorrect = self.MockPopen(
+            "Zstandard compressed data"\
+            "'incorrect.format.core', last modified: Wed Jun 29"\
+            " 19:56:56 2022, from Unix, original size modulo 2^32 11"
+        )
+
+
     # Core is not compressed and file is in the correct format
     @patch('teuthology.task.internal.subprocess.Popen')
     @patch('teuthology.task.internal.os')
@@ -112,5 +157,39 @@ def test_compressed_incorrect_format(self, m_os, m_subproc_popen):
         self.the_function(None, self.m_remote)
         assert self.m_remote.get_file.called == False
 
+    # Core is zstd-compressed and file is in the correct format
+    @patch('teuthology.task.internal.subprocess.Popen')
+    @patch('teuthology.task.internal.os')
+    def test_zstd_compressed_correct_format(self, m_os, m_subproc_popen):
+        m_subproc_popen.side_effect = [
+                self.zstd_compressed_correct,
+                self.uncompressed_correct
+            ]
+        m_os.path.join.return_value = self.core_dump_path
+        m_os.path.sep = self.core_dump_path
+        m_os.path.isdir.return_value = True
+        m_os.path.dirname.return_value = self.core_dump_path
+        m_os.path.exists.return_value = True
+        m_os.listdir.return_value = [self.core_dump_path]
+        self.the_function(None, self.m_remote)
+        assert self.m_remote.get_file.called
+
+    # Core is compressed and file is in the wrong format
+    @patch('teuthology.task.internal.subprocess.Popen')
+    @patch('teuthology.task.internal.os')
+    def test_zstd_compressed_incorrect_format(self, m_os, m_subproc_popen):
+        m_subproc_popen.side_effect = [
+                self.zstd_compressed_incorrect,
+                self.uncompressed_incorrect
+            ]
+        m_os.path.join.return_value = self.core_dump_path
+        m_os.path.sep = self.core_dump_path
+        m_os.path.isdir.return_value = True
+        m_os.path.dirname.return_value = self.core_dump_path
+        m_os.path.exists.return_value = True
+        m_os.listdir.return_value = [self.core_dump_path]
+        self.the_function(None, self.m_remote)
+        assert self.m_remote.get_file.called == False
+
     def teardown(self):
-        os.remove(self.core_dump_path)
+        os.remove(self.core_dump_path)