"""conti_ab — 콘티 A/B 외부 판정 결정론 테스트 (E2E6 ⑧). 판정 완성도=E2E/육안 — 여기는 블라인드·순서 뒤집기·동점 규칙·팩 중립성만 잠근다. """ from pathlib import Path from typing import Any, Dict, List import pytest from app.modules.pipeline.conti_ab import ( decide_winner, resolve_prompt_version, run_conti_ab_outer_judge, ) def test_decide_winner_agreement_and_tie(): ab = [{"order": "AB", "winner_branch": "A"}, {"order": "BA", "winner_branch": "A"}] assert decide_winner(ab) == "A" bb = [{"order": "AB", "winner_branch": "B"}, {"order": "BA", "winner_branch": "B"}] assert decide_winner(bb) == "B" split = [{"order": "AB", "winner_branch": "A"}, {"order": "BA", "winner_branch": "B"}] assert decide_winner(split) == "tie" assert decide_winner([]) == "tie" def _pngs(tmp_path): a = tmp_path / "a.png" b = tmp_path / "b.png" a.write_bytes(b"\x89PNG\r\n\x1a\n" + b"A" * 8) b.write_bytes(b"\x89PNG\r\n\x1a\n" + b"B" * 8) return a, b def test_outer_judge_blind_and_order_flip(tmp_path): """라벨은 항상 1/2(블라인드), 2회차는 입력 순서 뒤집기 — '항상 Candidate 1' 선택 판정은 순서 매핑으로 A/B 갈려 tie.""" a, b = _pngs(tmp_path) calls: List[Dict[str, Any]] = [] def fake(step_tag, system, parts, schema, **kw): calls.append({"parts": parts, "system": system}) return {"winner": "1", "reason_ko": "근거"} out = run_conti_ab_outer_judge( prompt="SAMPLE PROMPT", sel_a=a, sel_b=b, call_structured_fn=fake, ) assert out["winner"] == "tie" # AB→A, BA→B assert [v["winner_branch"] for v in out["verdicts"]] == ["A", "B"] assert len(calls) == 2 for c in calls: texts = [p["text"] for p in c["parts"] if p.get("type") == "text"] # 블라인드 — 어느 쪽이 콘티 사용본인지 노출 금지 joined = " ".join(texts) + " " + c["system"] assert "conti" not in joined.lower() or "internal pipeline" in \ c["system"] assert any(t == "Candidate 1:" for t in texts) assert any(t == "Candidate 2:" for t in texts) # 순서 뒤집기 실증: 이미지 첨부 순서가 2회차에 반대 def img_bytes(call): return [p["image_url"]["url"][-8:] for p in call["parts"] if p.get("type") == "image_url"] assert img_bytes(calls[0]) != img_bytes(calls[1]) def test_outer_judge_consistent_winner(tmp_path): a, b = _pngs(tmp_path) seen = [] def fake(step_tag, system, parts, schema, **kw): # 항상 '콘티 미사용본(b 바이트)'을 고르는 판정 시뮬레이션 imgs = [p["image_url"]["url"] for p in parts if p.get("type") == "image_url"] pick = "1" if imgs[0].endswith("pCQkJCQkJCQg==") else "2" # b 파일 seen.append(pick) return {"winner": pick, "reason_ko": ""} out = run_conti_ab_outer_judge( prompt="SAMPLE", sel_a=a, sel_b=b, call_structured_fn=fake) assert out["winner"] == "B" assert seen == ["2", "1"] def test_decision_fingerprint_drift_axes(tmp_path): """Codex 8e70d4c0 B1: prompt/sel bytes/모델 어느 하나 변경=지문 드리프트.""" from app.modules.pipeline.conti_ab import decision_fingerprint a, b = _pngs(tmp_path) base = dict(prompt="P", sel_a=a, sel_b=b, judge_model_physical="SAMPLE-m") fp = decision_fingerprint(**base) assert fp == decision_fingerprint(**base) # 결정론 assert fp != decision_fingerprint(**{**base, "prompt": "P2"}) assert fp != decision_fingerprint( **{**base, "judge_model_physical": "SAMPLE-m2"}) b.write_bytes(b"\x89PNG\r\n\x1a\n" + b"C" * 8) assert fp != decision_fingerprint(**base) # sel bytes 변경 def test_resolve_or_run_outer_reuses_decision_zero_calls(tmp_path): """지문 일치 사전 결정 = outer 0콜 동일 winner (resume 재현성).""" from app.modules.pipeline.conti_ab import ( decision_fingerprint, resolve_or_run_outer, ) a, b = _pngs(tmp_path) fp = decision_fingerprint( prompt="P", sel_a=a, sel_b=b, judge_model_physical="SAMPLE-m") def boom(*args, **kw): raise AssertionError("outer judge 호출되면 안 됨") out = resolve_or_run_outer( prompt="P", sel_a=a, sel_b=b, prior_decision={"fingerprint": fp, "winner": "B", "outer": {"winner": "B"}}, judge_model_physical="SAMPLE-m", call_structured_fn=boom, ) assert out["winner"] == "B" and out["judged"] is False # 지문 불일치(모델 변경) = 재판정 calls = [] def fake(step_tag, system, parts, schema, **kw): calls.append(1) return {"winner": "1", "reason_ko": ""} out2 = resolve_or_run_outer( prompt="P", sel_a=a, sel_b=b, prior_decision={"fingerprint": fp, "winner": "B", "outer": {"winner": "B"}}, judge_model_physical="SAMPLE-m2", call_structured_fn=fake, ) assert out2["judged"] is True and len(calls) == 2 assert out2["winner"] == "A" # AB→A/BA→B 불일치=tie→A 정책 def test_resolve_or_run_outer_judge_failure_raises(tmp_path): """사전 결정 없을 때 판정 실패=raise (A 고착 fail-open 제거 — 샷 실패 격리 후 resume 재판정).""" from app.modules.pipeline.conti_ab import resolve_or_run_outer a, b = _pngs(tmp_path) def broken(*args, **kw): raise RuntimeError("SAMPLE transient") with pytest.raises(RuntimeError): resolve_or_run_outer( prompt="P", sel_a=a, sel_b=b, prior_decision=None, judge_model_physical="SAMPLE-m", call_structured_fn=broken, ) def test_scene_image_hash_sensitive_to_physical_ab_judge_model(monkeypatch): """Codex 8e70d4c0 N3: 물리 판정 모델 교체=hash 드리프트 (ON시).""" from app.core.config import settings from app.core.steps.image_steps import SceneImagePipelineStep step = object.__new__(SceneImagePipelineStep) step.project_config = {} monkeypatch.setattr(settings, "still_recipe_mode", "v1") monkeypatch.setattr(settings, "still_conti_ab_enabled", True) monkeypatch.setattr(settings, "gemini_text_model", "SAMPLE-a") h1 = SceneImagePipelineStep._config_hash(step) monkeypatch.setattr(settings, "gemini_text_model", "SAMPLE-b") h2 = SceneImagePipelineStep._config_hash(step) assert h1 != h2 def test_pack_scenario_neutral_and_framing_priority(): repo = Path(__file__).resolve().parents[3] ver = resolve_prompt_version("1") text = (repo / "prompts" / "_base" / "conti_ab_judge" / ver / "judge_system.md").read_text(encoding="utf-8") n = " ".join(text.lower().split()) assert "strict priority order" in n assert "shot text's staging" in n # 블라인드 계약 — 내부 선택지 추측 금지 명시 assert "must not try to guess" in n for word in ("rooftop", "옥탑", "모니터", "conti", "sketch reference"): assert word not in n def test_unknown_version_raises(): with pytest.raises(ValueError): resolve_prompt_version("99")