"""SAFETY 다단 후퇴 사다리 (still_safety_fallback_enabled) 계약 시험.

계약(2026-08-16 사용자 지시 "거부 시 grok 전환, 그래도 안 되면 연화 진행"):
- OFF(default) = 기존 2회(원문 → 연화) 동작 그대로 — 회귀 가드.
- ON = primary → 원 모델 재시도 → 교차 백엔드 → 연화(원 모델) → 연화(교차).
- moderation 밖 오류는 primary 단계에서 즉시 전파(사다리 대상 아님).
- 후퇴 단계 자체 오류(grok 8k 상한 등)는 다음 단계로 넘어간다.
- 연화 저작은 1회만, 교차 클라이언트 생성도 1회만.
- 발화 흔적 = set_context 의 safety_ladder 메타(비 primary 단계만).
"""
from __future__ import annotations

from typing import Any, Dict, List, Optional

import pytest

from app.core.config import settings
from app.modules.pipeline.multiroll_gemini import make_nb2_gen_fn


class FakeClient:
    """gen_fn 계약(set_context/generate_image)만 흉내 — 호출 대본 실행."""

    def __init__(self, script: Optional[List[Any]] = None,
                 name: str = "fake") -> None:
        # script 항목: Exception 이면 raise, bytes 면 반환. 소진 후엔 성공.
        self.script = list(script or [])
        self.name = name
        self.calls: List[Dict[str, Any]] = []
        self._ctx: Dict[str, Any] = {}

    def set_context(self, **kwargs: Any) -> None:
        # ★실클라이언트와 동일한 update 의미(교체 아님) — sticky context
        # 결함(Codex BLOCK-1)을 가리지 않기 위한 필수 조건.
        self._ctx.update(kwargs)

    def generate_image(self, prompt: str, *, labeled_references=None,
                       aspect_ratio: str = "16:9"):
        self.calls.append({
            "prompt": prompt,
            "ctx": dict(self._ctx),
            "client": self.name,
        })
        if self.script:
            item = self.script.pop(0)
            if isinstance(item, Exception):
                raise item
        return b"PNG", 100


class FakeSanitizer:
    def __init__(self, sanitized: Optional[str] = "soft prompt") -> None:
        self.sanitized = sanitized
        self.calls: List[Dict[str, Any]] = []

    def sanitize(self, original_prompt, block_reason, block_categories,
                 attempt=1, **_kw):
        self.calls.append({
            "original": original_prompt, "reason": block_reason,
            "attempt": attempt,
        })
        return {"sanitized_prompt": self.sanitized}


def _moderr() -> RuntimeError:
    return RuntimeError("Content moderation blocked: SAFETY")


def _make(client, sanitizer, **kw):
    return make_nb2_gen_fn(
        project_id="p1", episode_id="e1",
        operation_type="still_recipe_roll",
        gemini_client=client, sanitizer=sanitizer,
        context_extra={"still_id": "sid1"}, **kw)


REFS = [("REF A", b"imgbytes")]


# ── OFF(default): 기존 동작 회귀 가드 ─────────────────────────────────

def test_off_moderation_sanitize_retry_then_success(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "still_safety_fallback_enabled", False)
    client = FakeClient(script=[_moderr()])
    san = FakeSanitizer()
    gen = _make(client, san)
    out = gen("t1", "original", REFS, tmp_path / "o.png")
    assert out.read_bytes() == b"PNG"
    assert [c["prompt"] for c in client.calls] == ["original", "soft prompt"]
    assert len(san.calls) == 1
    # OFF 경로는 safety_ladder 메타를 남기지 않는다
    assert all("safety_ladder" not in c["ctx"] for c in client.calls)


def test_off_moderation_twice_raises(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "still_safety_fallback_enabled", False)
    client = FakeClient(script=[_moderr(), _moderr()])
    gen = _make(client, FakeSanitizer())
    with pytest.raises(RuntimeError, match="moderation"):
        gen("t1", "original", REFS, tmp_path / "o.png")
    assert len(client.calls) == 2


# ── ON: 사다리 ────────────────────────────────────────────────────────

@pytest.fixture()
def ladder_on(monkeypatch):
    monkeypatch.setattr(settings, "still_safety_fallback_enabled", True)
    # 교차 클라이언트 생성을 가짜로 — 생성 횟수·방향 검증용
    import app.modules.llm.gemini_image_client as gic
    import app.modules.llm.grok_image_client as grc

    created: Dict[str, List[FakeClient]] = {"gemini": [], "grok": []}

    class FakeGemini(FakeClient):
        def __init__(self) -> None:
            super().__init__(name="gemini-cross")
            created["gemini"].append(self)

    class FakeGrok(FakeClient):
        def __init__(self) -> None:
            super().__init__(name="grok-cross")
            created["grok"].append(self)

    # ★세 번째 백엔드(seedream) 대역 — 안 끼우면 시험이 **바깥을 진짜로**
    #  부른다(돈 가드가 openrouter 를 잡았다, 2026-09-19).
    created["seedream"] = []

    class FakeSeedream(FakeClient):
        def __init__(self) -> None:
            super().__init__(name="seedream-third")
            created["seedream"].append(self)

    monkeypatch.setattr(gic, "GeminiImageClient", FakeGemini)
    monkeypatch.setattr(grc, "GrokImageClient", FakeGrok)
    monkeypatch.setattr(
        "app.modules.pipeline.cine_provider.build_cine_client",
        lambda name="", **k: FakeSeedream())
    return {"created": created, "FakeGrok": FakeGrok,
            "FakeSeedream": FakeSeedream}


def test_on_primary_success_single_call(tmp_path, ladder_on):
    client = FakeClient()
    gen = _make(client, FakeSanitizer())
    gen("t1", "original", REFS, tmp_path / "o.png")
    assert len(client.calls) == 1
    # primary 는 항상 None 으로 명시 청소(BLOCK-1) — 키는 있되 값 None
    assert client.calls[0]["ctx"].get("safety_ladder") is None
    assert not ladder_on["created"]["grok"]  # 교차 클라이언트 미생성


def test_on_no_stale_stage_leaks_to_next_shot(tmp_path, ladder_on):
    # (Codex BLOCK-1) 공유 클라이언트 + update 형 set_context: 앞 샷이
    # fallback 단계로 회복해도 다음 샷 primary 호출의 단계명은 None 이어야
    # 한다 — 잔류하면 발화 안 한 샷에 거짓 단계가 기록된다.
    client = FakeClient(script=[_moderr()])  # 샷1: retry 단계에서 회복
    gen1 = _make(client, FakeSanitizer())
    gen1("shot1", "original", REFS, tmp_path / "s1.png")
    assert client.calls[-1]["ctx"]["safety_ladder"] == "same_model_retry"
    gen2 = _make(client, FakeSanitizer())  # 같은 클라이언트 재사용
    gen2("shot2", "original", REFS, tmp_path / "s2.png")
    assert client.calls[-1]["ctx"].get("safety_ladder") is None


def test_on_full_ladder_order_then_raise(tmp_path, ladder_on):
    client = FakeClient(script=[_moderr(), _moderr(), _moderr()])
    san = FakeSanitizer()
    gen = _make(client, san)
    # 교차 클라이언트도 전부 거부하게
    with pytest.raises(RuntimeError, match="moderation") as ei:
        # cross 는 fixture 의 FakeGrok — 대본 주입이 안 되므로 성공해
        # 버린다. 거부시키려면 생성 후 스크립트를 심어야 하는데 생성
        # 시점이 사다리 내부라, generate_image 를 클래스 수준에서 거부로
        # 덮는다.
        FakeGrok = ladder_on["FakeGrok"]

        def _always_block(self, prompt, *, labeled_references=None,
                          aspect_ratio="16:9"):
            self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                               "client": self.name})
            raise _moderr()

        FakeGrok.generate_image = _always_block
        ladder_on["FakeSeedream"].generate_image = _always_block
        gen("t1", "original", REFS, tmp_path / "o.png")
    # 원 모델: primary·retry·sanitized 3회
    assert [c["prompt"] for c in client.calls] == [
        "original", "original", "soft prompt"]
    # 교차: cross_backend·cross_sanitized 2회 — 같은 인스턴스 재사용
    assert len(ladder_on["created"]["grok"]) == 1
    cross = ladder_on["created"]["grok"][0]
    assert [c["prompt"] for c in cross.calls] == ["original", "soft prompt"]
    # ★세 번째 백엔드는 **연화 전에** 원문으로 한 번 (2026-09-19).
    assert len(ladder_on["created"]["seedream"]) == 1
    third = ladder_on["created"]["seedream"][0]
    assert [c["prompt"] for c in third.calls] == ["original"], (
        "세 번째 백엔드가 연화된 글을 받았다 — 순서가 뒤집혔다")
    # 연화 저작은 1회만
    assert len(san.calls) == 1
    # 단계 메타: 비 primary 전부 기록
    stages = [c["ctx"].get("safety_ladder")
              for c in client.calls + cross.calls + third.calls]
    assert stages == [None, "same_model_retry", "sanitized",
                      "cross_backend", "cross_sanitized", "third_backend"]
    # 최종 예외는 moderation 사유
    assert "moderation" in str(ei.value).lower()


def test_on_recovery_at_same_model_retry(tmp_path, ladder_on):
    client = FakeClient(script=[_moderr()])
    gen = _make(client, FakeSanitizer())
    out = gen("t1", "original", REFS, tmp_path / "o.png")
    assert out.read_bytes() == b"PNG"
    assert len(client.calls) == 2
    assert client.calls[1]["ctx"]["safety_ladder"] == "same_model_retry"
    assert not ladder_on["created"]["grok"]  # 교차까지 안 갔다


def test_on_cross_stage_incompat_continues_to_sanitized(
        tmp_path, ladder_on):
    # 교차 단계의 무호출·결정론적 비호환(grok 8k 상한)만 삼키고 연화로
    # 넘어간다 — 실제 예외 클래스로 검증(Codex BLOCK-3b)
    from app.modules.llm.grok_image_client import GrokPromptOverBudget

    FakeGrok = ladder_on["FakeGrok"]

    def _over_budget(self, prompt, *, labeled_references=None,
                     aspect_ratio="16:9"):
        self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                           "client": self.name})
        raise GrokPromptOverBudget("grok 텍스트 총량 9000B > 7900B")

    FakeGrok.generate_image = _over_budget
    client = FakeClient(script=[_moderr(), _moderr()])  # primary·retry 거부
    gen = _make(client, FakeSanitizer())
    out = gen("t1", "original", REFS, tmp_path / "o.png")
    assert out.read_bytes() == b"PNG"
    # ★2026-09-19 — 이제 연화보다 **세 번째 백엔드가 먼저**다.
    #  삼키는 규칙(무호출·결정론적 비호환만)은 그대로고, 회복 지점만 바뀐다.
    third = ladder_on["created"]["seedream"][0]
    assert third.calls[-1]["prompt"] == "original", (
        "세 번째 백엔드가 연화된 글을 받았다 — 순서가 뒤집혔다")
    assert third.calls[-1]["ctx"]["safety_ladder"] == "third_backend"
    # 연화까지 안 갔다 — 요청한 것을 지우지 않았다.
    assert "soft prompt" not in [c["prompt"] for c in client.calls]


def test_on_primary_grok_over_budget_propagates_no_cross(
        tmp_path, ladder_on):
    # (Codex R2 BLOCK) moderation 이 한 번도 없는 GrokPromptOverBudget 는
    # 즉시 전파 — primary 가 Grok 이고 프롬프트가 상한 초과인 입력에서
    # 유료 교차(Gemini)가 발화하면 OFF(즉시 실패)와 어긋난다.
    from app.modules.llm.grok_image_client import GrokPromptOverBudget

    FakeGrok = ladder_on["FakeGrok"]
    primary = FakeGrok()
    primary.script = [GrokPromptOverBudget("grok 텍스트 총량 9000B > 7900B")]
    san = FakeSanitizer()
    gen = _make(primary, san)
    with pytest.raises(GrokPromptOverBudget):
        gen("t1", "long original", REFS, tmp_path / "o.png")
    assert len(primary.calls) == 1  # primary 1회에서 끝
    assert not san.calls  # 연화 미진행
    assert not ladder_on["created"]["gemini"]  # 유료 교차 미생성


def test_on_non_moderation_error_mid_ladder_propagates(tmp_path, ladder_on):
    # (Codex BLOCK-3b) moderation 밖 오류(네트워크 등)는 후퇴 단계에서도
    # 즉시 전파 — 삼키면 원인 오독+비용 창이 열린다
    client = FakeClient(script=[_moderr(), RuntimeError("connection reset")])
    san = FakeSanitizer()
    gen = _make(client, san)
    with pytest.raises(RuntimeError, match="connection reset"):
        gen("t1", "original", REFS, tmp_path / "o.png")
    assert len(client.calls) == 2  # primary + retry 까지만
    assert not san.calls  # 유료 연화 미진행
    assert not ladder_on["created"]["grok"]


def test_on_budget_exceeded_mid_ladder_propagates(tmp_path, ladder_on):
    # (Codex BLOCK-3a) 예산 소진은 batch abort 계약 — 어느 단계든 즉시
    # 전파, 후속 단계·유료 연화 진행 금지
    from app.core.image_call_budget import ImageCallBudgetExceeded

    client = FakeClient(
        script=[_moderr(),
                ImageCallBudgetExceeded(cap=1, used=1, source="test")])
    san = FakeSanitizer()
    gen = _make(client, san)
    with pytest.raises(ImageCallBudgetExceeded):
        gen("t1", "original", REFS, tmp_path / "o.png")
    assert len(client.calls) == 2
    assert not san.calls
    assert not ladder_on["created"]["grok"]


def test_on_non_moderation_at_primary_raises_immediately(
        tmp_path, ladder_on):
    client = FakeClient(script=[RuntimeError("connection reset")])
    gen = _make(client, FakeSanitizer())
    with pytest.raises(RuntimeError, match="connection reset"):
        gen("t1", "original", REFS, tmp_path / "o.png")
    assert len(client.calls) == 1
    assert not ladder_on["created"]["grok"]


def test_on_no_sanitizer_stops_after_the_last_provider(tmp_path, ladder_on):
    """★2026-09-19 — 연화 저작기가 없으면 **제공자를 다 써 본 뒤** 멈춘다.

    옛 이름은 `..._stops_after_cross` 였다. 제공자 단계가 하나 늘었으니
    멈추는 자리도 한 칸 뒤다 — 계약이 바뀐 것이지 결함이 아니다.
    """
    FakeGrok = ladder_on["FakeGrok"]

    def _always_block(self, prompt, *, labeled_references=None,
                      aspect_ratio="16:9"):
        self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                           "client": self.name})
        raise _moderr()

    FakeGrok.generate_image = _always_block
    ladder_on["FakeSeedream"].generate_image = _always_block
    client = FakeClient(script=[_moderr(), _moderr()])
    gen = _make(client, None)
    with pytest.raises(RuntimeError, match="moderation"):
        gen("t1", "original", REFS, tmp_path / "o.png")
    assert [c["prompt"] for c in client.calls] == ["original", "original"]
    assert len(ladder_on["created"]["grok"][0].calls) == 1
    assert len(ladder_on["created"]["seedream"][0].calls) == 1


def test_on_empty_sanitized_stops_ladder(tmp_path, ladder_on):
    FakeGrok = ladder_on["FakeGrok"]

    def _always_block(self, prompt, *, labeled_references=None,
                      aspect_ratio="16:9"):
        self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                           "client": self.name})
        raise _moderr()

    FakeGrok.generate_image = _always_block
    # ★세 번째 백엔드도 막아야 연화 단계까지 간다 (2026-09-19).
    ladder_on["FakeSeedream"].generate_image = _always_block
    client = FakeClient(script=[_moderr(), _moderr()])
    san = FakeSanitizer(sanitized="")
    gen = _make(client, san)
    with pytest.raises(RuntimeError, match="moderation"):
        gen("t1", "original", REFS, tmp_path / "o.png")
    assert len(san.calls) == 1
    assert len(client.calls) == 2  # 연화 단계는 안 돌았다
    # 제공자는 다 써 봤다 — 빈 연화 때문에 멈춘 것이지 덜 써서가 아니다.
    assert len(ladder_on["created"]["seedream"][0].calls) == 1


# ── 실클라이언트 projection (Codex BLOCK-1) ──────────────────────────

def test_real_client_projects_safety_ladder_to_db_and_opik_meta():
    # 네트워크 없음 — _log_ctx/_opik_meta property 만 검증
    from app.modules.llm.gemini_image_client import GeminiImageClient

    c = GeminiImageClient(api_key="test-key")
    c.set_context(project_id="p", still_id="s1", multiroll_tag="t_a",
                  safety_ladder="cross_backend")
    assert c._log_ctx["metadata"]["safety_ladder"] == "cross_backend"
    assert c._opik_meta["safety_ladder"] == "cross_backend"
    # primary 의 명시 청소: None 을 실으면 projection 양쪽에서 사라진다
    c.set_context(safety_ladder=None)
    assert "safety_ladder" not in c._log_ctx.get("metadata", {})
    assert "safety_ladder" not in c._opik_meta


# ── 영속 provenance 헬퍼 (Codex BLOCK-2) ─────────────────────────────

def _prov_svc(row):
    from unittest.mock import MagicMock

    from app.services.scene_persistence_service import (
        ScenePersistenceService,
    )

    svc = ScenePersistenceService.__new__(ScenePersistenceService)
    db = MagicMock()
    db.query.return_value.filter.return_value.first.return_value = row
    svc._db = db
    return svc


def test_provenance_none_without_ladder_metadata():
    svc = _prov_svc(("gemini-x", "prompt", '{"still_id": "s1"}'))
    assert svc.safety_ladder_call_provenance("cid") is None
    assert svc.safety_ladder_call_provenance(None) is None
    assert _prov_svc(None).safety_ladder_call_provenance("cid") is None


def test_provenance_returns_actual_model_and_prompt():
    svc = _prov_svc((
        "x-ai/grok-imagine-image-2.0", "soft prompt",
        '{"still_id": "s1", "safety_ladder": "cross_sanitized"}'))
    prov = svc.safety_ladder_call_provenance("cid")
    assert prov == {
        "stage": "cross_sanitized",
        "model_name": "x-ai/grok-imagine-image-2.0",
        "user_prompt": "soft prompt",
    }


def test_on_cross_direction_grok_primary_falls_to_gemini(
        tmp_path, ladder_on):
    # 기본 클라이언트가 (몽키패치된) GrokImageClient 인스턴스면 교차는
    # Gemini 쪽이어야 한다 — 상속 관계(Grok ⊂ Gemini) 순서 가드.
    FakeGrok = ladder_on["FakeGrok"]
    primary = FakeGrok()
    primary.script = [_moderr(), _moderr()]
    gen = _make(primary, FakeSanitizer())
    out = gen("t1", "original", REFS, tmp_path / "o.png")
    assert out.read_bytes() == b"PNG"
    assert len(ladder_on["created"]["gemini"]) == 1
    assert ladder_on["created"]["gemini"][0].calls[0]["ctx"][
        "safety_ladder"] == "cross_backend"


# ── ★세 번째 백엔드의 **운반 오류**가 샷을 죽였다 (2026-09-19 실측) ──────

def test_third_backend_timeout_falls_through_to_the_next_stage(
        tmp_path, ladder_on):
    """실측 그대로 — seedream 이 HTTP 524(시간 초과)를 내자 롤이 죽었다.

    S72sh43: 롤 b 가 제미나이 두 번·grok 한 번 막힌 뒤 seedream 으로 갔고,
    seedream 이 2분 반을 못 버티고 524 를 냈다. 사다리가 검열 오류만 삼키는
    계약이라 그대로 던졌고, **다른 롤이 멀쩡히 성공해 있었는데도 샷 전체가
    실패**했다. 다음 샷은 「앞 샷 선정본 없음」으로 연결이 끊겼다.

    세 번째 백엔드는 **대체 제공자**다 — 응답을 못 하면 「그 대체가 안 됐다」
    이지 샷이 실패한 게 아니다. 다음 계획 단계(연화)로 넘어가야 한다.
    """
    import urllib.error

    FakeGrok = ladder_on["FakeGrok"]

    def _block(self, prompt, *, labeled_references=None, aspect_ratio="16:9"):
        self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                           "client": self.name})
        raise _moderr()

    def _timeout(self, prompt, *, labeled_references=None, aspect_ratio="16:9"):
        self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                           "client": self.name})
        raise urllib.error.HTTPError("https://openrouter.ai/api/v1/images",
                                     524, "timeout", {}, None)

    FakeGrok.generate_image = _block
    ladder_on["FakeSeedream"].generate_image = _timeout
    # primary·retry 거부, 연화 단계에서 성공
    client = FakeClient(script=[_moderr(), _moderr()])
    gen = _make(client, FakeSanitizer())
    out = gen("t1", "original", REFS, tmp_path / "o.png")

    assert out.read_bytes() == b"PNG", "524 에서 멈추면 샷이 죽는다"
    third = ladder_on["created"]["seedream"][0]
    assert len(third.calls) == 1, "세 번째 백엔드는 한 번 시도했다"
    assert client.calls[-1]["ctx"]["safety_ladder"] == "sanitized", (
        "524 뒤에 다음 계획 단계(연화)로 넘어가야 한다")


def test_a_non_transport_error_at_the_third_backend_still_propagates(
        tmp_path, ladder_on):
    """★넘기는 것은 **운반 오류뿐**이다 — 엉뚱한 오류는 그대로 던진다.

    Codex BLOCK-3b 의 계약(검열 밖 오류는 즉시 전파)을 세 번째 단계에서만
    좁게 풀었다. 코드 결함 같은 오류까지 삼키면 원인이 오독된다.
    """
    FakeGrok = ladder_on["FakeGrok"]

    def _block(self, prompt, *, labeled_references=None, aspect_ratio="16:9"):
        self.calls.append({"prompt": prompt, "ctx": dict(self._ctx),
                           "client": self.name})
        raise _moderr()

    def _bug(self, prompt, *, labeled_references=None, aspect_ratio="16:9"):
        raise ValueError("코드 결함")

    FakeGrok.generate_image = _block
    ladder_on["FakeSeedream"].generate_image = _bug
    client = FakeClient(script=[_moderr(), _moderr()])
    gen = _make(client, FakeSanitizer())
    with pytest.raises(ValueError, match="코드 결함"):
        gen("t1", "original", REFS, tmp_path / "o.png")


def test_a_transport_error_outside_the_third_backend_still_propagates(
        tmp_path, ladder_on):
    """★다른 단계의 운반 오류는 그대로다 — 좁힌 것은 세 번째 단계뿐."""
    import urllib.error

    client = FakeClient(script=[urllib.error.HTTPError(
        "https://x", 524, "timeout", {}, None)])
    gen = _make(client, FakeSanitizer())
    with pytest.raises(urllib.error.HTTPError):
        gen("t1", "original", REFS, tmp_path / "o.png")