# -*- coding: utf-8 -*-
"""数据库单写队列测试（不联网）: uv run python tests/test_database.py"""

import os
import sqlite3
import sys
import tempfile
import threading
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")

from reaction_board.database import Database, DatabaseClosed  # noqa: E402


def make_db(tmp):
    db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup")
    db.initialize()
    return db


def test_write_read_transaction():
    with tempfile.TemporaryDirectory() as tmp:
        db = make_db(tmp)
        try:
            assert db.schema_version == 6
            db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')"))
            assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"

            def boom(conn):
                conn.execute("UPDATE kv SET v='2' WHERE k='a'")
                raise RuntimeError("boom")
            try:
                db.transaction(boom)
            except RuntimeError as exc:
                assert str(exc) == "boom"
            else:
                raise AssertionError("事务应抛错")
            assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
        finally:
            db.close()


def test_concurrent_writes_are_serialized():
    with tempfile.TemporaryDirectory() as tmp:
        db = make_db(tmp)
        try:
            db.write(lambda c: c.execute(
                "CREATE TABLE counter (id INTEGER PRIMARY KEY, n INTEGER NOT NULL)"
            ))
            db.write(lambda c: c.execute("INSERT INTO counter(id, n) VALUES (1, 0)"))
            errors = []

            def worker():
                try:
                    for _ in range(10):
                        db.transaction(
                            lambda conn: conn.execute(
                                "UPDATE counter SET n = n + 1 WHERE id = 1"
                            )
                        )
                except Exception as exc:  # noqa: BLE001
                    errors.append(exc)

            threads = [threading.Thread(target=worker) for _ in range(8)]
            for t in threads:
                t.start()
            for t in threads:
                t.join()
            assert not errors, errors
            assert db.read(lambda c: c.execute("SELECT n FROM counter").fetchone()[0]) == 80
        finally:
            db.close()


def test_close_and_reinitialize():
    with tempfile.TemporaryDirectory() as tmp:
        db = make_db(tmp)
        db.close()
        db.close()  # 幂等
        try:
            db.write(lambda c: None)
        except DatabaseClosed:
            pass
        else:
            raise AssertionError("关闭后写入应失败")
        try:
            db.read(lambda c: None)
        except DatabaseClosed:
            pass
        else:
            raise AssertionError("关闭后读取应失败")

        db2 = make_db(tmp)
        try:
            assert db2.schema_version == 6
            assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 6
        finally:
            db2.close()


def test_wal_enabled():
    with tempfile.TemporaryDirectory() as tmp:
        db = make_db(tmp)
        try:
            mode = db.read(lambda c: c.execute("PRAGMA journal_mode").fetchone()[0])
            assert str(mode).lower() == "wal"
            assert db.read(lambda c: c.execute("PRAGMA foreign_keys").fetchone()[0]) == 1
        finally:
            db.close()


def test_migration_backup_on_existing_db():
    """已有数据升级时由 Database 自动备份。"""
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "app.db"
        # 先建一个旧库（只有 schema_version v0），Database.initialize 会升级并备份
        conn = sqlite3.connect(path)
        conn.execute("CREATE TABLE legacy (x INTEGER)")
        conn.commit()
        conn.close()
        db = Database(path, backup_dir=Path(tmp) / "backup")
        try:
            applied = db.initialize()
            assert applied == [1, 2, 3, 4, 5, 6]
            assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1
        finally:
            db.close()


def main():
    test_write_read_transaction()
    test_concurrent_writes_are_serialized()
    test_close_and_reinitialize()
    test_wal_enabled()
    test_migration_backup_on_existing_db()
    print("OK: 数据库单写队列与迁移入口测试全部通过")


if __name__ == "__main__":
    main()