"""outlook_dedup 단위 테스트 — Phase 4.3.

apply_outlook_merge의 UPSERT 이관 로직 (EntityAlias / RelationParticipant /
EntityEpisodeLink / CharacterOutlook / ImageAsset / SceneStill / EntityCanon).
주요 검증:
- 기존 DELETE-only 동작이 UPSERT 이관으로 전환되어 데이터 유실 방지
- keep에 이미 존재하면 remove 측 행을 DELETE (unique constraint 준수)
"""
from __future__ import annotations

from typing import Any, Dict, List
from unittest.mock import MagicMock

import pytest

from app.modules.pipeline import outlook_dedup as od


class FakeDB:
    """executemany-like 기록용 경량 fake — SQL + params 히스토리 수집.

    Claude Review Important #4 반영: response key는 가장 구체적(긴) 키부터 매칭하여
    SQL fragment가 다른 쿼리의 부분 문자열로도 매칭되는 false-positive를 방지.
    """

    def __init__(self, select_responses: Dict[str, List[tuple]] | None = None):
        self._responses = select_responses or {}
        self.executed: List[tuple[str, Dict[str, Any]]] = []
        self.committed = False
        self.rolled_back = False

    def execute(self, statement, params: Dict[str, Any] | None = None):
        sql = str(statement)
        self.executed.append((sql, params or {}))

        result = MagicMock()
        # 긴 키부터 매칭 — "SELECT id, episode_id, ..."가 "SELECT episode_id, ..."를
        # 포함하는 상황에서 더 구체적인 키 우선
        sorted_keys = sorted(self._responses.keys(), key=len, reverse=True)
        for key in sorted_keys:
            if key in sql:
                rows = self._responses[key]
                result.fetchone.return_value = rows[0] if rows else None
                result.fetchall.return_value = rows
                result.scalar.return_value = (rows[0][0] if rows else 0)
                return result

        result.fetchone.return_value = None
        result.fetchall.return_value = []
        result.scalar.return_value = 0
        return result

    def commit(self):
        self.committed = True

    def rollback(self):
        self.rolled_back = True


def _executed_sqls(db: FakeDB) -> List[str]:
    return [sql for sql, _ in db.executed]


# ── _resolve_outlook_id ──


def test_resolve_outlook_id_returns_id_for_matching_name():
    db = FakeDB({"SELECT id FROM entity_canon": [("outlook_123",)]})
    result = od._resolve_outlook_id(db, "p1", "검은 정장")
    assert result == "outlook_123"


def test_resolve_outlook_id_returns_none_when_not_found():
    db = FakeDB()
    assert od._resolve_outlook_id(db, "p1", "없음") is None


# ── _migrate_entity_aliases (Phase 4.3 핵심) ──


def test_migrate_entity_aliases_moves_unique_aliases_to_keep():
    """remove의 alias가 keep에 없으면 canon_id를 UPDATE."""
    db = FakeDB({
        "SELECT alias FROM entity_alias WHERE canon_id = :kid": [],
        "SELECT id, alias FROM entity_alias WHERE canon_id = :rid": [
            ("alias_1", "검정정장"),
            ("alias_2", "블랙수트"),
        ],
    })

    od._migrate_entity_aliases(db, keep_id="K", remove_id="R")

    sqls = _executed_sqls(db)
    # 각 alias에 UPDATE 1회씩
    assert sum("UPDATE entity_alias SET canon_id" in s for s in sqls) == 2
    # DELETE는 없어야 함 (keep에 중복 없음)
    assert not any("DELETE FROM entity_alias WHERE id" in s for s in sqls)


def test_migrate_entity_aliases_deletes_duplicates_on_keep():
    """remove의 alias가 keep에 이미 있으면 remove 측 행을 DELETE (unique 준수)."""
    db = FakeDB({
        "SELECT alias FROM entity_alias WHERE canon_id = :kid": [("검정정장",)],
        "SELECT id, alias FROM entity_alias WHERE canon_id = :rid": [
            ("alias_dup", "검정정장"),   # 이미 keep에 있음 → DELETE
            ("alias_new", "블랙수트"),   # 없음 → UPDATE
        ],
    })

    od._migrate_entity_aliases(db, keep_id="K", remove_id="R")

    sqls = _executed_sqls(db)
    assert sum("DELETE FROM entity_alias WHERE id" in s for s in sqls) == 1
    assert sum("UPDATE entity_alias SET canon_id" in s for s in sqls) == 1


def test_migrate_entity_aliases_handles_empty_remove_set():
    """remove에 alias 없으면 UPDATE / DELETE 모두 0."""
    db = FakeDB({
        "SELECT alias FROM entity_alias WHERE canon_id = :kid": [("기존",)],
        "SELECT id, alias FROM entity_alias WHERE canon_id = :rid": [],
    })

    od._migrate_entity_aliases(db, keep_id="K", remove_id="R")

    sqls = _executed_sqls(db)
    assert not any("UPDATE entity_alias" in s for s in sqls)
    assert not any("DELETE FROM entity_alias" in s for s in sqls)


# ── _migrate_relation_participants ──


def test_migrate_relation_participants_update_and_role_aware_dedup():
    """canon_id UPDATE + (relation_id, canon_id, participant_role) 기준 dedup.

    Codex Review Important #2: role을 포함해 dedup해야 base/variant가 모두 keep이 된
    self-collapsed relation에서 양 role이 보존됨.
    """
    db = FakeDB()

    od._migrate_relation_participants(db, keep_id="K", remove_id="R")

    sqls = _executed_sqls(db)
    assert len(sqls) == 2
    assert "UPDATE relation_participant SET canon_id" in sqls[0]
    assert db.executed[0][1] == {"kid": "K", "rid": "R"}
    # role을 포함한 dedup (Codex #2)
    assert "DELETE FROM relation_participant" in sqls[1]
    assert "GROUP BY relation_id, canon_id, participant_role" in sqls[1]
    assert db.executed[1][1] == {"kid": "K"}


# ── _migrate_entity_episode_links ──


def test_migrate_entity_episode_links_updates_when_no_collision():
    """keep에 해당 episode link 없으면 canon_id UPDATE."""
    db = FakeDB({
        "SELECT episode_id, t2i_appearance_count FROM entity_episode_link": [],
        "SELECT id, episode_id, t2i_appearance_count FROM entity_episode_link": [
            ("link_1", "ep1", 5),
        ],
    })

    od._migrate_entity_episode_links(db, "p1", keep_id="K", remove_id="R")

    sqls = _executed_sqls(db)
    assert any("UPDATE entity_episode_link SET canon_id" in s for s in sqls)
    assert not any("DELETE FROM entity_episode_link WHERE id" in s for s in sqls)


def test_migrate_entity_episode_links_merges_counts_on_collision():
    """keep에 동일 episode link가 있으면 t2i_appearance_count 합산 후 remove DELETE."""
    sql_patterns = {
        "SELECT episode_id, t2i_appearance_count FROM entity_episode_link": [
            ("ep1", 3),
        ],
        "SELECT id, episode_id, t2i_appearance_count FROM entity_episode_link": [
            ("link_1", "ep1", 5),
        ],
    }
    db = FakeDB(sql_patterns)

    od._migrate_entity_episode_links(db, "p1", keep_id="K", remove_id="R")

    update_calls = [e for e in db.executed if "UPDATE entity_episode_link SET t2i_appearance_count" in e[0]]
    assert len(update_calls) == 1
    assert update_calls[0][1]["cnt"] == 8  # 3 + 5 합산

    delete_calls = [e for e in db.executed if "DELETE FROM entity_episode_link WHERE id" in e[0]]
    assert len(delete_calls) == 1
    assert delete_calls[0][1] == {"id": "link_1"}


# ── _merge_single_outlook (통합) ──


def test_merge_single_outlook_skips_when_resolve_fails():
    """keep 또는 remove가 없으면 병합 skip."""
    db = FakeDB()  # 모든 SELECT가 None 리턴

    result = od._merge_single_outlook(db, "p1", "없는_keep", "없는_remove")

    assert result is False
    # entity_canon DELETE는 호출 안 됨
    assert not any("DELETE FROM entity_canon" in s for s, _ in db.executed)


def test_merge_single_outlook_skips_self_merge():
    """Codex Review Critical #1: keep_id == remove_id면 skip하여 keep 자기 삭제 방지."""
    db = FakeDB()

    original_resolve = od._resolve_outlook_id
    # 같은 이름의 두 canon이 있어 resolve가 같은 id를 두 번 반환하는 상황 재현
    od._resolve_outlook_id = lambda d, p, name: "SAME_ID"
    try:
        result = od._merge_single_outlook(db, "p1", "A", "A")
    finally:
        od._resolve_outlook_id = original_resolve

    assert result is False
    # 어떤 migration/cleanup SQL도 호출되지 않아야 함
    destructive = [
        sql for sql, _ in db.executed
        if "DELETE" in sql or "UPDATE" in sql
    ]
    assert destructive == []


def test_merge_single_outlook_deletes_canon_last():
    """EntityCanon DELETE는 모든 migration 후 마지막에 호출."""
    # 각 SELECT가 순서대로 id를 돌려주도록
    db = FakeDB({
        "SELECT id FROM entity_canon": [("outlook_keep",), ("outlook_remove",)],
        "SELECT character_id FROM character_outlook": [],
        "SELECT id, character_id FROM character_outlook": [],
        "SELECT COUNT(*) FROM image_asset": [(0,)],
        "SELECT id, visible_entities_json, t2i_variations_json, t2i_prompt_cinematic": [],
        "SELECT episode_id, t2i_appearance_count FROM entity_episode_link": [],
        "SELECT id, episode_id, t2i_appearance_count FROM entity_episode_link": [],
        "SELECT alias FROM entity_alias WHERE canon_id = :kid": [],
        "SELECT id, alias FROM entity_alias WHERE canon_id = :rid": [],
    })

    # _resolve_outlook_id가 키 "SELECT id FROM entity_canon"에서 첫 행만 리턴하므로
    # 두 번째 호출에서 keep_id가 잘못 매치되는 문제가 있다. 테스트 단순화를 위해
    # _resolve_outlook_id를 monkeypatch.
    original_resolve = od._resolve_outlook_id
    od._resolve_outlook_id = lambda db, pid, name: {"K_NAME": "K_ID", "R_NAME": "R_ID"}.get(name)
    try:
        result = od._merge_single_outlook(db, "p1", "K_NAME", "R_NAME")
    finally:
        od._resolve_outlook_id = original_resolve

    assert result is True
    final_sqls = [sql for sql, _ in db.executed]
    # Claude Review Important #3: 마지막 SQL이 DELETE FROM entity_canon이어야 함
    assert "DELETE FROM entity_canon" in final_sqls[-1], (
        f"entity_canon DELETE가 마지막에 있어야 하나 실제 마지막 SQL은: {final_sqls[-1]!r}"
    )


# ── apply_outlook_merge (public API) ──


def test_apply_outlook_merge_commits_on_success():
    """성공 시 단일 commit."""
    db = FakeDB()
    # _resolve_outlook_id를 monkey patch하여 None 리턴 → skip 모든 그룹
    original_resolve = od._resolve_outlook_id
    od._resolve_outlook_id = lambda *args, **kwargs: None
    try:
        count = od.apply_outlook_merge(
            db, "p1", [{"keep": "a", "remove": ["b"], "reason": ""}]
        )
    finally:
        od._resolve_outlook_id = original_resolve

    assert count == 0  # 모두 resolve 실패
    assert db.committed is True
    assert db.rolled_back is False


def test_apply_outlook_merge_rollbacks_on_exception():
    """예외 발생 시 rollback + re-raise."""
    db = FakeDB()
    original_resolve = od._resolve_outlook_id
    # 첫 호출에서 예외 발생
    od._resolve_outlook_id = MagicMock(side_effect=RuntimeError("DB broke"))
    try:
        with pytest.raises(RuntimeError, match="DB broke"):
            od.apply_outlook_merge(
                db, "p1", [{"keep": "a", "remove": ["b"], "reason": ""}]
            )
    finally:
        od._resolve_outlook_id = original_resolve

    assert db.rolled_back is True
    assert db.committed is False


def test_apply_outlook_merge_counts_successful_merges():
    """성공한 merge 수를 리턴."""
    db = FakeDB()
    call_count = {"n": 0}

    def _fake_merge(db_arg, pid, keep, remove):
        call_count["n"] += 1
        return True  # 항상 성공

    original_merge = od._merge_single_outlook
    od._merge_single_outlook = _fake_merge
    try:
        count = od.apply_outlook_merge(
            db,
            "p1",
            [
                {"keep": "a", "remove": ["b", "c"], "reason": ""},
                {"keep": "d", "remove": ["e"], "reason": ""},
            ],
        )
    finally:
        od._merge_single_outlook = original_merge

    assert count == 3
    assert call_count["n"] == 3
    assert db.committed is True


# ── _rewrite_scene_still_references ──


def test_rewrite_scene_still_replaces_marker_in_all_prompt_fields():
    """Codex Review Important #3: t2i_prompt_closeup까지 포함해 모든 prompt 필드에서 marker 교체."""
    import json as _json

    vis = _json.dumps([{"entity_name": "검정정장", "short_id": "O01"}])
    variations = _json.dumps([{"t2i_prompt": "a man in [검정정장]"}])

    db = FakeDB({
        "SELECT id, visible_entities_json, t2i_variations_json, "
        "       t2i_prompt_cinematic, t2i_prompt_closeup":
            [(
                "still_1",
                vis,
                variations,
                "wide: man wearing [검정정장]",
                "closeup: face and [검정정장]",
            )],
    })

    od._rewrite_scene_still_references(db, "p1", "BLACK_SUIT", "검정정장")

    update_calls = [e for e in db.executed if "UPDATE scene_still SET visible_entities_json" in e[0]]
    assert len(update_calls) == 1
    params = update_calls[0][1]
    assert "BLACK_SUIT" in params["vars"]
    assert "검정정장" not in params["vars"]
    assert params["t2i_cin"] == "wide: man wearing [BLACK_SUIT]"
    assert params["t2i_close"] == "closeup: face and [BLACK_SUIT]"
    assert "BLACK_SUIT" in params["vis"]


def test_rewrite_scene_still_no_change_when_name_absent():
    """해당 name이 없으면 UPDATE 호출 없음."""
    import json as _json

    vis = _json.dumps([{"entity_name": "흰셔츠", "short_id": "O02"}])
    variations = _json.dumps([{"t2i_prompt": "a man in [흰셔츠]"}])

    db = FakeDB({
        "SELECT id, visible_entities_json, t2i_variations_json, "
        "       t2i_prompt_cinematic, t2i_prompt_closeup":
            [("still_1", vis, variations, "plain text", "closeup plain")],
    })

    od._rewrite_scene_still_references(db, "p1", "KEEP", "REMOVE")

    update_calls = [e for e in db.executed if "UPDATE scene_still SET visible_entities_json" in e[0]]
    assert len(update_calls) == 0