tests: added test_validate_error_intermediate_backups()

Author: gsmolk

tests/validate_test.py

@@ -411,6 +411,100 @@ def test_validate_corrupted_intermediate_backups(self):
         # Clean after yourself
         self.del_test_dir(module_name, fname)
 
+    # @unittest.skip("skip")
+    def test_validate_error_intermediate_backups(self):
+        """
+        make archive node, take FULL, PAGE1, PAGE2 backups,
+        change backup status of FULL and PAGE1 to ERROR,
+        run validate on PAGE1
+        purpouse of this test is to be sure that not only
+        CORRUPT backup descendants can be orphanized
+        """
+        fname = self.id().split('.')[3]
+        node = self.make_simple_node(
+            base_dir=os.path.join(module_name, fname, 'node'),
+            initdb_params=['--data-checksums'],
+            pg_options={'wal_level': 'replica'}
+            )
+        backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
+        self.init_pb(backup_dir)
+        self.add_instance(backup_dir, 'node', node)
+        self.set_archiving(backup_dir, 'node', node)
+        node.slow_start()
+
+        # FULL
+        backup_id_1 = self.backup_node(backup_dir, 'node', node)
+
+        # PAGE1
+        backup_id_2 = self.backup_node(
+            backup_dir, 'node', node, backup_type='page')
+
+        # PAGE2
+        backup_id_3 = self.backup_node(
+            backup_dir, 'node', node, backup_type='page')
+
+        # Change FULL backup status to ERROR
+        control_path = os.path.join(
+            backup_dir, 'backups', 'node', backup_id_1, 'backup.control')
+
+        with open(control_path, 'r') as f:
+            actual_control = f.read()
+
+        print(actual_control)
+
+        new_control_file = ''
+        for line in actual_control.splitlines():
+            new_control_file += line.replace(
+                'status = OK', 'status = ERROR')
+            new_control_file += '\n'
+
+        with open(control_path, 'wt') as f:
+            f.write(new_control_file)
+            f.flush()
+            f.close()
+
+        print('HELLO')
+        print(new_control_file)
+
+        # Validate PAGE1
+        try:
+            self.validate_pb(
+                backup_dir, 'node', backup_id=backup_id_2)
+            self.assertEqual(
+                1, 0,
+                "Expecting Error because backup has status ERROR.\n "
+                "Output: {0} \n CMD: {1}".format(
+                    repr(self.output), self.cmd))
+        except ProbackupException as e:
+            self.assertTrue(
+                'WARNING: Backup {0} is orphaned because '
+                'his parent {1} has status: ERROR'.format(
+                    backup_id_2, backup_id_1) in e.message and
+                'INFO: Validating parents for backup {0}'.format(
+                    backup_id_2) in e.message and
+                'WARNING: Backup {0} has status ERROR. Skip validation.'.format(
+                    backup_id_1) and
+                'ERROR: Backup {0} is orphan.'.format(backup_id_2) in e.message,
+                '\n Unexpected Error Message: {0}\n '
+                'CMD: {1}'.format(
+                    repr(e.message), self.cmd))
+
+        self.assertEqual(
+            'ERROR',
+            self.show_pb(backup_dir, 'node', backup_id_1)['status'],
+            'Backup STATUS should be "ERROR"')
+        self.assertEqual(
+            'ORPHAN',
+            self.show_pb(backup_dir, 'node', backup_id_2)['status'],
+            'Backup STATUS should be "ORPHAN"')
+        self.assertEqual(
+            'ORPHAN',
+            self.show_pb(backup_dir, 'node', backup_id_3)['status'],
+            'Backup STATUS should be "ORPHAN"')
+
+        # Clean after yourself
+        self.del_test_dir(module_name, fname)
+
     # @unittest.skip("skip")
     def test_validate_corrupted_intermediate_backups_1(self):
         """