"""사람 대표가 있을 때 **JIT 저장 제어**를 실제 루프로 두 번 태운다.

2026-09-20 Codex 종료 조건. 앞선 시험들은 `auto_set_primary` 나 소스
문자열만 봤고, **서비스 루프를 두 번 돌린** 시험이 없었다.

## 이 시험이 잠그는 것

  · 신규 산출 G 를 **한 번만** 저장하고, 다음 방문은 그 G 를 재사용
  · 사람 대표 H 가 있어도 비교 대상은 **파이프라인 산출**이다
  · 재사용으로 끝나는 방문도 **CP 대표를 실제 대표(H)로** 한 번 맞춘다
  · 그 다음 무변경 방문은 **CP 도 무기록**

## 이 시험이 **재지 않는** 것 (범위 정정, Codex)

  · **prev 계보** — 샷이 하나뿐이라 후속 샷이 없다
  · **대표 선택 자체의 정당성** — `auto_set_primary` 시험이 담당한다
  · **새 이미지 파일 수·provider 무구매** — 생성 경계가 대역이다.
    이것은 **JIT 저장 제어**의 무료 검증이지 캐시 전체의 증명이 아니다

★DB 대역은 **저장에 따라 상태가 바뀐다**. 조회 조건은 열·연산자·값을
 **구조로** 확인한다 — bind 값만 보면 조건을 뒤집어도 통과한다.
"""
from __future__ import annotations

import json
from unittest.mock import MagicMock, patch

import pytest

from app.services.image_service_helpers import MANUAL_UPLOAD_PROMPT

from tests.services.test_still_cine_stage import (  # 본보기 재사용
    EID,
    PID,
    _FLAG_PATCHES,
    _still,
    _write_cp,
)

TAG = "S1sh1"
STILL_ID = "still-1"


class _Asset:
    def __init__(self, id_, path, *, prompt, primary):
        self.id = id_
        self.file_path = str(path)
        self.prompt_used = prompt
        self.is_primary = primary
        self.still_id = STILL_ID


class _Q:
    """조회를 **구조로** 가른다 — 열·연산자·값을 다 본다.

    ★첫 판은 bind 값이 `uploaded` 인지만 봤다. 그러면 실제 조건을
     `!=` 에서 `==` 로 **뒤집어도 대역이 계속 G 를 준다** — 시험이 결함을
     못 잡는다(Codex). 그래서 왼쪽 열·연산자·오른쪽 값을 함께 본다.
    """

    def __init__(self, db):
        self._db = db
        self._non_upload = False
        self._primary = False

    @staticmethod
    def _parts(a):
        """(왼쪽 열 이름, 연산자 이름, 오른쪽 bind 값) — 모르면 None."""
        left = getattr(getattr(a, "left", None), "key", None)
        op = getattr(getattr(a, "operator", None), "__name__", None)
        val = getattr(getattr(a, "right", None), "value", None)
        return left, op, val

    def filter(self, *args):
        for a in args:
            left, op, val = self._parts(a)
            if left == "prompt_used":
                if op == "ne" and val == MANUAL_UPLOAD_PROMPT:
                    self._non_upload = True
                elif op is not None:
                    raise AssertionError(
                        f"prompt_used 조건이 뜻밖의 모양이다: {op} {val!r}")
            elif left == "is_primary":
                if op == "eq" and val == 1:
                    self._primary = True
                elif op is not None:
                    raise AssertionError(
                        f"is_primary 조건이 뜻밖의 모양이다: {op} {val!r}")
        return self

    def order_by(self, *a):
        return self

    def all(self):
        return []

    def count(self):
        return 0

    def first(self):
        if self._non_upload:
            return self._db.pipeline
        # ★primary 조회는 `is_primary == 1` 을 **요구**한다 — 세우기만 하고
        #  안 읽으면 그 조건이 빠지거나 뒤집혀도 대역이 H 를 준다(Codex).
        if not self._primary:
            raise AssertionError(
                "ImageAsset 조회에 prompt_used 도 is_primary 조건도 없다 — "
                "대역이 무엇을 묻는지 모른다")
        return self._db.human


class _DB:
    """★저장에 따라 **상태가 바뀐다** — 처음엔 파이프라인 산출이 없다.

    첫 판은 G 를 처음부터 심어 두어 「신규 저장 → 다음 조회에서 재사용」을
    못 잠갔다(Codex). 이제 `_save` 가 `pipeline` 을 채운다.
    """

    def __init__(self, human, pipeline=None):
        self.human, self.pipeline = human, pipeline

    def query(self, model):
        return _Q(self)


@pytest.fixture
def _wired(tmp_path):
    """H(사람 대표) · G(파이프라인 산출) · 레시피 SEL 을 놓는다."""
    recipe = tmp_path / "scene" / "recipe"
    recipe.mkdir(parents=True, exist_ok=True)
    h = tmp_path / "h.png"
    h.write_bytes(b"HUMAN-BYTES")
    g = recipe / f"{TAG}_sel.png"
    g.write_bytes(b"PIPELINE-BYTES")
    human = _Asset("H", h, prompt=MANUAL_UPLOAD_PROMPT, primary=1)
    pipeline = _Asset("G", g, prompt="Create ONE FINAL…", primary=0)
    return tmp_path, recipe, g, _DB(human, pipeline)


def _run(tmp_path, *, db, scene_cp, sel, persistence):
    from app.services.still_recipe_service import run_still_recipe_generation

    _write_cp(tmp_path, "shot_ref_classify",
              {"shots": {}, "scenes": {}, "world_anchor_en": ""})
    _write_cp(tmp_path, "shot_continuity", {"pose_canon": []})
    _write_cp(tmp_path, "shot_conti_light", {"contis": {}})
    # ★완료 샷이 JIT 검증 갈래를 타려면 record 가 있어야 한다 — 없으면
    #  「record 없어 검증 없이 skip」으로 빠진다(기존 시험과 같은 준비).
    recipe = tmp_path / "scene" / "recipe"
    recipe.mkdir(parents=True, exist_ok=True)
    rj = recipe / "records.json"
    if not rj.exists():
        rj.write_text(json.dumps({TAG: {"selected": "a", "totals": {}}},
                                 ensure_ascii=False), encoding="utf-8")

    def _gen(**kwargs):
        rec = dict(kwargs["records"].data.get(kwargs["rec_key"]) or {})
        return sel, rec

    over = [("still_jit_verify_enabled", True),
            ("still_cine_transform_enabled", False)]
    patches = [
        patch("app.core.config.settings.projects_dir", str(tmp_path)),
        patch("app.modules.pipeline.multiroll_gemini.make_nb2_gen_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_judge_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.multiroll_gemini.make_gemini_critique_fn",
              return_value=MagicMock()),
        patch("app.modules.pipeline.still_recipe.run_branch_select",
              side_effect=_gen),
    ] + [patch(f"app.core.config.settings.{n}", v, create=True)
         for n, v in _FLAG_PATCHES + over]
    from contextlib import ExitStack

    with ExitStack() as st:
        for p in patches:
            st.enter_context(p)
        return run_still_recipe_generation(
            db=db, project_id=PID, episode_id=EID,
            stills=[_still(STILL_ID, 1, 1)], stills_orm=[],
            entity_lookup={}, ref_image_map={},
            reference_svc=MagicMock(),
            scene_ref_image_map={}, scene_ref_asset_id_map={},
            staging_map={}, scene_cp=scene_cp,
            persistence_svc=persistence, progress=MagicMock(),
            project_config=None, scene_dir=tmp_path / "scene",
            already_done_stills={STILL_ID}, target_scenes=None,
        )


def _seed_cp(tmp_path, *, primary_id=None, primary_path=None, seed=None):
    from app.modules.image_checkpoint import ImageCheckpointManager

    cp = ImageCheckpointManager(tmp_path / "cp", "scene_image")
    cp.mark_completed(STILL_ID, seed if seed is not None else {
        "asset_ids": [primary_id], "primary_id": primary_id,
        "primary_path": str(primary_path), "recipe_tag": TAG})
    return cp


def _spy_writes(cp):
    """`mark_completed` 를 감싸 **쓰기 횟수**를 센다.

    ★첫 판은 `len(get_completed(...))` 를 썼는데 그것은 **row 의 필드
     개수**다 — 무기록을 전혀 검증하지 않았다(Codex).
    """
    calls = []
    orig = cp.mark_completed

    def _wrapped(item_id, result):
        calls.append(item_id)
        return orig(item_id, result)

    cp.mark_completed = _wrapped                      # type: ignore[method-assign]
    return calls


def _persistence(saved, db, recipe):
    def _save(scene_result, lineage):
        a = _Asset(f"G{len(saved) + 1}", scene_result["file_path"],
                   prompt="Create ONE FINAL…", primary=0)
        saved.append(a)
        db.pipeline = a           # ★저장이 **상태를 바꾼다**
        return a

    p = MagicMock()
    p.safety_ladder_call_provenance.return_value = None
    p.save_single_scene_asset.side_effect = _save
    return p


def test_a_new_output_is_stored_once_and_then_reused(_wired):
    """① 신규 G 저장 → 두 번째 방문은 **그 G 를 재사용**(저장 추가 0)."""
    tmp_path, recipe, sel, db = _wired
    db.pipeline = None                                # 아직 산출이 없다
    cp = _seed_cp(tmp_path, primary_id="H", primary_path=tmp_path / "h.png")
    saved: list = []
    per = _persistence(saved, db, recipe)

    _run(tmp_path, db=db, scene_cp=cp, sel=sel, persistence=per)
    assert len(saved) == 1, f"첫 방문이 한 번 저장하지 않았다: {len(saved)}"
    assert db.pipeline is not None and db.pipeline.id == "G1"
    # ★**새 저장 순간에도** CP 대표는 사람 것이어야 한다 — 「G1 로 잘못
    #  적었다가 다음 방문에 복구」되는 것으로는 안 된다(Codex).
    row = cp.get_completed(STILL_ID)
    assert row["primary_id"] == "H", (
        f"새 저장이 CP 대표를 가로챘다: {row['primary_id']}")
    assert row.get("produced_id") == "G1", (
        "이번에 보관한 산출이 따로 안 남았다")

    _run(tmp_path, db=db, scene_cp=cp, sel=sel, persistence=per)
    assert len(saved) == 1, (
        f"두 번째 방문이 또 저장했다 (1 → {len(saved)})")


def test_an_upload_after_the_fact_corrects_the_checkpoint_once(_wired):
    """② 기존 G · CP 대표 G 에서 사람이 H 를 올린 상태 —

    첫 재사용은 **CP 만 한 번** 정정하고, 다음 재사용은 **CP 도 무기록**.
    """
    tmp_path, recipe, sel, db = _wired          # db.pipeline = G (이미 있음)
    seed = {"asset_ids": ["G"], "primary_id": "G",
            "primary_path": str(sel), "recipe_tag": TAG,
            "produced_id": "G", "produced_path": str(sel),
            "audit": {"note": "임의의 중첩 감사 메타"}}
    cp = _seed_cp(tmp_path, seed=seed)
    saved: list = []
    per = _persistence(saved, db, recipe)
    writes = _spy_writes(cp)

    _run(tmp_path, db=db, scene_cp=cp, sel=sel, persistence=per)
    assert saved == [], f"재사용인데 저장했다: {len(saved)}"
    assert len(writes) == 1, f"CP 정정이 한 번이 아니다: {len(writes)}"
    # ★**원래 payload 에서 대표 두 칸만 바뀐 것**과 통째로 견준다 —
    #  칸마다 시험을 늘리는 대신 한 번에 본다(Codex).
    row = cp.get_completed(STILL_ID)
    assert row == {**seed, "primary_id": "H",
                   "primary_path": str(tmp_path / "h.png")}

    _run(tmp_path, db=db, scene_cp=cp, sel=sel, persistence=per)
    assert saved == []
    assert len(writes) == 1, (
        f"무변경 방문이 CP 를 또 썼다: {len(writes)}")


def test_the_stub_rejects_an_inverted_condition(_wired):
    """★대역이 **조건 뒤집기**를 잡는지 — 시험이 스스로 속지 않게.

    첫 판은 bind 값만 봐서 `!=` 를 `==` 로 뒤집어도 통과했다.
    """
    _tmp, _recipe, _sel, db = _wired
    from app.models.project import ImageAsset

    with pytest.raises(AssertionError):
        db.query(ImageAsset).filter(
            ImageAsset.prompt_used == MANUAL_UPLOAD_PROMPT)
    with pytest.raises(AssertionError):
        db.query(ImageAsset).filter(ImageAsset.is_primary != 1)
    # 아무 조건도 없으면 무엇을 묻는지 모른다 — 답하지 않는다
    with pytest.raises(AssertionError):
        db.query(ImageAsset).filter().first()