"""장부 → 중앙 획득 **한 곳**. ★유료 0 — 검색·다운로드·판정을 갈아 끼운다. Codex 끝점 셋 (09-01) — ①active caller 가 `inputs_from_ledger` **한 입구만** 소비하는지 ②`auto_completed`/`not_applicable`/`skipped`/`refused` 가 **durable 산출에 보존**되는지 ③provider 앞에서 **구매 상한·신원·재개 문**을 지나는지 그리고 Codex 재지적 (09-01) — **상한에 닿아도 사람을 안 기다린다.** """ from __future__ import annotations import inspect import json import pytest from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_outlook_binding as ob from app.modules.pipeline import reference_acquisition as ra from app.modules.pipeline import search_grounded_ref as sgr from app.modules.pipeline.grounding_entity_contract import (FACET_BINDING, PRODUCER_PAYLOAD) P3 = {"outlooks": [{"outlook_id": "O01"}], "scene_assignments": [{"scene_index": 1, "assignments": [ {"character_id": "C01", "outlook_id": "O01"}]}]} def _row(rsid, screen="obligation", seg="scene-1", brief="긴 겉옷"): return {"research_subject_id": rsid, "owner_type": "outlook", "screen": screen, FACET_BINDING: {"local_id": "c0#2", "final_id": "O01", "owner_type": "outlook", "parent_local_id": "c0#1", "parent_owner_type": "character", "parent_final_id": "C01"}, PRODUCER_PAYLOAD: {"coarse_type_label": "두루마기", "visual_brief": brief, "search_terms_native": ["두루마기"], "language_lock_native": "ko"}, "source_evidence": {"source_quote": "원문", "occurrences": [{"source_span": { "segment_id": seg}}]}} @pytest.fixture def journal(tmp_path): from app.modules.pipeline import grounding_chunk_journal as cj return cj.ChunkJournal(tmp_path / "j.json", contract={"v": 1}) class _Spy: """검색·다운로드·판정을 **세는** 대역. ★바깥으로 한 번도 안 나간다. ★★대역 모양을 **지어내지 않는다.** 앞 판에서 제가 판정 payload 를 만들어 냈더니 실물이 못 읽고 2라운드로 갔다 — 그런 줄 모르고 「상한을 넘겨 샀다」로 읽을 뻔했다. 아래는 `coarse_type_pick` 이 실제로 읽는 칸이고, `TestTheFakeIsShapedLikeTheRealThing` 이 그것을 잠근다. """ def __init__(self, match=True): self.searched = 0 self.match = match def search(self, **kw): self.searched += 1 return {"queries": ["q"], "images": [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": "", "source_website_url": "", "caption": ""} for i in range(3)]} def download(self, url, dest, fallback_url=""): dest.write_bytes(b"x") return True def judge(self, got): # ★enum 값을 **손으로 안 적는다** — production 것을 그대로 쓴다. # 앞 판에서 `True` 를 넣었더니 실물이 「enum 밖」으로 버렸고, # 그 탓에 안 골라진 것을 「상한을 넘겨 샀다」로 읽을 뻔했다. yes, no = ctp.TYPE_MATCH[0], ctp.TYPE_MATCH[1] return {"j1": {"verdicts": [ {"index": i, "object_type_match": (yes if self.match else no), "visible": True} for i in range(1, len(got) + 1)]}} def _buy(got): """구매 갈래 줄만. ★`rows` 는 이제 **장부 전부**를 덮는다.""" return [r for r in got["rows"] if r["disposition"] in (ca.DISP_ACQUIRED, ca.DISP_CAP_REACHED, ca.DISP_UNCONFIRMED)] def _run(ledger, journal, tmp_path, cap=10, spy=None): spy = spy or _Spy() got = ca.run(ledger, journal=journal, cap=cap, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) return got, spy class TestTheFakeIsShapedLikeTheRealThing: """★대역이 실물과 갈리면 **실물에서 안 도는 것**을 통과시킨다.""" def test_the_search_fake_uses_the_real_argument_names(self): want = set(inspect.signature(sgr.search_reference_images).parameters) want.discard("client") assert {"directive_native", "terms_native"} <= want def test_the_download_fake_matches(self): p = list(inspect.signature(sgr.download_candidate).parameters) assert p[:3] == ["url", "dest", "fallback_url"] def test_the_verdict_enum_comes_from_production(self): assert ctp.TYPE_MATCH[0] != ctp.TYPE_MATCH[1] assert True not in ctp.TYPE_MATCH, "★bool 은 enum 값이 아니다" def test_a_matching_judgement_actually_selects(self, journal, tmp_path): """★양성 대조 — 이 대역으로 **정말 골라지는지**. 안 골라지면 위 시험들의 「한 번만 샀다」가 재검색 실패를 잘못 읽은 것이다.""" led = ob.bind([_row("rs1")], P3) got, spy = _run(led, journal, tmp_path) assert _buy(got)[0]["acquisition"]["status"] == ra.STATUS_SELECTED assert spy.searched == 1, "★한 라운드에 골랐어야 한다" def test_a_non_matching_judgement_retries_once_then_stops(self, journal, tmp_path): """★음성 대조 — 종류가 안 맞으면 실물은 **최대 2라운드**를 돈다.""" led = ob.bind([_row("rs1")], P3) got, spy = _run(led, journal, tmp_path, spy=_Spy(match=False)) assert spy.searched == ctp.MAX_ROUNDS # ★사용자 5단계 ⑤ (2026-09-03): 두 라운드 다 안 맞아도 가장 닮은 한 장을 고른다 assert _buy(got)[0]["outcome"] == ra.STATUS_SELECTED class TestItGoesThroughTheOneDoor: def test_the_module_reads_the_ledger_only_through_the_adapter(self): """★★장부를 **여기서 다시 해석하지 않는다** — 입구가 둘이면 갈린다.""" import ast tree = ast.parse(inspect.getsource(ca.run)) calls = {ast.unparse(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)} assert "_aa.inputs_from_ledger" in calls for banned in ("ob.acquisition_targets", "ob.accounting", "_aa.acquisition_inputs"): assert banned not in calls, f"★{banned} 를 직접 부른다" def test_a_not_target_row_never_reaches_the_provider(self, journal, tmp_path): led = ob.bind([_row("rs1", screen="not_target")], P3) got, spy = _run(led, journal, tmp_path) assert spy.searched == 0 assert _buy(got) == [] and got["not_applicable"] == 1 class TestEveryLaneSurvivesIntoTheOutput: def test_all_five_buckets_are_written(self, journal, tmp_path): rows = [_row("rs1"), # 산다 _row("rs2", seg="scene-9"), # 자동 완료 _row("rs3", screen="not_target"), # 비대상 _row("rs4", brief="")] # 저작 재료 없음 rows[3][PRODUCER_PAYLOAD] = {"coarse_type_label": "두루마기"} led = ob.bind(rows, P3) got, spy = _run(led, journal, tmp_path) assert len(_buy(got)) == 1 and spy.searched == 1 assert got["auto_completed"] == 1 assert got["not_applicable"] == 1 assert got["skipped"] == 1 assert [r["research_subject_id"] for r in got["rows"] if r["disposition"] == ca.DISP_SKIPPED] == ["rs4"] assert got["ledger_rows"] == 4 def test_the_output_is_json_serialisable(self, journal, tmp_path): """★durable 이려면 **적을 수 있어야** 한다.""" led = ob.bind([_row("rs1")], P3) got, _spy = _run(led, journal, tmp_path) assert json.loads(json.dumps(got, ensure_ascii=False, default=str)) def test_the_evidence_rides_with_every_row(self, journal, tmp_path): led = ob.bind([_row("rs1")], P3) got, _spy = _run(led, journal, tmp_path) assert all(r["source_evidence"]["source_quote"] == "원문" for r in got["rows"]) class TestEveryLedgerRowSurvivesInFull: """★★Codex BLOCK 09-01 — 수만 남으면 **원행을 못 찾는다.**""" def _four(self): rows = [_row("rs1"), # 산다 _row("rs2", seg="scene-9"), # 자동 완료 _row("rs3", screen="not_target"), # 비대상 _row("rs4", brief="")] # 저작 재료 없음 rows[3][PRODUCER_PAYLOAD] = {"coarse_type_label": "두루마기"} return rows def test_every_id_appears_exactly_once(self, journal, tmp_path): led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) cov = ca.ledger_coverage(led, got) assert cov["ok"], cov assert cov["ledger"] == cov["result"] == 4 assert sorted(r["research_subject_id"] for r in got["rows"]) == \ ["rs1", "rs2", "rs3", "rs4"] def test_the_original_row_rides_with_every_disposition(self, journal, tmp_path): """★★screen·producer payload·facet 좌표가 **그대로** 남아야 한다.""" led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) for r in got["rows"]: orig = r["ledger_row"] assert orig, f"★{r['research_subject_id']} 의 원행이 없다" assert orig["research_subject_id"] == r["research_subject_id"] assert orig["screen"] and orig["owner_type"] assert orig[FACET_BINDING]["parent_final_id"] == "C01" assert orig[PRODUCER_PAYLOAD]["coarse_type_label"] == "두루마기" def test_each_disposition_is_a_known_one(self, journal, tmp_path): led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) seen = {r["disposition"] for r in got["rows"]} assert seen <= set(ca.DISPOSITIONS) assert seen == {ca.DISP_ACQUIRED, ca.DISP_AUTO_DONE, ca.DISP_NOT_APPLICABLE, ca.DISP_SKIPPED} assert sum(got["dispositions"].values()) == 4 def test_the_counts_are_derived_from_the_rows(self, journal, tmp_path): """★수를 따로 세면 두 수가 갈린다 — 행에서 파생해야 한다.""" led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) for name, disp in (("auto_completed", ca.DISP_AUTO_DONE), ("not_applicable", ca.DISP_NOT_APPLICABLE), ("skipped", ca.DISP_SKIPPED), ("refused", ca.DISP_REFUSED)): assert got[name] == sum(1 for r in got["rows"] if r["disposition"] == disp), name def test_not_applicable_is_not_a_failed_search(self, journal, tmp_path): """★★Codex BLOCK 09-01 — 「애초에 필요 없었다」를 「못 찾았다」로 적으면 UI·통계에서 **불필요한 대상이 검색 실패처럼** 보인다. `grounding_outlook_binding.auto_completed` 계약이 이미 둘을 갈라 놓았는데 내가 한 상태(`retryable`)로 뭉갰다. """ led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) by = {r["disposition"]: r for r in got["rows"]} na = by[ca.DISP_NOT_APPLICABLE] assert na["outcome"] is None, "★비대상에 획득 결과를 만들었다" assert na["outcome"] != ra.STATUS_UNAVAILABLE assert na["status"] is None, "★비대상에 획득 상태를 만들었다" assert na["downstream_blocked"] is False assert ca.acquisition_outcome_of(na) is None # ★음성 대조 — **필요했는데** 못 구한 것은 unavailable 이어야 한다 need = by[ca.DISP_AUTO_DONE] assert need["outcome"] == ra.STATUS_UNAVAILABLE assert ca.acquisition_outcome_of(need) == ra.STATUS_UNAVAILABLE assert by[ca.DISP_SKIPPED]["outcome"] == ra.STATUS_UNAVAILABLE def test_counting_missed_references_excludes_the_not_needed( self, journal, tmp_path): """★끝점에서 — 「참조를 못 구한 것」을 셀 때 비대상이 안 섞인다.""" led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) missed = [r for r in got["rows"] if ca.acquisition_outcome_of(r) == ra.STATUS_UNAVAILABLE] assert {r["disposition"] for r in missed} == {ca.DISP_AUTO_DONE, ca.DISP_SKIPPED} assert ca.DISP_NOT_APPLICABLE not in {r["disposition"] for r in missed} def test_the_terminal_check_catches_a_wrong_outcome_either_way(self): """★양성 대조 — 두 축을 **따로** 본다는 것을 확인한다.""" bad_na = {"rows": [{"disposition": ca.DISP_NOT_APPLICABLE, "outcome": ra.STATUS_UNAVAILABLE, "downstream_blocked": False}]} assert ca.unfinished_rows(bad_na), "★비대상에 결과가 붙었는데 지나갔다" bad_need = {"rows": [{"disposition": ca.DISP_AUTO_DONE, "outcome": None, "downstream_blocked": False}]} assert ca.unfinished_rows(bad_need), "★필요한데 결과가 없다" ok = {"rows": [{"disposition": ca.DISP_NOT_APPLICABLE, "outcome": None, "downstream_blocked": False}, {"disposition": ca.DISP_AUTO_DONE, "outcome": ra.STATUS_UNAVAILABLE, "downstream_blocked": False}]} assert ca.unfinished_rows(ok) == [] def test_no_row_of_any_lane_waits_for_a_person(self, journal, tmp_path): led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) assert ca.unfinished_rows(got) == [] assert all(r["downstream_blocked"] is False for r in got["rows"]) def test_the_coverage_endpoint_actually_catches_a_hole(self, journal, tmp_path): """★양성 대조 — 줄을 하나 빼면 이 끝점이 **잡아야** 한다.""" led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) holed = {**got, "rows": got["rows"][:-1]} assert ca.ledger_coverage(led, holed)["ok"] is False assert ca.ledger_coverage(led, holed)["missing"] doubled = {**got, "rows": got["rows"] + got["rows"][:1]} assert ca.ledger_coverage(led, doubled)["duplicated"] def test_the_whole_result_is_json_serialisable(self, journal, tmp_path): led = ob.bind(self._four(), P3) got, _spy = _run(led, journal, tmp_path) assert json.loads(json.dumps(got, ensure_ascii=False, default=str)) class TestTheGatesStandBeforeTheProvider: def test_the_cap_stops_buying_but_not_the_run(self, journal, tmp_path): """★★상한은 **더 안 사는 문**이지 주행을 멈추는 문이 아니다.""" led = ob.bind([_row("rs1"), _row("rs2")], P3) got, spy = _run(led, journal, tmp_path, cap=1) assert spy.searched == 1, "★상한을 넘겨 샀다" assert len(_buy(got)) == 2, "★상한에 걸린 줄이 사라졌다" assert _buy(got)[0]["acquisition"]["status"] == ra.STATUS_SELECTED pur = got["purchases"] assert pur["cap_reached"] is True assert pur["unbought_after_cap"] == 1 # ★★이름이 뜻을 그대로 말한다 — 「대상 2개를 다 **다뤘고**, 그 중 # provider 경계를 지난 것은 1개」. 앞 판은 둘 다 `attempted` 라 # 불러서 서로 모순되게 읽혔다. assert pur["targets_total"] == 2 assert pur["targets_processed"] == 2 assert pur["dispatch_attempted_this_run"] == 1 assert pur["reused_this_run"] == 0 def test_what_was_already_paid_for_survives_the_cap(self, journal, tmp_path): """★★산 것을 **판정보다 먼저** 남긴다 — 앞 판은 예외로 날렸다.""" led = ob.bind([_row("rs1"), _row("rs2")], P3) got, _spy = _run(led, journal, tmp_path, cap=1) paid = _buy(got)[0]["acquisition"] assert paid["chosen"] is not None and paid["rounds"] assert _buy(got)[0]["outcome"] == ra.STATUS_SELECTED def test_the_capped_row_finishes_without_a_person(self, journal, tmp_path): """★★Codex 09-01 — 상한이 **필수 HITL 상태**를 만들면 안 된다.""" led = ob.bind([_row("rs1"), _row("rs2")], P3) got, _spy = _run(led, journal, tmp_path, cap=1) capped = _buy(got)[1] assert capped["why_unbought"] == ca.WHY_CAP_REACHED assert capped["outcome"] == ra.STATUS_UNAVAILABLE assert capped["downstream_blocked"] is False assert ca.unfinished_rows(got) == [], "★사람을 기다리는 줄이 있다" def test_no_row_ever_waits_for_a_person(self, journal, tmp_path): for cap in (0, 1, 2, 9): from app.modules.pipeline import grounding_chunk_journal as cj j = cj.ChunkJournal(tmp_path / f"j{cap}.json", contract={"v": 1}) led = ob.bind([_row("rs1"), _row("rs2", brief="다른 것")], P3) got, _spy = _run(led, j, tmp_path, cap=cap) assert ca.unfinished_rows(got) == [], f"cap={cap}" assert len(_buy(got)) == 2, f"cap={cap} — 줄이 사라졌다" assert ca.ledger_coverage(led, got)["ok"], f"cap={cap}" def test_the_same_identity_is_bought_once(self, journal, tmp_path): """★★재개 — 같은 신원은 **다시 안 산다**.""" led = ob.bind([_row("rs1")], P3) spy = _Spy() for _ in range(3): ca.run(led, journal=journal, cap=5, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) assert spy.searched == 1 assert journal.reused() >= 2 again = ca.run(led, journal=journal, cap=5, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) # ★★재개 전후를 맞댄다 — 이번 판에 **한 번도 안 보냈고**, 한 번 # 되썼고, 누계는 그대로 1 이다 assert again["purchases"]["dispatch_attempted_this_run"] == 0 assert again["purchases"]["reused_this_run"] == 1 assert again["purchases"]["bought_or_uncertain_epoch"] == 1 assert again["purchases"]["targets_processed"] == 1 def test_a_changed_query_is_a_different_purchase(self): """★음성 대조 — 질의가 바뀌면 **다른 것**이다.""" def _id(directive="가", hint="h", sha="s"): return ca.identity_of({"target": {"subject_id": "x", "directive_native": directive}, "narrow_hint": hint}, contract_sha=sha) assert len({_id(), _id(directive="나"), _id(hint="다른 힌트"), _id(sha="다른 계약")}) == 4 def test_a_changed_pack_or_round_is_a_different_purchase(self): """★★팩·라운드가 바뀌면 **옛 구매를 되쓰면 안 된다.** 앞 판은 이 모듈의 버전 문자열만 접어서, 팩을 바꿔도 신원이 같았다 — 옛 참조가 영구히 봉인된다. 지금은 production 한 곳(`acquisition_ contract_sha`)이 만든 해시를 받는다. """ assert ra.acquisition_contract_sha(rounds=1) \ != ra.acquisition_contract_sha(rounds=ctp.MAX_ROUNDS) assert ra.acquisition_contract_sha(rounds=ctp.MAX_ROUNDS) \ != ra.acquisition_contract_sha(rounds=ctp.MAX_ROUNDS, coarse_version="다른 팩") def test_the_run_folds_the_production_contract_into_the_identity( self, journal, tmp_path): led = ob.bind([_row("rs1")], P3) got, _spy = _run(led, journal, tmp_path) sha = got["purchases"]["acquisition_contract"] assert sha == ra.acquisition_contract_sha(rounds=ctp.MAX_ROUNDS) # ★대상을 **지어내지 않는다** — 프로덕션 입구가 낸 것을 그대로 쓴다 from app.modules.pipeline import grounding_acquisition_adapter as aa one = aa.inputs_from_ledger(ob.bind([_row("rs1")], P3))["targets"][0] assert _buy(got)[0]["identity"] == ca.identity_of(one, contract_sha=sha) def test_a_different_round_count_does_not_reuse(self, journal, tmp_path): """★끝점에서 잰다 — 라운드를 바꾸면 **또 산다**(되쓰지 않는다).""" led = ob.bind([_row("rs1")], P3) spy = _Spy() for r in (1, 2): ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, rounds=r) assert spy.searched == 2, "★라운드가 달라졌는데 옛 구매를 되썼다" assert journal.reused() == 0 def test_an_unconfirmed_prior_call_never_waits_for_a_person( self, journal, tmp_path, monkeypatch): """★★「샀는지 모른다」 — 다시 사지도, 사람을 기다리지도 않는다.""" from app.modules.pipeline import grounding_chunk_journal as cj led = ob.bind([_row("rs1"), _row("rs2", brief="다른 것")], P3) spy = _Spy() calls = {"n": 0} real = cj.buy_or_reuse def _fake(j, ident, *, cap, send, stop_check=None, **_kw): calls["n"] += 1 if calls["n"] == 1: raise cj.NeedsHumanDecision("앞 판 답을 못 받았다") return real(j, ident, cap=cap, send=send, stop_check=stop_check) monkeypatch.setattr(cj, "buy_or_reuse", _fake) got = ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) assert len(_buy(got)) == 2 assert _buy(got)[0]["why_unbought"] == ca.WHY_PRIOR_UNCONFIRMED assert _buy(got)[0]["outcome"] == ra.STATUS_UNAVAILABLE assert got["purchases"]["unconfirmed_prior_calls"] == 1 assert ca.unfinished_rows(got) == [], "★사람을 기다린다" # ★뒤 대상은 **그대로 산다** — 하나가 미궁이라고 판이 서지 않는다 assert _buy(got)[1]["outcome"] == ra.STATUS_SELECTED def test_it_does_not_rebuy_an_unconfirmed_identity(self, journal, tmp_path, monkeypatch): """★음성 대조 — 미궁이면 **provider 를 안 부른다**(다시 사지 않는다).""" from app.modules.pipeline import grounding_chunk_journal as cj led = ob.bind([_row("rs1")], P3) spy = _Spy() monkeypatch.setattr(cj, "buy_or_reuse", lambda *a, **k: (_ for _ in ()) .throw(cj.NeedsHumanDecision("미궁"))) got = ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) assert spy.searched == 0 assert got["purchases"]["dispatch_attempted_this_run"] == 0 assert got["purchases"]["bought_or_uncertain_epoch"] == 0 def test_a_stop_check_fires_before_any_spend(self, journal, tmp_path): led = ob.bind([_row("rs1")], P3) spy = _Spy() def _stop(): raise RuntimeError("멈춰라") with pytest.raises(RuntimeError, match="멈춰라"): ca.run(led, journal=journal, cap=5, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, stop_check=_stop) assert spy.searched == 0 class TestTheCostDoorTakesNoLooseValues: """★비용 문에 관용은 없다 (Codex NON-BLOCK 09-01).""" @pytest.mark.parametrize("bad", [True, False, "1", 1.0, -1, None]) def test_a_loose_cap_stops_before_anything_opens(self, journal, tmp_path, bad): led = ob.bind([_row("rs1")], P3) spy = _Spy() with pytest.raises(ca.BadPurchaseCap): ca.run(led, journal=journal, cap=bad, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) assert spy.searched == 0 assert journal.bought() == 0, "★장부를 건드렸다" def test_cap_zero_is_allowed_and_buys_nothing(self, journal, tmp_path): led = ob.bind([_row("rs1")], P3) got, spy = _run(led, journal, tmp_path, cap=0) assert spy.searched == 0 assert len(_buy(got)) == 1 assert _buy(got)[0]["why_unbought"] == ca.WHY_CAP_REACHED assert ca.unfinished_rows(got) == [] @pytest.mark.parametrize("bad", [True, "2", 2.0, 0, -1]) def test_a_loose_round_count_stops(self, journal, tmp_path, bad): led = ob.bind([_row("rs1")], P3) spy = _Spy() with pytest.raises(ca.BadRoundCount): ca.run(led, journal=journal, cap=5, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, rounds=bad) assert spy.searched == 0 def test_none_resolves_to_the_loops_own_default(self): assert ca.resolve_rounds(None) == ctp.MAX_ROUNDS class TestAProviderFailureDoesNotStrandTheRun: """★★Codex BLOCK 2 — provider 예외 → 다음 재개까지 **사람 입력 0**.""" @staticmethod def _escaping_once(monkeypatch): """★`acquire_one` **밖으로 튀는** 예외를 만든다. 검색·다운로드 실패는 `acquire_one` 이 이미 안에서 `retryable` 로 매듭짓는다(아래 시험이 그것을 확인한다). 새는 자리는 그 밖이다 — `buy_or_reuse` 가 장부에 `uncertain` 을 적고 예외를 **다시 올린다**. ★★그리고 **선언된 종류**만 접는다. 아무 `RuntimeError` 나 접으면 내 코드 결함이 provider 장애로 둔갑한다 (Codex 2026-09-01). """ from app.modules.pipeline import reference_acquisition_rounds as rr real, state = rr.acquire_one, {"n": 0} def _fake(*a, **k): state["n"] += 1 if state["n"] == 1: raise ca.ProviderCallUncertain("보냈는데 답이 안 왔다") return real(*a, **k) monkeypatch.setattr(rr, "acquire_one", _fake) def test_a_search_failure_is_already_terminal_inside_the_loop( self, journal, tmp_path): """★먼저 확인 — 흔한 실패는 loop 안에서 이미 자동 종결된다.""" led = ob.bind([_row("rs1")], P3) spy = _Spy() def _dead(**kw): spy.searched += 1 raise RuntimeError("provider 가 답을 안 줬다") got = ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=_dead, download=spy.download, judge=spy.judge) assert got["rows"][0]["outcome"] == ra.STATUS_UNAVAILABLE assert ca.unfinished_rows(got) == [] def test_the_failed_row_finishes_and_the_rest_go_on(self, journal, tmp_path, monkeypatch): led = ob.bind([_row("rs1"), _row("rs2", brief="다른 것")], P3) spy = _Spy() self._escaping_once(monkeypatch) got = ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) assert len(got["rows"]) == 2 assert _buy(got)[0]["why_unbought"] == ca.WHY_CURRENT_UNCONFIRMED assert _buy(got)[0]["outcome"] == ra.STATUS_UNAVAILABLE assert _buy(got)[1]["outcome"] == ra.STATUS_SELECTED assert ca.unfinished_rows(got) == [] def test_the_next_resume_still_needs_no_person(self, journal, tmp_path, monkeypatch): """★★여기가 진짜 자리다 — 실패한 신원이 장부에 `uncertain` 으로 남은 **뒤** 다시 돌린다. 앞 판은 여기서 사람을 불렀다.""" from app.modules.pipeline import grounding_chunk_journal as cj led = ob.bind([_row("rs1"), _row("rs2", brief="다른 것")], P3) spy = _Spy() self._escaping_once(monkeypatch) ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) # ★장부에 미궁이 **정말로 남았는지** 먼저 본다 — 안 남았으면 아래 # 재개 시험은 아무것도 안 재는 것이다 assert [e for e in journal.entries.values() if e.get("status") == cj.STATUS_UNCERTAIN] again = ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) assert ca.unfinished_rows(again) == [], "★재개가 사람을 기다린다" assert _buy(again)[0]["why_unbought"] == ca.WHY_PRIOR_UNCONFIRMED # ★★미궁인 신원을 **다시 사지 않는다** — 그러면 상한을 넘긴다 assert again["purchases"]["unconfirmed_prior_calls"] == 1 def test_it_does_not_swallow_a_stop_or_a_defect_mid_send( self, journal, tmp_path, monkeypatch): """★★Codex 09-01 — 자동 종결이 **취소·코드 결함까지** 삼키면 안 된다. 여기가 위험한 자리다: `send()` 안에서 났으므로 「보냈다」 표시는 이미 서 있다. 그것만 보고 접으면 Ctrl-C 를 눌러도 다음 대상을 계속 산다. """ from app.core.errors import AppError from app.core.run_control import ABORT_CODES from app.modules.pipeline import reference_acquisition_rounds as rr led = ob.bind([_row("rs1"), _row("rs2", brief="다른 것")], P3) boom = [KeyboardInterrupt("사람이 멈췄다"), SystemExit(1), AttributeError("내 코드가 틀렸다"), TypeError("내 코드가 틀렸다"), AppError(code=sorted(ABORT_CODES)[0], message="주행이 섰다")] from app.modules.pipeline import grounding_chunk_journal as cj for n, exc in enumerate(boom): spy = _Spy() # ★판마다 **새 장부**다 — 앞 판이 남긴 미궁 때문에 `acquire_one` # 까지 못 가면 이 시험은 아무것도 안 재는 것이 된다 j = cj.ChunkJournal(tmp_path / f"sw{n}.json", contract={"v": 1}) def _raise(*a, _e=exc, **k): raise _e monkeypatch.setattr(rr, "acquire_one", _raise) with pytest.raises(type(exc)): ca.run(led, journal=j, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) def test_only_the_declared_kind_is_folded(self): """★★기본값이 **안 접는다** — 접을 것을 선언한다(allowlist). 앞 판은 「안 접을 것」을 적었는데, 그러면 `KeyError`·`ValueError`· `IndexError` 처럼 목록에 없는 코드 결함이 provider 장애로 둔갑한다. """ from app.core.errors import AppError from app.core.run_control import ABORT_CODES assert ca.must_not_swallow(ca.ProviderCallUncertain()) is False for exc in (KeyboardInterrupt(), SystemExit(), AttributeError(), TypeError(), KeyError("k"), ValueError(), IndexError(), RuntimeError("무엇인지 모른다"), OSError(), AppError(code=sorted(ABORT_CODES)[0], message="x")): assert ca.must_not_swallow(exc) is True, f"★{exc!r} 를 삼킨다" def test_only_a_transport_that_knows_may_wrap(self): """★★Codex NON-BLOCK — 「무엇이든 실패했으니 불확정」으로 감싸면 allowlist 가 다시 blacklist 가 된다.""" e = OSError("연결이 끊겼다") assert ca.wrap_if_unconfirmed(e, request_left_the_process=False) is e assert ca.must_not_swallow( ca.wrap_if_unconfirmed(e, request_left_the_process=False)) is True wrapped = ca.wrap_if_unconfirmed(e, request_left_the_process=True) assert isinstance(wrapped, ca.ProviderCallUncertain) assert ca.must_not_swallow(wrapped) is False def test_production_raises_it_nowhere_yet(self): """★지금 이것을 내는 production 자리는 **0곳**이다 — 배선 때 한 자리.""" import ast from pathlib import Path root = Path(__file__).resolve().parents[2] / "app" hits = [] for f in root.rglob("*.py"): try: tree = ast.parse(f.read_text(encoding="utf-8")) except SyntaxError: # noqa: PERF203 continue for n in ast.walk(tree): if not isinstance(n, ast.Raise) or n.exc is None: continue if "ProviderCallUncertain" in ast.unparse(n.exc): hits.append(f"{f.name}:{n.lineno}") assert hits == [], f"★배선됐다 — 운반층 한 자리인지 보라: {hits}" def test_an_undeclared_failure_stops_the_run(self, journal, tmp_path, monkeypatch): """★끝점에서 — 선언 안 된 것은 **판을 세운다**(자동 종결 아님).""" from app.modules.pipeline import reference_acquisition_rounds as rr led = ob.bind([_row("rs1")], P3) spy = _Spy() for exc in (KeyError("칸이 없다"), ValueError("값이 이상하다"), IndexError("범위 밖"), RuntimeError("모르겠다")): from app.modules.pipeline import grounding_chunk_journal as cj j = cj.ChunkJournal(tmp_path / f"u{type(exc).__name__}.json", contract={"v": 1}) monkeypatch.setattr(rr, "acquire_one", lambda *a, _e=exc, **k: (_ for _ in ()) .throw(_e)) with pytest.raises(type(exc)): ca.run(led, journal=j, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge) def test_a_stop_still_propagates_before_the_provider(self, journal, tmp_path): """★음성 대조 — 멈추라는 말은 **삼키지 않는다**.""" led = ob.bind([_row("rs1")], P3) spy = _Spy() def _stop(): raise KeyboardInterrupt("멈춰라") with pytest.raises(KeyboardInterrupt): ca.run(led, journal=journal, cap=9, workdir=tmp_path, rel_root=tmp_path, search=spy.search, download=spy.download, judge=spy.judge, stop_check=_stop) assert spy.searched == 0 class TestItNamesNoNewTerminalState: def test_the_outcome_comes_from_the_policy_function(self): """★종결 상태를 **여기서 짓지 않는다** — 두 곳이면 한쪽만 고쳐진다.""" import ast tree = ast.parse(inspect.getsource(ca)) calls = {ast.unparse(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)} assert "ra.acquisition_outcome" in calls assert "ra.downstream_blocked" in calls def test_no_human_waiting_state_is_declared(self): """★AST 로 **문자열 값**만 본다 — 설명 주석은 안 걸린다.""" import ast tree = ast.parse(inspect.getsource(ca)) vals = {n.value for n in ast.walk(tree) if isinstance(n, ast.Constant) and isinstance(n.value, str)} for banned in ("human_review", "awaiting_human", "needs_human", "manual_approval"): assert not [v for v in vals if banned in v], f"★{banned}" class TestItIsNotWiredYet: def test_no_step_imports_it(self): import ast from pathlib import Path root = Path(__file__).resolve().parents[2] / "app" hits = [] for f in root.rglob("*.py"): if f.name == "grounding_central_acquisition.py": continue try: tree = ast.parse(f.read_text(encoding="utf-8")) except SyntaxError: # noqa: PERF203 continue for n in ast.walk(tree): if (isinstance(n, ast.ImportFrom) and "grounding_central_acquisition" in str( n.module or "")): hits.append(f.name) assert hits == [], f"★벌써 배선됐다: {hits}" def test_it_builds_no_second_buyer(self): """★새 구매 구현을 **안 만든다** — 기존 라운드 기구에 넘긴다.""" import ast tree = ast.parse(inspect.getsource(ca)) calls = {ast.unparse(n.func) for n in ast.walk(tree) if isinstance(n, ast.Call)} assert "rr.acquire_one" in calls assert not [c for c in calls if "urlopen" in c or "requests" in c]