cmautobuy/client/test/test_db.py

"""SQLite 初始化和数据库迁移测试。"""

import sqlite3
import tempfile
import unittest
from pathlib import Path

from src.db import DatabaseVersionError, initialize_database, open_database
from src.db_schema import MIGRATION_1, MIGRATION_2, MIGRATION_3, MIGRATION_4


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",
    "idx_pdd_tasks_visible_list",
}


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_latest_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, 5)

    def test_v1_database_is_upgraded_without_losing_task_runs(self) -> None:
        connection = open_database(self.db_path)
        try:
            with connection:
                for statement in MIGRATION_1:
                    connection.execute(statement)
                connection.execute("PRAGMA user_version = 1")
                connection.execute(
                    "INSERT INTO pdd_tasks"
                    " (remote_task_id, task_type, goods_url, status, received_at,"
                    " created_at, updated_at) VALUES"
                    " ('TASK-OLD', 'collect', 'https://example.test', 'claimed',"
                    " '2026-08-06T00:00:00Z', '2026-08-06T00:00:00Z',"
                    " '2026-08-06T00:00:00Z')"
                )
                connection.execute(
                    "INSERT INTO task_runs"
                    " (task_id, attempt_id, attempt_no, device_address, run_status,"
                    " started_at, created_at, updated_at) VALUES"
                    " (1, 'ATTEMPT-OLD', 1, 'USB-001', 'succeeded',"
                    " '2026-08-06T00:00:00Z', '2026-08-06T00:00:00Z',"
                    " '2026-08-06T00:00:00Z')"
                )
        finally:
            connection.close()

        initialize_database(self.db_path)

        connection = open_database(self.db_path)
        try:
            columns = {
                row[1] for row in connection.execute("PRAGMA table_info(task_runs)")
            }
            attempt_id = connection.execute(
                "SELECT attempt_id FROM task_runs"
            ).fetchone()[0]
            version = connection.execute("PRAGMA user_version").fetchone()[0]
        finally:
            connection.close()
        self.assertIn("result_data", columns)
        self.assertEqual(attempt_id, "ATTEMPT-OLD")
        self.assertEqual(version, 5)

    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_v2_database_adds_safe_execution_mode_default(self) -> None:
        connection = open_database(self.db_path)
        try:
            with connection:
                for statement in MIGRATION_1 + MIGRATION_2:
                    connection.execute(statement)
                connection.execute("PRAGMA user_version = 2")
                connection.execute(
                    "INSERT INTO pdd_tasks"
                    " (remote_task_id, task_type, goods_url, status, received_at,"
                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (
                        "PUR-OLD",
                        "purchase",
                        "https://example.test",
                        "claimed",
                        "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                    ),
                )
        finally:
            connection.close()

        initialize_database(self.db_path)

        connection = open_database(self.db_path)
        try:
            row = connection.execute(
                "SELECT execution_mode FROM pdd_tasks"
                " WHERE remote_task_id = 'PUR-OLD'"
            ).fetchone()
            version = connection.execute("PRAGMA user_version").fetchone()[0]
        finally:
            connection.close()
        self.assertEqual(row[0], "dry_run")
        self.assertEqual(version, 5)

    def test_v3_database_adds_soft_remove_column_without_losing_tasks(self) -> None:
        connection = open_database(self.db_path)
        try:
            with connection:
                for statement in MIGRATION_1 + MIGRATION_2 + MIGRATION_3:
                    connection.execute(statement)
                connection.execute("PRAGMA user_version = 3")
                connection.execute(
                    "INSERT INTO pdd_tasks"
                    " (remote_task_id, task_type, goods_url, status, received_at,"
                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (
                        "COL-V3",
                        "collect",
                        "https://example.test/goods",
                        "succeeded",
                        "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                    ),
                )
        finally:
            connection.close()

        initialize_database(self.db_path)

        connection = open_database(self.db_path)
        try:
            columns = {
                row[1] for row in connection.execute("PRAGMA table_info(pdd_tasks)")
            }
            row = connection.execute(
                "SELECT remote_task_id, removed_at FROM pdd_tasks"
                " WHERE remote_task_id = 'COL-V3'"
            ).fetchone()
            version = connection.execute("PRAGMA user_version").fetchone()[0]
        finally:
            connection.close()
        self.assertIn("removed_at", columns)
        self.assertEqual(tuple(row), ("COL-V3", None))
        self.assertEqual(version, 5)

    def test_v4_database_backfills_shop_name_from_valid_pdd_data(self) -> None:
        connection = open_database(self.db_path)
        try:
            with connection:
                for statement in MIGRATION_1 + MIGRATION_2 + MIGRATION_3 + MIGRATION_4:
                    connection.execute(statement)
                connection.execute("PRAGMA user_version = 4")
                connection.execute(
                    "INSERT INTO pdd_tasks"
                    " (remote_task_id, task_type, goods_url, status, pdd_data,"
                    " received_at, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (
                        "COL-V4", "collect", "https://example.test/goods",
                        "succeeded", '{"shop_name":" 测试店铺 "}',
                        "2026-08-10T00:00:00Z", "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                    ),
                )
        finally:
            connection.close()

        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            row = connection.execute(
                "SELECT shop_name FROM pdd_tasks WHERE remote_task_id = 'COL-V4'"
            ).fetchone()
            version = connection.execute("PRAGMA user_version").fetchone()[0]
        finally:
            connection.close()
        self.assertEqual(row[0], "测试店铺")
        self.assertEqual(version, 5)

    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_execution_mode_constraint_is_enforced(self) -> None:
        initialize_database(self.db_path)
        connection = open_database(self.db_path)
        try:
            with self.assertRaises(sqlite3.IntegrityError):
                connection.execute(
                    "INSERT INTO pdd_tasks"
                    " (remote_task_id, task_type, execution_mode, goods_url,"
                    " status, received_at, created_at, updated_at)"
                    " VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (
                        "PUR-INVALID-MODE",
                        "purchase",
                        "unknown",
                        "https://example.test/goods",
                        "claimed",
                        "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                        "2026-08-10T00:00:00Z",
                    ),
                )
        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()