Merge pull request #1641 from nharraud/fix-upgrade-replay

upgrade: fix upgrade replay

b2share/modules/upgrade/api.py

@@ -75,6 +75,8 @@ def upgrade_to_last_version(verbose):
                 if last_migration.version == last_version:
                     click.secho('Already up to date.')
                     return
+            else:
+                last_failure = last_migration
             try:
                 last_success = next(mig for mig in all_migrations
                                     if mig.success)
@@ -93,6 +95,27 @@ def upgrade_to_last_version(verbose):
 Step = namedtuple('Step', ['condition', 'run'])
 
 
+def default_step_condition_factory(target_version, step_name):
+    """Create a default condition used by upgrade steps.
+
+    The created condition checks if the step was already successfully ran
+    and skip it if it is the case.
+    """
+    def default_step_condition(alembic, failed_migration, *args):
+        if failed_migration and failed_migration.version == target_version:
+            try:
+                step = next(step for step in failed_migration.data['steps']
+                            if step['name'] == step_name)
+                # skip the step if it already ran successfully
+                if step['status'] in {'success', 'skip'}:
+                    return False
+            except StopIteration:
+                # The step was not run at all.
+                pass
+        return True
+    return default_step_condition
+
+
 class UpgradeRecipe(object):
     """A B2SHARE upgrade which migrates from one B2SHARE version to another.
 
@@ -137,13 +160,24 @@ def decorator(step_func):
             assert step_func.__doc__ is not None
             # check for duplicate step names
             assert step_func.__name__ not in self._step_names
+            final_condition = condition
+            # Use the default condition if none is provided
+            if condition is None:
+                final_condition = default_step_condition_factory(
+                    self.dst_version, step_func.__name__
+                )
             self.steps.append(
-                Step(condition=condition, run=step_func)
+                Step(condition=final_condition, run=step_func)
             )
             self._step_names.add(step_func.__name__)
             return step_func
         return decorator
 
+    def remove_step(self, step_func):
+        """Remove a step from the list of upgrade steps."""
+        self.steps = filter(lambda step: step.run != step_func, self.steps)
+        self._step_names.remove(step_func.__name__)
+
 
     @classmethod
     def load(cls):
@@ -190,6 +224,10 @@ def run(self, failed_migration=None, verbose=None):
                     step_log['status'] = 'skip'
             except BaseException as e:
                 db.session.rollback()
+                migration.data['steps'].append(dict(
+                    name=step.run.__name__,
+                    status='error'
+                ))
                 migration.data['error'] = traceback.format_exc()
                 migration.data['status'] = 'error'
                 if not db.engine.dialect.has_table(db.engine,

tests/b2share_unit_tests/b2share_db_load_data.sql

@@ -512,7 +512,7 @@ INSERT INTO records_metadata (created, updated, id, json, version_id) VALUES ('2
 INSERT INTO records_metadata (created, updated, id, json, version_id) VALUES ('2017-11-04 09:04:47.367063', '2017-11-04 09:04:48.096744', '9544056e-dff2-505f-b561-1f11127b817e', '{"resource_types": [{"resource_type_general": "Text"}], "community": "0afede87-2bf2-4d89-867e-d2ee57251c62", "_pid": [{"type": "b2rec", "value": "9544056edff2505fb5611f11127b817e"}], "creators": [{"creator_name": "Daniel Zeman"}], "descriptions": [{"description_type": "Abstract", "description": "This is an unpublished draft record description."}], "$schema": "http://localhost:5000/communities/0afede87-2bf2-4d89-867e-d2ee57251c62/schemas/0#/draft_json_schema", "_deposit": {"id": "1033083fedf4408fb5611f23527a926d", "pid": {"type": "b2rec", "revision_id": 0, "value": "1033083fedf4408fb5611f23527a926d"}, "owners": [1], "status": "published", "created_by": 1}, "community_specific": {"2a01ee91-36fe-4edb-9734-73d22ac78821": {"ling_resource_type": ["Text"], "language_code": "eng"}}, "contact_email": "[email protected]", "titles": [{"title": "This is an unpublished a draft record"}], "publication_state": "draft", "open_access": true}', 2);
 --- Soft deleted deposit and its non deleted record
 INSERT INTO records_metadata (created, updated, id, json, version_id) VALUES ('2016-12-21 13:55:36.136869', '2016-12-21 15:04:21.185352', '9d2cba26-de44-42af-9633-140e35e8a357', NULL, 2);
-INSERT INTO records_metadata (created, updated, id, json, version_id) VALUES ('2016-12-21 13:55:36.136869', '2016-12-21 15:04:21.185352', 'ed6c104a-1243-4ed8-86c5-2b1691b81f37', '{"titles": [{"title": "test_upload"}], "open_access": true, "_pid": [{"type": "b2rec", "value": "9d2cba26de4442af9633140e35e8a357"}, {"type": "ePIC_PID", "value": "http://hdl.handle.net/11304/a42592ce-aee6-4b6f-9bd1-8909c7625e03"}], "creators": [{"creator_name": "John Doe"}], "$schema": "https://trng-b2share.eudat.eu/api/communities/e9b9792e-79fb-4b07-b6b4-b9c2bd06d095/schemas/0#/json_schema", "license": {"license_uri": "http://creativecommons.org/publicdomain/mark/1.0/", "license": "Public Domain Mark (PD)"}, "community_specific": {}, "_files": [{"version_id": "994b69ac-9ed8-49b2-8a33-89c8e8508da5", "bucket": "02e429f2-3b18-45d4-9089-79dece4caba5", "key": "512MB.file", "checksum": "md5:aa559b4e3523a6c931f08f4df52d58f2", "ePIC_PID": "http://hdl.handle.net/11304/76166cb2-f453-4a07-b7a8-68bea9db51b3", "size": 536870912}], "community": "e9b9792e-79fb-4b07-b6b4-b9c2bd06d095", "publication_state": "published", "_oai": {"id": "oai:b2share.eudat.eu:b2rec/9d2cba26de4442af9633140e35e8a357", "sets": [], "updated": "2016-12-21T13:59:10Z"}, "descriptions": [{"description_type": "Abstract", "description": "One 512MB file"}], "_deposit": {"id": "9d2cba26de4442af9633140e35e8a357", "status": "published", "created_by": 2, "pid": {"type": "b2rec", "revision_id": 0, "value": "9d2cba26de4442af9633140e35e8a357"}, "owners": [2]}, "keywords": ["one gigabyte file"]}', 2);
+INSERT INTO records_metadata (created, updated, id, json, version_id) VALUES ('2016-12-21 13:55:36.136869', '2016-12-21 15:04:21.185352', 'ed6c104a-1243-4ed8-86c5-2b1691b81f37', '{"titles": [{"title": "test_upload"}], "open_access": true, "_pid": [{"type": "b2rec", "value": "9d2cba26de4442af9633140e35e8a357"}, {"type": "ePIC_PID", "value": "http://hdl.handle.net/11304/a42592ce-aee6-4b6f-9bd1-8909c7625e03"}], "creators": [{"creator_name": "John Doe"}], "$schema": "http://localhost:5000/communities/e9b9792e-79fb-4b07-b6b4-b9c2bd06d095/schemas/0#/json_schema", "license": {"license_uri": "http://creativecommons.org/publicdomain/mark/1.0/", "license": "Public Domain Mark (PD)"}, "community_specific": {}, "_files": [{"version_id": "994b69ac-9ed8-49b2-8a33-89c8e8508da5", "bucket": "02e429f2-3b18-45d4-9089-79dece4caba5", "key": "512MB.file", "checksum": "md5:aa559b4e3523a6c931f08f4df52d58f2", "ePIC_PID": "http://hdl.handle.net/11304/76166cb2-f453-4a07-b7a8-68bea9db51b3", "size": 536870912}], "community": "e9b9792e-79fb-4b07-b6b4-b9c2bd06d095", "publication_state": "published", "_oai": {"id": "oai:b2share.eudat.eu:b2rec/9d2cba26de4442af9633140e35e8a357", "sets": [], "updated": "2016-12-21T13:59:10Z"}, "descriptions": [{"description_type": "Abstract", "description": "One 512MB file"}], "_deposit": {"id": "9d2cba26de4442af9633140e35e8a357", "status": "published", "created_by": 2, "pid": {"type": "b2rec", "revision_id": 0, "value": "9d2cba26de4442af9633140e35e8a357"}, "owners": [2]}, "keywords": ["one gigabyte file"]}', 2);
 --- Mis-deleted record. The pid is not deleted but the record is.
 INSERT INTO records_metadata (created, updated, id, json, version_id) VALUES ('2016-12-21 13:55:36.136869', '2016-12-21 15:04:21.185352', '331acc49-d97a-4415-969d-97c82424fc99', NULL, 1);
 

tests/b2share_unit_tests/upgrade/expected_records.json

@@ -51,7 +51,7 @@
         "created": "2016-12-21 13:55:36.136869",
         "updated": "2016-12-21 15:04:21.185352",
         "id": "ed6c104a-1243-4ed8-86c5-2b1691b81f37",
-        "metadata": {"titles": [{"title": "test_upload"}], "open_access": true, "_pid": [{"type": "b2rec", "value": "9d2cba26de4442af9633140e35e8a357"}, {"type": "ePIC_PID", "value": "http://hdl.handle.net/11304/a42592ce-aee6-4b6f-9bd1-8909c7625e03"}], "creators": [{"creator_name": "John Doe"}], "$schema": "https://trng-b2share.eudat.eu/api/communities/e9b9792e-79fb-4b07-b6b4-b9c2bd06d095/schemas/0#/json_schema", "license": {"license_uri": "http://creativecommons.org/publicdomain/mark/1.0/", "license": "Public Domain Mark (PD)"}, "community_specific": {}, "_files": [{"version_id": "994b69ac-9ed8-49b2-8a33-89c8e8508da5", "bucket": "02e429f2-3b18-45d4-9089-79dece4caba5", "key": "512MB.file", "checksum": "md5:aa559b4e3523a6c931f08f4df52d58f2", "ePIC_PID": "http://hdl.handle.net/11304/76166cb2-f453-4a07-b7a8-68bea9db51b3", "size": 536870912}], "community": "e9b9792e-79fb-4b07-b6b4-b9c2bd06d095", "publication_state": "published", "_oai": {"id": "oai:b2share.eudat.eu:b2rec/9d2cba26de4442af9633140e35e8a357", "sets": [], "updated": "2016-12-21T13:59:10Z"}, "descriptions": [{"description_type": "Abstract", "description": "One 512MB file"}], "_deposit": {"id": "9d2cba26de4442af9633140e35e8a357", "status": "published", "created_by": 2, "pid": {"type": "b2rec", "revision_id": 0, "value": "9d2cba26de4442af9633140e35e8a357"}, "owners": [2]}, "keywords": ["one gigabyte file"]}
+        "metadata": {"titles": [{"title": "test_upload"}], "open_access": true, "_pid": [{"type": "b2rec", "value": "9d2cba26de4442af9633140e35e8a357"}, {"type": "ePIC_PID", "value": "http://hdl.handle.net/11304/a42592ce-aee6-4b6f-9bd1-8909c7625e03"}], "creators": [{"creator_name": "John Doe"}], "$schema": "http://localhost:5000/communities/e9b9792e-79fb-4b07-b6b4-b9c2bd06d095/schemas/0#/json_schema", "license": {"license_uri": "http://creativecommons.org/publicdomain/mark/1.0/", "license": "Public Domain Mark (PD)"}, "community_specific": {}, "_files": [{"version_id": "994b69ac-9ed8-49b2-8a33-89c8e8508da5", "bucket": "02e429f2-3b18-45d4-9089-79dece4caba5", "key": "512MB.file", "checksum": "md5:aa559b4e3523a6c931f08f4df52d58f2", "ePIC_PID": "http://hdl.handle.net/11304/76166cb2-f453-4a07-b7a8-68bea9db51b3", "size": 536870912}], "community": "e9b9792e-79fb-4b07-b6b4-b9c2bd06d095", "publication_state": "published", "_oai": {"id": "oai:b2share.eudat.eu:b2rec/9d2cba26de4442af9633140e35e8a357", "sets": [], "updated": "2016-12-21T13:59:10Z"}, "descriptions": [{"description_type": "Abstract", "description": "One 512MB file"}], "_deposit": {"id": "9d2cba26de4442af9633140e35e8a357", "status": "published", "created_by": 2, "pid": {"type": "b2rec", "revision_id": 0, "value": "9d2cba26de4442af9633140e35e8a357"}, "owners": [2]}, "keywords": ["one gigabyte file"]}
     },
     {
         "created": "2016-12-21 13:55:36.136869",

tests/b2share_unit_tests/upgrade/test_upgrade.py

@@ -214,50 +214,62 @@ def test_upgrade_from_v2_0_0(clean_app):
 
 def test_failed_and_repair_upgrade_from_v2_0_0(clean_app):
     """Test upgrade from an unknown B2SHARE version."""
+
+    def failing_step(alembic, verbose):
+        """Failing step used to make the upgrade fail."""
+        raise Exception('This error is on purpose')
+
     with clean_app.app_context():
+        expected_migrations = UpgradeRecipe.build_upgrade_path(
+            '2.0.0', current_migration_version)
         ext = clean_app.extensions['invenio-db']
         if db.engine.name == 'sqlite':
             raise pytest.skip('upgrades are not supported on sqlite.')
 
         # bring db to v2.0.1 state
         db_create_v2_0_1()
+        # Add a table which shouldn't exist in 2.0.1 version. This will
+        # Make the 2.0.0->2.1.0 migration fail.
         Migration.__table__.create(db.engine)
         db.session.commit()
         result = upgrade_run(clean_app)
         assert result.exception.__class__ == sa.exc.ProgrammingError
 
         # Drop the problematic table
         Migration.__table__.drop(db.engine)
-        # Make a record impossible to migrate
-        query = ("UPDATE {tablename} "
-                 "SET json={value} "
-                 "WHERE id='{id}'::uuid")
-        db.session.execute(
-            query.format(
-                tablename=RecordMetadata.__tablename__,
-                id='331acc49-d97a-4415-969d-97c82424fc99',
-                value='\'{"wrong": "field"}\''
-            )
-        )
-        # Run again a failing upgrade.
-        result = upgrade_run(clean_app)
-        assert result.exit_code == -1
-        # This time the upgrade succeeded partially, the db schema is migrated
-        # But not the records.
-        migrations = Migration.query.all()
-        assert len(migrations) == 1
-        mig = migrations[0]
-        assert mig.version == '2.1.0'
-        assert not mig.success
-        assert not ext.alembic.compare_metadata()
-        # Fix the record
-        db.session.execute(
-            query.format(
-                tablename=RecordMetadata.__tablename__,
-                id='331acc49-d97a-4415-969d-97c82424fc99',
-                value='NULL'
-            )
-        )
+
+        # Fail every other migration one by one. Fixing the previous one at
+        # each iteration.
+        expected_number_of_migrations = 0
+        first_loop = True
+        for migration_idx  in range(len(expected_migrations)):
+            migration = expected_migrations[migration_idx]
+            # Add a failing step to the migration
+            migration.step()(failing_step)
+            # Run again a failing upgrade.
+            result = upgrade_run(clean_app)
+            assert result.exit_code == -1
+            # This time the upgrade succeeded partially, the db schema is migrated
+            # But not the records.
+            migrations = Migration.query.order_by(
+                Migration.updated.asc()).all()
+            # The first failing migration is expected to save only one
+            # "Migration" row in the database, every other migration should
+            # have 2 migrations (the successful previous migration and the
+            # next failing one).
+            if first_loop:
+                expected_number_of_migrations += 1
+                first_loop = False
+            else:
+                expected_number_of_migrations += 2
+
+            assert len(migrations) == expected_number_of_migrations
+            number_of_migrations = len(migrations)
+            last_migration = migrations[-1]
+            assert last_migration.version == migration.dst_version
+            assert not last_migration.success
+            # Remove the failing migration step
+            migration.remove_step(failing_step)
         # Migrate successfully
         result = upgrade_run(clean_app)
         assert result.exit_code == 0
@@ -266,17 +278,15 @@ def test_failed_and_repair_upgrade_from_v2_0_0(clean_app):
 
         # check that the migration information have been saved
         migrations = Migration.query.order_by(Migration.created.asc()).all()
-        expected_migrations = UpgradeRecipe.build_upgrade_path(
-            '2.0.0', current_migration_version)
         # Failed release + total succeeded releases
-        assert len(migrations) == 1 + len(expected_migrations)
+        expected_number_of_migrations += 1
+        assert len(migrations) == expected_number_of_migrations
+        # Check that every migration ran successfully
         assert [mig.version for mig in migrations if mig.success] == \
             [mig.dst_version for mig in expected_migrations]
-        # assert mig.version == current_migration_version
-        # assert mig.success
 
     with clean_app.app_context():
-        repeat_upgrade(clean_app, ext.alembic, 1 + len(expected_migrations))
+        repeat_upgrade(clean_app, ext.alembic, expected_number_of_migrations)
 
     with clean_app.app_context():
         validate_loaded_data(clean_app, ext.alembic)