"""Task A4 — CaptureQueue.flush (DB insert + rename, 독립 세션, diagnostics) 테스트.

PG theroad_test. 스키마 생성은 conftest session-autouse `_ensure_d6_test_db_schema`
가 Base.metadata.create_all(checkfirst=True) 로 담당(모델 컬럼 자동 반영). 격리는
매 테스트 고유 uuid 프로젝트 + project_id 스코프 조회.
"""

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

import pytest

from app.core.config import settings
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 import queue as queue_mod
from app.services.image_capture.context import GenerationContext
from app.services.image_capture.queue import CAPTURE_DIAG, 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="Capture 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="pose_guide") -> GenerationContext:
    # queue 는 flush 시점에만 ctx 참조 — 여기선 placeholder queue 로 ctx 구성 후
    # 실제 CaptureQueue(ctx) 로 교체할 필요 없음(테스트는 CaptureQueue 직접 생성).
    return GenerationContext(
        project_id=pid, episode_id=eid, stage=stage, queue=None,  # type: ignore[arg-type]
    )


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

    ctx = _ctx(pid, eid)
    q = CaptureQueue(ctx)
    spool = write_spool(b"PNGBYTES", pid, "pose_guide")
    q.append(spool, {
        "pipeline_role": "pose_guide",
        "input_image_ids": ["bg-1", "base-2"],
        "generation_call_id": "call-1",
        "candidate_index": 0,
        "pipeline_metadata": {"group_id": "g1"},
    })

    n = q.flush()
    assert n == 1
    # spool 파일은 final 로 rename — 사라짐
    assert not Path(spool).exists()

    db = SessionLocal()
    try:
        rows = db.query(ImageAsset).filter(
            ImageAsset.project_id == pid, ImageAsset.is_intermediate == True  # noqa: E712
        ).all()
        assert len(rows) == 1
        row = rows[0]
        assert row.is_intermediate is True
        assert row.pipeline_role == "pose_guide"
        assert row.stage == "pose_guide"
        assert json.loads(row.input_image_ids) == ["bg-1", "base-2"]
        assert row.generation_call_id == "call-1"
        assert row.candidate_index == 0
        assert json.loads(row.pipeline_metadata_json) == {"group_id": "g1"}
        assert row.status == "generated"
        assert row.episode_id == eid
        # file_path 는 ImagePathType 으로 read 시 절대화 — 실재해야 함
        assert Path(row.file_path).exists()
        assert Path(row.file_path).read_bytes() == b"PNGBYTES"
        # reference_image_ids 오염 금지 — 기본값 유지(input 은 input_image_ids 로)
        assert "bg-1" not in (row.reference_image_ids or "")
    finally:
        db.close()


def test_flush_persists_disposition_and_attempt_index(monkeypatch):
    """Wave2 — disposition/attempt_index 가 거부본/재시도 메타로 영속화된다."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    ctx = _ctx(pid, eid)
    q = CaptureQueue(ctx)
    spool = write_spool(b"REJECTED", pid, "pose_guide")
    q.append(spool, {
        "pipeline_role": "i2i_candidate",
        "disposition": "rejected",
        "attempt_index": 2,
    })
    assert q.flush() == 1

    db = SessionLocal()
    try:
        row = db.query(ImageAsset).filter(
            ImageAsset.project_id == pid, ImageAsset.is_intermediate == True  # noqa: E712
        ).one()
        assert row.pipeline_role == "i2i_candidate"
        assert row.disposition == "rejected"
        assert row.attempt_index == 2
    finally:
        db.close()


def test_flush_disposition_defaults_none(monkeypatch):
    """disposition/attempt_index 미지정 시 NULL (기존 capture 경로 byte-identical)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    ctx = _ctx(pid, eid)
    q = CaptureQueue(ctx)
    spool = write_spool(b"X", pid, "pose_guide")
    q.append(spool, {"pipeline_role": "pose_guide"})
    assert q.flush() == 1

    db = SessionLocal()
    try:
        row = db.query(ImageAsset).filter(
            ImageAsset.project_id == pid, ImageAsset.is_intermediate == True  # noqa: E712
        ).one()
        assert row.disposition is None
        assert row.attempt_index is None
    finally:
        db.close()


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

    ctx = _ctx(pid, eid)
    q = CaptureQueue(ctx)
    spool = write_spool(b"X", pid, "pose_guide")
    q.append(spool, {"pipeline_role": "pose_guide"})
    assert q.flush() == 1
    # 두 번째 flush — 큐 비었으므로 0
    assert q.flush() == 0


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

    before = CAPTURE_DIAG["flush_errors"]

    # SessionLocal 을 monkeypatch 해 즉시 raise → flush 가 삼키고 diag 증가
    class _Boom:
        def __call__(self):
            raise RuntimeError("db down")

    monkeypatch.setattr(queue_mod, "SessionLocal", _Boom())

    ctx = _ctx(pid, eid)
    q = CaptureQueue(ctx)
    spool = write_spool(b"X", pid, "pose_guide")
    q.append(spool, {"pipeline_role": "pose_guide"})

    # raise 안 함 (non-fatal)
    n = q.flush()
    assert n == 0
    assert CAPTURE_DIAG["flush_errors"] == before + 1


def test_safe_segment_blocks_path_traversal():
    """stage 가 path 세그먼트로 탈출하지 않게 정화(Codex BLOCKING 2)."""
    from app.services.image_capture.queue import _generated_dir, _safe_segment

    assert _safe_segment("pose_guide") == "pose_guide"
    assert _safe_segment("../bad/stage") == ".._bad_stage"  # 슬래시→_, 단일 세그먼트
    assert _safe_segment("a/b\\c") == "a_b_c"
    assert _safe_segment("..") == "unknown"
    assert _safe_segment(".") == "unknown"
    assert _safe_segment("") == "unknown"
    assert _safe_segment(None) == "unknown"
    # 악성 stage 도 항상 generated root 바로 아래 1뎁스(상위 탈출 없음)
    malicious = _generated_dir("pid", "eid", "../../etc")
    gen_root = _generated_dir("pid", "eid", "x").parent
    assert malicious.parent == gen_root


def test_flush_calls_register_models(monkeypatch):
    """flush 가 ORM 사용 전 register_models 를 호출(FK 미해결 drop 방지, BLOCKING 1)."""
    db = SessionLocal()
    try:
        pid, eid = _seed_base(db)
    finally:
        db.close()

    import app.core.database as db_mod

    called = {"n": 0}
    _orig = db_mod.register_models

    def _spy():
        called["n"] += 1
        return _orig()

    monkeypatch.setattr(db_mod, "register_models", _spy)

    ctx = _ctx(pid, eid)
    q = CaptureQueue(ctx)
    spool = write_spool(b"X", pid, "pose_guide")
    q.append(spool, {"pipeline_role": "pose_guide"})

    assert q.flush() == 1
    assert called["n"] >= 1  # flush 가 register_models 를 호출했다