feat(c2): integrate Mythic command and control (sprint 8) #11
@@ -92,27 +92,22 @@ def _run_upgrade(engine, migration_mod):
|
||||
with engine.begin() as conn:
|
||||
ctx = MigrationContext.configure(conn)
|
||||
ops = Operations(ctx)
|
||||
# Patch op module for the migration
|
||||
import alembic.op as alembic_op
|
||||
original_proxy = alembic_op._proxy # type: ignore[attr-defined]
|
||||
alembic_op._proxy = ops # type: ignore[attr-defined]
|
||||
ops._install_proxy() # type: ignore[attr-defined]
|
||||
try:
|
||||
migration_mod.upgrade()
|
||||
finally:
|
||||
alembic_op._proxy = original_proxy # type: ignore[attr-defined]
|
||||
ops._remove_proxy() # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def _run_downgrade(engine, migration_mod):
|
||||
with engine.begin() as conn:
|
||||
ctx = MigrationContext.configure(conn)
|
||||
ops = Operations(ctx)
|
||||
import alembic.op as alembic_op
|
||||
original_proxy = alembic_op._proxy # type: ignore[attr-defined]
|
||||
alembic_op._proxy = ops # type: ignore[attr-defined]
|
||||
ops._install_proxy() # type: ignore[attr-defined]
|
||||
try:
|
||||
migration_mod.downgrade()
|
||||
finally:
|
||||
alembic_op._proxy = original_proxy # type: ignore[attr-defined]
|
||||
ops._remove_proxy() # type: ignore[attr-defined]
|
||||
|
||||
|
||||
class TestMigration0006Upgrade:
|
||||
|
||||
Reference in New Issue
Block a user