zhouchengcheng/ctms
1
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

修复工作流表删除迁移幂等

Browse Source
This commit is contained in:
Cheng Zhou
2026-05-11 08:54:45 +08:00
parent 4592ff375d
commit f6a7a3d6ee
2 changed files with 21 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/alembic/versions/20260116_05_remove_document_workflows.py
+12 -4
View File
@@ -20,10 +20,18 @@ depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None: def upgrade() -> None:
op.drop_table("workflow_actions") bind = op.get_bind()
op.drop_table("version_workflows") inspector = sa.inspect(bind)
op.drop_table("workflow_nodes") tables = set(inspector.get_table_names())
op.drop_table("workflow_templates")
if "workflow_actions" in tables:
op.drop_table("workflow_actions")
if "version_workflows" in tables:
op.drop_table("version_workflows")
if "workflow_nodes" in tables:
op.drop_table("workflow_nodes")
if "workflow_templates" in tables:
op.drop_table("workflow_templates")
op.execute("DROP TYPE IF EXISTS workflow_action_type") op.execute("DROP TYPE IF EXISTS workflow_action_type")
op.execute("DROP TYPE IF EXISTS workflow_status") op.execute("DROP TYPE IF EXISTS workflow_status")
backend/tests/test_migration_safety.py
+9
View File
@@ -8,6 +8,15 @@ def test_is_locked_migration_is_idempotent():
assert 'if "is_locked" in columns:' in source assert 'if "is_locked" in columns:' in source
def test_removed_workflow_tables_are_dropped_conditionally():
source = Path("alembic/versions/20260116_05_remove_document_workflows.py").read_text(encoding="utf-8")
assert 'if "workflow_actions" in tables:' in source
assert 'if "version_workflows" in tables:' in source
assert 'if "workflow_nodes" in tables:' in source
assert 'if "workflow_templates" in tables:' in source
def test_migration_state_check_script_exists(): def test_migration_state_check_script_exists():
source = Path("scripts/check_migration_state.py").read_text(encoding="utf-8") source = Path("scripts/check_migration_state.py").read_text(encoding="utf-8")