cmautobuy/client/test/test_db.py

"""SQLite 初始化和 v1 数据库结构测试。"""

import sqlite3
import tempfile
import unittest
from pathlib import Path

from src.db import DatabaseVersionError, initialize_database, open_database


EXPECTED_TABLES = {
    "pdd_tasks",
    "task_runs",
    "outbox_events",
    "app_settings",
}

EXPECTED_INDEXES = {
    "idx_pdd_tasks_list",
    "idx_pdd_tasks_status_type",
    "idx_pdd_tasks_goods_id",
    "idx_task_runs_task",
    "idx_outbox_pending",
}


class DatabaseInitializationTests(unittest.TestCase):
    """每个测试都使用独立临时数据库，不接触真实 data 目录。"""

    def setUp(self) -> None:
        self._temporary_directory = tempfile.TemporaryDirectory()
        self.db_path = (
            Path(self._temporary_directory.name) / "nested" / "client.db"
        )

    def tearDown(self) -> None:
        self._temporary_directory.cleanup()

    def test_initialize_creates_v1_tables_and_indexes(self) -> None:
        result_path = initialize_database(self.db_path)

        self.assertEqual(result_path, self.db_path)
        self.assertTrue(self.db_path.is_file())

        connection = open_database(self.db_path)
        try:
            tables = {
                row[0]
                for row in connection.execute(
                    "SELECT name FROM sqlite_master WHERE type = 'table'"
                )
            }
            indexes = {
                row[0]
                for row in connection.execute(
                    "SELECT name FROM sqlite_master WHERE type = 'index'"
                )
            }
            version = connection.execute("PRAGMA user_version").fetchone()[0]
        finally:
            connection.close()

        self.assertTrue(EXPECTED_TABLES.issubset(tables))
        self.assertTrue(EXPECTED_INDEXES.issubset(indexes))
        self.assertEqual(version, 1)

    def test_initialize_can_run_twice_without_losing_data(self) -> None:
        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            with connection:
                connection.execute(
                    "INSERT INTO app_settings"
                    " (setting_key, value_json, updated_at) VALUES (?, ?, ?)",
                    ("automation.dry_run", "true", "2026-08-06T00:00:00Z"),
                )
        finally:
            connection.close()

        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            value = connection.execute(
                "SELECT value_json FROM app_settings WHERE setting_key = ?",
                ("automation.dry_run",),
            ).fetchone()[0]
        finally:
            connection.close()

        self.assertEqual(value, "true")

    def test_new_connection_uses_required_pragmas(self) -> None:
        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            foreign_keys = connection.execute("PRAGMA foreign_keys").fetchone()[0]
            journal_mode = connection.execute("PRAGMA journal_mode").fetchone()[0]
            busy_timeout = connection.execute("PRAGMA busy_timeout").fetchone()[0]
        finally:
            connection.close()

        self.assertEqual(foreign_keys, 1)
        self.assertEqual(journal_mode.lower(), "wal")
        self.assertEqual(busy_timeout, 5000)

    def test_pdd_task_check_constraints_are_enforced(self) -> None:
        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            invalid_values = (
                ("unknown", "claimed", None, None),
                ("collect", "unknown", None, None),
                ("collect", "claimed", -1, None),
                ("purchase", "claimed", None, 0),
            )
            for index, values in enumerate(invalid_values):
                with self.subTest(values=values):
                    with self.assertRaises(sqlite3.IntegrityError):
                        connection.execute(
                            "INSERT INTO pdd_tasks"
                            " (remote_task_id, task_type, goods_url, price_cent,"
                            " quantity, status, received_at, created_at, updated_at)"
                            " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                            (
                                f"TASK-{index}",
                                values[0],
                                "https://example.test/goods",
                                values[2],
                                values[3],
                                values[1],
                                "2026-08-06T00:00:00Z",
                                "2026-08-06T00:00:00Z",
                                "2026-08-06T00:00:00Z",
                            ),
                        )
                    connection.rollback()
        finally:
            connection.close()

    def test_task_run_and_outbox_foreign_keys_are_enforced(self) -> None:
        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            with self.assertRaises(sqlite3.IntegrityError):
                connection.execute(
                    "INSERT INTO task_runs"
                    " (task_id, attempt_id, attempt_no, device_address, run_status,"
                    " started_at, created_at, updated_at)"
                    " VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (
                        999,
                        "attempt-1",
                        1,
                        "device-1",
                        "running",
                        "2026-08-06T00:00:00Z",
                        "2026-08-06T00:00:00Z",
                        "2026-08-06T00:00:00Z",
                    ),
                )
            connection.rollback()

            with self.assertRaises(sqlite3.IntegrityError):
                connection.execute(
                    "INSERT INTO outbox_events"
                    " (task_id, event_type, idempotency_key, payload_json,"
                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)",
                    (
                        999,
                        "collect_result",
                        "TASK-1:attempt-1:result-v1",
                        "{}",
                        "2026-08-06T00:00:00Z",
                        "2026-08-06T00:00:00Z",
                    ),
                )
            connection.rollback()
        finally:
            connection.close()

    def test_newer_database_version_is_rejected(self) -> None:
        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            connection.execute("PRAGMA user_version = 99")
        finally:
            connection.close()

        with self.assertRaisesRegex(DatabaseVersionError, "数据库版本 99"):
            initialize_database(self.db_path)


if __name__ == "__main__":
    unittest.main()