"""판정이 **미확정을 합격으로 바꾸지 않는지**. 앞 판에 실제로 그런 일이 있었다 — 못 읽은 것을 「없다」로 읽고 통과를 냈다. 그래서 어긋남·미확정·맞음을 **따로** 세고, 앞의 둘이 **모두 0** 일 때만 통과다. """ from __future__ import annotations import json import pytest from tools.grounding_audit import canary_report as rp SENTINEL_OK = {"files_unchanged": True, "fingerprint_unchanged": True, "db_rows_unchanged": True} @pytest.fixture def root(monkeypatch, tmp_path): monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) d = tmp_path / "canary_a1b2c3d4" d.mkdir() return d def _write(root, *, this, skipped=("s1", "s2"), prior=None): """★★`canary_run.json` 은 **production 이 쓰는 모양**으로 짓는다. 앞 판은 `skipped` 를 `[{"step": ...}]` dict 목록으로 지어 놨는데, 실제 `canary_run.py` 는 **문자열 목록**을 쓴다. 그래서 시험은 다 초록인데 실제 산출에서는 `TypeError` 로 바로 터졌다 (Codex 2026-09-01). 모양이 production 과 같은지는 `TestTheFixtureMatchesWhatProductionWrites` 가 따로 잰다 — 여기서 또 지어내면 같은 일이 반복된다. """ led = list(prior or [{"attempt_id": "old", "status": "stopped", "used": 15, "ceiling": 92}]) led.append(this) (root / "pipeline_attempts.json").write_text( json.dumps(led, ensure_ascii=False), encoding="utf-8") (root / "canary_run.json").write_text(json.dumps( {"plan": {"skipped": list(skipped)}}, # ★문자열 목록 ensure_ascii=False), encoding="utf-8") def _good_attempt(**over): got = {"attempt_id": "new", "status": "completed", "used": 20, "image_used": 0, "approved_image_calls": 0, "ceiling": 92, "cumulative_before": 15, "scope_cap": 77, "started_kst": "2026-09-01T11:00:00+09:00", "per_step": {"done1": {"counted": 0}, "s1": {"counted": 0}, "s2": {"counted": 0}, "new1": {"counted": 20}}} got.update(over) return got def _db(monkeypatch, states, *, completed_at="2026-09-01T00:00:00+00"): def fake(run_id, sql): if "updated_at" in sql: return [[k, v, completed_at] for k, v in states.items()] return [[k, v] for k, v in states.items()] monkeypatch.setattr(rp, "_psql", fake) class TestItPassesOnlyWhenNothingIsUnknown: def test_a_clean_run_passes(self, root, monkeypatch): _write(root, this=_good_attempt()) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"] == {"맞음": 6, "어긋남": 0, "미확정": 0} assert got["passed"] is True def test_an_unreadable_db_is_undecided_not_a_pass(self, root, monkeypatch): """★★「못 읽었다」를 **「없다」로 읽지 않는다**.""" _write(root, this=_good_attempt()) def boom(run_id, sql): raise RuntimeError("연결 안 됨") monkeypatch.setattr(rp, "_psql", boom) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["미확정"] == 2 assert got["passed"] is False assert any("못 읽었다" in a["잰 것"] for a in got["axes"]) def test_a_rebought_step_is_a_mismatch(self, root, monkeypatch): a = _good_attempt() a["per_step"]["done1"] = {"counted": 3} _write(root, this=a) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 1 assert got["passed"] is False def test_a_skipped_step_with_no_record_is_a_mismatch(self, root, monkeypatch): """★기록이 **아예 없으면** 뒤가 막힌다 — 그것이 이 판의 목적이다.""" _write(root, this=_good_attempt()) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) bad = [a for a in got["axes"] if a["verdict"] == "어긋남"] assert len(bad) == 1 and "기록 없음" in bad[0]["잰 것"] def test_an_image_purchase_is_a_mismatch(self, root, monkeypatch): _write(root, this=_good_attempt(image_used=1)) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 1 assert got["★cost_split"]["이미지"] == 1 def test_an_open_attempt_is_a_mismatch(self, root, monkeypatch): _write(root, this=_good_attempt(status="running")) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 1 def test_a_changed_sentinel_is_a_mismatch(self, root, monkeypatch): _write(root, this=_good_attempt()) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={ "★sentinel": {**SENTINEL_OK, "fingerprint_unchanged": False}}) assert got["tally"]["어긋남"] == 1 class TestWhatItCountsAsAlreadyDone: def test_it_reads_the_db_instead_of_a_hand_written_list(self, root, monkeypatch): """★11 을 손으로 적지 않는다 — 적으면 다음 판에서 거짓말이 된다.""" _write(root, this=_good_attempt()) _db(monkeypatch, {"a": "completed", "b": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert "앞서 끝난 2개" in got["axes"][1]["axis"] def test_a_step_finished_inside_this_attempt_does_not_count_as_prior( self, root, monkeypatch): """★이번 판에서 끝난 것을 「앞서 끝난 것」으로 세면, 이번에 산 것이 재구매로 잘못 읽힌다.""" _write(root, this=_good_attempt()) _db(monkeypatch, {"new1": "completed", "s1": "not_applicable", "s2": "not_applicable"}, completed_at="2026-09-01T11:30:00+09:00") # ★시작보다 **뒤** got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 0 assert "앞서 끝난 0개" in got["axes"][1]["axis"] class TestTheStartingNumbersComeFromTheLedgerNotMyHand: """★★★앞 판은 `15/77` 을 손으로 박아 뒀다 (Codex 2026-09-01). 다음 판의 정본은 `42/50` 이라, 성공해도 어긋남이 된다. """ def test_the_next_run_starts_at_forty_two(self, root, monkeypatch): prior = [{"attempt_id": "a", "status": "stopped", "used": 7}, {"attempt_id": "b", "status": "stopped", "used": 8}, {"attempt_id": "c", "status": "stopped", "used": 27}] this = _good_attempt(cumulative_before=42, scope_cap=50, used=4) _write(root, this=this, prior=prior) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 0, got["axes"][0] assert "cumulative 42" in got["axes"][0]["axis"] assert got["★cost_split"]["이 run 누계"] == 46 def test_a_stale_hardcoded_start_is_a_mismatch(self, root, monkeypatch): """★양성 대조 — 앞 값을 그대로 들고 오면 잡혀야 한다.""" prior = [{"attempt_id": "a", "status": "stopped", "used": 42}] _write(root, this=_good_attempt(cumulative_before=15, scope_cap=77), prior=prior) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 1 def test_an_open_prior_attempt_is_not_counted_into_the_start( self, root, monkeypatch): """★열린 판은 얼마 썼는지 **모른다** — 앞 누계에 안 넣는다.""" prior = [{"attempt_id": "a", "status": "stopped", "used": 7}, {"attempt_id": "b", "status": "running", "used": None}] _write(root, this=_good_attempt(cumulative_before=7, scope_cap=85), prior=prior) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert "cumulative 7" in got["axes"][0]["axis"] # ★열린 판이 있으므로 ④축은 어긋남이다 assert got["tally"]["어긋남"] == 1 class TestTimeIsComparedAsAMomentNotAsText: """★★★DB 는 UTC, 장부는 KST 로 적힌다 (Codex 실측 2026-09-01). 글자로 견주면 `2026-09-01T02:26+00` < `2026-09-01T11:21+09` 라서, **뒤에** 끝난 것이 「앞서 완료」로 읽힌다 — 이번에 산 스텝이 재구매로 세어진다. """ def test_a_utc_stamp_after_the_kst_start_is_not_prior(self, root, monkeypatch): # 02:26+00 == 11:26 KST — 시작(11:21 KST)보다 **뒤**다 _write(root, this=_good_attempt( started_kst="2026-09-01T11:21:00+09:00", per_step={"paid1": {"counted": 3}, "s1": {"counted": 0}, "s2": {"counted": 0}})) _db(monkeypatch, {"paid1": "completed", "s1": "not_applicable", "s2": "not_applicable"}, completed_at="2026-09-01T02:26:00+00") got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert "앞서 끝난 0개" in got["axes"][1]["axis"] assert got["tally"]["어긋남"] == 0 def test_a_utc_stamp_before_the_kst_start_is_prior(self, root, monkeypatch): """★음성 대조 — 진짜로 앞선 것은 앞선 것으로 세어야 한다.""" _write(root, this=_good_attempt( started_kst="2026-09-01T11:21:00+09:00", per_step={"paid1": {"counted": 3}, "s1": {"counted": 0}, "s2": {"counted": 0}})) _db(monkeypatch, {"paid1": "completed", "s1": "not_applicable", "s2": "not_applicable"}, completed_at="2026-09-01T01:26:00+00") # 10:26 KST — 앞 got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert "앞서 끝난 1개" in got["axes"][1]["axis"] assert got["tally"]["어긋남"] == 1 # ★재구매로 잡힌다 def test_a_naive_stamp_stops_instead_of_guessing(self, root, monkeypatch): """★시간대가 없으면 **짐작하지 않는다** — 미확정으로 간다.""" _write(root, this=_good_attempt()) _db(monkeypatch, {"done1": "completed", "s1": "not_applicable", "s2": "not_applicable"}, completed_at="2026-09-01 09:00:00") got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["미확정"] == 1 assert got["passed"] is False class TestTheFixtureMatchesWhatProductionWrites: """★★★내 fixture 가 **실제와 반대 모양**을 지어 결함을 초록으로 잠갔다. `plan.skipped` 를 dict 목록으로 지어 놓아서, 시험은 다 통과하는데 실제 산출에서는 `TypeError` 로 바로 터졌다 (Codex 2026-09-01). """ def test_the_real_writer_produces_strings(self, monkeypatch, tmp_path): from tools.grounding_audit import canary_cost_table as ct from tools.grounding_audit import canary_run as cr if not ct.settings_came_from_backend_env(): pytest.skip("★`backend/` 에서 돌려야 잰다") monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) monkeypatch.setenv("THEROAD_CANARY_TEMPLATE_URL", "postgresql://사용자:암호@어딘가:5432/theroad") got = cr.run(live=False)["plan"]["skipped"] assert got and all(isinstance(x, str) for x in got), got def test_acceptance_reads_what_the_real_writer_wrote(self, monkeypatch, tmp_path): """★끝점 — **production 이 쓴 파일**을 그대로 먹여 터지지 않아야 한다.""" from tools.grounding_audit import canary_cost_table as ct from tools.grounding_audit import canary_run as cr if not ct.settings_came_from_backend_env(): pytest.skip("★`backend/` 에서 돌려야 잰다") monkeypatch.setenv("THEROAD_CANARY_ROOT", str(tmp_path)) monkeypatch.setenv("THEROAD_CANARY_TEMPLATE_URL", "postgresql://사용자:암호@어딘가:5432/theroad") run = cr.run(live=False) # ★진짜 writer d = tmp_path / "canary_a1b2c3d4" d.mkdir(exist_ok=True) (d / "canary_run.json").write_text( json.dumps(run, ensure_ascii=False, default=str), encoding="utf-8") skipped = run["plan"]["skipped"] (d / "pipeline_attempts.json").write_text(json.dumps( [{"attempt_id": "old", "status": "stopped", "used": 15, "ceiling": 92}, _good_attempt(per_step={s: {"counted": 0} for s in skipped})], ensure_ascii=False), encoding="utf-8") _db(monkeypatch, {s: "not_applicable" for s in skipped}) got = rp.acceptance("a1b2c3d4", report={"★sentinel": SENTINEL_OK}) assert got["tally"]["어긋남"] == 0, got["axes"] class TestOpikIsASubsetNotTheSourceOfTruthForMoney: """★★★Opik 대조에서 **두 번 틀렸다** (2026-09-01). ①canary 태그로만 찾아 1건 — 실제 26건 ②top-level trace 만 세어 8 — **span** 을 세면 26 ③시각 파싱이 틀려 0 — 0 은 결함보다 **재는 오류**다 그리고 litellm 의 `OpikLogger` 에는 `log_success_event` 뿐이다. **「Opik 에 없다」는 「안 샀다」가 아니다.** """ def test_the_opik_logger_really_has_no_failure_hook(self): """★양성 대조 — 주장을 **그 클래스에서** 확인한다.""" from litellm.integrations.opik.opik import OpikLogger assert hasattr(OpikLogger, "log_success_event") own = {n for n in vars(OpikLogger) if "log" in n and "event" in n} assert not any("failure" in n for n in own), own # ★부모의 실패 훅은 아무것도 안 한다 from litellm.integrations.custom_logger import CustomLogger import inspect body = inspect.getsource(CustomLogger.log_failure_event) assert body.strip().endswith("pass") def test_without_a_binding_key_it_refuses_instead_of_saying_zero( self, monkeypatch): """★결속 고리가 없으면 **0 이 아니라 「못 찾았다」**여야 한다.""" monkeypatch.setattr(rp, "_opik_get", lambda path, **kw: {"content": []}) got = rp.opik_traces("", project_id="", episode_id="") assert got["ok"] is False and "못 읽은 것" in got["★means"] def test_the_report_says_the_ledger_is_the_source_of_truth(self, monkeypatch): monkeypatch.setattr(rp, "_opik_get", lambda path, **kw: {"content": []}) got = rp.opik_traces("r", project_id="p", attempts=[]) assert "장부" in got["★sot"] assert "실패한 유료 호출은 Opik 에 안 남는다" in got["★blind_spot"] def test_it_counts_llm_spans_not_only_top_level_traces(self, monkeypatch): """★span 을 세는지 — trace 만 세면 8, span 을 세면 26 이었다.""" def fake(path, **kw): if path == "traces": if kw.get("page", 1) > 1: return {"content": []} return {"content": [{"id": "t1", "project_id": "p", "start_time": "2026-09-01T02:00:00.1Z"}]} return {"content": [{"type": "llm"}, {"type": "llm"}, {"type": "general"}]} monkeypatch.setattr(rp, "_opik_get", fake) got = rp.opik_traces("r", project_id="p", attempts=[ {"attempt_id": "a", "used": 2, "started_kst": "2026-09-01T10:00:00+09:00", "finished_kst": "2026-09-01T12:00:00+09:00"}]) row = got["per_attempt"][0] assert row["opik_llm"] == 2 and row["traces"] == 1 assert row["★agrees"] is True def test_an_attempt_without_times_is_not_counted_as_a_gap(self, monkeypatch): monkeypatch.setattr(rp, "_opik_get", lambda path, **kw: {"content": []}) got = rp.opik_traces("r", project_id="p", attempts=[ {"attempt_id": "old", "used": 7}]) row = got["per_attempt"][0] assert row["opik_llm"] is None and "시각이 없어" in row["★why"]