"""goal#3 (2026-07-01) — outdoor 항공뷰 체인 lineage post-hoc resolve 결정론 테스트. ``_resolve_outdoor_chain_lineage`` 는 aerial_base→shot_blocking→camera_sketch 의 ``input_image_ids`` 를 구조키(group_id + producer_stage + shot_index)로 채우고 (annotate) camera_sketch ImageAsset UUID 를 composition_guides 엔트리에 회수한다. DB·구조키 기반 결정론 로직 → 단위 테스트 대상(캔버스 시각 품질은 별도 육안). """ import json import uuid from datetime import datetime, timezone from app.core.database import SessionLocal from app.core.steps.outdoor_site_layout_step import _resolve_outdoor_chain_lineage from app.models.catalog import ProjectRegistry, UserAccount from app.models.project import Episode, ImageAsset def _now() -> str: return datetime.now(timezone.utc).isoformat() def _seed_base(db) -> tuple[str, str]: uid = str(uuid.uuid4()) db.add(UserAccount( id=uid, username=f"u_{uid[:8]}", display_name="t", password_hash="x", role="admin", created_at=_now(), updated_at=_now())) db.commit() pid = str(uuid.uuid4()) db.add(ProjectRegistry( id=pid, name="OSL Chain Test", created_by=uid, created_at=_now(), updated_at=_now())) eid = str(uuid.uuid4()) db.add(Episode( id=eid, project_id=pid, episode_number=1, title="Pilot", source_filename="p.pdf", source_path="p.pdf", created_at=_now(), updated_at=_now())) db.commit() return pid, eid def _add_chain_asset(db, pid, eid, *, role, created_at, meta) -> str: aid = str(uuid.uuid4()) # asset_type='generated' + is_intermediate=True — 실데이터(capture sink) 와 동일, # resolver MINOR 필터(Codex 정렬) 통과 조건. db.add(ImageAsset( id=aid, project_id=pid, episode_id=eid, asset_type="generated", file_path=f"{aid}.png", pipeline_role=role, stage="outdoor_site_layout", is_intermediate=True, pipeline_metadata_json=json.dumps(meta), created_at=created_at)) db.commit() return aid def _inputs(db, asset_id): row = db.get(ImageAsset, asset_id) return json.loads(row.input_image_ids) if row.input_image_ids else None def test_chain_lineage_blocking_and_sketch_resolve(): """aerial→blocking→sketch 완전 체인: blocking.input=[aerial], sketch.input=[blocking](parent_stage=shot_blocking) + guide_asset_id 회수.""" db = SessionLocal() try: pid, eid = _seed_base(db) gid = "osl-shared-l03-s10" aerial = _add_chain_asset( db, pid, eid, role="outdoor_aerial_base", created_at="2026-06-30T19:11:18.000000+00:00", meta={"group_id": gid, "producer_stage": "aerial_base", "parent_stage": None}) blocking = _add_chain_asset( db, pid, eid, role="outdoor_shot_blocking", created_at="2026-06-30T19:16:07.917613+00:00", meta={"group_id": gid, "producer_stage": "shot_blocking", "parent_stage": "aerial_base", "shot_index": 8}) sketch = _add_chain_asset( db, pid, eid, role="outdoor_camera_sketch", created_at="2026-06-30T19:16:07.918231+00:00", meta={"group_id": gid, "producer_stage": "camera_sketch", "parent_stage": "shot_blocking", "shot_index": 8}) cg = {"10:8": {"mode": "sketch", "group_id": gid}} diag = _resolve_outdoor_chain_lineage(db, pid, eid, cg) # aerial = root → input 미설정 assert _inputs(db, aerial) is None # blocking.input = [aerial] assert _inputs(db, blocking) == [aerial] # sketch.input = [blocking] (parent_stage=shot_blocking) assert _inputs(db, sketch) == [blocking] # camera_sketch UUID → composition_guides guide_asset_id assert cg["10:8"]["guide_asset_id"] == sketch assert diag["guide_asset_resolved"] == [{"shot": "10:8", "asset": sketch}] assert {"asset": blocking, "role": "outdoor_shot_blocking", "input": aerial} in diag["annotated"] finally: db.close() def test_chain_lineage_blocking_skipped_sketch_points_to_aerial(): """단순 구도(블로킹 skip): sketch.parent_stage=aerial_base → sketch.input=[aerial].""" db = SessionLocal() try: pid, eid = _seed_base(db) gid = "osl-single-l11-s10-sh3" aerial = _add_chain_asset( db, pid, eid, role="outdoor_aerial_base", created_at="2026-06-30T20:00:00.000000+00:00", meta={"group_id": gid, "producer_stage": "aerial_base", "parent_stage": None}) sketch = _add_chain_asset( db, pid, eid, role="outdoor_camera_sketch", created_at="2026-06-30T20:01:00.000000+00:00", meta={"group_id": gid, "producer_stage": "camera_sketch", "parent_stage": "aerial_base", "shot_index": 3}) cg = {"10:3": {"mode": "sketch", "group_id": gid}} _resolve_outdoor_chain_lineage(db, pid, eid, cg) assert _inputs(db, sketch) == [aerial] assert cg["10:3"]["guide_asset_id"] == sketch finally: db.close() def test_chain_lineage_rerun_picks_parent_before_child(): """재실행 누적 — blocking 은 자기 created_at 이하의 가장 최근 aerial 을 가리킨다 (같은 run 의 부모, 후행 run aerial 로 잘못 연결 금지).""" db = SessionLocal() try: pid, eid = _seed_base(db) gid = "osl-shared-l03-s10" aerial_old = _add_chain_asset( db, pid, eid, role="outdoor_aerial_base", created_at="2026-06-30T10:00:00.000000+00:00", meta={"group_id": gid, "producer_stage": "aerial_base"}) blocking_old = _add_chain_asset( db, pid, eid, role="outdoor_shot_blocking", created_at="2026-06-30T10:05:00.000000+00:00", meta={"group_id": gid, "producer_stage": "shot_blocking", "parent_stage": "aerial_base", "shot_index": 8}) aerial_new = _add_chain_asset( db, pid, eid, role="outdoor_aerial_base", created_at="2026-06-30T11:00:00.000000+00:00", meta={"group_id": gid, "producer_stage": "aerial_base"}) _resolve_outdoor_chain_lineage(db, pid, eid, {}) # 옛 blocking → 옛 aerial (created_at <= blocking 인 가장 최근) assert _inputs(db, blocking_old) == [aerial_old] assert _inputs(db, blocking_old) != [aerial_new] finally: db.close() def test_chain_lineage_future_parent_not_linked(): """★Codex NARROW 1 — parent 가 child 이후에만 존재(future)하면 input 미설정 + unresolved 진단. future parent 를 wrong edge 로 붙이지 않는다.""" db = SessionLocal() try: pid, eid = _seed_base(db) gid = "osl-shared-l03-s10" # blocking 이 aerial 보다 먼저 생성된 비정상 순서(부분실패/재실행 잔재). blocking = _add_chain_asset( db, pid, eid, role="outdoor_shot_blocking", created_at="2026-06-30T09:00:00.000000+00:00", meta={"group_id": gid, "producer_stage": "shot_blocking", "parent_stage": "aerial_base", "shot_index": 8}) _add_chain_asset( # aerial 은 blocking 이후 → future parent db, pid, eid, role="outdoor_aerial_base", created_at="2026-06-30T10:00:00.000000+00:00", meta={"group_id": gid, "producer_stage": "aerial_base"}) diag = _resolve_outdoor_chain_lineage(db, pid, eid, {}) # future aerial 은 붙지 않음 → blocking.input 미설정 + unresolved 진단. assert _inputs(db, blocking) is None assert any(u["asset"] == blocking and u["reason"] == "no_aerial_parent" for u in diag["unresolved"]) assert diag["annotated"] == [] finally: db.close() def test_chain_lineage_db_none_noop(): """db=None → no-op(진단 빈 구조), raise 없음.""" cg = {"10:8": {"mode": "sketch", "group_id": "g"}} diag = _resolve_outdoor_chain_lineage(None, "p", "e", cg) assert diag["annotated"] == [] assert "guide_asset_id" not in cg["10:8"] def test_chain_lineage_sketch_unresolved_when_no_assets(): """체인 asset 없음 → composition_guides guide_asset_id 미설정 + unresolved 진단.""" db = SessionLocal() try: pid, eid = _seed_base(db) cg = {"10:8": {"mode": "sketch", "group_id": "osl-shared-l99-s10"}} diag = _resolve_outdoor_chain_lineage(db, pid, eid, cg) assert "guide_asset_id" not in cg["10:8"] # asset 0 → 함수 조기반환(guide_asset_unresolved 도 기록 전). 빈 진단. assert diag["annotated"] == [] finally: db.close()