worldquant-alpha-system/backend/tests/test_ppac_candidate_migration.py

"""Historical PPAC classification changes only matching cached check types."""

from datetime import datetime, timezone
from pathlib import Path

import sqlalchemy as sa
from alembic import command
from alembic.config import Config
from cryptography.fernet import Fernet


def test_ppac_backfill_preserves_evidence_across_batches_and_downgrade(tmp_path, monkeypatch):
    database = tmp_path / "ppac.db"
    monkeypatch.setenv("DATABASE_URL", f"sqlite+aiosqlite:///{database}")
    monkeypatch.setenv("ADMIN_PASSWORD", "migration-test-only")
    monkeypatch.setenv("ENCRYPTION_KEY", Fernet.generate_key().decode())
    monkeypatch.setenv("WQ_EMAIL", "")
    monkeypatch.setenv("WQ_PASSWORD", "")
    root = Path(__file__).resolve().parents[1]
    config = Config(str(root / "alembic.ini"))
    config.set_main_option("script_location", str(root / "migrations"))
    command.upgrade(config, "0021")
    engine = sa.create_engine(f"sqlite:///{database}")
    alphas = sa.Table("alphas", sa.MetaData(), autoload_with=engine)
    theme = {"name": "PURE_POWER_POOL_THEME", "result": "FAIL"}
    other = {"name": "LOW_SHARPE", "result": "FAIL"}
    patterns = [
        ([theme], "FAIL_1", "PPAC_CANDIDATE"),
        ([{**theme, "result": "fail"}, {"name": "REGULAR_SUBMISSION", "result": "FAIL"}], "FAIL_1", "PPAC_CANDIDATE"),
        ([theme, other], "FAIL_2", "FAIL_2"),
        ([theme, theme], "FAIL_2", "FAIL_2"),
        ([other], "FAIL_1", "FAIL_1"),
        ([{**theme, "result": "PASS"}], "PASS", "PASS"),
        ([{**theme, "result": "WARNING"}], "PRE_CHECK", "PRE_CHECK"),
        ([{}], "PENDING", "PENDING"),
        (None, "PENDING", "PENDING"),
    ]
    with engine.begin() as db:
        db.execute(alphas.insert(), [
            {"id": f"ppac{i:04}", "status": "UNSUBMITTED", "hidden": False, "settings": {}, "os_metrics": {},
             "is_metrics": {"checks": patterns[i % len(patterns)][0]}, "checks": patterns[i % len(patterns)][0],
             "check_type": patterns[i % len(patterns)][1], "synced_at": datetime(2026, 9, 13, tzinfo=timezone.utc),
             "raw": {"is": {"checks": patterns[i % len(patterns)][0]}}}
            for i in range(503)
        ])
        original = db.execute(sa.select(alphas).order_by(alphas.c.id)).mappings().all()
    for target, position in [("0022", 2), ("0021", 1), ("0022", 2)]:
        (command.upgrade if target == "0022" else command.downgrade)(config, target)
        with engine.connect() as db:
            rows = db.execute(sa.select(alphas).order_by(alphas.c.id)).mappings().all()
            assert len(rows) == 503
            for i, row in enumerate(rows):
                assert dict(row) == {**original[i], "check_type": patterns[i % len(patterns)][position]}
    command.upgrade(config, "head")
    command.check(config)
    engine.dispose()