[PBCKP-304] cfs tests moved back to build

Author: Ivan Lazarev

tests/__init__.py

@@ -35,9 +35,9 @@ def load_tests(loader, tests, pattern):
         suite.addTests(loader.loadTestsFromModule(compatibility))
     suite.addTests(loader.loadTestsFromModule(checkdb))
     suite.addTests(loader.loadTestsFromModule(config))
-#    suite.addTests(loader.loadTestsFromModule(cfs_backup))
-#    suite.addTests(loader.loadTestsFromModule(cfs_restore))
-#    suite.addTests(loader.loadTestsFromModule(cfs_validate_backup))
+    suite.addTests(loader.loadTestsFromModule(cfs_backup))
+    suite.addTests(loader.loadTestsFromModule(cfs_restore))
+    suite.addTests(loader.loadTestsFromModule(cfs_validate_backup))
     suite.addTests(loader.loadTestsFromModule(compression))
     suite.addTests(loader.loadTestsFromModule(delete))
     suite.addTests(loader.loadTestsFromModule(delta))

tests/cfs_backup.py

@@ -4,21 +4,22 @@
 import shutil
 
 from .helpers.cfs_helpers import find_by_extensions, find_by_name, find_by_pattern, corrupt_file
-from .helpers.ptrack_helpers import ProbackupTest, ProbackupException
+from .helpers.ptrack_helpers import ProbackupTest, ProbackupException, is_test_result_ok
 
-module_name = 'cfs_backup'
 tblspace_name = 'cfs_tblspace'
 
 
 class CfsBackupNoEncTest(ProbackupTest, unittest.TestCase):
     # --- Begin --- #
     @unittest.skipUnless(ProbackupTest.enterprise, 'skip')
     def setUp(self):
+        self.module_name = self.id().split('.')[1]
         self.fname = self.id().split('.')[3]
+
         self.backup_dir = os.path.join(
-            self.tmp_path, module_name, self.fname, 'backup')
+            self.tmp_path, self.module_name, self.fname, 'backup')
         self.node = self.make_simple_node(
-            base_dir="{0}/{1}/node".format(module_name, self.fname),
+            base_dir="{0}/{1}/node".format(self.module_name, self.fname),
             set_replication=True,
             ptrack_enable=True,
             initdb_params=['--data-checksums'],
@@ -1159,10 +1160,12 @@ def test_broken_file_pg_compression_into_tablespace_dir(self):
         )
 
 #    # --- End ---#
-#    @unittest.skipUnless(ProbackupTest.enterprise, 'skip')
-#    def tearDown(self):
-#        self.node.cleanup()
-#        self.del_test_dir(module_name, self.fname)
+    @unittest.skipUnless(ProbackupTest.enterprise, 'skip')
+    def tearDown(self):
+        module_name = self.id().split('.')[1]
+        fname = self.id().split('.')[3]
+        if is_test_result_ok(self):
+            self.del_test_dir(module_name, fname)
 
 
 #class CfsBackupEncTest(CfsBackupNoEncTest):

tests/cfs_restore.py

@@ -13,22 +13,21 @@
 import shutil
 
 from .helpers.cfs_helpers import find_by_name
-from .helpers.ptrack_helpers import ProbackupTest, ProbackupException
-
-
-module_name = 'cfs_restore'
+from .helpers.ptrack_helpers import ProbackupTest, ProbackupException, is_test_result_ok
 
 tblspace_name = 'cfs_tblspace'
 tblspace_name_new = 'cfs_tblspace_new'
 
 
 class CfsRestoreBase(ProbackupTest, unittest.TestCase):
     def setUp(self):
+        self.module_name = self.id().split('.')[1]
         self.fname = self.id().split('.')[3]
-        self.backup_dir = os.path.join(self.tmp_path, module_name, self.fname, 'backup')
+
+        self.backup_dir = os.path.join(self.tmp_path, self.module_name, self.fname, 'backup')
 
         self.node = self.make_simple_node(
-            base_dir="{0}/{1}/node".format(module_name, self.fname),
+            base_dir="{0}/{1}/node".format(self.module_name, self.fname),
             set_replication=True,
             initdb_params=['--data-checksums'],
             pg_options={
@@ -60,9 +59,12 @@ def setUp(self):
     def add_data_in_cluster(self):
         pass
 
+    @unittest.skipUnless(ProbackupTest.enterprise, 'skip')
     def tearDown(self):
-        self.node.cleanup()
-        self.del_test_dir(module_name, self.fname)
+        module_name = self.id().split('.')[1]
+        fname = self.id().split('.')[3]
+        if is_test_result_ok(self):
+            self.del_test_dir(module_name, fname)
 
 
 class CfsRestoreNoencEmptyTablespaceTest(CfsRestoreBase):
@@ -102,8 +104,7 @@ def test_restore_empty_tablespace_from_fullbackup(self):
         tblspace = self.node.safe_psql(
             "postgres",
             "SELECT * FROM pg_tablespace WHERE spcname='{0}'".format(tblspace_name)
-        )
-        tblspace = str(tblspace)
+        ).decode("UTF-8")
         self.assertTrue(
             tblspace_name in tblspace and "compression=true" in tblspace,
             "ERROR: The tablespace not restored or it restored without compressions"
@@ -212,7 +213,7 @@ def test_restore_from_fullbackup_to_new_location(self):
         self.node.cleanup()
         shutil.rmtree(self.get_tblspace_path(self.node, tblspace_name))
 
-        node_new = self.make_simple_node(base_dir="{0}/{1}/node_new_location".format(module_name, self.fname))
+        node_new = self.make_simple_node(base_dir="{0}/{1}/node_new_location".format(self.module_name, self.fname))
         node_new.cleanup()
 
         try:
@@ -255,7 +256,7 @@ def test_restore_from_fullbackup_to_new_location_5_jobs(self):
         self.node.cleanup()
         shutil.rmtree(self.get_tblspace_path(self.node, tblspace_name))
 
-        node_new = self.make_simple_node(base_dir="{0}/{1}/node_new_location".format(module_name, self.fname))
+        node_new = self.make_simple_node(base_dir="{0}/{1}/node_new_location".format(self.module_name, self.fname))
         node_new.cleanup()
 
         try:

tests/helpers/ptrack_helpers.py

@@ -1,6 +1,7 @@
 # you need os for unittest to work
 import os
 import gc
+import unittest
 from sys import exit, argv, version_info
 import subprocess
 import shutil
@@ -172,6 +173,30 @@ def slow_start(self, replica=False):
 
         sleep(0.5)
 
+
+def is_test_result_ok(test_case):
+    # sources of solution:
+    # 1. python versions 2.7 - 3.10, verified on 3.10, 3.7, 2.7, taken from:
+    # https://tousu.in/qa/?qa=555402/unit-testing-getting-pythons-unittest-results-in-a-teardown-method&show=555403#a555403
+    #
+    # 2. python versions 3.11+ mixin, verified on 3.11, taken from: https://stackoverflow.com/a/39606065
+
+    if hasattr(test_case, '_outcome'):  # Python 3.4+
+        if hasattr(test_case._outcome, 'errors'):
+            # Python 3.4 - 3.10  (These two methods have no side effects)
+            result = test_case.defaultTestResult()  # These two methods have no side effects
+            test_case._feedErrorsToResult(result, test_case._outcome.errors)
+        else:
+            # Python 3.11+
+            result = test_case._outcome.result
+    else:  # Python 2.7, 3.0-3.3
+        result = getattr(test_case, '_outcomeForDoCleanups', test_case._resultForDoCleanups)
+
+    ok = all(test != test_case for test, text in result.errors + result.failures)
+
+    return ok
+
+
 class ProbackupTest(object):
     # Class attributes
     enterprise = is_enterprise()