"""GROUNDING-V2 §2-3b — production mode 배선. 통과 조건: `legacy` 주행 산출이 **바이트 단위 불변** · `shadow_plan` 을 켠 것만으로 하류가 **stale 되지 않음**. ★그리고 **`v2` 는 반드시 움직여야 한다** — 안 움직이면 legacy 완료 CP 가 v2 진입에서 그대로 재사용된다. """ import pytest from app.core.errors import AppError from app.core.grounding_mode import ( GROUNDING_MODE_LEGACY, GROUNDING_MODE_SHADOW_PLAN, GROUNDING_MODE_V2, fingerprint_value, touches_production_fingerprint, ) from app.core.step_runner import compute_config_hash _BASE = {"project_id": "p", "style_rules": {"a": 1}, "model": "x"} class TestShadowDoesNotStaleButV2Does: """★관측 스위치는 하류를 안 굽고, 실제로 바뀌는 모드는 굽는다.""" def test_no_key_equals_legacy_equals_shadow_plan(self): bare = compute_config_hash(_BASE) assert compute_config_hash({**_BASE, "grounding_mode": GROUNDING_MODE_LEGACY}) == bare assert compute_config_hash({**_BASE, "grounding_mode": GROUNDING_MODE_SHADOW_PLAN}) == bare def test_v2_moves_the_hash(self): """★이게 핵심 — 안 움직이면 legacy 완료 CP 가 v2 에서 재사용된다.""" bare = compute_config_hash(_BASE) assert compute_config_hash({**_BASE, "grounding_mode": GROUNDING_MODE_V2}) != bare def test_v2_differs_from_both_free_modes(self): v2 = compute_config_hash({**_BASE, "grounding_mode": GROUNDING_MODE_V2}) for m in (GROUNDING_MODE_LEGACY, GROUNDING_MODE_SHADOW_PLAN): assert compute_config_hash({**_BASE, "grounding_mode": m}) != v2 def test_it_agrees_with_the_touches_production_fingerprint_contract(self): """★두 자리가 같은 계약을 말해야 한다 — 예전 안은 여기서 자기모순이었다.""" bare = compute_config_hash(_BASE) for mode in (GROUNDING_MODE_LEGACY, GROUNDING_MODE_SHADOW_PLAN, GROUNDING_MODE_V2): moved = compute_config_hash({**_BASE, "grounding_mode": mode}) != bare assert moved is touches_production_fingerprint(mode), ( f"{mode}: 지문 이동={moved} 인데 계약은 " f"{touches_production_fingerprint(mode)} 라고 한다") class TestFingerprintValue: def test_free_modes_are_dropped(self): assert fingerprint_value(GROUNDING_MODE_LEGACY) is None assert fingerprint_value(GROUNDING_MODE_SHADOW_PLAN) is None def test_v2_is_kept(self): assert fingerprint_value(GROUNDING_MODE_V2) == "v2" def test_none_is_dropped(self): assert fingerprint_value(None) is None @pytest.mark.parametrize("bad", ["shadowplan", "v_2", "true", "", " "]) def test_typo_fails_closed_here_too(self, bad): """★지문 계산이 오타를 조용히 접으면 같은 오타가 계속 같은 지문을 만든다.""" with pytest.raises(AppError): fingerprint_value(bad) def test_case_and_space_are_normalized_for_known_values(self): assert fingerprint_value(" V2 ") == "v2" assert fingerprint_value(" LEGACY ") is None class TestNothingElseChanged: """★공용 함수를 건드렸으니 기존 계약이 그대로인지 본다.""" @pytest.mark.parametrize("key", [ "style_rules", "model", "project_id", "target_scenes", "still_image_backend", ]) def test_other_keys_still_move_the_hash(self, key): assert compute_config_hash({**_BASE, key: "달라진 값"}) != compute_config_hash(_BASE) def test_hash_is_still_sixteen_hex(self): h = compute_config_hash(_BASE) assert len(h) == 16 and all(c in "0123456789abcdef" for c in h) def test_key_order_still_does_not_matter(self): assert compute_config_hash({"a": 1, "b": 2}) == compute_config_hash({"b": 2, "a": 1}) def test_unhashable_values_are_still_deterministic(self): class _Odd: pass assert compute_config_hash({"x": _Odd()}) == compute_config_hash({"x": _Odd()}) def test_empty_and_none_are_unchanged(self): assert compute_config_hash(None) == compute_config_hash({}) def test_caller_config_is_not_mutated(self): """★원본 dict 를 건드리면 호출자가 나중에 다른 것을 본다.""" cfg = {**_BASE, "grounding_mode": GROUNDING_MODE_SHADOW_PLAN} compute_config_hash(cfg) assert cfg["grounding_mode"] == GROUNDING_MODE_SHADOW_PLAN class TestTypoStopsBeforeAnySpending: """★`compute_config_hash` 안의 방어만으로는 **늦다**. 기존 완료 CP 가 있으면 `_check_cp_mismatch` 가 hash 를 먼저 계산해 실행 전에 서지만, **fresh/no-CP** 경로는 `_evaluate_resume_decision` 이 곧바로 RERUN_SELF 를 돌려주고 첫 hash 계산은 `save_checkpoint()` — 즉 `_execute` 와 유료 호출이 **끝난 뒤**다. 새 에피소드의 오타 하나가 돈을 다 쓰고 실패한다. """ def _runner(self, mode_value): from app.core.step_runner import StepRunner calls = {"execute": 0} class _Probe(StepRunner): def _execute(self, mode="resume"): calls["execute"] += 1 return {"ok": True} r = _Probe.__new__(_Probe) # __init__ 은 DB 를 만진다 r.project_config = {"project_id": "p", "grounding_mode": mode_value} return r, calls @pytest.mark.parametrize("mode", ["legacy", "shadow_plan", "v2"]) def test_valid_modes_pass_the_preflight(self, mode): r, _calls = self._runner(mode) r._preflight_grounding_mode() # 안 서야 한다 def test_missing_key_passes(self): from app.core.step_runner import StepRunner r = StepRunner.__new__(StepRunner) r.project_config = {"project_id": "p"} r._preflight_grounding_mode() def test_preflight_runs_at_the_front_of_run_not_in_validate_mode(self): """★`validate_mode` 에 넣으면 안 된다 — 제일 비싼 스텝이 override 하면서 `super()` 를 안 부른다.""" import inspect from app.core.step_runner import StepRunner src = inspect.getsource(StepRunner.run) assert "_preflight_grounding_mode" in src, "run() 안에 preflight 가 없다" assert src.index("_preflight_grounding_mode") < src.index("self.validate_mode"), \ "preflight 가 validate_mode 뒤에 있다 — override 경로가 빠진다" vm = inspect.getsource(StepRunner.validate_mode) assert "_preflight_grounding_mode" not in vm, \ "validate_mode 에 넣으면 override 하는 스텝이 빠진다" def test_the_expensive_step_still_gets_the_preflight(self): """★`SceneImagePipelineStep.validate_mode` 는 `super()` 를 안 부른다. 그래서 preflight 가 `run()` 최선두에 있어야 그 경로도 덮인다. """ import inspect from app.core.steps.image_steps import SceneImagePipelineStep vm = inspect.getsource(SceneImagePipelineStep.validate_mode) assert "super()" not in vm, ( "override 가 super() 를 부르게 바뀌었다 — 이 시험의 전제를 다시 보라") assert SceneImagePipelineStep.run is StepRunner_run(), \ "run() 을 override 했다면 preflight 가 그 경로에도 있는지 확인하라" def StepRunner_run(): from app.core.step_runner import StepRunner return StepRunner.run class TestEnvBranchIsFolded: """★`resolve_grounding_mode` 는 키가 없거나 None 이면 ENV 로 떨어진다. dict 키만 보고 접으면 **ENV=v2 + 키 없음** 일 때 실제 실행은 v2 인데 지문은 legacy 와 같아져 **막으려던 CP 재사용이 그대로 돌아온다** (실제로 재현했다). """ def test_env_v2_without_key_moves_the_hash(self, monkeypatch): monkeypatch.setenv("GROUNDING_MODE", "v2") legacy = compute_config_hash({**_BASE, "grounding_mode": "legacy"}) assert compute_config_hash(_BASE) != legacy @pytest.mark.parametrize("free", ["legacy", "shadow_plan"]) def test_env_free_modes_without_key_stay_identical(self, monkeypatch, free): monkeypatch.setenv("GROUNDING_MODE", free) monkeypatch.delenv("GROUNDING_MODE", raising=False) bare = compute_config_hash(_BASE) monkeypatch.setenv("GROUNDING_MODE", free) assert compute_config_hash(_BASE) == bare def test_config_key_beats_env(self, monkeypatch): monkeypatch.setenv("GROUNDING_MODE", "v2") # ★config 가 legacy 면 실효 모드도 legacy 다 — 지문도 그래야 한다. monkeypatch.delenv("GROUNDING_MODE", raising=False) bare = compute_config_hash(_BASE) monkeypatch.setenv("GROUNDING_MODE", "v2") assert compute_config_hash({**_BASE, "grounding_mode": "legacy"}) == bare def test_none_value_falls_through_to_env(self, monkeypatch): monkeypatch.setenv("GROUNDING_MODE", "v2") with_none = compute_config_hash({**_BASE, "grounding_mode": None}) monkeypatch.delenv("GROUNDING_MODE", raising=False) assert with_none != compute_config_hash(_BASE) def test_env_typo_fails_closed(self, monkeypatch): from app.core.errors import AppError monkeypatch.setenv("GROUNDING_MODE", "shadowplan") with pytest.raises(AppError): compute_config_hash(_BASE) class TestPublicRunActuallyStops: """★helper 를 직접 부르면 「execute 0회」는 **자명하다**. public `run()` 을 태워야 한다. 그리고 `_execute` 만 세면 부족하다 — preflight **뒤에 오는 모든 문** (validate_mode · gate/applicability · claim · execute · save)이 한 번도 안 돌아야 「아무것도 안 바꿨다」가 성립한다 (Codex). """ def _probe(self, cfg): from app.core.step_runner import StepRunner seen = {k: 0 for k in ( "validate_mode", "check_applicability", "load_checkpoint", "claim", "execute", "save_checkpoint", "invalidate_downstream", "clear_checkpoint", "cleanup_artifacts", )} def _spy(name, ret=None): def _f(*a, **k): seen[name] += 1 return ret return _f class _Probe(StepRunner): def _execute(self, mode="resume"): seen["execute"] += 1 return {"ok": True} def validate_mode(self, mode): seen["validate_mode"] += 1 r = _Probe.__new__(_Probe) r.project_config = cfg for name, ret in (("check_applicability", True), ("load_checkpoint", None), ("save_checkpoint", None), ("invalidate_downstream", None), ("clear_checkpoint", None), ("cleanup_artifacts", None)): setattr(r, name, _spy(name, ret)) return r, seen def test_invalid_config_mode_stops_public_run_touching_nothing(self): from app.core.errors import AppError r, seen = self._probe({"project_id": "p", "grounding_mode": "shadowplan"}) with pytest.raises(AppError): r.run("resume") assert sum(seen.values()) == 0, f"preflight 뒤 문이 돌았다: {seen}" def test_invalid_env_mode_stops_public_run_touching_nothing(self, monkeypatch): from app.core.errors import AppError monkeypatch.setenv("GROUNDING_MODE", "v_2") r, seen = self._probe({"project_id": "p"}) with pytest.raises(AppError): r.run("resume") assert sum(seen.values()) == 0, f"preflight 뒤 문이 돌았다: {seen}" def test_a_valid_mode_gets_past_the_preflight(self): """positive control — preflight 가 **아무것도 못 지나가게** 하는 것은 아니다.""" r, seen = self._probe({"project_id": "p", "grounding_mode": "shadow_plan"}) try: r.run("resume") except Exception: pass # 그 뒤 어디서 서든 상관없다 assert seen["validate_mode"] == 1, "유효 모드인데 preflight 가 막았다"