94 lines
3.9 KiB
Python
94 lines
3.9 KiB
Python
"""adversarial_verdicts 表：写入 / 续跑查询 / agent_config 作废 / 正确率聚合。"""
|
||
|
||
from app.question_gen.run_store import QuestionGenStore
|
||
|
||
|
||
def _store(tmp_path):
|
||
return QuestionGenStore(str(tmp_path / "q.db"))
|
||
|
||
|
||
def _row(**kw):
|
||
base = {
|
||
"question_id": "v1_Action Recognition_0001", "round": 0, "stage": "cheat",
|
||
"question_hash": "h1", "agent_prediction": "B", "agent_correct": False,
|
||
"verdict": "passed", "pair_id": None, "agent_config": "cfg1",
|
||
}
|
||
base.update(kw)
|
||
return base
|
||
|
||
|
||
def test_table_created(tmp_path):
|
||
store = _store(tmp_path)
|
||
cols = {r[1] for r in store._conn.execute("PRAGMA table_info(adversarial_verdicts)")}
|
||
assert {"question_id", "round", "stage", "question_hash", "agent_prediction",
|
||
"agent_correct", "verdict", "pair_id", "agent_config"} <= cols
|
||
store.close()
|
||
|
||
|
||
def test_record_and_resume_lookup(tmp_path):
|
||
store = _store(tmp_path)
|
||
store.record_verdict(**_row(stage="cheat"))
|
||
done = store.completed_stages("v1_Action Recognition_0001", "h1", "cfg1")
|
||
assert done == {"cheat"}
|
||
# 不同 hash 视为未完成
|
||
assert store.completed_stages("v1_Action Recognition_0001", "h2", "cfg1") == set()
|
||
store.close()
|
||
|
||
|
||
def test_agent_config_change_invalidates(tmp_path):
|
||
store = _store(tmp_path)
|
||
store.record_verdict(**_row(stage="cheat"))
|
||
store.invalidate_stale_config("v1_Action Recognition_0001", "cfg2")
|
||
# 旧 config 行必须被真正删除（不能只靠 cfg2 查空——no-op 也满足那个弱断言）
|
||
assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg1") == set()
|
||
cfg1_rows = store._conn.execute(
|
||
"SELECT COUNT(*) FROM adversarial_verdicts WHERE agent_config='cfg1'"
|
||
).fetchone()[0]
|
||
assert cfg1_rows == 0
|
||
assert store.completed_stages("v1_Action Recognition_0001", "h1", "cfg2") == set()
|
||
store.close()
|
||
|
||
|
||
def test_upsert_same_key_overwrites(tmp_path):
|
||
store = _store(tmp_path)
|
||
store.record_verdict(**_row(agent_prediction="A"))
|
||
store.record_verdict(**_row(agent_prediction="C"))
|
||
rows = store._conn.execute(
|
||
"SELECT agent_prediction FROM adversarial_verdicts "
|
||
"WHERE question_id=? AND question_hash=? AND stage=?",
|
||
("v1_Action Recognition_0001", "h1", "cheat"),
|
||
).fetchall()
|
||
assert len(rows) == 1 and rows[0][0] == "C"
|
||
store.close()
|
||
|
||
|
||
def test_cheat_accuracy_aggregation(tmp_path):
|
||
store = _store(tmp_path)
|
||
store.record_verdict(**_row(question_id="q1", question_hash="a", agent_correct=True))
|
||
store.record_verdict(**_row(question_id="q2", question_hash="b", agent_correct=False))
|
||
store.record_verdict(**_row(question_id="q3", question_hash="c", agent_correct=True))
|
||
assert store.cheat_agent_accuracy(round_no=0) == 2 / 3
|
||
store.close()
|
||
|
||
|
||
def test_final_passed_question_ids_survives_both_gates(tmp_path):
|
||
store = _store(tmp_path)
|
||
# q1 太简单被作弊门剔除；q2 过两门；q3 被翻转门剔除（filtered_no_flip）
|
||
store.record_verdict(**_row(question_id="q1", question_hash="a", stage="cheat",
|
||
verdict="filtered_too_easy"))
|
||
store.record_verdict(**_row(question_id="q2", question_hash="b", stage="cheat",
|
||
verdict="passed"))
|
||
store.record_verdict(**_row(question_id="q3", question_hash="c", stage="cheat",
|
||
verdict="passed"))
|
||
store.record_verdict(**_row(question_id="q3", question_hash="c", stage="flip_mirror",
|
||
verdict="filtered_no_flip"))
|
||
passed = store.final_passed_question_ids(
|
||
{"q1": "a", "q2": "b", "q3": "c"}, "cfg1"
|
||
)
|
||
assert passed == {"q2"} # 仅 q2：cheat=passed 且无 filtered_no_flip
|
||
# stale hash 不泄漏（当前 hash 不匹配旧行）
|
||
assert store.final_passed_question_ids({"q2": "stale"}, "cfg1") == set()
|
||
# stale config 不泄漏
|
||
assert store.final_passed_question_ids({"q2": "b"}, "cfgX") == set()
|
||
store.close()
|