"""**사람 입력 0**으로 양쪽이 다 끝난다. ★유료 0. 사용자 확정 (2026-08-31): 「궁극적 목적은 자동화이니 HITL을 무조건 필요한 요소로 하면 안 된다. 차후 UI에서 수동 수정하게 고칠 예정이니 지금은 무조건 HITL이 없어야 한다.」 selected 참조를 쓰고 간다 reference_unavailable 참조 없이 간다 — **기록은 남긴다** ★어느 쪽도 사람을 안 기다린다. 다만 **잘못된 부류의 사진을 억지로 쓰지 않는다** — 그건 「아무거나 고른다」가 아니라 「없다고 적고 간다」다. """ from __future__ import annotations import ast import inspect import pytest from app.modules.pipeline import reference_acquisition as ra from app.modules.pipeline import reference_acquisition_rounds as rr TARGET = {"subject_id": "x", "directive_native": "찾을 것", "terms_native": ["낱말"], "language_lock_native": "ko"} def _judge(verdicts): return {"verdicts": [{"index": i, "object_type_match": m, "visible": v} for i, (m, v) in enumerate(verdicts, 1)]} def _found(n): return {"queries": ["q"], "images": [{"image_url": f"https://x/{i}.jpg", "thumbnail_url": "", "source_website_url": "", "caption": ""} for i in range(n)]} class TestBothOutcomesFinishWithoutAPerson: def _run(self, tmp_path, match): return rr.acquire_one( TARGET, workdir=tmp_path, rel_root=tmp_path, search=lambda **kw: _found(2), download=lambda u, d, f="": (d.write_bytes(b"x"), True)[1], judge=lambda c: {"j1": _judge([(match, True)] * len(c))}) def test_a_selection_finishes(self, tmp_path): got = self._run(tmp_path, "yes") assert got["status"] == ra.STATUS_SELECTED assert got["downstream_blocked"] is False assert ra.acquisition_outcome(got["status"]) == ra.STATUS_SELECTED def test_a_wrong_kind_still_finishes_with_the_closest_one(self, tmp_path): """★★여기가 앞 계약에서 **사람을 기다리던** 자리다. 사용자 5단계 ⑤ (2026-09-03): 기준에 맞는 것이 없어도 **가장 닮은 한 장**을 자동으로 고른다 — 사람 없이 끝까지.""" got = self._run(tmp_path, "no") assert got["status"] == ra.STATUS_SELECTED and got["match_quality"] == "closest" assert got["downstream_blocked"] is False, "★사람을 기다린다" assert got["chosen"] is not None and got["chosen"]["forced_pick"] is True def test_the_forced_pick_is_marked_not_hidden(self, tmp_path): """★가장 닮은 것을 고르되 **기준에 맞은 것처럼 적지 않는다** — 기록이 갈라 말한다.""" got = self._run(tmp_path, "no") assert got.get("chosen_path") and got["match_quality"] == "closest" assert got["rounds"][-1]["decision"]["next"] == "select_closest" def test_the_record_says_why_the_last_round_ended(self, tmp_path): got = self._run(tmp_path, "no") assert got["rounds"][-1]["decision"]["why"] assert got["status"] == ra.STATUS_SELECTED class TestNoProductionCodeWaitsForAPerson: """★production 실행 경로에 사람 승인·verdict·대기·GO 가 없다.""" def test_nothing_returns_a_human_review_state(self): src = inspect.getsource(ra) for banned in ("human_review", "awaiting_human", "needs_human", "manual_approval"): assert banned not in src, f"★사람 대기 상태가 있다: {banned}" def test_the_loop_never_asks_for_input(self): for mod in (ra, rr): tree = ast.parse(inspect.getsource(mod)) for n in ast.walk(tree): if isinstance(n, ast.Call) and \ getattr(n.func, "id", "") == "input": raise AssertionError(f"★{mod.__name__} 이 입력을 기다린다") def test_missing_reference_is_allowed(self): assert ra.ALLOW_MISSING_REFERENCE is True class TestTheJudgeIsNotAskedForDetails: """★★VLM 에게 시대·형태 세부를 안 묻는다 (사용자 확정 2026-08-31). 「너무 협소하게 판단해 VLM이 판단할 수 없어. 절대로 사람도 잘 못하는데.」 """ def test_the_kind_is_the_coarse_label_only(self): from tools.grounding_audit import ref_canary as rc t = {"surface_form": "코", "coarse_type_label": "신발의 앞부분", "visual_brief": "검정 고무신 앞부분으로, 끝이 위를 향해 살짝 들린 " "둥근 모양이다."} got = rc.kind_name_of(t) assert got == "신발의 앞부분" assert "고무신" not in got and "1960" not in got, \ "★시대·형태 세부가 심판에게 간다" def test_the_appearance_never_reaches_the_judge(self): from tools.grounding_audit import ref_canary as rc brief = "발목까지 내려오며 앞을 끈으로 여미는 긴 겉옷이다" got = rc.kind_name_of({"surface_form": "긴 겉옷", "coarse_type_label": "사람이 입는 겉옷", "visual_brief": brief}) assert brief not in got def test_the_search_still_gets_the_details(self): """★positive control — 세부는 **검색 쪽에** 그대로 쓴다. ★★**나가는 것**을 본다. 앞 판은 `_write_brief` 의 소스에 낱말이 있나만 봐서, 저작 조립이 다른 함수로 옮겨가자 깨졌다 — 그리고 그때 「검색이 세부를 받나」는 **여전히 안 재고 있었다**. """ from tools.grounding_audit import ref_canary as rc brief = "발목까지 내려오며 앞을 끈으로 여미는 긴 겉옷이다" t = {"surface_form": "긴 겉옷", "owner_type": "outlook", "coarse_type_label": "사람이 입는 겉옷", "visual_brief": brief} out = rc.brief_outbound(t, world_facts="W", source_text="S", narrow=False) assert brief in out["user"], "★검색 저작에서 겉모습을 뺐다" assert brief not in rc.kind_name_of(t), "★심판에게도 간다" def test_the_pack_asks_for_the_label(self): from app.modules.pipeline import grounding_chunk as gc body = gc.load_text("system.md") assert "coarse_type_label" in body schema = gc.build_chunk_payload(["scene-1"], {"scene-1": "글"}, "규칙")["schema"] props = schema["properties"]["rows"]["items"] assert "coarse_type_label" in props["properties"] assert "coarse_type_label" in props["required"] def test_the_pack_forbids_the_detail_axes(self): from app.modules.pipeline import grounding_chunk as gc body = gc.load_text("system.md") i = body.index("coarse_type_label") block = body[i:i + 600] for w in ("연도", "나라", "재질", "생김새"): assert w in block, f"★{w} 를 금지한다고 안 적혀 있다" class TestTheOutcomeIsWrittenNotJustComputable: """★★★**함수로만 있으면 없는 것과 같다** (Codex BLOCK 2026-08-31). 앞 판은 `acquisition_outcome()` 이 있는데 산출에 안 적혀서, 얼어붙은 기록에는 raw status 뿐이었다. 「못 구한 것은 `reference_unavailable` 로 적힌다」는 보고가 실제 기록과 달랐다. ★raw status 를 **지우지 않는다** — 왜 못 구했는지는 감사에 필요하다. `outcome` 은 그 위에 얹는 **하류가 보는 한 칸**이다. """ @staticmethod def _run(tmp_path, *, found): from app.modules.pipeline import reference_acquisition_rounds as rar def _search(**kw): return {"queries": [["질의"]], "images": ([{"url": "https://x/1.png"}] if found else [])} return rar.acquire_one( {"subject_id": "S1", "directive_native": "찾을 것", "terms_native": ["말"], "language_lock_native": "ko"}, workdir=tmp_path, rel_root=tmp_path, search=_search, download=lambda u, d, f="": (d.write_bytes(b"x"), True)[1], judge=lambda c: {"j": {"by_candidate": [ {"index": i + 1, "object_type_match": "yes", "visible": True} for i, _ in enumerate(c)]}} if found else {}) def test_a_failed_acquisition_writes_the_outcome(self, tmp_path): from app.modules.pipeline import reference_acquisition as ra got = self._run(tmp_path, found=False) assert "outcome" in got, "★처분이 산출에 안 적힌다" assert got["outcome"] == ra.STATUS_UNAVAILABLE assert got["status"] != got["outcome"], \ "★raw status 를 덮어썼다 — 왜 못 구했는지가 사라진다" assert got["downstream_blocked"] is False def test_both_fields_are_present_on_every_path(self, tmp_path): for found in (True, False): got = self._run(tmp_path / f"w{found}", found=found) assert got.get("status") and got.get("outcome"), \ f"★found={found} 에서 칸이 빈다: {got.get('status')!r} " \ f"{got.get('outcome')!r}" def test_the_outcome_is_the_one_mapping(self, tmp_path): """★두 벌로 만들면 한쪽만 고쳐진다 — 산출은 그 함수를 **부른다**.""" import inspect from app.modules.pipeline import reference_acquisition_rounds as rar src = inspect.getsource(rar.acquire_one) assert '"outcome": ra.acquisition_outcome(status)' in src class TestTheStepDocMatchesThePolicy: """★★고친 규칙이 **다른 자리에서 되살아나지 않게** 한다. 금지어 lint 로 막지 않는다 — 그러면 설명을 못 쓰게 된다. 대신 **정책 함수에 물어** 그 답과 문서가 같은 방향인지 본다 (Codex 2026-09-01). """ def test_the_policy_says_no_match_does_not_block(self): for st in (ra.STATUS_NO_MATCH, ra.STATUS_RETRYABLE): assert ra.downstream_blocked(st) is False assert ra.acquisition_outcome(st) == ra.STATUS_UNAVAILABLE def test_the_step_doc_does_not_claim_the_opposite(self): """★부정문은 지우지 않는다 — **긍정으로 쓰인 자리**만 본다.""" import re from app.core.steps import reference_acquisition_step as step doc = inspect.getdoc(step.ReferenceAcquisitionStep) or "" assert "하류" in doc, "★이 문서가 하류를 아예 안 말한다" for line in doc.splitlines(): if "하류를 막" not in line: continue # ★「★가 있으면 봐준다」 같은 빠져나갈 구멍을 두지 않는다 — # 그러면 「★하류를 막는다」가 그냥 지나간다 assert re.search(r"(안|않)\s*(막|는다)", line), \ f"★정책과 반대로 적혀 있다: {line.strip()}"