"""Phase C — Task C0 토대 결정론 테스트.

두 책임을 잠근다:
  1. queue.flush 가 ctx.scene_index 를 pipeline_metadata_json["scene_index"] 로
     non-overwrite fold (ImageAsset 에 scene_index 컬럼이 없어 still-less 중간물의
     씬 grouping 키가 필요). 사이트 명시 metadata 가 우선(setdefault).
  2. call_gpt_image_bytes / generate_floor_plan_image 의 optional
     capture_input_image_ids threading (default None = byte-identical, 명시 전달 시
     capture_generated_image 까지 도달).

결정론만 TDD. PG theroad_test, conftest 가 스키마 생성(create_all). 격리는 uuid 프로젝트.
"""

import base64
import json
import uuid
from datetime import datetime, timezone
from types import SimpleNamespace

from app.core.database import SessionLocal
from app.models.catalog import ProjectRegistry, UserAccount
from app.models.project import Episode, ImageAsset
from app.services.image_capture.context import GenerationContext, generation_context
from app.services.image_capture.queue import CaptureQueue
from app.services.image_capture.spool import write_spool


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _seed_base(db) -> tuple[str, str]:
    uid = str(uuid.uuid4())
    db.add(UserAccount(
        id=uid, username=f"u_{uid[:8]}", display_name="t", password_hash="x",
        role="admin", created_at=_now(), updated_at=_now(),
    ))
    db.commit()
    pid = str(uuid.uuid4())
    db.add(ProjectRegistry(
        id=pid, name="Phase C Test", created_by=uid, created_at=_now(), updated_at=_now(),
    ))
    eid = str(uuid.uuid4())
    db.add(Episode(
        id=eid, project_id=pid, episode_number=1, title="Pilot",
        source_filename="p.pdf", source_path="p.pdf", created_at=_now(), updated_at=_now(),
    ))
    db.commit()
    return pid, eid


def _ctx(pid, eid, *, stage="zoom_continuity", scene_index=None) -> GenerationContext:
    return GenerationContext(
        project_id=pid, episode_id=eid, stage=stage, queue=None,  # type: ignore[arg-type]
        scene_index=scene_index,
    )


def _intermediate_rows(pid):
    db = SessionLocal()
    try:
        return db.query(ImageAsset).filter(
            ImageAsset.project_id == pid, ImageAsset.is_intermediate == True  # noqa: E712
        ).all()
    finally:
        db.close()


# ───────────────────────── 1. queue scene_index fold ─────────────────────────

def test_flush_folds_scene_index_merging_with_existing_metadata():
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    ctx = _ctx(pid, eid, scene_index=12)
    q = CaptureQueue(ctx)
    spool = write_spool(b"PNGBYTES", pid, "zoom_continuity")
    q.append(spool, {
        "pipeline_role": "zoom_continuity_crop",
        "pipeline_metadata": {"group_id": "g1"},
    })
    assert q.flush() == 1

    rows = _intermediate_rows(pid)
    assert len(rows) == 1
    pm = json.loads(rows[0].pipeline_metadata_json)
    assert pm == {"group_id": "g1", "scene_index": 12}


def test_flush_folds_scene_index_when_no_site_metadata():
    """still-less 중간물(aerial base 등): metadata 가 유일 grouping 키 → scene_index 만 fold."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    ctx = _ctx(pid, eid, stage="outdoor_site_layout", scene_index=7)
    q = CaptureQueue(ctx)
    spool = write_spool(b"PNGBYTES", pid, "outdoor_site_layout")
    q.append(spool, {"pipeline_role": "outdoor_aerial_base"})  # pipeline_metadata 없음
    assert q.flush() == 1

    rows = _intermediate_rows(pid)
    assert len(rows) == 1
    assert json.loads(rows[0].pipeline_metadata_json) == {"scene_index": 7}


def test_flush_scene_index_none_preserves_existing_behavior():
    """scene_index None 이면 키 부재 — 기존 동작 보존(metadata 없으면 json None)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    # (a) metadata 없음 + scene_index None → pipeline_metadata_json None
    ctx = _ctx(pid, eid, scene_index=None)
    q = CaptureQueue(ctx)
    q.append(write_spool(b"X", pid, "zoom_continuity"), {"pipeline_role": "zoom_continuity_crop"})
    assert q.flush() == 1
    rows = _intermediate_rows(pid)
    assert len(rows) == 1
    assert rows[0].pipeline_metadata_json is None

    # (b) metadata 있음 + scene_index None → 그대로 보존(scene_index 키 없음)
    ctx2 = _ctx(pid, eid, scene_index=None)
    q2 = CaptureQueue(ctx2)
    q2.append(write_spool(b"Y", pid, "zoom_continuity"),
              {"pipeline_role": "r", "pipeline_metadata": {"x": 1}})
    assert q2.flush() == 1
    pms = [json.loads(r.pipeline_metadata_json) for r in _intermediate_rows(pid)
           if r.pipeline_metadata_json is not None]
    assert {"x": 1} in pms
    assert all("scene_index" not in pm for pm in pms)


def test_flush_scene_index_fold_is_non_overwrite():
    """사이트가 명시한 scene_index 가 ctx 값보다 우선(setdefault)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    ctx = _ctx(pid, eid, scene_index=12)
    q = CaptureQueue(ctx)
    q.append(write_spool(b"Z", pid, "zoom_continuity"),
             {"pipeline_role": "r", "pipeline_metadata": {"scene_index": 99}})
    assert q.flush() == 1
    rows = _intermediate_rows(pid)
    assert len(rows) == 1
    assert json.loads(rows[0].pipeline_metadata_json)["scene_index"] == 99


# ──────────────── 2. capture_input_image_ids threading (primitive) ────────────────

def _recording_client(decoded: bytes = b"\x89PNG\r\n" + b"\x00" * 4096):
    b64 = base64.b64encode(decoded).decode("ascii")
    calls = []

    class _Images:
        def generate(self, **kwargs):
            assert "image" not in kwargs
            calls.append(("generate", dict(kwargs)))
            return SimpleNamespace(data=[SimpleNamespace(b64_json=b64)])

        def edit(self, **kwargs):
            kwargs.pop("image")
            calls.append(("edit", dict(kwargs)))
            return SimpleNamespace(data=[SimpleNamespace(b64_json=b64)])

    return SimpleNamespace(images=_Images()), calls


def test_primitive_capture_input_image_ids_default_none_byte_identical():
    from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes

    client, calls = _recording_client()
    with generation_context("p-c0", "e-c0", "zoom_continuity") as q:
        call_gpt_image_bytes(
            client, mode="generate", prompt="p", ref_paths=None,
            call_kwargs={"model": "gpt-image-2.5-sunburst", "size": "1024x1024", "quality": "high", "n": 1},
            capture_role="zoom_continuity_crop",
        )
        _spool, meta = q._items[0]
        assert meta["input_image_ids"] is None  # default = None

    # default None 은 images.generate kwargs 에 input_image_ids 를 누출하지 않음(byte-identical)
    _method, kwargs = calls[0]
    assert "input_image_ids" not in kwargs
    assert kwargs == {"model": "gpt-image-2.5-sunburst", "size": "1024x1024", "quality": "high", "n": 1, "prompt": "p"}


def test_primitive_capture_input_image_ids_threads_to_capture():
    from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes

    client, calls = _recording_client()
    with generation_context("p-c0b", "e-c0b", "zoom_continuity") as q:
        call_gpt_image_bytes(
            client, mode="generate", prompt="p", ref_paths=None,
            call_kwargs={"model": "gpt-image-2.5-sunburst", "size": "1024x1024"},
            capture_role="zoom_continuity_crop",
            capture_input_image_ids=["asset-1", "asset-2"],
        )
        _spool, meta = q._items[0]
        assert meta["input_image_ids"] == ["asset-1", "asset-2"]
    # 입력 UUID 는 capture meta 로만 — images API kwargs 에는 안 들어감
    _method, kwargs = calls[0]
    assert "input_image_ids" not in kwargs


# ──────────── 2b. capture_input_image_ids threading (generate_floor_plan_image) ────────────

def test_floor_plan_image_threads_input_image_ids():
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    client, _calls = _recording_client()
    with generation_context("p-c0c", "e-c0c", "registered_pose_guide") as q:
        generate_floor_plan_image(
            prompt="draw pose guide", openai_client=client, ref_paths=None,
            model="gpt-image-2.5-sunburst", size="1024x1024", quality="high",
            capture_role="registered_pose_guide",
            capture_input_image_ids=["bg-asset-1"],
        )
        _spool, meta = q._items[0]
        assert meta["pipeline_role"] == "registered_pose_guide"
        assert meta["input_image_ids"] == ["bg-asset-1"]


def test_floor_plan_image_fixed_budget_meta_wins_over_caller_extra():
    """고정 메타(budget_source/ref_count)는 caller capture_extra_metadata 가 못 덮는다
    (budget label 불변 규칙 — Codex minor 보강)."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    client, _calls = _recording_client()
    with generation_context("p-c0e", "e-c0e", "registered_pose_guide") as q:
        generate_floor_plan_image(
            prompt="p", openai_client=client, ref_paths=None,
            model="gpt-image-2.5-sunburst", size="1024x1024", quality="high",
            capture_role="registered_pose_guide",
            capture_extra_metadata={"budget_source": "HACKED", "ref_count": 999,
                                    "group_id": "g1"},
        )
        _spool, meta = q._items[0]
        pm = meta["pipeline_metadata"]
        assert pm["budget_source"] == "location_floor_plan.generate"  # 고정 메타 우선
        assert pm["ref_count"] == 0
        assert pm["group_id"] == "g1"  # 그 외 extra 는 보존


def test_floor_plan_image_default_input_image_ids_none():
    """default None — 기존 호출자(role 미지정/input 미지정) byte-identical."""
    from app.modules.pipeline.location_floor_plan import generate_floor_plan_image

    client, calls = _recording_client()
    with generation_context("p-c0d", "e-c0d", "floor_plan") as q:
        generate_floor_plan_image(
            prompt="p", openai_client=client, ref_paths=None,
            model="gpt-image-2.5-sunburst", size="1024x1024", quality="high",
            capture_role="floor_plan_image",
        )
        _spool, meta = q._items[0]
        assert meta["input_image_ids"] is None
    _method, kwargs = calls[0]
    assert kwargs == {"model": "gpt-image-2.5-sunburst", "prompt": "p", "size": "1024x1024", "quality": "high", "n": 1}