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_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"}],
            [{"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_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 !== 'imported'" in html


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"]}
            )