From f6a7a3d6ee8c47a213086880e1b2377dace9d183 Mon Sep 17 00:00:00 2001 From: Cheng Zhou Date: Mon, 11 May 2026 08:54:45 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=B7=A5=E4=BD=9C=E6=B5=81?= =?UTF-8?q?=E8=A1=A8=E5=88=A0=E9=99=A4=E8=BF=81=E7=A7=BB=E5=B9=82=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../20260116_05_remove_document_workflows.py | 16 ++++++++++++---- backend/tests/test_migration_safety.py | 9 +++++++++ 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/backend/alembic/versions/20260116_05_remove_document_workflows.py b/backend/alembic/versions/20260116_05_remove_document_workflows.py index 9c62459f..395beb14 100644 --- a/backend/alembic/versions/20260116_05_remove_document_workflows.py +++ b/backend/alembic/versions/20260116_05_remove_document_workflows.py @@ -20,10 +20,18 @@ depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: - op.drop_table("workflow_actions") - op.drop_table("version_workflows") - op.drop_table("workflow_nodes") - op.drop_table("workflow_templates") + bind = op.get_bind() + inspector = sa.inspect(bind) + tables = set(inspector.get_table_names()) + + 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_status") diff --git a/backend/tests/test_migration_safety.py b/backend/tests/test_migration_safety.py index 0ff5545a..66a08db2 100644 --- a/backend/tests/test_migration_safety.py +++ b/backend/tests/test_migration_safety.py @@ -8,6 +8,15 @@ def test_is_locked_migration_is_idempotent(): 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(): source = Path("scripts/check_migration_state.py").read_text(encoding="utf-8")