yuxuanhui
57 lines
2.8 KiB
Python
57 lines
2.8 KiB
Python
"""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()
|