"""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()