feat: migrate advisor database schema to qyqy base

tests/unit/test_advisor_migration_contract.py

@@ -0,0 +1,64 @@
+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_adv_profile_tags"
+
+    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