Skip to content

Commit

Permalink
Fix 'alembic check' with multidb env
Browse files Browse the repository at this point in the history
  • Loading branch information
spladug committed Nov 22, 2023
1 parent 8ba8e7b commit c1bb019
Show file tree
Hide file tree
Showing 2 changed files with 52 additions and 1 deletion.
5 changes: 4 additions & 1 deletion alembic/command.py
Original file line number Diff line number Diff line change
Expand Up @@ -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}"
Expand Down
48 changes: 48 additions & 0 deletions tests/test_command.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 = (
Expand Down

0 comments on commit c1bb019

Please sign in to comment.