"""scene_detail runner retry path unit tests. 2026-05-11 — `SceneDetailStep._run_shots_with_retry` 의 shot-level typed failure + retry 동작 검증. _analyze_one 을 mock 으로 swap 해서 LLM 호출 우회. 검증 케이스: - 모든 shot 1차 성공 → results 채워짐 / failed=0 / retry 진입 X - 1 shot 1차 AppError → retry 2차 성공 → results 채워짐 / failed=0 - 1 shot 1차+2차 모두 AppError → typed summary AppError raise - 1 shot 1차 unexpected exception (Exception, not AppError) → 같은 retry path """ from __future__ import annotations from typing import Any, Dict from unittest.mock import MagicMock import pytest from app.core.errors import AppError def _make_step(): """SceneDetailStep instance — _run_shots_with_retry 만 호출하면 되므로 minimal mock. __new__ 로 instance 생성 후 logger 만 attach.""" from app.core.steps.detail_steps import SceneDetailStep step = SceneDetailStep.__new__(SceneDetailStep) return step def _mk_seg(scene_index: int) -> Dict[str, Any]: return {"scene_index": scene_index} def _mk_shot(shot_index: int) -> Dict[str, Any]: return {"shot_index": shot_index} def _mk_result(scene_index: int, shot_index: int) -> Dict[str, Any]: return {"scene_index": scene_index, "_shot_index": shot_index, "t2i_variations": [{"t2i_prompt": "ok"}]} def test_all_shots_primary_success_no_retry(monkeypatch): """1차에서 모든 shot 성공 → retry path 진입 X, failed=0.""" step = _make_step() tasks = [(_mk_seg(1), _mk_shot(1)), (_mk_seg(2), _mk_shot(2))] expected = { (1, 1): _mk_result(1, 1), (2, 2): _mk_result(2, 2), } def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): return expected[(seg["scene_index"], sh["shot_index"])] monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) results, failed, failed_shots = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) assert failed == 0 # 순서 무관 — 둘 다 포함 assert {r["_shot_index"] for r in results} == {1, 2} def test_shot_primary_fail_retry_recovers(monkeypatch): """1 shot 의 1차 AppError → retry 2차 성공 → final failed=0.""" step = _make_step() tasks = [(_mk_seg(13), _mk_shot(5)), (_mk_seg(1), _mk_shot(1))] call_counts: Dict[tuple, int] = {(13, 5): 0, (1, 1): 0} def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): key = (seg["scene_index"], sh["shot_index"]) call_counts[key] += 1 # S13/5: 1차 fail, 2차 success. if key == (13, 5) and call_counts[key] == 1: raise AppError( code="step.scene_detail.contract_violation_visible_id_not_in_prompt", message="S13_Shot5 mock contract violation", status_code=400, ) return _mk_result(*key) monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) results, failed, failed_shots = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) assert failed == 0 assert {r["_shot_index"] for r in results} == {1, 5} # S13/5 는 2회 호출 (primary + retry) assert call_counts[(13, 5)] == 2 # S1/1 은 1회 (primary 성공) assert call_counts[(1, 1)] == 1 # 2026-08-04: retry 로 회복된 shot 은 unresolved 목록에 남으면 안 된다. # 남으면 다음 resume 이 그 키를 실패로 읽어 **이미 성공한 shot 을 다시 # 유료로 부른다**(시도 이력과 미해결 목록의 혼동). assert failed_shots == [] def test_shot_primary_and_retry_both_fail_typed_summary(monkeypatch): """1 shot 의 1차+2차 모두 AppError → raise 하지 않고 typed 목록으로 반환. 2026-08-04 계약 변경: 잔존 실패가 있어도 성공분을 버리지 않는다. 무결성은 `step_manifest` 의 `allow_partial_downstream: False` 가 partial 일 때 하류를 차단하는 것으로 지킨다 (raise 로 전량 폐기하던 이전 방식은 255 중 1 실패에 254 건의 유료 호출을 버렸다). """ step = _make_step() tasks = [(_mk_seg(13), _mk_shot(5))] def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): raise AppError( code="step.scene_detail.contract_violation_visible_id_not_in_prompt", message="S13_Shot5 mock persistent contract violation", status_code=400, ) monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) results, failed, failed_shots = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) assert failed == 1 assert results == [] # typed 정보가 목록으로 남는다 (primary + retry 두 stage). codes = {f.get("code") for f in failed_shots} assert "step.scene_detail.contract_violation_visible_id_not_in_prompt" in codes keys = {(f["scene_index"], f["shot_index"]) for f in failed_shots} assert keys == {(13, 5)} def test_shot_primary_unexpected_exception_caught(monkeypatch): """1차에서 AppError 가 아닌 일반 Exception 도 typed failure 로 보존 + retry 진입.""" step = _make_step() tasks = [(_mk_seg(9), _mk_shot(3))] call_count = {"n": 0} def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): call_count["n"] += 1 if call_count["n"] == 1: raise RuntimeError("unexpected runtime error") return _mk_result(9, 3) monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) results, failed, failed_shots = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) assert failed == 0 assert results == [_mk_result(9, 3)] assert call_count["n"] == 2 def test_echo_mismatch_shot_skipped_in_outer_retry(monkeypatch): """area-frame-spatial-contract T6: scene_detail.frame_spatial_contract_echo_mismatch 는 inner _check_prompts retry 가 이미 1회 처리 → outer _run_shots_with_retry 에서 추가 retry 안 함. 같은 batch 의 다른 contract violation 은 정상 retry. """ step = _make_step() tasks = [(_mk_seg(7), _mk_shot(3)), (_mk_seg(8), _mk_shot(2))] call_counts: Dict[tuple, int] = {(7, 3): 0, (8, 2): 0} def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): key = (seg["scene_index"], sh["shot_index"]) call_counts[key] += 1 if key == (7, 3): # echo mismatch — non-retryable raise AppError( code="scene_detail.frame_spatial_contract_echo_mismatch", message="S7_Shot3 mock echo mismatch", status_code=400, ) if key == (8, 2) and call_counts[key] == 1: # 다른 contract violation — 정상 retry raise AppError( code="step.scene_detail.contract_violation_visible_id_not_in_prompt", message="S8_Shot2 mock contract violation", status_code=400, ) return _mk_result(seg["scene_index"], sh["shot_index"]) monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) _results, _failed, failed_shots = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) # echo mismatch shot 은 outer retry 진입 X — 1회만 호출 assert call_counts[(7, 3)] == 1 # 다른 violation 은 outer retry 진입 — 2회 호출 assert call_counts[(8, 2)] == 2 # typed 목록에 echo mismatch code 가 남는다 (2026-08-04: raise → 반환) assert any( "frame_spatial_contract_echo_mismatch" in (f.get("code") or "") for f in failed_shots ) def test_existing_failed_count_propagated(monkeypatch): """caller 가 이미 failed=2 를 넘기고 tasks 가 비어도 그 값이 그대로 반환된다. legacy callers (edited_results pre-fill 등) 의 failed 인자 누적 흐름 보장. 2026-08-04: 잔존 failed 가 있어도 raise 하지 않는다 — 호출부가 partial 로 넘겨 성공분을 보존한다. """ step = _make_step() # tasks 없음 — retry_tasks 가 비어 sleep/retry skip → failed 그대로 전달. results, failed, failed_shots = step._run_shots_with_retry( tasks=[], ctx=None, system="sys", schema={}, results=[], failed=2, retry_sleep_seconds=0.0, ) assert failed == 2 assert results == [] # 이 경로는 typed 실패를 기록한 적이 없으므로 목록은 비어 있다. assert failed_shots == [] def test_unresolved_excludes_retry_recovered_shots(monkeypatch): """혼합 케이스 — S1 은 retry 로 회복, S2 는 지속 실패. 2026-08-04 회귀 가드: 반환되는 목록은 **시도 이력이 아니라 미해결 키**여야 한다. S1 이 남으면 다음 resume 이 이미 성공한 S1 을 재사용에서 제외해 다시 유료로 부른다 (Codex BLOCKING-1 최소 재현). """ step = _make_step() tasks = [(_mk_seg(1), _mk_shot(1)), (_mk_seg(2), _mk_shot(2))] calls: Dict[tuple, int] = {(1, 1): 0, (2, 2): 0} def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): key = (seg["scene_index"], sh["shot_index"]) calls[key] += 1 if key == (2, 2): raise AppError( code="step.scene_detail.contract_violation_visible_id_not_in_prompt", message="S2_Shot2 persistent", status_code=400, ) if key == (1, 1) and calls[key] == 1: raise AppError( code="step.scene_detail.contract_violation_visible_id_not_in_prompt", message="S1_Shot1 transient", status_code=400, ) return _mk_result(*key) monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) results, failed, unresolved = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) assert failed == 1 assert {r["_shot_index"] for r in results} == {1} # 미해결은 S2 뿐 — S1 은 retry 로 회복됐으므로 제외된다. assert {(f["scene_index"], f["shot_index"]) for f in unresolved} == {(2, 2)} def test_retry_carries_the_contract_violation_it_failed_on(monkeypatch): """★재시도는 직전 계약 위반 문구를 싣는다 (2026-09-18 컨트리로드). 손 클로즈업 3샷이 같은 입력으로 두 번 모두 인물 표식을 빼서 base_id_missing 으로 섰다. 계약 위반이 아닌 실패(네트워크 등)는 싣지 않는다. """ step = _make_step() tasks = [(_mk_seg(9), _mk_shot(8)), (_mk_seg(2), _mk_shot(3))] seen: Dict[tuple, list] = {(9, 8): [], (2, 3): []} why = "S9_Shot8 (variation 0): subject 'C03' policy='id_and_outlook_required' requires base C## in t2i_prompt, but missing." def _fake_analyze_one(seg, sh, ctx, system, schema, corrections=None): key = (seg["scene_index"], sh["shot_index"]) seen[key].append(corrections) if len(seen[key]) == 1: if key == (9, 8): raise AppError( code="step.scene_detail.contract_violation.subject_reference_policy.base_id_missing", message=why, status_code=400) raise RuntimeError("connection reset") return _mk_result(*key) monkeypatch.setattr(step, "_analyze_one", _fake_analyze_one) results, failed, _ = step._run_shots_with_retry( tasks=tasks, ctx=None, system="sys", schema={}, results=[], failed=0, retry_sleep_seconds=0.0, ) assert failed == 0 assert seen[(9, 8)][0] is None, "첫 시도에는 고칠 것이 없다" assert seen[(9, 8)][1] and any(why in c for c in seen[(9, 8)][1]), seen[(9, 8)] assert seen[(2, 3)][1] is None, "계약 위반이 아닌 실패까지 수정 요청으로 싣지 않는다"