"""asset_readiness preflight — Group 1 #2 회귀 가드. 본 테스트는 dataclass 동작 + ENV/settings flag parsing + assert_episode_asset_readiness 의 fail-fast 분기 로직 검증. compute_episode_asset_readiness 의 DB 통합은 query mock 으로 표현 (대규모 fixture 는 e2e 영역이라 본 단위 테스트 범위 외). 시나리오 의존성 0 — placeholder C01/L01/P01 만 사용. """ from __future__ import annotations import os from unittest.mock import MagicMock, patch import pytest from app.core.asset_readiness import ( AssetMissEntry, AssetReadinessCard, _is_text_only_allowed, assert_episode_asset_readiness, ) # ───────────────────────────────────────────── # AssetReadinessCard dataclass 동작 # ───────────────────────────────────────────── def test_card_is_ready_when_no_missing_and_expected_present(): card = AssetReadinessCard( project_id="p", episode_id="e", expected_ids=["reference:E1"], registered_ids=["reference:E1"], disk_found_ids=["reference:E1"], missing=[], ) assert card.is_ready is True assert card.missing_count == 0 def test_card_not_ready_when_missing(): card = AssetReadinessCard( project_id="p", episode_id="e", expected_ids=["reference:E1"], registered_ids=[], disk_found_ids=[], missing=[AssetMissEntry( entity_id="E1", short_id="C01", name="x", asset_type="reference", reason="no_db_row", )], ) assert card.is_ready is False assert card.missing_count == 1 def test_card_not_ready_when_no_expected(): """expected 가 0 인 episode 도 is_ready=False — 빈 episode 가 silent OK 아닌 명시 진단 (image generation 은 entity 1+ 전제).""" card = AssetReadinessCard(project_id="p", episode_id="e") assert card.is_ready is False # expected_ids 비어 있으면 False def test_card_summary_format(): card = AssetReadinessCard( project_id="p", episode_id="abcdef0123456789", expected_ids=["a", "b", "c"], registered_ids=["a", "b"], disk_found_ids=["a"], missing=[ AssetMissEntry("e1", "C01", "x", "reference", "disk_missing"), AssetMissEntry("e2", "C02", "y", "composite", "no_db_row"), ], ) s = card.summary() assert "expected=3" in s assert "registered=2" in s assert "disk=1" in s assert "missing=2" in s def test_card_missing_summary_truncates_long_lists(): misses = [ AssetMissEntry(f"e{i}", f"C{i:02d}", "x", "reference", "no_db_row") for i in range(8) ] card = AssetReadinessCard(project_id="p", episode_id="e", missing=misses) s = card.missing_summary(max_items=3) assert "+5 more" in s assert "C00" in s assert "C03" not in s # 4번째부터는 짧게. # ───────────────────────────────────────────── # _is_text_only_allowed — ENV / settings dual-source # ───────────────────────────────────────────── @pytest.mark.parametrize("raw,expected", [ ("1", True), ("true", True), ("True", True), ("TRUE", True), ("yes", True), ("on", True), ("0", False), ("false", False), ("False", False), ("no", False), ("off", False), ]) def test_text_only_env_recognized_values(raw, expected): with patch.dict(os.environ, {"ALLOW_TEXT_ONLY_WITHOUT_REFS": raw}, clear=False): assert _is_text_only_allowed() is expected def test_text_only_env_typo_falls_back_to_settings(): """unrecognized ENV 값은 silent True 가 아니라 settings default fallback.""" with patch.dict(os.environ, {"ALLOW_TEXT_ONLY_WITHOUT_REFS": "garbage"}, clear=False): # settings default = False → typo 은 False (block 유지). assert _is_text_only_allowed() is False def test_text_only_default_false(): """ENV 미설정 + settings default False → block.""" if "ALLOW_TEXT_ONLY_WITHOUT_REFS" in os.environ: del os.environ["ALLOW_TEXT_ONLY_WITHOUT_REFS"] with patch("app.core.asset_readiness.os.environ.get", return_value=None): # settings 의 default 는 False. assert _is_text_only_allowed() is False # ───────────────────────────────────────────── # assert_episode_asset_readiness — fail-fast block 분기 # ───────────────────────────────────────────── def test_assert_returns_card_when_ready(): """is_ready=True 인 card 면 raise 없이 반환.""" db = MagicMock() fake_card = AssetReadinessCard( project_id="p", episode_id="e", expected_ids=["reference:E1"], registered_ids=["reference:E1"], disk_found_ids=["reference:E1"], missing=[], ) with patch( "app.core.asset_readiness.compute_episode_asset_readiness", return_value=fake_card, ): out = assert_episode_asset_readiness(db, "p", "e") assert out is fake_card def test_assert_blocks_when_missing_default_settings(): """missing > 0 + settings default (block) → AppError raise.""" from app.core.errors import AppError db = MagicMock() fake_card = AssetReadinessCard( project_id="p", episode_id="e", expected_ids=["reference:E1"], missing=[AssetMissEntry("E1", "C01", "x", "reference", "no_db_row")], ) with patch( "app.core.asset_readiness.compute_episode_asset_readiness", return_value=fake_card, ), patch("app.core.asset_readiness._is_text_only_allowed", return_value=False): with pytest.raises(AppError) as excinfo: assert_episode_asset_readiness(db, "p", "e") assert excinfo.value.code == "image.assets_not_ready" assert "참조 이미지" in excinfo.value.message # 디버깅 우회 hint. assert "ALLOW_TEXT_ONLY_WITHOUT_REFS" in excinfo.value.message def test_assert_passes_when_text_only_allowed_via_opt_in(caplog): """missing 있어도 opt-in flag True 면 raise 안 함 (warning log 만).""" import logging as _logging db = MagicMock() fake_card = AssetReadinessCard( project_id="p", episode_id="e", expected_ids=["reference:E1"], missing=[AssetMissEntry("E1", "C01", "x", "reference", "disk_missing")], ) with patch( "app.core.asset_readiness.compute_episode_asset_readiness", return_value=fake_card, ), patch("app.core.asset_readiness._is_text_only_allowed", return_value=True): with caplog.at_level(_logging.WARNING, logger="app.core.asset_readiness"): out = assert_episode_asset_readiness(db, "p", "e") assert out is fake_card # warning log 가 explicit 명시 (silent fallback 아님). warnings = [r for r in caplog.records if r.levelname == "WARNING"] assert any("opt-in bypass" in r.message for r in warnings) assert any("ALLOW_TEXT_ONLY_WITHOUT_REFS=true" in r.message for r in warnings) def test_assert_blocks_when_no_expected_entities(): """expected_ids 가 비어 있는 case (entity scope 비어 있음) 도 block. 이는 image generation 진입 전 데이터 부족 명시 — silent OK 아닌 fail-fast. Claude review IMPORTANT (Group 1 #2): AppError code 도 같이 검증. """ from app.core.errors import AppError db = MagicMock() fake_card = AssetReadinessCard(project_id="p", episode_id="e") with patch( "app.core.asset_readiness.compute_episode_asset_readiness", return_value=fake_card, ), patch("app.core.asset_readiness._is_text_only_allowed", return_value=False): with pytest.raises(AppError) as excinfo: assert_episode_asset_readiness(db, "p", "e") # missing_count=0 이지만 expected_ids 도 0 이라 is_ready=False → block 진입. # AppError code 가 정확한지 검증. assert excinfo.value.code == "image.assets_not_ready", ( f"잘못된 AppError code raised: {excinfo.value.code}" ) # ───────────────────────────────────────────── # Settings str type — Pydantic ValidationError 회귀 가드 (Codex IMPORTANT) # ───────────────────────────────────────────── def test_settings_field_is_str_type_not_bool(): """Codex review IMPORTANT: ENV ALLOW_TEXT_ONLY_WITHOUT_REFS=garbage 가 Pydantic Settings 의 bool field type 이면 ValidationError 로 import 가 실패. str field 로 두고 helper 가 lenient parse 해야 한다. """ from app.core.config import Settings field_info = Settings.model_fields.get("allow_text_only_without_refs") assert field_info is not None, "allow_text_only_without_refs field 누락" # type 이 str 이어야 함 (bool 이면 typo ENV → ValidationError → app 부팅 실패). annotation = field_info.annotation assert annotation is str, ( f"allow_text_only_without_refs annotation={annotation!r} != str — " f"Pydantic 이 ENV 자동 parsing 단계에서 typo 를 ValidationError 처리해 " f"앱 import 가 실패한다." ) def test_settings_garbage_env_does_not_break_import(): """Codex review IMPORTANT: ENV typo 가 settings module reload 시 ValidationError 안 남. Claude review IMPORTANT (Group 1 #2 iter#2): sys.modules 처리 robust — original module 객체 보존 후 finally 에서 무조건 복원. 재import 가 실패해도 downstream test 의 모듈 상태가 깨지지 않음. """ import sys original_config = sys.modules.get("app.core.config") try: with patch.dict(os.environ, {"ALLOW_TEXT_ONLY_WITHOUT_REFS": "garbage"}, clear=False): if "app.core.config" in sys.modules: del sys.modules["app.core.config"] from app.core import config as _cfg # noqa _cfg.Settings() # explicit fresh instance — typo ENV 도 통과해야. except Exception as exc: pytest.fail(f"Settings import/instantiation failed with garbage ENV: {exc}") finally: # 무조건 원래 module 복원 — re-import 실패 위험 없음. if original_config is not None: sys.modules["app.core.config"] = original_config elif "app.core.config" in sys.modules: del sys.modules["app.core.config"] # ───────────────────────────────────────────── # state_variant scope (Codex BLOCKING) — 추출 로직 # ───────────────────────────────────────────── def test_collect_state_variants_returns_empty_when_no_cp(tmp_path, monkeypatch): """shot_staging cp 부재 시 silent skip 0 — 빈 리스트 반환 (expected 0).""" from app.core.asset_readiness import _collect_expected_state_variants monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) pairs = _collect_expected_state_variants("nonexistent_project", "ep1", []) assert pairs == [] def test_collect_state_variants_extracts_from_cp(tmp_path, monkeypatch): """shot_staging cp 의 immobilized subject_state 인물 → (cid, state) pair 추출. Area #2 W5 (2026-05-17): v13 shape 으로 cascade — character_angles 안 ``gaze_direction_kind`` + ``subject_state`` 3 field 구조. immobilized 판정은 SOT helper ``is_immobilized_state`` 단일 path. """ import json from app.core.asset_readiness import _collect_expected_state_variants monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "shot_staging" cp_dir.mkdir(parents=True) cp_data = { "shots": [ {"character_angles": [ {"character": "name_A", "gaze_direction_kind": "closed_eyes", "subject_state": "dead"}, {"character": "name_B", "gaze_direction_kind": "closed_eyes", "subject_state": "severely_injured"}, {"character": "name_C", "gaze_direction_kind": "camera", "subject_state": "alive"}, # not immobilized. {"character": "name_A", "gaze_direction_kind": "closed_eyes", "subject_state": "dead"}, # dup → 1번만. ]}, ] } (cp_dir / "manifest.json").write_text(json.dumps(cp_data), encoding="utf-8") expected_entities = [ {"id": "uuid_A", "name": "name_A", "type": "character", "short_id": "C01"}, {"id": "uuid_B", "name": "name_B", "type": "character", "short_id": "C02"}, {"id": "uuid_C", "name": "name_C", "type": "character", "short_id": "C03"}, ] pairs = _collect_expected_state_variants("P1", "E1", expected_entities) assert ("uuid_A", "dead") in pairs assert ("uuid_B", "severely_injured") in pairs # name_C 는 subject_state 가 immobilized 아니므로 제외. assert not any(p[0] == "uuid_C" for p in pairs) # 중복 제거. assert len([p for p in pairs if p[0] == "uuid_A"]) == 1 def test_collect_state_variants_handles_data_wrapper_shape(tmp_path, monkeypatch): """Claude review IMPORTANT (Group 1 #2 iter#2): cp 가 ``{"data": {"shots": [...]}}`` 형식 (pipeline-v4 manifest) 일 때도 정확히 추출.""" import json from app.core.asset_readiness import _collect_expected_state_variants monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "shot_staging" cp_dir.mkdir(parents=True) cp_data = { "data": { # pipeline-v4 manifest wrapper "shots": [ {"character_angles": [ # Area #2 W5: v13 3 field shape {"character": "name_A", "gaze_direction_kind": "closed_eyes", "subject_state": "dead"}, ]}, ] } } (cp_dir / "manifest.json").write_text(json.dumps(cp_data), encoding="utf-8") expected_entities = [ {"id": "uuid_A", "name": "name_A", "type": "character", "short_id": "C01"}, ] pairs = _collect_expected_state_variants("P1", "E1", expected_entities) assert ("uuid_A", "dead") in pairs def test_collect_state_variants_handles_data_null(tmp_path, monkeypatch): """Claude review IMPORTANT: cp 의 data 가 null/list 인 edge case → 빈 리스트 (silent crash 안 함).""" import json from app.core.asset_readiness import _collect_expected_state_variants monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "shot_staging" cp_dir.mkdir(parents=True) # data null — AttributeError 회귀 위험 → 빈 리스트로 safe handle. (cp_dir / "manifest.json").write_text(json.dumps({"data": None}), encoding="utf-8") pairs = _collect_expected_state_variants("P1", "E1", []) assert pairs == [] # data 가 list 인 case. (cp_dir / "manifest.json").write_text(json.dumps({"data": []}), encoding="utf-8") pairs = _collect_expected_state_variants("P1", "E1", []) assert pairs == [] # shots 가 dict (잘못된 type) 인 case. (cp_dir / "manifest.json").write_text(json.dumps({"shots": {}}), encoding="utf-8") pairs = _collect_expected_state_variants("P1", "E1", []) assert pairs == [] def test_collect_state_variants_uses_normalized_name_match(tmp_path, monkeypatch): """Codex review Critical (Group 1 #2 iter#2): producer 와 동일 normalization (build_name_index / lookup_name) — 괄호/공백 변형도 매칭.""" import json from app.core.asset_readiness import _collect_expected_state_variants monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "shot_staging" cp_dir.mkdir(parents=True) # cp 가 entity 등록 이름과 다른 표기 (e.g. 괄호 안 정보 추가) 사용. cp_data = { "shots": [ {"character_angles": [ # Area #2 W5: v13 3 field shape {"character": "name_A (변형)", "gaze_direction_kind": "closed_eyes", "subject_state": "dead"}, # 괄호 노이즈. ]}, ] } (cp_dir / "manifest.json").write_text(json.dumps(cp_data), encoding="utf-8") expected_entities = [ {"id": "uuid_A", "name": "name_A", "type": "character", "short_id": "C01"}, ] pairs = _collect_expected_state_variants("P1", "E1", expected_entities) # name_matcher 가 괄호 제거된 base 매칭 → uuid_A 추출 성공. assert ("uuid_A", "dead") in pairs, ( "name_matcher 가 producer 와 동일 normalization 안 적용 — drift 회귀" ) def test_collect_state_variants_handles_corrupted_cp(tmp_path, monkeypatch, caplog): """corrupted JSON cp → warning log + 빈 리스트 (silent skip 안 함, expected 0).""" import logging from app.core.asset_readiness import _collect_expected_state_variants monkeypatch.setattr("app.core.config.settings.projects_dir", str(tmp_path)) cp_dir = tmp_path / "P1" / "checkpoints" / "episodes" / "E1" / "shot_staging" cp_dir.mkdir(parents=True) (cp_dir / "manifest.json").write_text("not valid json {", encoding="utf-8") with caplog.at_level(logging.WARNING, logger="app.core.asset_readiness"): pairs = _collect_expected_state_variants("P1", "E1", []) assert pairs == [] warnings = [r for r in caplog.records if r.levelname == "WARNING"] assert any("shot_staging cp parse failed" in r.message for r in warnings), ( "corrupted cp 가 silent fallback — warning log 명시 안 됨" ) # ───────────────────────────────────────────── # AssetMissEntry reason 분류 — silent skip 0 검증 # ───────────────────────────────────────────── def test_miss_entry_reason_categories(): """missing 의 reason 은 silent skip 패턴 별로 명시.""" valid_reasons = {"no_db_row", "no_file_path", "disk_missing"} # 본 가드 테스트 — asset_readiness 코드 안에서 사용한 reason 들이 valid set. e1 = AssetMissEntry("e", "C01", "x", "reference", "no_db_row") e2 = AssetMissEntry("e", "C01", "x", "reference", "no_file_path") e3 = AssetMissEntry("e", "C01", "x", "reference", "disk_missing") for entry in (e1, e2, e3): assert entry.reason in valid_reasons def test_miss_entry_supports_optional_file_path(): """disk_missing reason 은 file_path 보존 (디버깅용 — 어디 disk 가 비었는지).""" entry = AssetMissEntry( entity_id="e", short_id="C01", name="x", asset_type="reference", reason="disk_missing", file_path="/path/to/missing.png", ) assert entry.file_path == "/path/to/missing.png" # no_db_row / no_file_path 는 file_path default empty. entry2 = AssetMissEntry("e", "C01", "x", "reference", "no_db_row") assert entry2.file_path == ""