"""★★canary ② 도구가 **끝점에서** 재나 · **아무것도 안 사나**. Codex 승인 범위 (2026-09-02): 새 검색 0 · 새 받기 0 · VLM 0 · 최종 이미지 생성 0. 이미지 provider 함수는 **부르지 않고**, 전송 직전 payload 만 붙잡는다. """ from __future__ import annotations import hashlib import json from pathlib import Path import pytest from app.modules.pipeline import grounding_reference_bundle as rb from tools.grounding_audit import canary_payload_probe as pp class TestItBuysNothing: def test_no_provider_name_anywhere(self): import inspect src = inspect.getsource(pp) for banned in ("generate_and_validate_scene", "gemini_image", "generate_image", "search_reference_images", "download_candidate", "call_structured", "responses.create", "requests.", "httpx"): assert banned not in src, f"★{banned} 를 부른다" def test_it_does_not_even_take_a_provider(self): import inspect for fn in (pp.run, pp.measure_shot): names = set(inspect.signature(fn).parameters) assert not (names & {"gemini_client", "provider", "generate", "search", "download"}), fn.__name__ def test_it_only_reads(self): import inspect src = inspect.getsource(pp.measure_shot) for banned in ("write_text", "write_bytes", "mkdir", "unlink"): assert banned not in src, f"★{banned} — 재는 도구가 쓴다" class TestItRefusesToInventVisibility: """★★★「visibility 의 답을 따르라」인데 그 답이 없으면 **선다**. 씬 단위로 내리거나 내가 목록을 만들어 넣으면 그것은 재는 것이 아니라 **만드는 것**이다. """ def test_no_shot_director_stops(self): with pytest.raises(pp.VisibilityMissing, match="shot_director"): pp.shots_from(None) with pytest.raises(pp.VisibilityMissing): pp.shots_from({"data": {}}) def test_an_empty_shot_list_stops(self): with pytest.raises(pp.VisibilityMissing, match="샷이 하나도"): pp.shots_from({"data": {"scenes": [{"scene_index": 1, "shots": []}]}}) def test_it_takes_the_answer_as_given(self): got = pp.shots_from({"data": {"scenes": [ {"scene_index": 1, "shots": [ {"shot_index": 0, "visible_entity_ids": ["LP01", "C01"]}, {"shot_index": 1, "visible_entity_ids": []}]}]}}) assert [g["visible"] for g in got] == [["LP01", "C01"], []] assert [g["shot_index"] for g in got] == [0, 1] def _cp(tmp_path, rows): return {"data": {"rows": rows}} def _row(fid, purpose, owner, blob, *, verified=True): from app.modules.pipeline import reference_acquisition as ra return {"research_subject_id": f"rs_{fid}", "identity": f"acq-{fid}", "disposition": "acquired", "status": ra.STATUS_SELECTED, "outcome": ra.STATUS_SELECTED, "why": "", "why_unbought": None, "grounding_fidelity": { "state": (ra.FIDELITY_VERIFIED if verified else ra.FIDELITY_REJECTED)}, "acquisition": {"chosen": {"path": f"pics/{fid}.png", "url": f"https://x.invalid/{fid}"}}, "ledger_row": {"research_subject_id": f"rs_{fid}", "owner_type": owner, "final_id": fid, "parent_final_id": None, "purpose": purpose, "covers": [fid]}} class TestWhatLandsInThePayload: @pytest.fixture def world(self, tmp_path): pics = tmp_path / "pics" pics.mkdir() same = b"\x89PNG\r\n\x1a\n" + b"SAME" * 4 (pics / "LP01.png").write_bytes(same) (pics / "LP04.png").write_bytes(same) # ★같은 bytes (pics / "LP03.png").write_bytes(b"\x89PNG\r\n\x1a\nCCC") rows = [_row("LP01", "detail", "location_part", same), _row("LP04", "detail", "location_part", same, verified=False), _row("LP03", "detail", "location_part", b"c")] return tmp_path, _cp(tmp_path, rows) def test_every_selected_one_lands(self, world): """★HITL 0 — 미확인 줄도 붙는다.""" root, cp = world got = pp.measure_shot(cp, ["LP01", "LP04"], root=root) subs = sorted(s for r in got["refs"] for s in r["subjects"]) assert subs == ["LP01", "LP04"], f"★{subs}" def test_the_same_bytes_carry_both_subjects(self, world): """★HITL 0: 같은 bytes 의 두 줄이 다 붙고, 실린 참조는 두 주체를 든다.""" root, cp = world got = pp.measure_shot(cp, ["LP01", "LP04"], root=root) assert sorted(s for r in got["refs"] for s in r["subjects"]) == ["LP01", "LP04"] def test_it_records_order_role_bytes_and_sha(self, world): root, cp = world got = pp.measure_shot(cp, ["LP01", "LP03"], root=root) assert len(got["refs"]) == 2 assert [r["order"] for r in got["refs"]] == [0, 1] for r in got["refs"]: assert r["bytes"] > 0 assert len(r["sha256"]) == 64 assert r["role"], "★role 이 비었다" assert r["subjects"] and r["purposes"] def test_the_sha_is_of_the_bytes_that_actually_go(self, world): """★★적힌 것이 아니라 **실제로 실린 bytes** 의 지문이다.""" root, cp = world got = pp.measure_shot(cp, ["LP01"], root=root) want = hashlib.sha256((root / "pics" / "LP01.png").read_bytes() ).hexdigest() assert got["refs"][0]["sha256"] == want def test_nothing_visible_lands_nothing(self, world): root, cp = world got = pp.measure_shot(cp, [], root=root) assert got["refs"] == [] and got["attached"] == 0 class TestTheVerdictsSplitTheAxes: def _got(self, **over): base = { "rows": { "LP01": {"owner_type": "location_part", "outcome": "selected", "fidelity": "verified", "usable": True}, "LP04": {"owner_type": "location_part", "outcome": "selected", "fidelity": "rejected", "usable": False}, "P01": {"owner_type": "prop", "outcome": "selected", "fidelity": "verified", "usable": True}}, "sidecar_lane": ["LP01"], "policy_lane": ["P01"], "shots": [{"visible": ["C01", "LP01"], "refs": [{"sha256": "a", "subjects": ["LP01"]}]}]} base.update(over) return base def test_a_clean_run_passes_every_axis(self): assert all(v["통과"] for v in pp.verdicts(self._got())) def test_an_unverified_in_the_payload_fails(self): got = self._got(sidecar_lane=["LP01", "LP04"]) bad = [v for v in pp.verdicts(got) if not v["통과"]] assert any("확인 안 된" in v["축"] for v in bad) def test_a_double_lane_attachment_fails(self): got = self._got(sidecar_lane=["LP01", "P01"]) bad = [v for v in pp.verdicts(got) if not v["통과"]] assert any("두 lane" in v["축"] for v in bad) def test_an_empty_background_lane_is_not_a_pass(self): """★★빈손이 모든 축을 지나가면 안 된다.""" got = self._got(sidecar_lane=[]) bad = [v for v in pp.verdicts(got) if not v["통과"]] assert any("sidecar 에 실린다" in v["축"] for v in bad) def test_a_leaked_sha_fails(self): got = self._got(shots=[{"visible": ["LP01"], "refs": [{"sha256": "a", "subjects": ["LP01", "LP04"]}]}]) bad = [v for v in pp.verdicts(got) if not v["통과"]] assert any("새지 않는다" in v["축"] for v in bad) def test_a_skipped_context_row_does_not_hide_the_verified_detail_row(self): """★canary ② 실측 — `LP01#context`(건너뜀) 이 `LP01#detail`(확인됨) 을 final_id 키로 **덮어** 확인된 것이 「미확인」으로 읽혔다.""" rows = { "LP01#detail": {"final_id": "LP01", "purpose": "detail", "owner_type": "location_part", "outcome": "selected", "fidelity": "verified", "usable": True}, "LP01#context": {"final_id": "LP01", "purpose": "context", "owner_type": "location_part", "outcome": "reference_unavailable", "fidelity": "unverified", "usable": False}, "rs_P01": {"final_id": "P01", "purpose": "", "owner_type": "prop", "outcome": "selected", "fidelity": "verified", "usable": True}} got = self._got(rows=rows) assert all(v["통과"] for v in pp.verdicts(got)), pp.verdicts(got) def test_but_a_final_id_with_no_verified_row_still_fails(self): """일부러 확인된 줄을 빼면 두 축이 잡아야 한다.""" rows = { "LP01#context": {"final_id": "LP01", "purpose": "context", "owner_type": "location_part", "outcome": "reference_unavailable", "fidelity": "unverified", "usable": False}, "rs_P01": {"final_id": "P01", "purpose": "", "owner_type": "prop", "outcome": "selected", "fidelity": "verified", "usable": True}} bad = [v["축"] for v in pp.verdicts(self._got(rows=rows)) if not v["통과"]] assert any("확인 안 된" in a for a in bad), bad assert any("새지 않는다" in a for a in bad), bad class TestAVerifiedButInvisibleSubjectIsNotAFailure: """★canary ② 실측 — `LP05` 는 확인됐지만 shot_director 가 어느 샷에도 보이게 두지 않았다. 참조는 보이는 샷에만 붙으니 그것은 결함이 아니라 production visibility 의 답이다. 다만 **따로 적힌다**.""" def _got(self, **over): base = { "rows": { "LP01#detail": {"final_id": "LP01", "purpose": "detail", "owner_type": "location_part", "outcome": "selected", "fidelity": "verified", "usable": True}, "LP05#detail": {"final_id": "LP05", "purpose": "detail", "owner_type": "location_part", "outcome": "selected", "fidelity": "verified", "usable": True}}, "sidecar_lane": ["LP01"], "policy_lane": [], "shots": [{"visible": ["C01", "LP01"], "refs": [{"sha256": "a", "subjects": ["LP01"]}]}, {"visible": ["C02"], "refs": []}]} base.update(over) return base def test_it_passes_and_is_written_down_separately(self): vs = pp.verdicts(self._got()) assert all(v["통과"] for v in vs), vs axis = next(v for v in vs if "sidecar 에 실린다" in v["축"]) assert axis["기대"] == ["LP01"] assert axis["확인됐지만 어느 샷에도 안 보여 안 실림"] == ["LP05"] def test_but_a_visible_verified_one_that_did_not_land_fails(self): """일부러 LP05 를 보이게 하고 sidecar 에서 빼면 잡혀야 한다.""" got = self._got(shots=[{"visible": ["C01", "LP01"], "refs": [{"sha256": "a", "subjects": ["LP01"]}]}, {"visible": ["C02", "LP05"], "refs": []}]) bad = [v for v in pp.verdicts(got) if not v["통과"]] assert any("sidecar 에 실린다" in v["축"] for v in bad), bad def test_nothing_visible_at_all_is_still_not_a_pass(self): """★빈손 — 보이는 배경이 하나도 없으면 「통과」가 아니다.""" got = self._got(sidecar_lane=[], shots=[{"visible": ["C01"], "refs": []}]) bad = [v for v in pp.verdicts(got) if not v["통과"]] assert any("sidecar 에 실린다" in v["축"] for v in bad), bad class TestTheAuditListsKeepTheSubject: def test_unavailable_context_and_selected_detail_stay_apart(self): items = [{"research_subject_id": "LP01#context", "final_id": "LP01", "fidelity": "unverified", "outcome": "reference_unavailable"}] got = pp._subject_triples(items) assert got == [{"research_subject_id": "LP01#context", "final_id": "LP01", "purpose": "context", "fidelity": "unverified", "outcome": "reference_unavailable"}] def test_a_prop_subject_without_purpose_gets_an_empty_purpose(self): got = pp._subject_triples([{"research_subject_id": "rs_bd", "final_id": "P01"}]) assert got[0]["purpose"] == "" and got[0]["final_id"] == "P01" class TestRowsAreKeyedBySubjectNotFinalId: def test_two_purposes_of_one_final_id_are_both_kept(self, tmp_path): detail = _row("LP01", "detail", "location_part", b"x") context = _row("LP01", "context", "location_part", b"", verified=False) context["research_subject_id"] = "LP01#context" context["disposition"] = "skipped" context["outcome"] = "reference_unavailable" context.pop("acquisition", None) detail["research_subject_id"] = "LP01#detail" rows = pp.rows_by_subject(_cp(tmp_path, [detail, context])) assert set(rows) == {"LP01#detail", "LP01#context"} assert rows["LP01#detail"]["usable"] is True assert rows["LP01#detail"]["final_id"] == "LP01" assert rows["LP01#context"]["usable"] is False class TestWhatWasDeclaredIsWhatWentOut: """★sidecar 가 **적은** 지문과 실제로 **실린 bytes** 의 지문이 같나. 둘을 갈라 적어야 「적힌 대로 나갔다」를 말할 수 있다 — 하나만 적으면 그 주장을 스스로 못 확인한다. """ def test_the_declared_and_the_loaded_sha_agree(self, tmp_path): pics = tmp_path / "pics" pics.mkdir() (pics / "LP01.png").write_bytes(b"\x89PNG\r\n\x1a\nHELLO") cp = _cp(tmp_path, [_row("LP01", "detail", "location_part", b"x")]) got = pp.measure_shot(cp, ["LP01"], root=tmp_path) one = got["refs"][0] assert one["declared_sha256"], "★적힌 지문이 비었다" assert one["sha256"] == one["declared_sha256"] def test_both_are_recorded_separately(self): import inspect src = inspect.getsource(pp.measure_shot) assert '"declared_sha256"' in src and '"sha256"' in src