Require alembic db migration scripts to be idempotent

With this patch alembic migration script should always be idempotent.
This is forced by new functional test which makes sure that migrations
starting from first alembic migration script from the stable/2024.1
release is idempotent.

Closes-bug: #2100770
Change-Id: Ia14f3748245b59850bc21cbc87e04ffbdbb5850f

doc/source/contributor/alembic_migrations.rst

@@ -161,6 +161,31 @@ neutron-db-manage command can be used to generate migration scripts for you to
 complete. The operations in the template are those supported by the Alembic
 migration library.
 
+The neutron-db-manage command is designed for sequential database upgrades.
+Typically, it's executed once to reach a specific schema version. Subsequent
+executions should then resume from that achieved point. This design
+theoretically prevents scenarios where the same upgrade script runs multiple
+times, such as attempting to create an already existing table.
+However, due to potential real-world issues such as loss of the recorded
+database version or backporting of scripts, Alembic migration scripts must be
+idempotent. Therefore, scripts that modify the schema, such as adding a new
+table, should first check for the object's existence, for example:
+
+.. code-block:: python
+
+        import sqlalchemy as sa
+
+        from neutron.db import migration
+
+        def upgrade():
+            migration.add_column_if_not_exists(
+                'networks',
+                sa.Column('qinq', sa.Boolean(), server_default=None)
+            )
+
+Module `neutron.db.migration` provides helper functions to create table or add
+column to the existing table in the idempotent way.
+
 
 .. _neutron-db-manage-without-devstack:
 

neutron/db/migration/__init__.py

@@ -118,6 +118,21 @@ def schema_has_column(table_name, column_name):
                            insp.get_columns(table_name)]
 
 
+@raise_if_offline
+def create_table_if_not_exists(table_name, *args, **kwargs):
+    table = None
+    if not schema_has_table(table_name):
+        table = op.create_table(table_name, *args, **kwargs)
+    return table
+
+
+@raise_if_offline
+def add_column_if_not_exists(table_name, column, **kwargs):
+    """Add column only if it not exists in the schema."""
+    if not schema_has_column(table_name, column.name):
+        op.add_column(table_name, column, **kwargs)
+
+
 @raise_if_offline
 def alter_column_if_exists(table_name, column_name, **kwargs):
     """Alter a column only if it exists in the schema."""

neutron/db/migration/alembic_migrations/versions/2024.1/expand/0e6eff810791_add_port_hardware_offload_extension_type.py

@@ -13,7 +13,6 @@
 #    under the License.
 #
 
-from alembic import op
 import sqlalchemy as sa
 
 from neutron.db import migration
@@ -36,14 +35,15 @@ neutron_milestone = [migration.RELEASE_2024_1]
 
 
 def upgrade():
-    op.create_table('porthardwareoffloadtype',
-                    sa.Column('port_id',
-                              sa.String(36),
-                              sa.ForeignKey('ports.id',
-                                            ondelete="CASCADE"),
-                              primary_key=True,
-                              index=True),
-                    sa.Column('hardware_offload_type',
-                              sa.String(255),
-                              nullable=True)
-                    )
+    migration.create_table_if_not_exists(
+        'porthardwareoffloadtype',
+        sa.Column('port_id',
+                  sa.String(36),
+                  sa.ForeignKey('ports.id',
+                                ondelete="CASCADE"),
+                  primary_key=True,
+                  index=True),
+        sa.Column('hardware_offload_type',
+                  sa.String(255),
+                  nullable=True)
+    )

neutron/db/migration/alembic_migrations/versions/2024.2/expand/5bcb7b31ec7d_add_port_trusted_attribute.py

@@ -38,7 +38,7 @@ neutron_milestone = [migration.RELEASE_2024_2]
 
 
 def upgrade():
-    port_trusted_table = op.create_table(
+    port_trusted_table = migration.create_table_if_not_exists(
         'porttrusted',
         sa.Column('port_id',
                   sa.String(36),
@@ -49,6 +49,11 @@ def upgrade():
                   sa.Boolean,
                   nullable=True))
 
+    if port_trusted_table is None:
+        # Table was already created before so no need to insert any data
+        # to it now
+        return
+
     # A simple model of the ml2_port_bindings table, just to get and update
     # binding:profile fields where needed
     port_binding_table = sa.Table(

neutron/db/migration/alembic_migrations/versions/2025.1/expand/ad80a9f07c5c_add_vlan_qinq_column_to_the_network_.py

@@ -13,7 +13,6 @@
 #    under the License.
 #
 
-from alembic import op
 import sqlalchemy as sa
 
 from neutron.db import migration
@@ -34,7 +33,7 @@ neutron_milestone = [migration.RELEASE_2025_1]
 
 
 def upgrade():
-    op.add_column(
+    migration.add_column_if_not_exists(
         'networks',
         sa.Column('qinq', sa.Boolean(), server_default=None)
     )

neutron/tests/functional/db/test_migrations.py

@@ -505,22 +505,10 @@ class TestWalkDowngrade(oslotest_base.BaseTestCase):
             return True
 
 
-class TestWalkMigrations(testlib_api.MySQLTestCaseMixin,
-                         testlib_api.SqlTestCaseLight):
-    '''This will add framework for testing schema migration
-       for different backends.
-
-    '''
+class _BaseTestWalkMigrations(object):
 
     BUILD_SCHEMA = False
 
-    def execute_cmd(self, cmd=None):
-        with subprocess.Popen(cmd, stdout=subprocess.PIPE,
-                              stderr=subprocess.STDOUT, shell=True) as proc:
-            output = proc.communicate()[0]
-            self.assertEqual(0, proc.returncode, 'Command failed with '
-                             'output:\n%s' % output)
-
     def _get_alembic_config(self, uri):
         db_config = migration.get_neutron_config()
         self.script_dir = alembic_script.ScriptDirectory.from_config(db_config)
@@ -530,6 +518,22 @@ class TestWalkMigrations(testlib_api.MySQLTestCaseMixin,
                                               group='database')
         return db_config
 
+
+class TestWalkMigrations(_BaseTestWalkMigrations,
+                         testlib_api.MySQLTestCaseMixin,
+                         testlib_api.SqlTestCaseLight):
+    '''This will add framework for testing schema migration
+       for different backends.
+
+    '''
+
+    def execute_cmd(self, cmd=None):
+        with subprocess.Popen(cmd, stdout=subprocess.PIPE,
+                              stderr=subprocess.STDOUT, shell=True) as proc:
+            output = proc.communicate()[0]
+            self.assertEqual(0, proc.returncode, 'Command failed with '
+                             'output:\n%s' % output)
+
     def _revisions(self):
         """Provides revisions and its parent revisions.
 
@@ -578,3 +582,55 @@ class TestWalkMigrations(testlib_api.MySQLTestCaseMixin,
 
         if upgrade_dest:
             migration.do_alembic_command(config, 'upgrade', upgrade_dest)
+
+
+class TestMigrationsIdempotency(_BaseTestWalkMigrations,
+                                testlib_api.MySQLTestCaseMixin,
+                                testlib_api.SqlTestCaseLight):
+    '''This class tests if the migration scripts are idempotent
+    '''
+
+    def _revert_alembic_version(self, target_versions=None):
+        alembic_version = sqlalchemy.Table(
+            'alembic_version', sqlalchemy.MetaData(),
+            sqlalchemy.Column('version_num', sqlalchemy.String(32)))
+
+        with self.engine.begin() as conn:
+            # Revision "5c85685d616d" is the head of the CONTRACT branch,
+            # it is from Newton release and we don't allow any new CONTRACT
+            # DB upgrades, so let's don't bother with that branch
+            conn.execute(
+                alembic_version.delete().where(
+                    alembic_version.c.version_num != '5c85685d616d'
+                )
+            )
+            if target_versions:
+                conn.execute(
+                    alembic_version.insert(),
+                    [{'version_num': tv} for tv in target_versions]
+                )
+
+    # NOTE(slaweq): this workaround is taken from Manila patch:
+    # https://review.opendev.org/#/c/291397/
+    # Set 5 minutes timeout for case of running it on very slow nodes/VMs.
+    # Note, that this test becomes slower with each addition of new DB
+    # migration. On fast nodes it can take about 5-10 secs having Mitaka set of
+    # migrations.
+    @test_base.set_timeout(600)
+    def test_db_upgrade_is_idempotent(self):
+        """Tests if Alembic upgrade scripts are idempotent.
+
+        This function tests if running Alembic upgrade scripts multiple times
+        results in the same database state. It does this by first upgrading the
+        database to the latest revision, then reverting it to a previous state,
+        and finally upgrading it again to the latest revision.
+        """
+        url_str = render_url_str(self.engine.url)
+        config = self._get_alembic_config(url_str)
+        migration.do_alembic_command(config, 'upgrade', 'heads')
+
+        # Now lets get back with revision to the 2023.2 HEAD ('89c58a70ceba')
+        # and then test again upgrade from from that point through all next
+        # releases, starting from 2024.1 if db migration scripts are idempotent
+        self._revert_alembic_version(["89c58a70ceba"])
+        migration.do_alembic_command(config, 'upgrade', 'heads')