"""**진짜 production 조립**이 묶음을 싣나. ★유료 0. Codex 지적 (2026-08-31): > helper 만 직접 부르는 유료 시험은 **우회 시험**이다. 실제 길은 > `generate_single_scene_image → _build_single_scene_prompt_and_refs → > build_scene_attached_refs → validate_attached_refs → provider` 다. 맞다. 그래서 여기서는 **`build_scene_attached_refs` 를 진짜로 부른다**. fixture 꼴은 `tests/services/test_single_scene_uses_helper.py` 것을 따르되 **그 함수를 가짜로 바꾸지 않는다**. """ from __future__ import annotations import copy import hashlib from pathlib import Path from unittest.mock import MagicMock import pytest from app.modules.pipeline import grounding_reference_bundle as gb from app.services.prompt_service import make_labeled_ref_payload from app.services.scene_generation_coordinator import build_scene_attached_refs DETAIL = b"detail-photo" CONTEXT = b"context-photo" def _sha(b): return hashlib.sha256(b).hexdigest() def _member(purpose, img, *, ident=None, outcome=gb.OUTCOME_SELECTED, root=None, sha=None): """★durable sidecar 모양 — **좌표**다. bytes 는 CP 에 못 담는다. `source=file` 을 쓴다 — `source=asset` loader 는 D cutover 에서 잇는다. """ m = {"subject_final_id": "LP01", "purpose": purpose, "outcome": outcome, "member_identity": ident or f"id-{purpose}", "label": f"{purpose} 사진"} if img is not None: p = Path(root) / f"grounding-{purpose}.png" p.write_bytes(img) m["source"] = "file" m["path"] = str(p) m["content_sha256"] = sha or _sha(img) return m def _svc(): """참조 서비스 — 기존 시험과 같은 꼴. ★조립 함수는 **안 바꾼다**.""" s = MagicMock() s.build_scene_ref_image_map.return_value = {} s.detect_state_variant_sids.return_value = set() s.load_identity_family_by_sid.return_value = {} s.resolve_refs_for_prompt_set.return_value = make_labeled_ref_payload( labeled_refs=[], ref_roles=[], ref_role_metadata=[], attached_meta=[]) s.build_prev_shot_background_ref.return_value = None s.resolve_chain_bg_asset_id.return_value = None return s def _still(): s = MagicMock() s.camera_json = "{}" s.scene_index = 8 s.still_index = 0 s.shot_index = 4 return s def _rpc(members=None, *, write_required=True): rpc = {"asset_requirements": {"required_refs": []}} if members is None: return rpc if write_required: gb.write_sidecar(rpc, members) # ★원자적 producer 한 벌 else: rpc[gb.RPC_MEMBERS_KEY] = [dict(x) for x in members] return rpc def _run(rpc): """★**진짜** `build_scene_attached_refs` 를 부른다.""" return build_scene_attached_refs( still=_still(), episode_id="ep", still_data={"scene_index": 8, "shot_index": 4, "still_frame_prompt": "a shot", "beat_title": "", "render_prompt_card": rpc}, visible_entities=[], ref_image_map={}, cached_style_context="S", cached_entity_text_map={}, scene_paths_by_index_by_id={}, location_scene_history={}, background_chain_bg_map={}, dep_detail_map={}, staging={"camera_direction": "medium shot", "framing_scale": "medium"}, entity_lookup={}, project_id="p", project_config={}, reference_svc=_svc()) class TestTheBundleLandsThroughTheRealAssembly: def test_a_detail_member_reaches_attached_meta(self, tmp_path): rpc = _rpc([_member("detail", DETAIL, root=tmp_path)]) _prompt, refs, meta, _kinds = _run(rpc) assert len(refs) == 1, f"★{len(refs)}장" assert meta == [("background", gb.meta_value(subject_final_id="LP01", purposes=["detail"], member_identities=["id-detail"]))] def test_two_members_both_land(self, tmp_path): rpc = _rpc([_member("context", CONTEXT, root=tmp_path), _member("detail", DETAIL, root=tmp_path)]) _p, refs, meta, _k = _run(rpc) assert len(refs) == 2 and len(meta) == 2 assert all(k == "background" for k, _v in meta) def test_the_same_photo_becomes_one(self, tmp_path): rpc = _rpc([_member("context", DETAIL, root=tmp_path), _member("detail", DETAIL, root=tmp_path)]) _p, refs, meta, _k = _run(rpc) assert len(refs) == 1, f"★합쳐지지 않았다: {len(refs)}" got = gb.parse_meta_value(meta[0][1]) assert got["purposes"] == ["context", "detail"] def test_unavailable_lands_nothing(self, tmp_path): rpc = _rpc([_member("detail", None, outcome=gb.OUTCOME_UNAVAILABLE)]) _p, refs, meta, _k = _run(rpc) assert refs == [] and meta == [] class TestNoSidecarIsByteIdentical: """★★비회귀의 **끝점 증명** — 기존 CP 는 한 글자도 안 바뀐다.""" @pytest.mark.parametrize("rpc", [ {"asset_requirements": {"required_refs": []}}, {"asset_requirements": {"required_refs": []}, "other": 1}, ]) def test_the_whole_return_is_the_same(self, rpc): base = _run({"asset_requirements": {"required_refs": []}}) got = _run(copy.deepcopy(rpc)) assert got == base, "★sidecar 없는데 반환이 달라졌다" def test_nothing_is_attached(self): _p, refs, meta, _k = _run({"asset_requirements": {"required_refs": []}}) assert refs == [] and meta == [] class TestTheAssemblyStopsOnABrokenSidecar: """★조용히 지나가지 않는다.""" def test_a_missing_required_list_stops(self, tmp_path): """★★만드는 쪽이 요구를 안 적었으면 **선다** — 소비하는 쪽은 안 만든다.""" rpc = _rpc([_member("detail", DETAIL, root=tmp_path)], write_required=False) with pytest.raises(gb.BundleContractError) as e: _run(rpc) assert "grounding_required_attachments" in str(e.value) def test_a_wrong_required_list_stops(self, tmp_path): rpc = _rpc([_member("detail", DETAIL, root=tmp_path)]) rpc[gb.RPC_REQUIRED_KEY] = [{"kind": "background", "value": "엉뚱"}] with pytest.raises(gb.BundleContractError) as e: _run(rpc) assert "다르다" in str(e.value) def test_a_photo_that_does_not_match_the_hash_stops(self, tmp_path): """★★좌표가 가리키는 사진이 **적힌 것과 다르면** 선다. ★적힌 해시는 다른 사진의 것으로 두고 파일은 진짜를 둔다 — 실제로 「옛 자산이 붙을 뻔한」 상황이다. """ want = _sha(b"a-different-photo") rpc = _rpc([_member("detail", DETAIL, root=tmp_path, sha=want)]) with pytest.raises(gb.BundleContractError) as e: _run(rpc) # ★문구 낱말이 아니라 **무엇을 말하는지**로 잡는다 — 적힌 해시와 # 실제로 읽은 해시가 **둘 다** 나와야 사람이 원인을 안다 # ([[feedback-no-literal-substring-meaning]]). msg = str(e.value) assert want in msg, "★적힌 해시가 없다" assert _sha(DETAIL) in msg, "★실제로 읽은 해시가 없다" def test_a_missing_photo_stops(self, tmp_path): """★좌표가 가리키는 파일이 **없으면** 선다.""" rpc = _rpc([_member("detail", DETAIL, root=tmp_path)]) Path(rpc[gb.RPC_MEMBERS_KEY][0]["path"]).unlink() with pytest.raises(gb.BundleContractError) as e: _run(rpc) assert "못 읽었다" in str(e.value) class TestTheSeamOrder: """★붙는 자리가 **프롬프트를 짓기 전**이어야 프롬프트·검사까지 간다.""" def test_attach_comes_before_the_payload(self): import inspect src = inspect.getsource(build_scene_attached_refs) assert src.index("_grounding_attach(") < \ src.index("make_labeled_ref_payload(") def test_the_consumer_does_not_write_the_required_list(self): import inspect assert "write_required_to_rpc" not in inspect.getsource( gb.attach_from_rpc) class TestTheSecondGateActuallyFires: """★★★`assert_from_rpc` 는 **둘째 문**이다 — 붙이는 쪽이 빠져도 잡는다. ★codegraph 는 `assert_from_rpc as _grounding_assert` 라는 **별칭 import** 때문에 caller 를 **0 으로 봤다**(measuring tool 의 맹점 — 내 memory 에 적힌 부류다). 그래서 「배선됐다」를 **AST 로** 보고, **작동하는지**는 여기서 붙이는 쪽을 무력화해 확인한다. """ def test_it_is_wired_with_an_alias(self): """★별칭 import 라 글자 검색으로는 안 보인다 — AST 로 본다.""" import ast import inspect from app.services import scene_generation_coordinator as sgc tree = ast.parse(inspect.getsource(sgc)) aliases = {a.asname or a.name for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and "grounding_reference_bundle" in str(n.module or "") for a in n.names} assert "_grounding_assert" in aliases, f"★안 이어졌다: {aliases}" def test_it_stops_when_the_attach_is_bypassed(self, tmp_path, monkeypatch): """★★붙이는 쪽이 **빠져도** 둘째 문이 잡는다. 이것이 두 문을 나눈 이유다 — 한 자리가 빠지면 다른 자리가 잡는다. """ from app.services import scene_generation_coordinator as sgc rpc = _rpc([_member("detail", DETAIL, root=tmp_path)]) # ★붙이는 쪽을 **무력화**한다 — 조립이 그 호출을 빠뜨린 상황 monkeypatch.setattr(sgc, "_GROUNDING_ATTACH_DISABLED", True, raising=False) import app.modules.pipeline.grounding_reference_bundle as _gb monkeypatch.setattr(_gb, "attach_from_rpc", lambda *a, **k: 0, raising=True) with pytest.raises(gb.BundleContractError) as e: _run(rpc) assert "안 붙었다" in str(e.value), str(e.value) def test_it_passes_when_nothing_is_required(self): """★요구가 없으면 둘째 문도 없다 — 옛 CP 가 안 막힌다.""" _p, refs, meta, _k = _run({"asset_requirements": {"required_refs": []}}) assert refs == [] and meta == []