"""★★★VLM 은 **「우리가 찾던 종류가 맞나」만** 본다 (사용자 확정 2026-08-30). 「우리가 찾던 게 맞냐는 해당 오브젝트 종류만 보는것 뿐이야 상세히는 인간도 몰라 전문가가 아니면. 자동차인지, 화폐인지 등등 만」 그래서 옛 pick 과 다르다: - 옛것: 0~100 품질 점수 · 시대/국적 판정 · 평범함 · 관광지 여부 → **평균 최고점** - 여기: `object_type_match` · `visible` → **eligibility 합의 + 결정적 index** ★평균을 안 쓰는 이유는 실측이다 — Gemini 가 0~100 을, GPT 가 사실상 0~10 을 써서 평균이 한쪽에 지배됐다(주유소 그룹 100 대 10). """ import pytest from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline.coarse_type_pick import (combine_coarse_verdicts, eligibility_table, load_pack) def _v(*rows): return {"verdicts": [{"index": i, "object_type_match": m, "visible": v} for i, (m, v) in enumerate(rows, 1)]} class TestThePromptAsksOnlyForTheKind: @staticmethod def _text(): return load_pack()["stems"]["system"]["content"] def test_it_asks_the_two_questions(self): t = self._text() assert "object_type_match" in t and "visible" in t def test_it_says_what_it_is_not_asking(self): """★안 묻는 것을 **적어 둬야** 모델이 알아서 판정하지 않는다.""" t = self._text() assert "What you are NOT asked" in t for phrase in ("year", "country", "model", "tourist", "good photograph"): assert phrase in t, phrase def test_it_asks_resemblance_of_form_but_never_quality_or_era(self): """★v2: 닮은 정도(similarity)는 묻는다 — 마지막에 한 장을 고르려고. 좋고 나쁨·시대·나라는 여전히 안 묻는다.""" t = self._text().lower() assert "similarity" in t and "resemblance" in t assert "year" in t and "country" in t and "not** asked" in t def test_it_says_a_specialist_would_be_needed(self): assert "specialist" in self._text() class TestEligibilityIsConservative: def test_all_judges_must_say_yes(self): got = combine_coarse_verdicts( {"gemini": _v(("yes", True)), "grok": _v(("yes", True))}, 1) assert got["chosen_index"] == 1 and got["eligible"] == [1] def test_one_no_blocks_it(self): got = combine_coarse_verdicts( {"gemini": _v(("yes", True)), "grok": _v(("no", True))}, 1) assert got["chosen_index"] == 0 assert "안 고른다" in got["reason"] def test_unsure_is_not_yes(self): """★모르겠다는 맞다가 아니다.""" got = combine_coarse_verdicts( {"gemini": _v(("yes", True)), "grok": _v(("unsure", True))}, 1) assert got["chosen_index"] == 0 def test_not_visible_blocks_it(self): got = combine_coarse_verdicts( {"gemini": _v(("yes", False)), "grok": _v(("yes", True))}, 1) assert got["chosen_index"] == 0 class TestItIsDeterministicNotAnAverage: def test_the_lowest_eligible_index_wins(self): """★순위는 **검색**이 정한다 — VLM 이 더 나은 쪽을 고르지 않는다.""" j = _v(("yes", True), ("yes", True), ("yes", True)) got = combine_coarse_verdicts({"gemini": j, "grok": j}, 3) assert got["eligible"] == [1, 2, 3] assert got["chosen_index"] == 1 def test_the_same_input_gives_the_same_answer(self): j = _v(("no", True), ("yes", True), ("yes", True)) a = combine_coarse_verdicts({"gemini": j, "grok": j}, 3) b = combine_coarse_verdicts({"grok": j, "gemini": j}, 3) assert a["chosen_index"] == b["chosen_index"] == 2 def test_there_is_no_score_anywhere_in_the_output(self): j = _v(("yes", True)) got = combine_coarse_verdicts({"gemini": j}, 1) assert "score" not in repr(got) class TestAJudgeAnswerIsAtomic: def test_a_short_answer_removes_that_judge(self): got = combine_coarse_verdicts( {"gemini": _v(("yes", True)), "grok": {"verdicts": []}}, 1) assert got["judges"] == ["gemini"] assert "grok" in got["rejected_judges"] assert got["single_judge"] is True def test_a_duplicate_index_removes_that_judge(self): bad = {"verdicts": [{"index": 1, "object_type_match": "yes", "visible": True}, {"index": 1, "object_type_match": "no", "visible": True}]} got = combine_coarse_verdicts({"gemini": bad}, 2) assert got["judges"] == [] and got["chosen_index"] == 0 def test_an_enum_violation_removes_that_judge(self): bad = {"verdicts": [{"index": 1, "object_type_match": "아마도", "visible": True}]} got = combine_coarse_verdicts({"gemini": bad}, 1) assert "gemini" in got["rejected_judges"] def test_no_valid_judge_fails_closed(self): """★조용히 1번을 집지 않는다.""" got = combine_coarse_verdicts({"gemini": {"verdicts": "x"}}, 2) assert got["chosen_index"] == 0 and "fail-closed" in got["reason"] def test_removals_are_written_down(self): got = combine_coarse_verdicts({"gemini": {"verdicts": []}}, 1) assert got["rejected_judges"], "★조용한 제외 금지" class TestTheAuditTableKeepsWhatEachJudgeSaid: def test_it_shows_every_candidate_and_judge(self): rows = eligibility_table( {"gemini": _v(("yes", True), ("no", True)), "grok": _v(("yes", True), ("yes", False))}, 2) assert [r["index"] for r in rows] == [1, 2] assert rows[1]["by_judge"]["gemini"]["object_type_match"] == "no" assert rows[1]["by_judge"]["grok"]["visible"] is False class TestTheRoundContract: """★「없으면 질의를 좁혀 **한 번 더**」 (사용자 확정 + Codex 계약). ★★핵심은 **「없다」와 「못 봤다」를 가르는 것**이다. provider 오류나 시간·예산 중단을 narrow retry 로 바꾸면 「다 보고 없었다」가 거짓이 된다. """ def test_the_cap_comes_from_the_existing_constant(self): """★★**새 상수를 만들지 않는다** (Codex 2026-08-30). 사용자가 3~5 로 확정했고 기존 `era_research.MAX_CANDIDATES` 가 **4** 라 그 안에 든다. 같은 뜻의 수를 둘 두면 **한쪽만 고쳐진다** — 처음에 5 로 뒀던 것을 되돌렸다. """ from app.modules.pipeline.era_research import MAX_CANDIDATES assert ctp.PER_ROUND_CAP == MAX_CANDIDATES assert 3 <= ctp.PER_ROUND_CAP <= 5, "사용자가 정한 범위" assert ctp.TOTAL_AUDIT_CAP == ctp.PER_ROUND_CAP * ctp.MAX_ROUNDS assert ctp.MAX_ROUNDS == 2 def test_it_is_still_not_the_outdoor_structure_contract(self): """★야외 구조물 판정용 상수(6·12)와는 여전히 다른 계약이다.""" from app.modules.pipeline import search_grounded_ref as legacy assert ctp.PER_ROUND_CAP != legacy.SEARCH_IMAGE_RESULTS assert ctp.TOTAL_AUDIT_CAP != legacy.MAX_PICK_CANDIDATES def test_an_eligible_first_round_does_not_retry(self): got = ctp.decide_next_round(1, {"chosen_index": 2, "reason": "골랐다"}, new_candidate_count=5) assert got["next"] == ctp.NEXT_SELECT def test_an_empty_first_round_narrows_and_retries(self): got = ctp.decide_next_round(1, {"chosen_index": 0}, new_candidate_count=5) assert got["next"] == ctp.NEXT_NARROW_RETRY def test_a_provider_error_is_not_the_same_as_finding_nothing(self): """★★이것을 narrow retry 로 바꾸면 「다 보고 없었다」가 거짓이 된다.""" # ★2026-09-03 (Codex BLOCK 3): 1라운드 실패는 **기록하고 넓혀 한 번 더** · 마지막 라운드에서 # 받은 것이 0 이면 retryable(못 봤다) · 받은 것이 있으면 강제로 한 장 got = ctp.decide_next_round(1, {"chosen_index": 0}, new_candidate_count=0, error="502") assert got["next"] == ctp.NEXT_NARROW_RETRY and got["error"] == "502" last = ctp.decide_next_round(ctp.MAX_ROUNDS, {"chosen_index": 0}, new_candidate_count=0, error="502", total_candidate_count=0) assert last["next"] == ctp.NEXT_RETRYABLE and last["error"] == "502" assert "없다" not in last["why"].replace("못 봤다", "") forced = ctp.decide_next_round(ctp.MAX_ROUNDS, {"chosen_index": 0}, new_candidate_count=0, error="502", total_candidate_count=2) assert forced["next"] == ctp.NEXT_SELECT_CLOSEST and forced["forced_reason"] == "judge_unavailable" def test_a_limit_is_not_the_same_either(self): got = ctp.decide_next_round(1, {"chosen_index": 0}, new_candidate_count=5, limit_kind="run_deadline") assert got["next"] == ctp.NEXT_NARROW_RETRY assert got["limit_kind"] == "run_deadline" last = ctp.decide_next_round(ctp.MAX_ROUNDS, {"chosen_index": 0}, new_candidate_count=0, limit_kind="run_deadline", total_candidate_count=0) assert last["next"] == ctp.NEXT_RETRYABLE and last["limit_kind"] == "run_deadline" def test_no_new_candidates_and_none_before_is_no_match(self): got = ctp.decide_next_round(2, {"chosen_index": 0}, new_candidate_count=0, total_candidate_count=0) assert got["next"] == ctp.NEXT_NO_MATCH assert "새 후보가 없다" in got["why"] def test_no_new_candidates_but_some_before_picks_the_closest(self): """★사용자 5단계 ⑤ — 받은 것이 있으면 마지막엔 반드시 한 장.""" got = ctp.decide_next_round(2, {"chosen_index": 0}, new_candidate_count=0, total_candidate_count=3) assert got["next"] == ctp.NEXT_SELECT_CLOSEST def test_two_rounds_of_wrong_kind_end_in_the_closest_pick(self): got = ctp.decide_next_round(2, {"chosen_index": 0}, new_candidate_count=5) assert got["next"] == ctp.NEXT_SELECT_CLOSEST def test_there_is_no_third_round(self): assert ctp.decide_next_round( 2, {"chosen_index": 0}, new_candidate_count=5 )["next"] != ctp.NEXT_NARROW_RETRY class TestDedupeKeepsTheLedger: """★중복을 **버리지 않는다** — 버리면 「받은 수」와 「본 수」가 갈라지는데 그 차이가 안 보인다.""" @staticmethod def _c(url, h=""): return {"url": url, "content_hash": h} def test_a_repeat_url_is_recorded_not_dropped(self): got = ctp.dedupe_candidates([self._c("https://a"), self._c("https://b")], seen_urls=["https://a"]) assert got["judged_count"] == 1 assert got["duplicate_count"] == 1 assert len(got["ledger"]) == 2, "★버리면 장부에서 사라진다" assert got["ledger"][0]["disposition"] == ctp.DISP_DUPLICATE def test_a_repeat_content_hash_counts_too(self): """★같은 사진이 다른 주소로 오는 일이 있다.""" got = ctp.dedupe_candidates([self._c("https://x", "h1")], seen_hashes=["h1"]) assert got["duplicate_count"] == 1 and got["judged_count"] == 0 def test_over_cap_is_recorded_not_silently_cut(self): n = ctp.PER_ROUND_CAP + 3 rows = [self._c(f"https://{i}") for i in range(n)] got = ctp.dedupe_candidates(rows) assert got["judged_count"] == ctp.PER_ROUND_CAP assert got["over_cap_count"] == 3 assert len(got["ledger"]) == n def test_fewer_than_asked_is_not_padded(self): """★provider 가 적게 줬다고 억지로 채우지 않는다.""" got = ctp.dedupe_candidates([self._c("https://a")]) assert got["judged_count"] == 1 def test_within_a_round_duplicates_are_caught_too(self): got = ctp.dedupe_candidates( [self._c("https://a"), self._c("https://a")]) assert got["judged_count"] == 1 and got["duplicate_count"] == 1 class TestTheNarrowHintKeepsTheCoordinates: """★좁힌다는 것은 **다른 것들**을 덜어내는 것이지, 대상을 못박는 시대·지역·모델 좌표를 버리는 것이 아니다 (Codex).""" @staticmethod def _text(): return ctp.load_narrow_hint() def test_it_says_one_subject_not_one_object(self): t = self._text() assert "one entity" in t assert "two intercity buses" in t, "★물리 개수로 줄이지 말라는 예" def test_it_keeps_era_region_and_model_in_the_query(self): t = self._text() assert "Keep the era, the region, the maker, the model" in t # ★고정 명사 예시(연도·도시·물건)는 금지 — 원리 문장만 (사용자 · 2026-09-03) assert "keeps all three" in t and not any(ch.isdigit() for ch in t) def test_it_says_what_to_drop(self): t = self._text() for phrase in ("neighbouring", "parts", "surrounding scene"): assert phrase in t, phrase def test_it_is_not_the_outdoor_wording(self): """★야외 전용 말투를 공용으로 올리면 안 된다.""" assert "structure" not in self._text().lower() class TestItWorksWithOneJudge: """★★★**새 다중 심판 비용을 넣지 않는다** (Codex 2026-08-30). 기존 `era_research` 는 **GPT LVM 단독 1심**이다. coarse 계약이 1심에서도 그대로 돌아야 한다 — 「살아남은 **모든** 심판이 yes+visible」이라는 규칙은 심판이 하나면 그 하나가 정하는 것이고, 그게 맞다. """ def test_one_judge_can_select(self): got = combine_coarse_verdicts( {"gpt": _v(("no", True), ("yes", True))}, 2) assert got["chosen_index"] == 2 assert got["single_judge"] is True, "★단독 판정을 **드러낸다**" def test_one_judge_can_also_refuse(self): got = combine_coarse_verdicts({"gpt": _v(("unsure", True))}, 1) assert got["chosen_index"] == 0 def test_one_judge_is_recorded_as_such(self): """★조용한 단독 판정 금지 — 감사에서 이중/단독을 구분해야 한다.""" one = combine_coarse_verdicts({"gpt": _v(("yes", True))}, 1) two = combine_coarse_verdicts( {"gpt": _v(("yes", True)), "gemini": _v(("yes", True))}, 1) assert one["single_judge"] is True and two["single_judge"] is False assert one["judges"] == ["gpt"] def test_the_contract_does_not_require_two(self): """★두 명을 요구하면 기존 1심 경로에 **새 비용**을 넣는 것이 된다.""" import inspect src = inspect.getsource(combine_coarse_verdicts) assert "len(valid) >= 2" not in src assert "len(valid) < 2" not in src