yuxuanhui
53 lines
2.4 KiB
Python
53 lines
2.4 KiB
Python
"""Upgrade cached classifications across batches while preserving platform evidence."""
|
|
|
|
from pathlib import Path
|
|
|
|
import sqlalchemy as sa
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from cryptography.fernet import Fernet
|
|
|
|
from app.models import now
|
|
|
|
|
|
def test_limit_migration_is_reversible_and_preserves_snapshots(tmp_path, monkeypatch):
|
|
path = tmp_path / "limits.db"
|
|
monkeypatch.setenv("DATABASE_URL", f"sqlite+aiosqlite:///{path}")
|
|
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, "0017")
|
|
engine = sa.create_engine(f"sqlite:///{path}")
|
|
table = sa.Table("alphas", sa.MetaData(), autoload_with=engine)
|
|
limit = {"name": "REGULAR_SUBMISSION", "result": "FAIL"}
|
|
patterns = [
|
|
([{"name": "PROD_CORRELATION", "result": "PASS"}, limit], "FAIL_1", "PASS"),
|
|
([{"name": "LOW_SHARPE", "result": "FAIL"}, limit], "FAIL_2", "FAIL_1"),
|
|
([limit], "FAIL_1", "PENDING"),
|
|
([{"name": "LOW_SHARPE", "result": "PASS"}, limit], "FAIL_1", "PRE_CHECK"),
|
|
([{"name": "UNKNOWN", "result": "FAIL"}], "FAIL_1", "FAIL_1"),
|
|
]
|
|
with engine.begin() as db:
|
|
db.execute(table.insert(), [
|
|
{"id": f"old{i:04}", "hidden": False, "settings": {}, "os_metrics": {},
|
|
"is_metrics": {"checks": patterns[i % 5][0]}, "checks": patterns[i % 5][0],
|
|
"check_type": patterns[i % 5][1], "synced_at": now(),
|
|
"raw": {"is": {"checks": patterns[i % 5][0]}}}
|
|
for i in range(503)
|
|
])
|
|
for version, position in [("0018", 2), ("0017", 1), ("0018", 2)]:
|
|
(command.upgrade if version == "0018" else command.downgrade)(config, version)
|
|
with engine.connect() as db:
|
|
rows = db.execute(sa.select(table).order_by(table.c.id)).mappings().all()
|
|
assert len(rows) == 503
|
|
for i, row in enumerate(rows):
|
|
assert row["check_type"] == patterns[i % 5][position]
|
|
assert row["checks"] == row["is_metrics"]["checks"] == row["raw"]["is"]["checks"] == patterns[i % 5][0]
|
|
command.upgrade(config, "head")
|
|
command.check(config)
|
|
engine.dispose()
|