"""回归测试: 预测唯一约束修复 —— live 与 backtest 可共存。 验证: 1. 唯一约束包含 mode + run_type 2. 同一场比赛 live 与 backtest 预测可共存,互不覆盖 3. _upsert_prediction 正确区分 run_type """ from __future__ import annotations import inspect from pydantic import BaseModel import pytest from src.db.models import Prediction, UniqueConstraint, CheckConstraint class TestUniqueConstraint: """验证唯一约束包含 mode + run_type。""" def test_constraint_columns(self): """唯一约束应包含 match_id, provider, model, mode, run_type。""" uc = [ c for c in Prediction.__table__.constraints if isinstance(c, UniqueConstraint) and "match" in c.name ] assert len(uc) == 1 cols = [c.name for c in uc[0].columns] assert cols == ["match_id", "provider", "model", "mode", "run_type"] def test_run_type_check_constraint(self): """应有 run_type 的 check constraint。""" cc = [ c for c in Prediction.__table__.constraints if isinstance(c, CheckConstraint) and "run_type" in c.name ] assert len(cc) == 1 def test_run_type_column_exists(self): """run_type 列应存在且 NOT NULL,默认 'live'。""" cols = {c.name: c for c in Prediction.__table__.columns} assert "run_type" in cols assert cols["run_type"].nullable is False # 默认值 assert cols["run_type"].default.arg == "live" if cols["run_type"].default else True class TestUpsertPredictionSignature: """验证 _upsert_prediction 函数签名包含 run_type。""" def test_signature_has_run_type(self): from src.llm.predict import _upsert_prediction sig = inspect.signature(_upsert_prediction) assert "run_type" in sig.parameters def test_signature_has_backtest_in_predict_match(self): from src.llm.predict import predict_match sig = inspect.signature(predict_match) assert "backtest" in sig.parameters def test_signature_has_backtest_in_predict_multi(self): from src.llm.agents.orchestrator import predict_match_multi sig = inspect.signature(predict_match_multi) assert "backtest" in sig.parameters class TestMigration: """验证迁移文件存在且内容正确。""" def test_migration_exists(self): import os path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0013_predictions_unique_constraint_mode_run_type.py" assert os.path.exists(path) def test_migration_adds_column_and_constraint(self): path = "/.octop/workspaces/CA7PFH/Profeto/alembic/versions/0013_predictions_unique_constraint_mode_run_type.py" content = open(path).read() assert 'run_type' in content assert 'uq_predictions_match_provider_model_mode_run_type' in content assert 'backtest' in content assert 'live' in content # 验证数据回填逻辑 assert "UPDATE predictions SET run_type = 'live'" in content class TestLiveBacktestCoexist: """验证 live 与 backtest 可共存(逻辑验证,无需数据库)。""" def test_different_run_type_allow_coexistence(self): """同一 match_id + provider + model + mode,不同 run_type 应可共存。 这是核心修复:之前唯一约束只有 (match_id, provider, model), backtest 会覆盖 live 预测。 """ # 模拟两行数据 class FakeRow: def __init__(self, **kw): for k, v in kw.items(): setattr(self, k, v) live = FakeRow(match_id=1, provider="openai", model="gpt-4o", mode="single", run_type="live") backtest = FakeRow(match_id=1, provider="openai", model="gpt-4o", mode="single", run_type="backtest") # 两者唯一键不同(因为 run_type 不同) live_key = (live.match_id, live.provider, live.model, live.mode, live.run_type) backtest_key = (backtest.match_id, backtest.provider, backtest.model, backtest.mode, backtest.run_type) assert live_key != backtest_key, "live 与 backtest 应有不同的唯一键" assert live_key == (1, "openai", "gpt-4o", "single", "live") assert backtest_key == (1, "openai", "gpt-4o", "single", "backtest") def test_same_run_type_prevents_duplicate(self): """相同 run_type 的重复预测仍应被约束阻止。""" key1 = (1, "openai", "gpt-4o", "single", "live") key2 = (1, "openai", "gpt-4o", "single", "live") assert key1 == key2, "相同 run_type 应有相同唯一键,应被约束阻止"