"""A5 (2026-07-02) — immobilized 그룹 prev 완성프레임 chaining. star-to-environment 계획(pure) + attach(교체/append/충돌/미해결 진단) + prompt render 분기(가이드 유/무 SOT 정렬) + lineage 구조필드 보존 + persistence post-hoc resolve. 전부 결정론 — 이미지 품질은 canary+육안. """ from __future__ import annotations from unittest.mock import MagicMock from app.core.config import settings from app.core.steps.visual_continuity_anchor_step import ( IMMOBILIZED_PREV_FRAME_LABEL, attach_immobilized_prev_frame_ref, ) from app.modules.pipeline.visual_continuity_anchor_plan import ( build_immobilized_prev_frame_plan, ) from app.services.prompt_service import make_labeled_ref_payload, resolve_ref_roles from app.services.scene_generation_coordinator import collect_actual_attached_refs from app.services.scene_persistence_service import ScenePersistenceService # ─────────────────────────── plan (pure) ─────────────────────────── def _members(*entries): return [ {"scene_index": si, "shot_index": shi, "role": role} for si, shi, role in entries ] def test_plan_star_to_environment_basic(): """env(sh10) 뒤 insert(sh16) → env 앵커. env 자신/env 이전 샷 엔트리 없음.""" plan = build_immobilized_prev_frame_plan({ "g1": _members((12, 3, "close_insert"), (12, 10, "environment"), (12, 16, "close_insert")), }) assert plan == { (12, 16): {"anchor_source": (12, 10), "group_id": "g1", "anchor_role": "environment"}, } def test_plan_nearest_preceding_env_and_env_chain(): """env 2개면 후속 멤버는 가장 가까운 선행 env, 뒤 env 는 앞 env 에 체이닝.""" plan = build_immobilized_prev_frame_plan({ "g1": _members((5, 1, "environment"), (5, 4, "close_insert"), (5, 7, "environment"), (5, 9, "close_insert")), }) assert plan[(5, 4)]["anchor_source"] == (5, 1) assert plan[(5, 7)]["anchor_source"] == (5, 1) assert plan[(5, 9)]["anchor_source"] == (5, 7) def test_plan_no_env_member_empty(): """environment 멤버 없는 그룹(전부 insert) → 계획 없음 (no-op).""" assert build_immobilized_prev_frame_plan({ "g1": _members((3, 2, "close_insert"), (3, 5, "close_insert")), }) == {} def test_plan_multi_group_first_gid_wins_and_unsorted_input(): """같은 샷이 두 그룹 멤버면 gid 정렬 순 첫 매핑 유지 + 입력 비정렬 허용.""" plan = build_immobilized_prev_frame_plan({ "g2": _members((7, 8, "close_insert"), (7, 2, "environment")), "g1": _members((7, 8, "close_insert"), (7, 4, "environment")), }) # g1 이 먼저 순회 → (7,8) 앵커는 g1 의 env (7,4) assert plan[(7, 8)]["group_id"] == "g1" assert plan[(7, 8)]["anchor_source"] == (7, 4) # ─────────────────────────── attach ─────────────────────────── def _lists(*, with_prev_role=None, with_zoom=False): labeled_refs = [("CHARACTER ref", b"c")] ref_roles = ["reference_face"] ref_role_metadata = [{"sid": "C05"}] attached_meta = [("character", "C05")] if with_prev_role: labeled_refs.append(("BG prev", b"old-bytes")) ref_roles.append(with_prev_role) ref_role_metadata.append({"loc": "L04"}) attached_meta.append(("background_prev_shot", "L04")) if with_zoom: labeled_refs.append(("zoomed", b"z")) ref_roles.append("previous_shot_same_frame_zoomed") ref_role_metadata.append({}) attached_meta.append(("background_prev_shot", "L04")) return labeled_refs, ref_roles, ref_role_metadata, attached_meta def _ctx(): return { "anchors_by_group": {"g1": {"character_short_id": "C05"}}, "members_by_shot": {(12, 10): ["g1"], (12, 16): ["g1"]}, "members_by_group": { "g1": _members((12, 10, "environment"), (12, 16, "close_insert")), }, } def _attach(monkeypatch, *, enabled=True, key=(12, 16), resolver=lambda k: b"env-frame", lists=None, sid_map=None): monkeypatch.setattr(settings, "immobilized_prev_frame_chain_enabled", enabled, raising=False) lr, rr, rm, am = lists or _lists() diag: dict = {} ok = attach_immobilized_prev_frame_ref( lr, rr, rm, am, scene_index=key[0], shot_index=key[1], anchor_ctx=_ctx(), source_bytes_resolver=resolver, source_still_id_by_key=sid_map if sid_map is not None else {(12, 10): "still-env"}, diag_out=diag, ) return ok, lr, rr, rm, am, diag def test_attach_flag_off_noop(monkeypatch): ok, lr, rr, rm, am, diag = _attach(monkeypatch, enabled=False) assert ok is False assert len(lr) == 1 and len(rr) == 1 and diag == {} def test_attach_non_member_noop(monkeypatch): ok, lr, rr, rm, am, diag = _attach(monkeypatch, key=(12, 10)) # env 자신 assert ok is False assert len(lr) == 1 and diag == {} def test_attach_appends_with_structural_metadata(monkeypatch): ok, lr, rr, rm, am, diag = _attach(monkeypatch) assert ok is True assert len(lr) == len(rr) == len(rm) == len(am) == 2 # parity assert lr[-1] == (IMMOBILIZED_PREV_FRAME_LABEL, b"env-frame") assert rr[-1] == "previous_shot_same_room" meta = rm[-1] assert meta["immobilized_prev_frame_anchor"] is True assert meta["pipeline_role"] == "immobilized_prev_frame" assert meta["source_still_id"] == "still-env" assert meta["anchor_source"] == [12, 10] assert meta["group_id"] == "g1" assert am[-1] == ("background_prev_shot", "") assert diag["resolved"] is True and diag["previous_frame_required"] is True def test_attach_replaces_existing_prev_shot_ref(monkeypatch): lists = _lists(with_prev_role="previous_shot_continuity") ok, lr, rr, rm, am, diag = _attach(monkeypatch, lists=lists) assert ok is True assert len(lr) == 2 # 교체 — append 아님 assert lr[1] == (IMMOBILIZED_PREV_FRAME_LABEL, b"env-frame") assert rr[1] == "previous_shot_continuity" # role 유지 (bytes/라벨/meta 만) assert rm[1]["immobilized_prev_frame_anchor"] is True assert rm[1]["loc"] == "L04" # 기존 metadata 병합 보존 assert diag["replaced_role"] == "previous_shot_continuity" def test_attach_zoom_conflict_noop(monkeypatch): lists = _lists(with_zoom=True) ok, lr, rr, rm, am, diag = _attach(monkeypatch, lists=lists) assert ok is False assert len(lr) == 2 # 불변 assert diag["resolved"] is False assert diag["reason_if_missing"] == "zoom_conflict" def test_attach_unresolved_anchor_diag(monkeypatch): ok, lr, rr, rm, am, diag = _attach(monkeypatch, resolver=lambda k: None) assert ok is False assert len(lr) == 1 # 불변 — stale fallback 없음 assert diag["resolved"] is False assert diag["reason_if_missing"] == "previous_frame_required_missing" assert diag["source_still_id"] == "still-env" # ─────────────────────────── prompt render ─────────────────────────── def _render(meta, roles): return resolve_ref_roles(make_labeled_ref_payload( labeled_refs=[(IMMOBILIZED_PREV_FRAME_LABEL, b"png")] if len(roles) == 1 else [(IMMOBILIZED_PREV_FRAME_LABEL, b"png"), ("POSE GUIDE", b"g")], ref_roles=roles, ref_role_metadata=[meta] + [{}] * (len(roles) - 1), attached_meta=[("x", "y")] * len(roles), )) def test_render_immobilized_prev_frame_no_guide(): """가이드 부재 → 프레임이 pose SOT ('image 1 wins').""" res = _render({"immobilized_prev_frame_anchor": True}, ["previous_shot_same_room"]) blob = "\n".join(res.ref_instructions) assert "has NOT moved" in blob assert "image 1 wins" in blob assert "do NOT copy image 1's composition" in blob def test_render_immobilized_prev_frame_with_guide_alignment(): """등록 pose 가이드 동반 → 가이드=자세 1차 SOT, 프레임=환경/연속성 (역할 분리).""" res = _render({"immobilized_prev_frame_anchor": True}, ["previous_shot_same_room", "immobilized_pose_guide"]) blob = "\n".join(res.ref_instructions) assert "attached pose guide" in blob assert "same configuration" in blob def test_render_continuity_role_variant(): """previous_shot_continuity role 에서도 동일 분기.""" res = _render({"immobilized_prev_frame_anchor": True}, ["previous_shot_continuity"]) assert "has NOT moved" in "\n".join(res.ref_instructions) # ─────────────────────────── lineage collect ─────────────────────────── def test_collect_preserves_prev_frame_structural_keys(): """asset_id 없는 prev-frame ref → unresolved 에 source_still_id/anchor_source 보존.""" payload = make_labeled_ref_payload( labeled_refs=[(IMMOBILIZED_PREV_FRAME_LABEL, b"png")], ref_roles=["previous_shot_same_room"], ref_role_metadata=[{ "immobilized_prev_frame_anchor": True, "pipeline_role": "immobilized_prev_frame", "source_still_id": "still-env", "anchor_source": [12, 10], "group_id": "g1", }], attached_meta=[("background_prev_shot", "")], ) out = collect_actual_attached_refs(payload) assert out["image_ids"] == [] u = out["unresolved"][0] assert u["pipeline_role"] == "immobilized_prev_frame" assert u["source_still_id"] == "still-env" assert u["anchor_source"] == [12, 10] # ─────────────────────────── persistence resolve ─────────────────────────── def _svc_with_first_returns(first_returns): instance = ScenePersistenceService.__new__(ScenePersistenceService) db = MagicMock() q = MagicMock() db.query.return_value = q q.filter.return_value = q q.order_by.return_value = q q.first.side_effect = first_returns instance._db = db instance._project_id = "p1" return instance def _prev_frame_unresolved(**over): base = { "role": "previous_shot_same_room", "label": IMMOBILIZED_PREV_FRAME_LABEL[:30], "pipeline_role": "immobilized_prev_frame", "source_still_id": "still-env", "anchor_source": [12, 10], "group_id": "g1", } base.update(over) return base def test_resolve_prev_frame_primary_found(): svc = _svc_with_first_returns([("scene-asset-uuid",)]) ids, refs, remaining = svc._resolve_prev_frame_asset_ids( [_prev_frame_unresolved()], "e1") assert ids == ["scene-asset-uuid"] assert refs[0]["asset_id"] == "scene-asset-uuid" assert refs[0]["resolved_from_unresolved"] is True assert refs[0]["pipeline_role"] == "immobilized_prev_frame" assert remaining == [] def test_resolve_prev_frame_primary_missing_kept_unresolved(): svc = _svc_with_first_returns([None]) ids, refs, remaining = svc._resolve_prev_frame_asset_ids( [_prev_frame_unresolved()], "e1") assert ids == [] and refs == [] assert remaining[0]["reason"] == "prev_frame_primary_not_found" def test_resolve_prev_frame_no_still_id_reason(): svc = _svc_with_first_returns([]) ids, refs, remaining = svc._resolve_prev_frame_asset_ids( [_prev_frame_unresolved(source_still_id=None)], "e1") assert remaining[0]["reason"] == "prev_frame_source_still_id_missing" def test_resolve_prev_frame_non_target_passthrough(): svc = _svc_with_first_returns([]) other = {"role": "previous_shot_continuity", "label": "prev"} ids, refs, remaining = svc._resolve_prev_frame_asset_ids([other], "e1") assert remaining == [other]