from __future__ import annotations

from pathlib import Path
from unittest.mock import patch

import pytest

import serve_l2_dashboard as dashboard


class FakeCursor:
    def __init__(self, update_rowcount: int, snapshot: dict[str, object]) -> None:
        self.update_rowcount = update_rowcount
        self.snapshot = snapshot
        self.rowcount = 0
        self.executions: list[tuple[str, object]] = []

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def execute(self, sql, params=None):
        self.executions.append((sql, params))
        self.rowcount = self.update_rowcount if sql.lstrip().startswith("UPDATE") else 0

    def fetchone(self):
        return self.snapshot


class FakeConnection:
    def __init__(self, cursor: FakeCursor) -> None:
        self._cursor = cursor
        self.committed = False
        self.rolled_back = False

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def cursor(self):
        return self._cursor

    def commit(self):
        self.committed = True

    def rollback(self):
        self.rolled_back = True


def test_review_update_uses_staging_version_and_claim_token():
    snapshot = {
        "staging_id": 42,
        "source_song_id": "song-1",
        "biz_review_status": "approved_import",
        "review_version": 8,
    }
    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        result = dashboard._update_staging_review(
            42, "approved_import", "ok", "alice", 7, "browser-token"
        )

    update_sql, params = cursor.executions[0]
    assert "WHERE staging_id = %s" in update_sql
    assert "review_version = %s" in update_sql
    assert "review_claim_token = %s" in update_sql
    assert params[-4:] == [42, 7, "browser-token", "alice"]
    assert result == snapshot
    assert conn.committed


def test_stale_review_returns_conflict_without_overwrite():
    snapshot = {
        "staging_id": 42,
        "biz_review_status": "rejected_duplicate",
        "reviewed_by": "bob",
        "review_version": 9,
    }
    cursor = FakeCursor(update_rowcount=0, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        with pytest.raises(dashboard.ReviewConflictError) as raised:
            dashboard._update_staging_review(
                42, "approved_import", "stale", "alice", 7, "browser-token"
            )

    assert raised.value.current_record == snapshot
    assert conn.rolled_back
    assert not conn.committed


def test_rejected_duplicate_persists_the_selected_candidate():
    snapshot = {
        "staging_id": 42,
        "matched_song_id": None,
        "recalled_candidates": '[{"id":"existing-song"}]',
        "biz_review_status": "rejected_duplicate",
        "review_version": 8,
    }
    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        dashboard._update_staging_review(
            42, "rejected_duplicate", "same", "alice", 7, "browser-token", "existing-song"
        )

    update_sql, params = cursor.executions[1]
    assert "matched_song_id = %s" in update_sql
    assert params[:3] == ["rejected_duplicate", "alice", "existing-song"]
    assert conn.committed


def test_rejected_duplicate_rejects_a_candidate_outside_recall_results():
    cursor = FakeCursor(
        update_rowcount=1,
        snapshot={"staging_id": 42, "matched_song_id": "candidate-a", "recalled_candidates": "[]"},
    )
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        with pytest.raises(ValueError, match="不在当前召回结果"):
            dashboard._update_staging_review(
                42, "rejected_duplicate", "", "alice", 7, "browser-token", "candidate-b"
            )

    assert conn.rolled_back
    assert not any(sql.lstrip().startswith("UPDATE") for sql, _params in cursor.executions)


def test_pending_review_restores_a_deleted_staging_row():
    snapshot = {
        "staging_id": 42,
        "biz_review_status": "pending",
        "staging_status": "staged",
        "review_version": 6,
    }
    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        dashboard._update_staging_review(
            42, "pending", "", "alice", 5, "alice-token"
        )

    update_sql, _params = cursor.executions[0]
    assert "staging_status = CASE WHEN staging_status = 'deleted' THEN 'staged'" in update_sql
    assert "biz_review_status = 'pending'" in update_sql
    assert conn.committed


def test_claim_conflict_exposes_current_owner():
    snapshot = {
        "staging_id": 42,
        "review_claimed_by": "bob",
        "review_version": 3,
    }
    cursor = FakeCursor(update_rowcount=0, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        with pytest.raises(dashboard.ReviewConflictError) as raised:
            dashboard._claim_staging_review(42, "alice", "alice-token")

    claim_sql, _params = cursor.executions[0]
    assert "reviewed_by = %s" not in claim_sql
    assert raised.value.current_record["review_claimed_by"] == "bob"
    assert conn.rolled_back


def test_deleted_review_can_be_claimed_for_undo_but_imported_review_cannot():
    snapshot = {
        "staging_id": 42,
        "biz_review_status": "deleted",
        "staging_status": "deleted",
        "review_version": 4,
    }
    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        dashboard._claim_staging_review(42, "alice", "alice-token")

    claim_sql, _params = cursor.executions[0]
    assert "staging_status <> 'imported'" in claim_sql
    assert "staging_status NOT IN ('imported', 'deleted')" not in claim_sql
    assert "biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')" in claim_sql
    assert conn.committed

    imported_cursor = FakeCursor(
        update_rowcount=0,
        snapshot={"staging_id": 43, "staging_status": "imported"},
    )
    imported_conn = FakeConnection(imported_cursor)
    with patch.object(dashboard, "_target_conn", return_value=imported_conn):
        with pytest.raises(dashboard.ReviewConflictError, match="已经入库"):
            dashboard._claim_staging_review(43, "alice", "alice-token")

    assert imported_conn.rolled_back
    assert not imported_conn.committed


def test_heartbeat_does_not_increment_review_version():
    snapshot = {"staging_id": 42, "review_version": 5}
    cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        dashboard._heartbeat_staging_review(42, "browser-token")

    update_sql, _params = cursor.executions[0]
    assert "review_version" not in update_sql
    assert conn.committed


class ImportCursor(FakeCursor):
    def __init__(self) -> None:
        super().__init__(update_rowcount=1, snapshot={})
        self.result_sets = [
            [{"staging_id": 42, "source_song_id": "song-1", "biz_review_status": "approved_import"}],
            [],
            [{"source_song_id": "song-1", "target_id": 9001}],
        ]

    def execute(self, sql, params=None):
        self.executions.append((sql, params))
        self.rowcount = 1

    def fetchall(self):
        return self.result_sets.pop(0)


def test_import_locks_only_unclaimed_staged_rows():
    cursor = ImportCursor()
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        result = dashboard._import_approved_staging()

    lock_sql, _params = cursor.executions[0]
    assert "FOR UPDATE" in lock_sql
    assert "staging_status = 'staged'" in lock_sql
    assert "review_claim_expires_at <= NOW()" in lock_sql
    assert result["inserted_count"] == 1
    assert result["source_ids"] == ["song-1"]
    assert conn.committed


def test_import_conflict_rolls_back_and_exposes_the_staging_row():
    cursor = ImportCursor()
    cursor.result_sets = [
        [{"staging_id": 42, "source_song_id": "song-1", "biz_review_status": "approved_import"}],
        [{
            "staging_id": 42,
            "source_song_id": "song-1",
            "source_table_name": "hk_song_platform",
            "name": "same song",
            "incoming_id": 9002,
            "id_conflict_target_id": None,
            "source_conflict_target_id": 9001,
        }],
    ]
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        with pytest.raises(dashboard.ImportConflictError) as raised:
            dashboard._import_approved_staging()

    assert raised.value.conflicts[0]["staging_id"] == 42
    assert raised.value.conflicts[0]["conflict_types"] == ["source_key"]
    assert conn.rolled_back
    assert not conn.committed
    assert not any(sql.lstrip().startswith("INSERT") for sql, _params in cursor.executions)


class DuplicateImportCursor:
    def __init__(self) -> None:
        self.rowcount = 0
        self.executions: list[tuple[str, object]] = []
        self.current = None

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def execute(self, sql, params=None):
        self.executions.append((sql, params))
        compact = " ".join(sql.split())
        self.rowcount = 0
        if "SELECT s.staging_id" in compact and "FOR UPDATE" in compact:
            self.current = [{
                "staging_id": 42,
                "source_song_id": "new-song",
                "source_table_name": "hk_song_platform",
                "name": "same song",
                "lyricist": "Alice/Bob",
                "composer": "Composer B",
                "biz_review_status": "rejected_duplicate",
                "matched_song_id": "existing-song",
            }]
        elif "SELECT id FROM" in compact and "source_table_name = %s" in compact:
            self.current = {"id": 9001}
        elif "SELECT id, lyricist, composer" in compact:
            self.current = {"id": 9001, "lyricist": "Alice", "composer": "Composer A"}
        elif compact.startswith("UPDATE"):
            self.current = None
            self.rowcount = 1
        else:
            raise AssertionError(f"unexpected SQL: {compact}")

    def fetchall(self):
        return self.current or []

    def fetchone(self):
        return self.current


def test_confirmed_duplicate_merges_authors_without_inserting_a_new_song():
    cursor = DuplicateImportCursor()
    conn = FakeConnection(cursor)

    with patch.object(dashboard, "_target_conn", return_value=conn):
        result = dashboard._import_approved_staging()

    sql_params = [(" ".join(sql.split()), params) for sql, params in cursor.executions]
    assert not any(sql.startswith("INSERT") for sql, _params in sql_params)
    assert any("SET lyricist = %s" in sql and params == ("Alice、Bob", 9001) for sql, params in sql_params)
    assert any("SET composer = %s" in sql and params == ("Composer A、Composer B", 9001) for sql, params in sql_params)
    assert any("SET staging_status = 'skipped'" in sql for sql, _params in sql_params)
    assert result["duplicate_count"] == 1
    assert result["author_merged_count"] == 1
    assert result["inserted_count"] == 0
    assert result["duplicate_source_ids"] == ["new-song"]
    assert conn.committed


def test_browsing_a_list_item_can_auto_claim_without_reloading_groups():
    html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")
    start = html.index("for (const btn of els.queryList.querySelectorAll('.query-item'))")
    end = html.index("function selectedQueryRows()", start)
    click_handler = html[start:end]

    assert "loadQueryRows()" in click_handler
    assert "claimCurrentReview" in click_handler
    assert "loadGroups()" not in click_handler


def test_staging_bigint_is_never_converted_to_javascript_number():
    html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")

    assert "Number(query.staging_id)" not in html
    assert "staging_id: String(query.staging_id)" in html


def test_dashboard_distinguishes_reviewed_from_submitted():
    html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")
    row = dashboard._staging_dashboard_row(
        {
            "staging_id": 42,
            "source_song_id": "song-1",
            "dedup_action": "review",
            "biz_review_status": "approved_import",
            "staging_status": "imported",
        }
    )

    assert row["staging_status"] == "imported"
    assert '<option value="reviewed">已审核</option>' in html
    assert '<option value="submitted">已提交</option>' in html
    assert "hasFinalReview && query.staging_status === 'staged'" in html


def test_dashboard_has_a_dedicated_deleted_filter():
    html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")

    assert '<option value="deleted">已删除</option>' in html
    source = Path("serve_l2_dashboard.py").read_text(encoding="utf-8")
    assert "elif decision == 'deleted':" in source
    assert "s.biz_review_status = 'deleted'" in source
    assert "s.staging_status = 'deleted'" in source


def test_review_access_code_issues_and_validates_edit_session():
    dashboard.REVIEW_SESSIONS.clear()
    with patch.object(dashboard, "REVIEW_ACCESS_CODE", "中文暗号"):
        login = dashboard._review_login("alice", "中文暗号")
        reviewer = dashboard._require_review_session(
            {"reviewer": "alice", "session_token": login["session_token"]}
        )

    assert reviewer == "alice"
    assert login["session_token"]


def test_wrong_access_code_and_mismatched_reviewer_are_rejected():
    dashboard.REVIEW_SESSIONS.clear()
    with patch.object(dashboard, "REVIEW_ACCESS_CODE", "correct"):
        with pytest.raises(dashboard.ReviewAuthError, match="暗号错误"):
            dashboard._review_login("alice", "wrong")
        login = dashboard._review_login("alice", "correct")
        with pytest.raises(dashboard.ReviewAuthError, match="昵称不匹配"):
            dashboard._require_review_session(
                {"reviewer": "bob", "session_token": login["session_token"]}
            )