"""`location_part` 참조 **묶음** — 맥락 + 상세. ★inert. 유료 0. Codex 가 **얼어붙은 코드로 직접 재현**한 결함 셋을 닫는다 (2026-08-31) — ①하류 SOT 가 뒤집혔다 — raw `status` 를 봤다. 정본은 `outcome` 이고 없거나 모르는 값이면 **조용히 건너뛰면 안 된다**(의무가 사라진다) ②exact 열쇠에 **실제 획득 신원**이 빠져, 옛 자산이 신원 바뀐 새 의무를 **만족시켰다** ③「같은 bytes 면 합친다」가 **구현이 없었다** — 2장이 나오고 역할 차례도 뒤집혔다 그리고 검사 입력을 **붙이는 쪽 목록과 갈랐다** — 같은 목록을 넘기면 둘 다 빠뜨려도 통과한다. """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_reference_bundle as gb from app.modules.pipeline.grounding_host_context import (PURPOSE_CONTEXT, PURPOSE_DETAIL) _UNSET = object() #: 좌표 → bytes. ★시험이 **기존 loader 자리**를 대신 채운다. BLOBS = {"a-img": b"img", "a-a": b"a", "a-b": b"b", "a-same": b"same"} def _sha(b): import hashlib return hashlib.sha256(b).hexdigest() def _load(coord): """★좌표를 받아 `(bytes, 실제로 읽은 자리)` 를 낸다.""" aid = str((coord or {}).get("asset_id") or "") return BLOBS.get(aid, b""), {"source": "asset", "asset_id": aid} def _m(purpose, *, lp="LP01", outcome=gb.OUTCOME_SELECTED, ident=_UNSET, img=b"img", **kw): """★durable sidecar 모양 — **bytes 가 아니라 좌표**다. CP 는 JSON 이라 bytes 를 못 담는다(Codex 재현: `json.dumps` 가 터진다). """ # ★`ident=""` 를 **그대로** 넘긴다 — `or` 로 접으면 빈 값 시험이 죽는다 d = {"subject_final_id": lp, "purpose": purpose, "outcome": outcome, "member_identity": (f"id-{purpose}" if ident is _UNSET else ident), "label": f"{purpose} 사진"} if img is not None: key = next((k for k, v in BLOBS.items() if v == img), None) assert key, f"★시험 blob 에 {img!r} 가 없다" d["source"] = "asset" d["asset_id"] = key d["content_sha256"] = _sha(img) d.update(kw) return d def _base(): return ([("배경판", b"plate")], [("background", "L01B01")], ["background_chain_ref"], [{"pipeline_role": "background_render"}]) class TestTheOutcomeIsTheSourceOfTruth: """★①raw `status` 가 아니라 **`outcome`** 이 의무를 정한다.""" def test_selected_becomes_required(self): _ph, req = gb.plan_bundle([_m(PURPOSE_CONTEXT)]) assert len(req) == 1 def test_unavailable_requires_nothing(self): """★못 구한 것이 **하류를 막으면 안 된다** — HITL 없음.""" ph, req = gb.plan_bundle([ _m(PURPOSE_CONTEXT, outcome=gb.OUTCOME_UNAVAILABLE, img=None), _m(PURPOSE_DETAIL, outcome=gb.OUTCOME_UNAVAILABLE, img=None)]) assert (ph, req) == ([], []) @pytest.mark.parametrize("bad", [None, "", "selected_maybe", "ok"]) def test_a_missing_or_unknown_outcome_stops(self, bad): """★★조용히 건너뛰면 **selected 의무가 사라진다**.""" with pytest.raises(gb.BundleContractError) as e: gb.plan_bundle([_m(PURPOSE_DETAIL, outcome=bad)]) assert "처분" in str(e.value) def test_the_raw_status_does_not_decide(self): """★raw status 는 **감사용**이다 — 의무를 안 바꾼다.""" a = gb.plan_bundle([_m(PURPOSE_DETAIL, status="no_match_after_retry")]) b = gb.plan_bundle([_m(PURPOSE_DETAIL, status="selected")]) assert a[1] == b[1] def test_selected_without_an_image_stops(self): with pytest.raises(gb.BundleContractError): gb.plan_bundle([_m(PURPOSE_DETAIL, img=None)]) class TestTheAcquisitionIdentityIsInTheKey: """★②없으면 **옛 자산이 새 의무를 만족시킨다**(Codex 재현).""" def test_a_different_identity_moves_the_value(self): a = gb.meta_value(subject_final_id="LP01", purposes=[PURPOSE_DETAIL], member_identities=["A"]) b = gb.meta_value(subject_final_id="LP01", purposes=[PURPOSE_DETAIL], member_identities=["B"]) assert a != b def test_an_old_attachment_does_not_satisfy_a_new_obligation(self): """★★재현된 그 상황 — 획득 계약이 바뀌면 **다시 붙어야** 한다.""" _ph_a, req_a = gb.plan_bundle([_m(PURPOSE_DETAIL, ident="A")]) _ph_b, req_b = gb.plan_bundle([_m(PURPOSE_DETAIL, ident="B")]) assert req_a != req_b, "★신원이 바뀌었는데 요구가 같다" gb.assert_grounding_attached(req_a, list(req_a)) with pytest.raises(gb.BundleContractError): gb.assert_grounding_attached(req_b, list(req_a)) def test_no_identity_stops(self): with pytest.raises(gb.BundleContractError): gb.plan_bundle([_m(PURPOSE_DETAIL, ident="")]) def test_the_value_round_trips(self): v = gb.meta_value(subject_final_id="LP01", purposes=[PURPOSE_DETAIL, PURPOSE_CONTEXT], member_identities=["B", "A"]) got = gb.parse_meta_value(v) assert got["subject_final_ids"] == ["LP01"] assert got["purposes"] == [PURPOSE_CONTEXT, PURPOSE_DETAIL] def test_the_identity_order_does_not_matter(self): assert gb.meta_value(subject_final_id="LP01", purposes=[PURPOSE_DETAIL], member_identities=["A", "B"]) == \ gb.meta_value(subject_final_id="LP01", purposes=[PURPOSE_DETAIL], member_identities=["B", "A"]) @pytest.mark.parametrize("other", [ "outdoor_canon:P1", "space_set_bg:g1:plate1", "L01B01", "", "grounding:LP01:detail"]) def test_other_sources_are_not_read_as_grounding(self, other): assert gb.parse_meta_value(other) is None class TestTheSamePhotoBecomesOne: """★③같은 사진이면 **한 장**으로 합친다 (사용자 확정).""" def test_same_bytes_give_one_image(self): ph, req = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"same"), _m(PURPOSE_DETAIL, img=b"same")]) assert len(ph) == 1 and len(req) == 1, f"★{len(ph)}장이 나왔다" def test_the_merged_one_gets_the_combined_role(self): ph, _r = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"same"), _m(PURPOSE_DETAIL, img=b"same")]) assert ph[0]["role"] == "grounding_context_detail_ref" assert ph[0]["purposes"] == [PURPOSE_CONTEXT, PURPOSE_DETAIL] def test_both_logical_members_survive_the_merge(self): """★★합쳐도 **논리 멤버 둘의 신원·계보**는 따로 남는다.""" ph, _r = gb.plan_bundle([ _m(PURPOSE_CONTEXT, img=b"same", ident="i1"), _m(PURPOSE_DETAIL, img=b"same", ident="i2")]) assert ph[0]["member_identities"] == ["i1", "i2"] assert ph[0]["content_sha256"] == _sha(b"same") def test_different_bytes_stay_two(self): ph, req = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) assert len(ph) == 2 and len(req) == 2 assert {m["role"] for m in ph} == {"background_general", "grounding_part_detail_ref"} def test_the_input_order_does_not_change_the_plan(self): """★차례가 바뀌어도 **같은 계획**이어야 한다.""" a = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) b = gb.plan_bundle([_m(PURPOSE_DETAIL, img=b"b"), _m(PURPOSE_CONTEXT, img=b"a")]) assert a == b class TestItIsAdditiveAndKeepsTheOrder: def test_the_existing_plate_survives(self): lr, am, rr, rm = _base() ph, _r = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) n = gb.insert_bundle_refs(lr, am, rr, rm, ph, load_bytes=_load) assert n == 2 assert len(lr) == len(am) == len(rr) == len(rm) == 3 assert ("background", "L01B01") in am, "★기존 배경판이 사라졌다" def test_the_canonical_order_is_kept(self): """★★앞 판은 `insert(0)` 을 되풀이해 **역할 차례가 뒤집혔다**.""" lr, am, rr, rm = _base() ph, _r = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) gb.insert_bundle_refs(lr, am, rr, rm, ph, load_bytes=_load) assert rr[:2] == [ph[0]["role"], ph[1]["role"]], \ f"★차례가 뒤집혔다: {rr[:2]}" def test_they_share_the_existing_kind(self): lr, am, rr, rm = _base() ph, _r = gb.plan_bundle([_m(PURPOSE_DETAIL)]) gb.insert_bundle_refs(lr, am, rr, rm, ph, load_bytes=_load) assert {k for k, _v in am} == {"background"} class TestTheValidatorReadsADurableList: """★★붙이는 쪽 목록을 그대로 넘기면 **둘 다 빠뜨려도 통과**한다.""" def test_the_required_list_is_produced_with_the_photos(self): ph, req = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) assert [m["meta_value"] for m in ph] == [v for _k, v in req], \ "★붙일 것과 요구할 것이 갈린다" def test_an_attached_bundle_passes(self): lr, am, rr, rm = _base() ph, req = gb.plan_bundle([_m(PURPOSE_CONTEXT, img=b"a"), _m(PURPOSE_DETAIL, img=b"b")]) gb.insert_bundle_refs(lr, am, rr, rm, ph, load_bytes=_load) gb.assert_grounding_attached(req, am) def test_a_missing_one_stops(self): _ph, req = gb.plan_bundle([_m(PURPOSE_DETAIL)]) with pytest.raises(gb.BundleContractError) as e: gb.assert_grounding_attached(req, [("background", "L01B01")]) assert "안 붙었다" in str(e.value) @pytest.mark.parametrize("waiver", [ "space_set_bg:g1:plate1", "outdoor_canon:P1", "L01B01"]) def test_no_existing_waiver_can_satisfy_it(self, waiver): _ph, req = gb.plan_bundle([_m(PURPOSE_DETAIL)]) with pytest.raises(gb.BundleContractError): gb.assert_grounding_attached(req, [("background", waiver)]) def test_the_wrong_purpose_does_not_satisfy_it(self): _ph, det = gb.plan_bundle([_m(PURPOSE_DETAIL)]) _ph2, ctx = gb.plan_bundle([_m(PURPOSE_CONTEXT)]) with pytest.raises(gb.BundleContractError): gb.assert_grounding_attached(det, ctx) class TestItIsWiredButGated: """★★★**뒤집은 시험** — 이제 활성 조립이 부른다. 다만 **문 뒤**다. 앞 판은 「`app/` 어디서도 안 부른다」를 잠갔다. Codex 지적 (2026-08-31): 「helper-only paid runner 는 **우회 시험**이다 — 실제 길은 `generate_single_scene_image → _build_single_scene_prompt_and_refs → build_scene_attached_refs → validate_attached_refs → provider` 다」. 맞다. 그래서 **실제 조립에 붙였다.** ★비회귀는 **등가**로 증명한다 — sidecar 가 없으면 네 목록이 글자까지 같다 (`test_bundle_end_to_end_inert.TestTheSidecarGateIsByteIdenticalWhenAbsent`). """ def test_the_production_assembly_calls_it(self): """★**실제 조립**이 부르는지 — 우회가 아니라는 증거.""" import inspect from app.services.scene_generation_coordinator import ( build_scene_attached_refs) src = inspect.getsource(build_scene_attached_refs) assert "attach_from_rpc" in src, "★실제 조립이 안 부른다" def test_it_is_called_before_the_payload_is_built(self): """★**payload 를 짓기 전**에 붙어야 프롬프트·검사까지 간다.""" import inspect from app.services.scene_generation_coordinator import ( build_scene_attached_refs) src = inspect.getsource(build_scene_attached_refs) assert src.index("_grounding_attach(") < \ src.index("make_labeled_ref_payload("), \ "★payload 를 지은 뒤에 붙는다 — 그러면 프롬프트에 안 실린다" def test_the_consumer_never_writes_the_required_list(self): """★★★소비하는 쪽이 요구를 **쓰면 SOT 가 아니다** (Codex 재현). 앞 판은 `attach_from_rpc` 가 같은 dict 에 쓰고 `assert_from_rpc` 가 그것을 읽었다 — 조립이 attach 를 **빠뜨리면 쓰기도 같이 빠져** 거짓 통과했다. """ import inspect src = inspect.getsource(gb.attach_from_rpc) assert "write_required_to_rpc" not in src, "★소비하는 쪽이 쓴다" assert "required_from_rpc" in src, "★적힌 것을 안 읽는다" def test_every_role_is_declared_now(self): """★★**뒤집은 시험** — 앞 판은 「아직 선언 안 됐다」를 잠갔다. Codex 확인 (2026-08-31): 「새 역할을 **선언하는 것 자체**는 기존 producer 가 안 내므로 **비회귀**」. 그래서 선언은 먼저 했고, 이제 이 목록이 **비어 있어야** 한다 — 안 그러면 부르는 순간 `RefRoleError` 로 선다. """ from app.services.prompt_service import REF_ROLE_VALUES assert gb.undeclared_roles(REF_ROLE_VALUES) == [] def test_declaring_them_is_non_regressive(self): """★기존 producer 가 이 값을 **안 낸다** — 그래서 선언이 안전하다.""" import ast from pathlib import Path as _P new_roles = set(gb.ROLE_BY_PURPOSES.values()) - {"background_general"} root = _P(__file__).resolve().parents[2] / "app" emitters = [] for f in root.rglob("*.py"): if f.name in ("prompt_service.py", "grounding_reference_bundle.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.Constant) and n.value in new_roles: emitters.append(str(f)) assert not emitters, f"★벌써 내는 곳이 있다: {sorted(set(emitters))}" def test_the_context_role_is_already_declared(self): from app.services.prompt_service import REF_ROLE_VALUES assert gb.ROLE_BY_PURPOSES[(PURPOSE_CONTEXT,)] in REF_ROLE_VALUES