group_fqcd_jr/tests/unit/test_advisor_migration_contract.py

from __future__ import annotations

import re
from pathlib import Path

from alembic.config import Config
from alembic.script import ScriptDirectory

ROOT = Path(__file__).resolve().parents[2]
VERSIONS = ROOT / "alembic" / "versions"
BASELINE = ROOT / "alembic" / "baseline_generated.sql"

ADVISOR_FILES = (
    "20260910_advisor_investment_goal.py",
    "20260910_advisor_product_industry_exposure.py",
    "20260910_advisor_portfolio_projection_checkpoint.py",
    "20260910_advisor_product_metric_snapshot.py",
    "20260910_advisor_product_asset_classification.py",
    "20260910_advisor_product_reference_snapshot.py",
    "20260910_advisor_offsite_fund_reference.py",
    "20260910_advisor_product_price_history.py",
    "20260910_advisor_product_governance_reference.py",
    "20260910_advisor_governance_monitor_and_quotes.py",
    "20260910_advisor_market_quote_resilience.py",
    "20260910_advisor_data_quality_backtest.py",
    "20260911_advisor_goal_conversation.py",
    "20260911_advisor_profile_tag_governance.py",
)


def created_tables(path: Path) -> set[str]:
    content = path.read_text(encoding="utf-8")
    return set(re.findall(r"CREATE TABLE(?: IF NOT EXISTS)?\s+`?([A-Za-z0-9_]+)`?", content))


def test_advisor_migrations_form_one_chain_from_qyqy_head() -> None:
    script = ScriptDirectory.from_config(Config(str(ROOT / "alembic.ini")))
    assert len(script.get_heads()) == 1
    assert script.get_heads()[0] == "20260911_merge_adv_risk_heads"

    first = (VERSIONS / ADVISOR_FILES[0]).read_text(encoding="utf-8")
    assert 'down_revision = "20260910_drop_review_separation"' in first
    for previous, current in zip(ADVISOR_FILES, ADVISOR_FILES[1:], strict=False):
        previous_content = (VERSIONS / previous).read_text(encoding="utf-8")
        current_content = (VERSIONS / current).read_text(encoding="utf-8")
        previous_revision = re.search(r'revision = "([^"]+)"', previous_content)
        assert previous_revision is not None
        assert f'down_revision = "{previous_revision.group(1)}"' in current_content


def test_advisor_migrations_only_create_additive_tables() -> None:
    baseline_tables = created_tables(BASELINE)
    advisor_tables: set[str] = set()
    for filename in ADVISOR_FILES:
        content = (VERSIONS / filename).read_text(encoding="utf-8")
        assert "ALTER TABLE" not in content
        assert "DROP TABLE" not in content
        advisor_tables.update(created_tables(VERSIONS / filename))

    assert advisor_tables
    assert not advisor_tables & baseline_tables
    assert "fin_sim_order" not in advisor_tables
    assert "fin_transaction" not in advisor_tables
    assert "fin_holding" not in advisor_tables