"""★★★네 소비 끝점 — **wrapper 만 부르지 않고 실제 공개 소비 경계**를 태운다. Codex: 「「어느 칸을 읽는지 소스 검사」보다 **실제 consumer endpoint** 가 낫습니다.」 그리고 「**B1 은 direct `assess_subjects`/`research_reference` 라 wrapper 시험만으로 덮이지 않습니다.**」 여기서 보는 것은 **기계적 계보**까지다 — 고증이 맞는지·나아졌는지는 §2-5·§2-6 에서 **사람**이 본다. """ import json from pathlib import Path from unittest.mock import patch import pytest from app.modules.pipeline import era_research as er _SUBJ = [{"subject_native": "1983년 서울역", "search_terms_native": ["가", "나"], "language_lock_native": "한국어로만 검색하세요", "reason_ko": "옛 역사"}] class TestB1EntityReference: """★B1 은 `assess_subjects` + `research_reference` **직호출**이다 — 공용 캐시 wrapper 를 안 탄다. 그래서 따로 태운다. ★임시 cache owner 를 **만들지 않는다**(Codex) — 중앙 SOT 전에 owner 가 또 생기면 안 된다. 지금은 direct public API 가 유지되는지만 본다. """ def _gen(self, tmp_path, *, subjects, meta, ref_bytes=b"\x89PNG"): from app.modules import reference_image_generator as rig seen = {} class _Gemini: _model = "gemini-3.1-flash-image" def generate_image(self, *, prompt, reference_images, aspect_ratio=None, **kw): seen["reference_images"] = reference_images return b"\x89PNG out", 10 def _assess(**kw): if subjects is None: raise RuntimeError("판별이 터졌다") return list(subjects) def _research(*, subject, out_path, **kw): if meta is None: return None Path(out_path).write_bytes(ref_bytes) return dict(meta) gen = rig.ReferenceImageGenerator.__new__(rig.ReferenceImageGenerator) gen._gemini = _Gemini() gen._language = "ko" gen._build_prompt = lambda *a, **k: "프롬프트" class _S: era_research_enabled = True with patch.object(rig, "settings", _S, create=True), \ patch("app.core.config.settings", _S), \ patch("app.modules.pipeline.era_research.assess_subjects", _assess), \ patch("app.modules.pipeline.era_research.research_reference", _research): try: out = gen.generate_for_entity( entity={"id": "e1", "name": "서울역", "entity_type": "location", "description": "옛 역사"}, world_guide={"era": "1983", "region": "KR"}, output_dir=tmp_path) except Exception as exc: # noqa: BLE001 out = {"_error": repr(exc)} return out, seen def test_selected_bytes_reach_reference_images(self, tmp_path): """★★고른 사진이 **실제로 생성 입력에 실리는가.** 소스 검사 아님.""" out, seen = self._gen( tmp_path, subjects=_SUBJ, meta={"subject": "1983년 서울역", "file": "x.png", "sha256": "s"}, ref_bytes=b"\x89PNG chosen") assert "_error" not in out, out.get("_error") assert seen["reference_images"] == [b"\x89PNG chosen"], \ "★고른 bytes 가 생성 입력에 안 실렸다" notes = json.loads(out["review_notes"])["era_research"] assert notes["researched"] is True assert notes["subject"] == "1983년 서울역" def test_no_subject_keeps_the_old_non_blocking_input(self, tmp_path): """★비대상이면 **기존과 같이** 참조 없이 그린다(late 계약).""" out, seen = self._gen(tmp_path, subjects=[], meta=None) assert "_error" not in out, out.get("_error") assert seen["reference_images"] is None notes = json.loads(out["review_notes"])["era_research"] assert notes["researched"] is False assert notes.get("reason") == "no_subject" def test_a_research_failure_is_also_non_blocking(self, tmp_path): out, seen = self._gen(tmp_path, subjects=_SUBJ, meta=None) assert "_error" not in out, out.get("_error") assert seen["reference_images"] is None assert json.loads(out["review_notes"])["era_research"][ "researched"] is False def test_an_assess_failure_is_recorded_not_swallowed(self, tmp_path): out, _seen = self._gen(tmp_path, subjects=None, meta=None) assert "_error" not in out, out.get("_error") notes = json.loads(out["review_notes"])["era_research"] assert "error" in notes, "★판별이 터진 것이 기록에 안 남았다" def test_the_direct_path_still_uses_the_public_api(self): """★step 6 전까지 두 공개 함수의 signature 를 유지한다.""" import inspect from app.modules import reference_image_generator as rig src = Path(rig.__file__).read_text(encoding="utf-8") assert "assess_subjects" in src and "research_reference" in src # ★임시 캐시 owner 를 만들지 않았다 assert "assess_and_research_cached" not in src assert "era_ref::" not in src and "era_assess::" not in src def test_the_public_functions_keep_their_shape(self): """★B1 이 부르는 두 함수의 인자·반환이 안 바뀌었다.""" import inspect a = inspect.signature(er.assess_subjects).parameters for need in ("step_tag", "subject_text", "world_facts_block"): assert need in a, need r = inspect.signature(er.research_reference).parameters for need in ("subject", "world_facts_block", "out_path", "step_tag"): assert need in r, need class TestTheCoarsePickerReachesTheDirectPath: """★★B1 은 wrapper 를 안 타지만 **같은 `research_reference`** 를 부른다 — 그래서 새 picker 가 그 direct path 에도 **닿는다**. 그것을 잰다.""" def test_the_picker_prompt_is_the_coarse_one(self): """★`research_reference` 안에서 어느 지문을 쓰는지 — 한 자리뿐이다.""" import inspect src = inspect.getsource(er.research_reference) assert "coarse_type_pick" in src assert "combine_coarse_verdicts" in src assert "build_pick_user_head(" not in src def test_the_judge_does_not_get_world_facts(self): """★심판에게 시대·지역을 안 준다 — 사용자 확정.""" import inspect src = inspect.getsource(er.research_reference) head = src[src.index("THE KIND OF THING"):src.index("PHOTOGRAPH")] assert "world_facts_block" not in head def test_the_search_query_still_carries_them(self): """★질의에는 그대로 남는다 — 검색은 좌표가 있어야 한다.""" import inspect src = inspect.getsource(er.research_reference) i = src.index("search_reference_images") assert "world_facts_block" in src[:i] class TestTheLateWrapperKeepsItsContract: """★groupbg·confined·plate 는 **wrapper** 를 탄다. 그 계약을 잰다.""" def test_the_three_call_sites_still_call_the_wrapper(self): """★소스 count 는 **보조**다 — 통과 근거가 아니다(Codex). 아래 `TestTheSharedConsumptionHelper` 가 진짜 끝점이다.""" src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") assert src.count("assess_and_research_cached") == 3 def test_the_wrapper_returns_what_they_read(self, tmp_path): """★셋 다 `path` 와 meta 칸을 읽는다.""" cache = {} calls = {"n": 0} def _research(*, subject, out_path, **kw): calls["n"] += 1 Path(out_path).write_bytes(b"\x89PNG") import hashlib return {"subject": "서울역", "file": "x.png", "sha256": hashlib.sha256(b"\x89PNG").hexdigest()} with patch.object(er, "assess_subjects", lambda **k: list(_SUBJ)), \ patch.object(er, "research_reference", _research): got = er.assess_and_research_cached( step_tag="era_research_plate", subject_text="옛 서울역", world_facts_block="KR / 1983", out_dir=tmp_path, cache_get=cache.get, cache_put=cache.__setitem__) assert got is not None for need in ("path", "subject", "sha256", "file"): assert need in got, need assert Path(got["path"]).is_file() def test_no_subject_is_still_non_blocking_none(self, tmp_path): """★late 계약은 **비차단**이다. fail-closed 는 중앙(step 5) 전환 때.""" cache = {} with patch.object(er, "assess_subjects", lambda **k: []): got = er.assess_and_research_cached( step_tag="t", subject_text="x", world_facts_block="KR", out_dir=tmp_path, cache_get=cache.get, cache_put=cache.__setitem__) assert got is None def test_a_failure_is_still_non_blocking_none(self, tmp_path): cache = {} with patch.object(er, "assess_subjects", lambda **k: list(_SUBJ)), \ patch.object(er, "research_reference", lambda **k: None): got = er.assess_and_research_cached( step_tag="t", subject_text="x", world_facts_block="KR", out_dir=tmp_path, cache_get=cache.get, cache_put=cache.__setitem__) assert got is None def test_the_late_path_folds_one_round(self, tmp_path): """★★late 는 1라운드다 — 키가 그것을 접는다.""" cache = {} with patch.object(er, "assess_subjects", lambda **k: list(_SUBJ)), \ patch.object(er, "research_reference", lambda **k: None): er.assess_and_research_cached( step_tag="t", subject_text="x", world_facts_block="KR", out_dir=tmp_path, cache_get=cache.get, cache_put=cache.__setitem__) # 실패 기록의 sha 가 1라운드 계약으로 난 키여야 한다 fails = [k for k in cache if k.startswith("era_fail::")] assert fails, "실패 기록이 없다" class TestNormalizeIsSharedButProjectionsAreNot: """★★★**세 소비자를 한 mutation helper 로 뭉치지 않는다** (Codex). 셋의 **효과가 다르다** — groupbg 는 프롬프트+참조+지문, confined 는 refs 자리+계보, plate 는 프롬프트+경로+meta. 정규화만 공통이다. """ _META = {"subject": "1983년 서울역", "path": "/tmp/eraref_ab.png", "file": "eraref_ab.png", "sha256": "deadbeef"} def test_normalize_reads_the_fields_once(self): env = er.normalize_era_reference(self._META) assert env["subject"] == "1983년 서울역" assert env["path"] == Path("/tmp/eraref_ab.png") assert env["file"] == "eraref_ab.png" and env["sha256"] == "deadbeef" assert env["role_text"] == er.build_ref_role("1983년 서울역") def test_normalize_has_no_side_effects(self): """★네트워크·캐시·생성 0 — 순수 변환이다.""" before = dict(self._META) er.normalize_era_reference(self._META) assert self._META == before def test_none_and_half_results_stay_none(self): assert er.normalize_era_reference(None) is None assert er.normalize_era_reference({}) is None assert er.normalize_era_reference({"subject": "x"}) is None assert er.normalize_era_reference({"path": "/tmp/a.png"}) is None class TestGroupbgProjection: """★프롬프트 role + path refs + **sha 지문**을 함께 낸다.""" def _env(self): return er.normalize_era_reference( TestNormalizeIsSharedButProjectionsAreNot._META) def test_selected_gives_all_three(self): prompt, refs, fp = er.project_groupbg_era("바탕", self._env()) assert er.build_ref_role("1983년 서울역") in prompt assert refs == [Path("/tmp/eraref_ab.png")] assert fp["era_research_sha"] == "deadbeef" def test_the_fingerprint_comes_with_the_reference(self): """★★따로 두면 참조만 갈리고 지문은 그대로여서 **옛 배경이 봉인**된다.""" _p, refs, fp = er.project_groupbg_era("바탕", self._env()) assert bool(refs) == bool(fp), "★참조와 지문이 따로 논다" def test_none_changes_nothing(self): prompt, refs, fp = er.project_groupbg_era("바탕", None) assert prompt == "바탕" and refs is None and fp is None def test_production_calls_it(self): src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") assert "project_groupbg_era" in src assert 'bg_prompt + "\\n\\n" + _era_role(' not in src, "★옛 조립이 남았다" def test_the_meta_consumes_the_fragment(self): """★★★**projection 이 낸 조각을 실제 meta 가 소비하는가.** 조각을 만들어 놓고 raw meta 로 **다시 조립**하면 두 벌이 되고, 「참조와 지문을 한 벌로 낸다」는 보호가 **실제 경로에는 없다**. 지금은 결과가 우연히 같을 뿐이었다(Codex). """ src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") assert "**(_gb_fp or {})" in src, "★조각을 안 쓴다" assert '{"era_research_sha": era_bg_meta["sha256"]}' not in src, \ "★raw meta 로 다시 조립한다 — 두 벌이다" def test_the_fragment_exists_when_era_is_off(self): """★꺼진 판에서도 meta 조립이 이것을 읽는다 — plate 에서 데였던 부류.""" src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") i = src.index("**(_gb_fp or {})") assert "_gb_fp: Optional[Dict[str, Any]] = None" in src[:i] def test_the_role_aliases_are_gone(self): """★projection 이 역할문을 만든다 — alias 가 남으면 두 벌의 씨앗이다.""" src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") assert "build_ref_role" not in src class TestConfinedProjection: """★refs **1번 자리** 삽입 + 계보 한 줄. `_attach` 는 그 줄만 소비한다.""" def _env(self): return er.normalize_era_reference( TestNormalizeIsSharedButProjectionsAreNot._META) def test_it_inserts_at_index_one(self): """★도면이 배치 권위이고 era 참조는 실물 look 앵커라 그 **뒤**다.""" refs, _row = er.project_confined_era( [("fp", "도면"), ("char", "인물")], self._env()) assert refs[0] == ("fp", "도면") assert refs[1][1] == Path("/tmp/eraref_ab.png") assert refs[2] == ("char", "인물") def test_the_lineage_row_carries_file_and_sha(self): _refs, row = er.project_confined_era([], self._env()) assert row["kind"] == "era_ref" assert row["file"] == "eraref_ab.png" and row["sha256"] == "deadbeef" def test_it_does_not_mutate_the_input(self): given = [("fp", "도면")] refs, _row = er.project_confined_era(given, self._env()) assert given == [("fp", "도면")], "★입력을 건드렸다" assert refs is not given def test_none_leaves_refs_alone(self): refs, row = er.project_confined_era([("fp", "도면")], None) assert refs == [("fp", "도면")] and row is None def test_production_calls_it_and_attach_uses_the_row(self): src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") assert "project_confined_era" in src assert '_attach(_cf_lineage["kind"]' in src, "★_attach 가 다시 조립한다" assert '_era_role_fn(_era_cf[' not in src, "★옛 조립이 남았다" class TestPlateProjection: """★프롬프트 role + `era_ref_path` + `era_meta` — `_run_bgfirst_bg` 가 **둘 다** 받아야 한다. 하나만 넘기면 참조는 실렸는데 계보가 없다.""" def _env(self): return er.normalize_era_reference( TestNormalizeIsSharedButProjectionsAreNot._META) def test_selected_gives_prompt_path_and_meta(self): raw = dict(TestNormalizeIsSharedButProjectionsAreNot._META) prompt, path, meta = er.project_plate_era("바탕", self._env(), raw) assert er.build_ref_role("1983년 서울역") in prompt assert path == Path("/tmp/eraref_ab.png") assert meta is raw, "★원본 meta 를 그대로 넘겨야 계보가 안 끊긴다" def test_path_and_meta_come_together(self): _p, path, meta = er.project_plate_era("바탕", self._env()) assert bool(path) == bool(meta) def test_none_changes_nothing(self): prompt, path, meta = er.project_plate_era("바탕", None) assert prompt == "바탕" and path is None and meta is None def test_production_calls_it_and_passes_both(self): src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") assert "project_plate_era" in src assert "era_ref_path=_pl_path" in src and "era_meta=_pl_meta" in src assert 'Path(_era_pl["path"])' not in src, "★옛 조립이 남았다" def test_the_variables_exist_when_era_is_off(self): """★`era_research_on` 이 꺼진 판에서도 아래 호출이 이 둘을 읽는다.""" src = Path("app/services/still_recipe_service.py").read_text( encoding="utf-8") i = src.index("era_ref_path=_pl_path") assert "_pl_path = None" in src[:i], "★꺼진 판에서 NameError 다"