Merge "Fix 'alembic check' with multidb env" into main

alembic/command.py

@@ -290,7 +290,10 @@ def retrieve_migrations(rev, context):
     # the revision_context now has MigrationScript structure(s) present.
 
     migration_script = revision_context.generated_revisions[-1]
-    diffs = migration_script.upgrade_ops.as_diffs()
+    diffs = []
+    for upgrade_ops in migration_script.upgrade_ops_list:
+        diffs.extend(upgrade_ops.as_diffs())
+
     if diffs:
         raise util.AutogenerateDiffsDetected(
             f"New upgrade operations detected: {diffs}"

docs/build/unreleased/1234.rst

@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, commands
+    :tickets: 1234
+
+    Fixed issue where the ``alembic check`` command did not function correctly
+    with upgrade structures that have multiple, top-level elements, as are
+    generated from the "multi-env" environment template.  Pull request courtesy
+    Neil Williams.

tests/test_command.py

@@ -26,6 +26,7 @@
 from alembic.testing import is_true
 from alembic.testing import mock
 from alembic.testing.env import _get_staging_directory
+from alembic.testing.env import _multidb_testing_config
 from alembic.testing.env import _no_sql_testing_config
 from alembic.testing.env import _sqlite_file_db
 from alembic.testing.env import _sqlite_testing_config
@@ -665,6 +666,53 @@ def test_check_changes_detected(self):
             )
 
 
+class CheckTestMultiDB(CheckTest):
+    def setUp(self):
+        self.engine1 = _sqlite_file_db(tempname="eng1.db")
+        self.engine2 = _sqlite_file_db(tempname="eng2.db")
+        self.engine3 = _sqlite_file_db(tempname="eng3.db")
+
+        self.env = staging_env(template="multidb")
+        self.cfg = _multidb_testing_config(
+            {
+                "engine1": self.engine1,
+                "engine2": self.engine2,
+                "engine3": self.engine3,
+            }
+        )
+
+    def _env_fixture(self):
+        env_file_fixture(
+            """
+
+import re
+from sqlalchemy import MetaData, engine_from_config
+
+db_names = config.get_main_option("databases", "")
+for db_name in re.split(r",\\s*", db_names):
+    engine = engine_from_config(
+        config.get_section(db_name),
+        prefix="sqlalchemy.",
+    )
+    connection = engine.connect()
+    metadata = MetaData()
+    context.configure(
+        connection=connection,
+        target_metadata=metadata,
+    )
+
+    try:
+        with context.begin_transaction():
+            context.run_migrations()
+    finally:
+        connection.close()
+        engine.dispose()
+
+
+"""
+        )
+
+
 class _StampTest:
     def _assert_sql(self, emitted_sql, origin, destinations):
         ins_expr = (