"""D6 T10 — HTTP response shape integration (real preflight + TestClient).

T8 Phase B (`tests/api/test_d6_endpoint_preflight.py`) 는 ``check_bg_catalog_freshness``
자체를 monkeypatch 하여 endpoint↔preflight wiring + ordering invariant 만 검증.
T10 은 real preflight 의 file I/O (master_plan + 4 consumer manifest 직접 기록 →
실제 hash 비교) 를 통과하는 full chain 422 shape 을 검증한다.

검증 시나리오:
  (a) catalog stale → 422 + STALE_UPSTREAM payload (code/upstream/expected/observed/
      remediation/force) full shape.
  (b) custom_prompt + binding-only stale + catalog OK → 200 (binding skip path,
      service mock 으로 service.generate_single_scene_image 가 정상 결과 반환).
  (c) custom_prompt + catalog stale → 422 (custom_prompt 도 catalog freshness 강제).

unit-style pivot (T6/T7/T8 와 동일 결정) — d6_episode_with_stale_* fixture 미존재.

spec: docs/superpowers/specs/2026-05-09-deterministic-bg-id-and-catalog-lineage.md §4.8
plan: T10 (unit-style pivot — tests/api/ + tmp manifest)
"""
from __future__ import annotations

import io
import json
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


_CATALOG_HASH = "cat_001"
_BINDING_HASH = "bind_001"
_CONSUMERS = (
    "floor_plan_prompt",
    "background_prompt",
    "background_render",
    "scene_detail",
)


# ──────────────────────────────────────────────────────────────────────
# fixtures (T8 Phase B + T8a 결합)
# ──────────────────────────────────────────────────────────────────────


@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": "T10 Shape 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:
    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 _write_manifest(pid, eid, step, data, schema_version=1):
    """settings.projects_dir 의 checkpoint 경로에 manifest 직접 기록."""
    p = (
        Path(settings.projects_dir) / pid / "checkpoints"
        / "episodes" / eid / step / "manifest.json"
    )
    p.parent.mkdir(parents=True, exist_ok=True)
    payload = {"schema_version": schema_version, "data": data}
    p.write_text(json.dumps(payload), encoding="utf-8")


def _setup_d6_manifests(pid, eid, *,
                         catalog_hash=_CATALOG_HASH,
                         binding_hash=_BINDING_HASH,
                         consumer_overrides=None):
    """master_plan + 4 consumer manifest 작성. consumer_overrides 로 특정 consumer
    의 stale 조건 시뮬레이션 (e.g. {'background_render': {'consumed_bg_catalog_hash':
    'STALE'}})."""
    _write_manifest(pid, eid, "background_master_plan", {
        "bg_catalog_hash": catalog_hash,
        "shot_binding_hash": binding_hash,
        "background_catalog": {},
        "shot_background_map": {},
    })
    overrides = consumer_overrides or {}
    for step in _CONSUMERS:
        consumer_data = {
            "consumed_bg_catalog_hash": catalog_hash,
            "consumed_shot_binding_hash": binding_hash,
        }
        consumer_data.update(overrides.get(step, {}))
        _write_manifest(pid, eid, step, consumer_data)


def _fake_image_response(project_id):
    return {
        "id": "img-fake",
        "asset_type": "scene",
        "file_path": "fake.png",
        "status": "generated",
        "review_notes": "",
        "created_at": datetime.now(timezone.utc).isoformat(),
    }


# ──────────────────────────────────────────────────────────────────────
# (a) catalog stale — 422 + full payload shape
# ──────────────────────────────────────────────────────────────────────


def test_stale_upstream_response_shape_full_payload(client, monkeypatch):
    """catalog stale → 422 + STALE_UPSTREAM payload 모든 required field 존재."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    # background_render 의 catalog hash 만 stale
    _setup_d6_manifests(project_id, episode_id, consumer_overrides={
        "background_render": {
            "consumed_bg_catalog_hash": "STALE_HASH_xyz",
            "consumed_shot_binding_hash": _BINDING_HASH,
        },
    })

    # check_scene_images_ready 는 preflight 통과 후 호출 — preflight raise 시 미도달
    monkeypatch.setattr(
        "app.core.pipeline_gate.check_scene_images_ready",
        lambda *a, **kw: None,
    )

    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()
    err = body["error"]
    assert err["code"] == "STALE_UPSTREAM"
    assert err["upstream"] == "background_render"
    assert err["expected_bg_catalog_hash"] == _CATALOG_HASH
    assert err["observed_bg_catalog_hash"] == "STALE_HASH_xyz"
    assert "remediation" in err
    assert "force" in err["remediation"].lower()
    assert "message" in err


# ──────────────────────────────────────────────────────────────────────
# (b) custom_prompt + binding-only stale + catalog OK → 200
# ──────────────────────────────────────────────────────────────────────


def test_custom_prompt_passes_with_binding_only_stale(client, monkeypatch):
    """custom_prompt path 는 binding stale 허용 — preflight 통과 → service 정상 호출."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    # scene_detail 의 binding 만 stale, catalog 는 OK
    _setup_d6_manifests(project_id, episode_id, consumer_overrides={
        "scene_detail": {
            "consumed_bg_catalog_hash": _CATALOG_HASH,
            "consumed_shot_binding_hash": "STALE_BINDING",
        },
    })

    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 _fake_image_response(project_id)

    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


# ──────────────────────────────────────────────────────────────────────
# (c) custom_prompt + catalog stale → 422 (catalog 는 항상 강제)
# ──────────────────────────────────────────────────────────────────────


def test_custom_prompt_blocked_on_catalog_stale(client, monkeypatch):
    """custom_prompt 도 catalog freshness 는 강제 — binding skip 만 허용."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    # background_render catalog 만 stale, binding 은 OK
    _setup_d6_manifests(project_id, episode_id, consumer_overrides={
        "background_render": {
            "consumed_bg_catalog_hash": "STALE_CATALOG",
            "consumed_shot_binding_hash": _BINDING_HASH,
        },
    })

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

    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 == 422, resp.text
    body = resp.json()
    assert body["error"]["code"] == "STALE_UPSTREAM"
    assert body["error"]["upstream"] == "background_render"
    assert body["error"]["observed_bg_catalog_hash"] == "STALE_CATALOG"


# ──────────────────────────────────────────────────────────────────────
# binding mismatch upstream 별 422 (default path — enforce_shot_binding=True)
# ──────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize("stale_step", _CONSUMERS)
def test_binding_mismatch_returns_422_with_upstream_per_consumer(
    client, monkeypatch, stale_step,
):
    """4 consumer 중 어느 하나의 binding hash 가 stale → upstream field 가 해당 consumer."""
    project_id = _create_project(client)
    episode_id = _create_episode(client, project_id)
    still_id = _insert_scene_still(project_id, episode_id)

    _setup_d6_manifests(project_id, episode_id, consumer_overrides={
        stale_step: {
            "consumed_bg_catalog_hash": _CATALOG_HASH,
            "consumed_shot_binding_hash": "STALE_BINDING_xyz",
        },
    })

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

    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()
    err = body["error"]
    assert err["code"] == "STALE_UPSTREAM"
    assert err["upstream"] == stale_step
    assert err["expected_shot_binding_hash"] == _BINDING_HASH
    assert err["observed_shot_binding_hash"] == "STALE_BINDING_xyz"