"""D6 T8 Phase B — endpoint preflight wiring + ordering invariant.

검증 대상:
  - generate_still_image (api/v1/images.py:387) — preflight 가 first signal.
    raise 시 check_scene_images_ready / service 미호출, HTTP 422 +
    STALE_UPSTREAM payload (handler serialization 정합).
  - generate_images batch (api/v1/images.py:598) — 동일 ordering,
    enforce_shot_binding=True 강제.

unit-style mock smoke (T6/T7/T8a 와 동일 패턴):
  - 진짜 SceneStill row + Episode row 는 helper 로 생성 (tests/test_images_api.py
    패턴 — _create_project + _create_episode).
  - check_bg_catalog_freshness / check_scene_images_ready / _make_scene_service 는
    monkeypatch (call tracking + raise injection).
  - 진짜 master_plan + 4 consumer manifest 는 만들지 않음 — preflight 동작은
    test_dispatcher_preflight 에서 이미 27 case cover.
  - 진짜 production E2E (manifest 4 set + scene image generation) 는 T13 wrong-order
    force smoke 에서 별도.

spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.8
plan: T8 Phase B (endpoint wiring + TestClient mock smoke)
"""
from __future__ import annotations

import io
import uuid
from datetime import datetime, timezone
from pathlib import Path

import pytest
from fastapi.testclient import TestClient
from fpdf import FPDF

from app.core.config import settings
from app.core.database import Base, engine, SessionLocal
from app.main import app
from app.models.project import SceneStill
from tests._safety_guards import safe_drop_all, safe_rmtree


# ──────────────────────────────────────────────────────────────────────
# fixtures (test_images_api.py 패턴 mirror)
# ──────────────────────────────────────────────────────────────────────


@pytest.fixture(autouse=True)
def _setup_db():
    Base.metadata.create_all(engine)
    with TestClient(app):
        pass
    yield
    safe_drop_all(engine, Base.metadata)
    proj_dir = Path(settings.projects_dir)
    if proj_dir.exists():
        safe_rmtree(proj_dir)


@pytest.fixture()
def client():
    with TestClient(app, raise_server_exceptions=False) as c:
        yield c


def _make_test_pdf(text="INT. OFFICE - DAY") -> bytes:
    pdf = FPDF()
    pdf.add_page()
    pdf.set_font("Helvetica", size=12)
    pdf.cell(200, 10, text=text)
    return pdf.output()


def _admin_login(client: TestClient):
    resp = client.post(
        "/api/v1/auth/login",
        json={"username": "admin", "password": "admin123"},
    )
    assert resp.status_code == 200


def _create_project(client: TestClient) -> str:
    _admin_login(client)
    resp = client.post("/api/v1/projects/", json={"name": "T8 Preflight Test"})
    assert resp.status_code == 200
    return resp.json()["id"]


def _create_episode(client: TestClient, project_id: str) -> str:
    pdf_bytes = _make_test_pdf()
    resp = client.post(
        f"/api/v1/projects/{project_id}/episodes/",
        data={"episode_number": "1", "title": "Pilot"},
        files={"file": ("pilot.pdf", io.BytesIO(pdf_bytes), "application/pdf")},
    )
    assert resp.status_code == 200
    return resp.json()["id"]


def _insert_scene_still(project_id: str, episode_id: str) -> str:
    """Minimal SceneStill row for endpoint testing."""
    db = SessionLocal()
    try:
        still_id = str(uuid.uuid4())
        still = SceneStill(
            id=still_id,
            project_id=project_id,
            episode_id=episode_id,
            still_index=1,
            screenplay_scene_heading="INT. TEST - DAY",
            scene_index=1,
            shot_index=1,
            shot_description="test shot",
            created_at=datetime.now(timezone.utc).isoformat(),
        )
        db.add(still)
        db.commit()
        return still_id
    finally:
        db.close()


def _raising_preflight(*args, **kwargs):
    """Stub `check_bg_catalog_freshness` that raises STALE_UPSTREAM."""
    from app.core.errors import StaleUpstreamError

    raise StaleUpstreamError(
        upstream="background_render",
        expected_bg_catalog_hash="EXPECTED_HASH",
        observed_bg_catalog_hash="STALE_HASH",
        remediation="POST /steps/background_render?mode=force",
    )


# ──────────────────────────────────────────────────────────────────────
# Single endpoint — generate_still_image
# ──────────────────────────────────────────────────────────────────────


def test_single_endpoint_preflight_raise_blocks_downstream(client, monkeypatch):
    """preflight raise → 422 + STALE_UPSTREAM payload + downstream 미호출."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    scene_ready_calls: list = []
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: scene_ready_calls.append(("scene_ready", a, kw)),
    )

    service_calls: list = []

    class _FakeService:
        def generate_single_scene_image(self, *a, **kw):
            service_calls.append(("service", a, kw))
            return {"id": "should-not-happen"}

    monkeypatch.setattr(
        "app.api.v1.images._make_scene_service",
        lambda *a, **kw: _FakeService(),
    )
    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        _raising_preflight,
    )

    resp = client.post(
        f"/api/v1/projects/{project_id}/stills/{still_id}/generate-image",
        json={},
    )

    assert resp.status_code == 422, resp.text
    body = resp.json()
    assert body["error"]["code"] == "STALE_UPSTREAM"
    assert body["error"]["upstream"] == "background_render"
    assert body["error"]["expected_bg_catalog_hash"] == "EXPECTED_HASH"
    assert body["error"]["observed_bg_catalog_hash"] == "STALE_HASH"
    assert "remediation" in body["error"]
    assert "background_render" in body["error"]["remediation"]

    # ordering invariant — preflight gates downstream
    assert scene_ready_calls == [], (
        f"check_scene_images_ready called despite preflight raise: {scene_ready_calls}"
    )
    assert service_calls == [], (
        f"service called despite preflight raise: {service_calls}"
    )


def test_single_endpoint_preflight_pass_proceeds_to_downstream(
    client, monkeypatch,
):
    """preflight 통과 → check_scene_images_ready + service 정상 호출."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    preflight_calls: list = []

    def _passing_preflight(pid, eid, *, enforce_shot_binding):
        preflight_calls.append((pid, eid, enforce_shot_binding))

    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        _passing_preflight,
    )

    scene_ready_calls: list = []
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: scene_ready_calls.append(("scene_ready", a, kw)),
    )

    service_calls: list = []

    class _FakeService:
        def generate_single_scene_image(self, sid, *, custom_prompt, ip):
            service_calls.append(("service", sid, custom_prompt))
            return {
                "id": "img-fake",
                "asset_type": "scene",
                "file_path": "fake.png",
                "status": "generated",
                "review_notes": "",
                "created_at": datetime.now(timezone.utc).isoformat(),
            }

    monkeypatch.setattr(
        "app.api.v1.images._make_scene_service",
        lambda *a, **kw: _FakeService(),
    )

    resp = client.post(
        f"/api/v1/projects/{project_id}/stills/{still_id}/generate-image",
        json={},
    )

    assert resp.status_code == 200, resp.text
    assert preflight_calls == [(project_id, episode_id, True)], (
        f"preflight called incorrectly: {preflight_calls}"
    )
    assert len(scene_ready_calls) == 1, (
        f"check_scene_images_ready not called: {scene_ready_calls}"
    )
    assert len(service_calls) == 1, (
        f"service not called: {service_calls}"
    )


def test_single_endpoint_custom_prompt_disables_binding_enforcement(
    client, monkeypatch,
):
    """custom_prompt 있으면 enforce_shot_binding=False (binding stale 허용)."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    preflight_calls: list = []

    def _passing_preflight(pid, eid, *, enforce_shot_binding):
        preflight_calls.append((pid, eid, enforce_shot_binding))

    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        _passing_preflight,
    )
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: None,
    )

    class _FakeService:
        def generate_single_scene_image(self, sid, *, custom_prompt, ip):
            return {
                "id": "img-fake",
                "asset_type": "scene",
                "file_path": "fake.png",
                "status": "generated",
                "review_notes": "",
                "created_at": datetime.now(timezone.utc).isoformat(),
            }

    monkeypatch.setattr(
        "app.api.v1.images._make_scene_service",
        lambda *a, **kw: _FakeService(),
    )

    resp = client.post(
        f"/api/v1/projects/{project_id}/stills/{still_id}/generate-image",
        json={"custom_prompt": "operator one-off prompt"},
    )

    assert resp.status_code == 200, resp.text
    assert preflight_calls == [(project_id, episode_id, False)], (
        f"custom_prompt 가 enforce_shot_binding=False 로 전달 안 됨: "
        f"{preflight_calls}"
    )


@pytest.mark.parametrize("blank_value", ["", "   ", "\t\n  "])
def test_single_endpoint_blank_custom_prompt_normalized_to_automatic_path(
    client, monkeypatch, blank_value,
):
    """blank/whitespace-only custom_prompt → endpoint 가 None 로 normalize.

    review iter1 BLOCKING: 이전에는 `body.custom_prompt is None` identity check 만
    써서 "" 가 enforce_shot_binding=False 로 흘러 binding 강제 우회됐음. 동시에
    service 의 `if custom_prompt:` truthy check 는 "" 를 automatic path 로 처리 →
    binding-stale generation 허용. fix: endpoint 가 strip + empty → None 로 정규화
    후 enforce_shot_binding 과 service 호출 양쪽에 일관 사용.
    """
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    preflight_calls: list = []

    def _passing_preflight(pid, eid, *, enforce_shot_binding):
        preflight_calls.append((pid, eid, enforce_shot_binding))

    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        _passing_preflight,
    )
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: None,
    )

    service_calls: list = []

    class _FakeService:
        def generate_single_scene_image(self, sid, *, custom_prompt, ip):
            service_calls.append(("service", sid, custom_prompt))
            return {
                "id": "img-fake",
                "asset_type": "scene",
                "file_path": "fake.png",
                "status": "generated",
                "review_notes": "",
                "created_at": datetime.now(timezone.utc).isoformat(),
            }

    monkeypatch.setattr(
        "app.api.v1.images._make_scene_service",
        lambda *a, **kw: _FakeService(),
    )

    resp = client.post(
        f"/api/v1/projects/{project_id}/stills/{still_id}/generate-image",
        json={"custom_prompt": blank_value},
    )

    assert resp.status_code == 200, resp.text
    # blank → 자동 경로 = enforce_shot_binding True (binding 강제 우회 차단)
    assert preflight_calls == [(project_id, episode_id, True)], (
        f"blank custom_prompt {blank_value!r} 가 enforce_shot_binding=True 로 호출 안 됨: "
        f"{preflight_calls}"
    )
    # service 에도 None 정규화된 값이 전달됨 (fallback path 의 의미론 정합)
    assert len(service_calls) == 1
    assert service_calls[0][2] is None, (
        f"blank custom_prompt 가 service 에 None 으로 정규화 안 됨: {service_calls[0][2]!r}"
    )


def test_single_endpoint_skips_preflight_when_still_not_found(
    client, monkeypatch,
):
    """still 미존재 → preflight skip (episode_id 도출 불가, 기존 404 path)."""
    project_id = _create_project(client)

    preflight_calls: list = []
    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        lambda *a, **kw: preflight_calls.append(("preflight", a, kw)),
    )
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: None,
    )

    # still 부재 — service 가 자체 404 raise 또는 invalid still_id 처리.
    resp = client.post(
        f"/api/v1/projects/{project_id}/stills/nonexistent-still/generate-image",
        json={},
    )

    # preflight 절대 호출 X (still 부재 → episode_id 도출 불가)
    assert preflight_calls == [], (
        f"preflight called for nonexistent still: {preflight_calls}"
    )


# ──────────────────────────────────────────────────────────────────────
# Batch endpoint — generate_images
# ──────────────────────────────────────────────────────────────────────


def test_batch_endpoint_preflight_raise_blocks_downstream(client, monkeypatch):
    """batch endpoint preflight raise → 422 + downstream (scene_ready /
    asset_readiness / submit_background_job) 미호출."""
    import app.api.v1.images as images_mod

    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)

    # gemini key 검증 통과 (test_generate_images_no_gemini_key 패턴)
    monkeypatch.setattr(images_mod.settings, "gemini_api_key", "test-key")

    scene_ready_calls: list = []
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: scene_ready_calls.append(("scene_ready", a, kw)),
    )

    asset_readiness_calls: list = []
    monkeypatch.setattr(
        "app.core.asset_readiness.assert_episode_asset_readiness",
        lambda *a, **kw: asset_readiness_calls.append(("asset_readiness", a)),
    )

    submit_calls: list = []
    monkeypatch.setattr(
        "app.api.v1.images.submit_background_job",
        lambda **kw: submit_calls.append(("submit", kw)) or True,
    )

    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        _raising_preflight,
    )

    resp = client.post(
        f"/api/v1/projects/{project_id}/episodes/{episode_id}/generate-images",
    )

    assert resp.status_code == 422, resp.text
    body = resp.json()
    assert body["error"]["code"] == "STALE_UPSTREAM"
    assert body["error"]["upstream"] == "background_render"

    assert scene_ready_calls == [], (
        f"check_scene_images_ready called despite preflight raise: {scene_ready_calls}"
    )
    assert asset_readiness_calls == [], (
        f"assert_episode_asset_readiness called: {asset_readiness_calls}"
    )
    assert submit_calls == [], (
        f"submit_background_job called: {submit_calls}"
    )


def test_batch_endpoint_preflight_always_enforces_shot_binding(
    client, monkeypatch,
):
    """batch endpoint 는 enforce_shot_binding=True 강제 (custom_prompt 의미 없음)."""
    import app.api.v1.images as images_mod

    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)

    monkeypatch.setattr(images_mod.settings, "gemini_api_key", "test-key")

    preflight_calls: list = []

    def _passing_preflight(pid, eid, *, enforce_shot_binding):
        preflight_calls.append((pid, eid, enforce_shot_binding))

    monkeypatch.setattr(
        "app.services.dispatcher_preflight.check_bg_catalog_freshness",
        _passing_preflight,
    )
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: None,
    )
    monkeypatch.setattr(
        "app.core.asset_readiness.assert_episode_asset_readiness",
        lambda *a, **kw: None,
    )
    monkeypatch.setattr(
        "app.api.v1.images.submit_background_job",
        lambda **kw: True,
    )

    resp = client.post(
        f"/api/v1/projects/{project_id}/episodes/{episode_id}/generate-images",
    )

    assert resp.status_code == 200, resp.text
    assert preflight_calls == [(project_id, episode_id, True)], (
        f"batch preflight 가 enforce_shot_binding=True 로 호출 안 됨: "
        f"{preflight_calls}"
    )