X-Financial/server/tests/test_schema_ownership.py

from __future__ import annotations

from sqlalchemy import create_engine, inspect

from app.db.schema_ownership import MIGRATION_OWNED_TABLES, create_legacy_schema


def test_create_legacy_schema_never_creates_migration_owned_tables() -> None:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    try:
        create_legacy_schema(engine)

        table_names = set(inspect(engine).get_table_names())
    finally:
        engine.dispose()

    assert MIGRATION_OWNED_TABLES == frozenset(
        {
            "ai_application_preview_decisions",
            "ai_decision_feedback",
            "ai_decisions",
            "approval_action_ledgers",
            "approval_task_events",
            "approval_tasks",
            "agent_asset_release_audit_samples",
            "agent_asset_release_labels",
            "agent_asset_release_observations",
            "auth_sessions",
            "attachment_association_jobs",
            "business_events",
            "commercial_cost_events",
            "commercial_admin_events",
            "commercial_billing_periods",
            "commercial_entitlements",
            "commercial_runtime_reservations",
            "expense_case_links",
            "expense_cases",
            "financial_connector_configs",
            "financial_connector_config_events",
            "financial_connector_events",
            "financial_connector_operational_events",
            "knowledge_onlyoffice_sessions",
            "memory_entries",
            "memory_evidence_links",
            "risk_observations",
            "risk_observation_feedback",
            "risk_disposition_events",
            "risk_dispositions",
            "profile_baseline_snapshots",
            "payment_reconciliation_cases",
            "payment_reconciliation_events",
            "savings_evidence_links",
            "savings_events",
            "savings_opportunities",
            "savings_realizations",
            "tenant_commercial_plans",
            "tenant_finance_report_configs",
            "tenant_finance_report_runs",
            "tenant_subscriptions",
            "tenants",
            "usage_meter_events",
            "few_shot_samples",
            "workflow_outcomes",
        }
    )
    assert table_names
    assert table_names.isdisjoint(MIGRATION_OWNED_TABLES)