import os import sys import unittest from unittest import mock sys.path.insert(0, os.path.dirname(__file__)) from _helpers import TempDirMixin from app import db class DbTests(TempDirMixin, unittest.TestCase): def test_init_db_is_idempotent_and_sets_pragmas(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) db.init_db(db_path) conn = db.connect(db_path) try: self.assertEqual(1, conn.execute("PRAGMA foreign_keys").fetchone()[0]) self.assertEqual("wal", conn.execute("PRAGMA journal_mode").fetchone()[0]) self.assertEqual(5000, conn.execute("PRAGMA busy_timeout").fetchone()[0]) tables = { row["name"] for row in conn.execute( "SELECT name FROM sqlite_master WHERE type = 'table'" ).fetchall() } self.assertTrue({"batches", "accounts", "tasks"}.issubset(tables)) task_columns = { row["name"] for row in conn.execute("PRAGMA table_info(tasks)").fetchall() } self.assertTrue( { "image_task_id", "image_task_key", "cover_reset_count", "cover_reset_at", "product_status", "product_status_note", "product_status_at", }.issubset(task_columns) ) finally: conn.close() self.assert_removed(temp_dir) def test_init_db_migrates_cover_reset_columns_for_legacy_tasks(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "legacy.sqlite") conn = db.connect(db_path) try: with conn: conn.executescript( """ CREATE TABLE batches ( id TEXT PRIMARY KEY, source_files_json TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'active', note TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE tasks ( id INTEGER PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id), source_file TEXT NOT NULL, source_file_abs TEXT NOT NULL, source_sheet TEXT NOT NULL, source_row INTEGER NOT NULL, row_key TEXT NOT NULL UNIQUE, account_name TEXT, alias TEXT NOT NULL, item_id TEXT NOT NULL, old_title TEXT, old_cover_path TEXT, new_title TEXT, new_cover_path TEXT, image_task_id TEXT, image_task_key TEXT, committed INTEGER NOT NULL DEFAULT 0, stage TEXT NOT NULL DEFAULT 'imported', status TEXT NOT NULL DEFAULT 'pending', last_error TEXT, collect_attempts INTEGER NOT NULL DEFAULT 0, generate_attempts INTEGER NOT NULL DEFAULT 0, apply_attempts INTEGER NOT NULL DEFAULT 0, imported_at TEXT NOT NULL, collected_at TEXT, generated_at TEXT, applied_at TEXT, updated_at TEXT NOT NULL, UNIQUE(batch_id, source_file_abs, source_sheet, source_row) ); INSERT INTO batches (id, source_files_json, status, note, created_at, updated_at) VALUES ('batch', '[]', 'active', NULL, '2026-07-11T00:00:00', '2026-07-11T00:00:00'); INSERT INTO tasks (batch_id, source_file, source_file_abs, source_sheet, source_row, row_key, account_name, alias, item_id, imported_at, updated_at) VALUES ('batch', 'input.xlsx', 'input.xlsx', 'Sheet1', 2, 'row-key', 'shop', 'alias', '51100639510', '2026-07-11T00:00:00', '2026-07-11T00:00:00'); """ ) db.init_db(conn=conn) columns = { row["name"] for row in conn.execute("PRAGMA table_info(tasks)").fetchall() } self.assertIn("cover_reset_count", columns) self.assertIn("cover_reset_at", columns) self.assertIn("product_status", columns) self.assertIn("product_status_note", columns) self.assertIn("product_status_at", columns) task = db.get_task(1, conn=conn) self.assertEqual(0, task.cover_reset_count) self.assertIsNone(task.cover_reset_at) self.assertIsNone(task.product_status) finally: conn.close() self.assert_removed(temp_dir) def test_account_batch_task_lifecycle(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], note="导入", path=db_path) batch = db.get_batch(batch_id, path=db_path) self.assertEqual("导入", batch.note) self.assertEqual([os.path.abspath("input.xlsx")], batch.source_files) account = db.add_account( "shop", "alias", "shopee.tw", 9222, note="备注", path=db_path, ) self.assertEqual("alias", account.alias) self.assertEqual("alias_cdb6fdbe", account.slug) db.update_account("alias", path=db_path, debug_port=9333) self.assertEqual(9333, db.get_account_by_alias("alias", path=db_path).debug_port) self.assertEqual(1, len(db.list_accounts(path=db_path))) count = db.insert_tasks( batch_id, [ { "source_file": "input.xlsx", "source_file_abs": os.path.abspath("input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) self.assertEqual(1, count) task = db.list_tasks(batch_id=batch_id, alias="alias", path=db_path)[0] self.assertEqual("imported", task.stage) self.assertEqual("pending", task.status) self.assertIsNone(task.image_task_id) self.assertIsNone(task.image_task_key) self.assertEqual(0, task.cover_reset_count) self.assertIsNone(task.cover_reset_at) db.mark_running(task.id, "collect", path=db_path) self.assertEqual("running", db.list_tasks(path=db_path)[0].status) db.mark_failed(task.id, "collect", "采集失败", path=db_path) failed = db.list_tasks(path=db_path)[0] self.assertEqual("imported", failed.stage) self.assertEqual("failed", failed.status) self.assertEqual(1, failed.collect_attempts) self.assertEqual("采集失败", failed.last_error) db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) collected = db.list_tasks(path=db_path)[0] self.assertEqual("collected", collected.stage) self.assertEqual("success", collected.status) self.assertEqual("旧标题", collected.old_title) db.set_generated(task.id, "新标题", "new.jpg", path=db_path) generated = db.list_tasks(path=db_path)[0] self.assertEqual("generated", generated.stage) self.assertEqual("新标题", generated.new_title) db.mark_failed(task.id, "generate", "标题需调整", path=db_path) db.update_generated_title(task.id, "人工标题", path=db_path) edited = db.list_tasks(path=db_path)[0] self.assertEqual("generated", edited.stage) self.assertEqual("pending", edited.status) self.assertEqual("人工标题", edited.new_title) self.assertIsNone(edited.last_error) db.set_applied(task.id, False, "按钮禁用", path=db_path) apply_failed = db.list_tasks(path=db_path)[0] self.assertEqual("generated", apply_failed.stage) self.assertEqual("failed", apply_failed.status) self.assertEqual(1, apply_failed.apply_attempts) db.set_applied(task.id, True, path=db_path) applied = db.list_tasks(path=db_path)[0] self.assertEqual("applied", applied.stage) self.assertEqual("success", applied.status) self.assertEqual(1, applied.committed) self.assertEqual(2, applied.apply_attempts) with self.assertRaises(db.DbError): db.update_generated_title(task.id, "线上后改标题", path=db_path) self.assert_removed(temp_dir) def test_product_status_snapshot_preserves_collection_lifecycle_and_content(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] db.set_collected(task.id, "旧标题", "old.jpg", db_path) db.set_product_status( task.id, "reviewing", "审核说明", "2026-07-18T09:00:00", path=db_path, ) status_only = db.get_task(task.id, path=db_path) self.assertEqual("collected", status_only.stage) self.assertEqual("success", status_only.status) self.assertEqual("旧标题", status_only.old_title) self.assertEqual("old.jpg", status_only.old_cover_path) self.assertEqual("reviewing", status_only.product_status) self.assertEqual("审核说明", status_only.product_status_note) self.assertEqual("2026-07-18T09:00:00", status_only.product_status_at) db.set_collected( task.id, "重新采集标题", "new-old.jpg", product_status_value="not-a-status", product_status_note="x" * 2200, product_status_at="2026-07-18T10:00:00", path=db_path, ) refreshed = db.get_task(task.id, path=db_path) self.assertEqual("unknown", refreshed.product_status) self.assertEqual("重新采集标题", refreshed.old_title) self.assertEqual("new-old.jpg", refreshed.old_cover_path) self.assertEqual("2026-07-18T10:00:00", refreshed.product_status_at) self.assertEqual(2000, len(refreshed.product_status_note)) self.assert_removed(temp_dir) def test_set_generated_cover_preserves_title_and_records_ai_success(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": row, "account_name": "shop", "alias": "alias", "item_id": str(51100639510 + row), } for row in (2, 3) ], path=db_path, ) tasks = db.list_tasks(batch_id=batch_id, path=db_path) for task in tasks: db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) db.set_generated(tasks[1].id, "已有新标题", None, path=db_path) before_with_title = db.get_task(tasks[1].id, path=db_path) db.set_generated_cover(tasks[0].id, "cover-a.jpg", path=db_path) db.set_generated_cover(tasks[1].id, "cover-b.jpg", path=db_path) without_title = db.get_task(tasks[0].id, path=db_path) with_title = db.get_task(tasks[1].id, path=db_path) self.assertIsNone(without_title.new_title) self.assertEqual("cover-a.jpg", without_title.new_cover_path) self.assertEqual("generated", without_title.stage) self.assertEqual("success", without_title.status) self.assertEqual(1, without_title.generate_attempts) self.assertIsNotNone(without_title.generated_at) self.assertEqual("已有新标题", with_title.new_title) self.assertEqual("cover-b.jpg", with_title.new_cover_path) self.assertEqual(before_with_title.generate_attempts + 1, with_title.generate_attempts) self.assert_removed(temp_dir) def test_image_task_helpers_and_reset_lifecycle(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) db.set_generated(task.id, "新标题", "new.jpg", path=db_path) key = db.ensure_image_task_key(task.id, path=db_path) self.assertTrue(key.startswith(f"cmshopee-task-{task.id}-")) self.assertEqual(key, db.ensure_image_task_key(task.id, path=db_path)) db.set_image_task_submitted(task.id, "cmhub-task-1", key, path=db_path) submitted = db.get_task(task.id, path=db_path) self.assertEqual("cmhub-task-1", submitted.image_task_id) self.assertEqual(key, submitted.image_task_key) title_only = db.reset_generated( task.id, reset_title=True, reset_cover=False, path=db_path, )["after"] self.assertEqual("cmhub-task-1", title_only.image_task_id) self.assertEqual(key, title_only.image_task_key) self.assertEqual(0, title_only.cover_reset_count) self.assertIsNone(title_only.cover_reset_at) cover_reset = db.reset_generated( task.id, reset_title=False, reset_cover=True, path=db_path, )["after"] self.assertIsNone(cover_reset.image_task_id) self.assertIsNone(cover_reset.image_task_key) self.assertEqual(1, cover_reset.cover_reset_count) self.assertIsNotNone(cover_reset.cover_reset_at) new_key = db.ensure_image_task_key(task.id, path=db_path) self.assertNotEqual(key, new_key) db.clear_image_task(task.id, path=db_path) cleared = db.get_task(task.id, path=db_path) self.assertIsNone(cleared.image_task_id) self.assertIsNone(cleared.image_task_key) self.assert_removed(temp_dir) def test_mark_failed_prepends_chinese_step_without_duplicate_prefix(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] db.mark_failed(task.id, "collect", "图片超过大小上限", path=db_path, step="read_cover") failed = db.get_task(task.id, path=db_path) self.assertEqual("读封面失败：图片超过大小上限", failed.last_error) db.mark_failed(task.id, "collect", "读封面失败：图片超过大小上限", path=db_path, step="read_cover") failed = db.get_task(task.id, path=db_path) self.assertEqual("读封面失败：图片超过大小上限", failed.last_error) db.mark_failed(task.id, "collect", "未知错误", path=db_path, step="custom_step") failed = db.get_task(task.id, path=db_path) self.assertEqual("custom_step失败：未知错误", failed.last_error) db.mark_failed(task.id, "collect", "保持原样", path=db_path) failed = db.get_task(task.id, path=db_path) self.assertEqual("保持原样", failed.last_error) self.assert_removed(temp_dir) def test_set_applied_prepends_failed_step_for_update_errors(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] db.set_generated(task.id, "新标题", "new.jpg", path=db_path) db.set_applied(task.id, False, "按钮禁用", path=db_path, step="click_update") failed = db.get_task(task.id, path=db_path) self.assertEqual("点击更新失败：按钮禁用", failed.last_error) self.assertEqual("failed", failed.status) self.assert_removed(temp_dir) def test_reset_generated_and_apply_status_keep_local_history(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] new_cover = os.path.join(temp_dir, "new.jpg") with open(new_cover, "wb") as fh: fh.write(b"jpeg") db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) db.set_generated(task.id, "新标题", new_cover, path=db_path) db.set_applied(task.id, True, path=db_path) reset_apply = db.reset_apply_status(task.id, path=db_path) self.assertEqual("applied", reset_apply["before"].stage) after_apply = reset_apply["after"] self.assertEqual("generated", after_apply.stage) self.assertEqual("pending", after_apply.status) self.assertEqual("新标题", after_apply.new_title) self.assertEqual(new_cover, after_apply.new_cover_path) self.assertEqual(1, after_apply.committed) reset_generated = db.reset_generated(task.id, path=db_path) after_generated = reset_generated["after"] self.assertEqual(new_cover, reset_generated["new_cover_path"]) self.assertIsNotNone(reset_generated["archived_file"]) self.assertIsNone(reset_generated["deleted_file"]) self.assertFalse(os.path.exists(new_cover)) self.assertTrue(os.path.exists(reset_generated["archived_file"])) self.assertEqual("generated", after_generated.stage) self.assertEqual("success", after_generated.status) self.assertIsNone(after_generated.new_title) self.assertIsNone(after_generated.new_cover_path) self.assertIsNone(after_generated.last_error) self.assertEqual(1, after_generated.committed) self.assertEqual(1, after_generated.cover_reset_count) self.assertIsNotNone(after_generated.cover_reset_at) self.assert_removed(temp_dir) def test_reset_generated_can_reset_title_or_cover_components(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", }, { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 3, "account_name": "shop", "alias": "alias", "item_id": "51100639511", }, ], path=db_path, ) first, second = db.list_tasks(batch_id=batch_id, path=db_path) first_cover = os.path.join(temp_dir, "first_new.jpg") second_cover = os.path.join(temp_dir, "second_new.jpg") for cover in [first_cover, second_cover]: with open(cover, "wb") as fh: fh.write(b"jpeg") db.set_collected(first.id, "旧标题A", "old-a.jpg", path=db_path) db.set_generated(first.id, "手动标题A", first_cover, path=db_path) db.mark_failed(first.id, "generate", "图片不满意", path=db_path) db.set_collected(second.id, "旧标题B", "old-b.jpg", path=db_path) db.set_generated(second.id, "手动标题B", second_cover, path=db_path) title_reset = db.reset_generated( first.id, reset_title=True, reset_cover=False, path=db_path, ) after_title = title_reset["after"] self.assertEqual("generated", after_title.stage) self.assertEqual("success", after_title.status) self.assertIsNone(after_title.new_title) self.assertEqual(first_cover, after_title.new_cover_path) self.assertIsNone(after_title.last_error) self.assertTrue(os.path.exists(first_cover)) self.assertEqual(0, after_title.cover_reset_count) self.assertIsNone(after_title.cover_reset_at) cover_reset = db.reset_generated( second.id, reset_title=False, reset_cover=True, delete_file=True, path=db_path, ) after_cover = cover_reset["after"] self.assertEqual("generated", after_cover.stage) self.assertEqual("success", after_cover.status) self.assertEqual("手动标题B", after_cover.new_title) self.assertIsNone(after_cover.new_cover_path) self.assertIsNone(cover_reset["deleted_file"]) self.assertIsNotNone(cover_reset["archived_file"]) self.assertFalse(os.path.exists(second_cover)) self.assertTrue(os.path.exists(cover_reset["archived_file"])) self.assertEqual(1, after_cover.cover_reset_count) self.assertIsNotNone(after_cover.cover_reset_at) with self.assertRaises(db.DbError): db.reset_generated(first.id, reset_title=False, reset_cover=False, path=db_path) with self.assertRaises(db.DbError): db.reset_generated(first.id, reset_title=True, reset_cover=False, delete_file=True, path=db_path) self.assert_removed(temp_dir) def test_reset_generated_archives_cover_with_collision_name(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] cover = os.path.join(temp_dir, "42_51100639510_new.jpg") collision = os.path.join(temp_dir, "42_51100639510_new_20260709094800.jpg") expected_archive = os.path.join(temp_dir, "42_51100639510_new_20260709094800_2.jpg") with open(cover, "wb") as fh: fh.write(b"current") with open(collision, "wb") as fh: fh.write(b"existing") db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) db.set_generated(task.id, "新标题", cover, path=db_path) with mock.patch("app.db._cover_archive_timestamp", return_value="20260709094800"): result = db.reset_generated( task.id, reset_title=False, reset_cover=True, path=db_path, ) self.assertEqual(expected_archive, result["archived_file"]) self.assertFalse(os.path.exists(cover)) self.assertTrue(os.path.exists(collision)) self.assertTrue(os.path.exists(expected_archive)) with open(collision, "rb") as fh: self.assertEqual(b"existing", fh.read()) with open(expected_archive, "rb") as fh: self.assertEqual(b"current", fh.read()) after = db.get_task(task.id, path=db_path) self.assertEqual("新标题", after.new_title) self.assertIsNone(after.new_cover_path) self.assertEqual(1, after.cover_reset_count) self.assert_removed(temp_dir) def test_reset_generated_permission_error_keeps_db_pointer(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] cover = os.path.join(temp_dir, "new.jpg") with open(cover, "wb") as fh: fh.write(b"jpeg") db.set_generated(task.id, "新标题", cover, path=db_path) with mock.patch("app.db.os.rename", side_effect=PermissionError): with self.assertRaisesRegex(db.DbError, "请先关闭正在查看的封面图片再重置"): db.reset_generated(task.id, reset_title=False, reset_cover=True, path=db_path) after = db.get_task(task.id, path=db_path) self.assertEqual(cover, after.new_cover_path) self.assertEqual("新标题", after.new_title) self.assertEqual(0, after.cover_reset_count) self.assertIsNone(after.cover_reset_at) self.assertTrue(os.path.exists(cover)) self.assert_removed(temp_dir) def test_update_generated_cover_switches_pointer_and_keeps_apply_history(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] current_cover = os.path.join(temp_dir, "current.jpg") selected_cover = os.path.join(temp_dir, "selected.jpg") for cover in (current_cover, selected_cover): with open(cover, "wb") as fh: fh.write(b"jpeg") db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) db.set_generated(task.id, "新标题", current_cover, path=db_path) key = db.ensure_image_task_key(task.id, path=db_path) db.set_image_task_submitted(task.id, "cmhub-task-1", key, path=db_path) db.set_applied(task.id, True, path=db_path) before = db.get_task(task.id, path=db_path) db.update_generated_cover(task.id, selected_cover, path=db_path) after = db.get_task(task.id, path=db_path) self.assertEqual(os.path.abspath(selected_cover), after.new_cover_path) self.assertEqual("generated", after.stage) self.assertEqual("pending", after.status) self.assertIsNone(after.last_error) self.assertEqual(before.generate_attempts, after.generate_attempts) self.assertEqual(before.apply_attempts, after.apply_attempts) self.assertEqual(before.committed, after.committed) self.assertEqual(before.applied_at, after.applied_at) self.assertEqual(before.image_task_id, after.image_task_id) self.assertEqual(before.image_task_key, after.image_task_key) with self.assertRaisesRegex(db.DbError, "新封面图片不存在"): db.update_generated_cover(task.id, os.path.join(temp_dir, "missing.jpg"), path=db_path) self.assert_removed(temp_dir) def test_update_generated_cover_rejects_running_task(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", } ], path=db_path, ) task = db.list_tasks(batch_id=batch_id, path=db_path)[0] cover = os.path.join(temp_dir, "selected.jpg") with open(cover, "wb") as fh: fh.write(b"jpeg") db.mark_running(task.id, "generate", path=db_path) with self.assertRaisesRegex(db.DbError, "任务正在运行"): db.update_generated_cover(task.id, cover, path=db_path) self.assert_removed(temp_dir) def test_cover_reset_history_persists_after_selecting_or_generating_cover(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", }, { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 3, "account_name": "shop", "alias": "alias", "item_id": "51100639511", }, ], path=db_path, ) first, second = db.list_tasks(batch_id=batch_id, path=db_path) first_cover = os.path.join(temp_dir, "first.jpg") selected_cover = os.path.join(temp_dir, "selected.jpg") regenerated_cover = os.path.join(temp_dir, "regenerated.jpg") for cover in (first_cover, selected_cover, regenerated_cover): with open(cover, "wb") as fh: fh.write(b"jpeg") db.set_generated(first.id, "新标题", first_cover, path=db_path) reset = db.reset_generated( first.id, reset_title=False, reset_cover=True, path=db_path, )["after"] self.assertEqual(1, reset.cover_reset_count) self.assertIsNotNone(reset.cover_reset_at) db.update_generated_cover(first.id, selected_cover, path=db_path) selected = db.get_task(first.id, path=db_path) self.assertEqual(1, selected.cover_reset_count) self.assertIsNotNone(selected.cover_reset_at) db.set_generated(first.id, selected.new_title, regenerated_cover, path=db_path) regenerated = db.get_task(first.id, path=db_path) self.assertEqual(1, regenerated.cover_reset_count) self.assertIsNotNone(regenerated.cover_reset_at) empty_reset = db.reset_generated( second.id, reset_title=False, reset_cover=True, path=db_path, )["after"] self.assertEqual(0, empty_reset.cover_reset_count) self.assertIsNone(empty_reset.cover_reset_at) self.assert_removed(temp_dir) def test_delete_batch_soft_hides_batch_and_tasks(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) db.insert_tasks( batch_id, [ { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "account_name": "shop", "alias": "alias", "item_id": "51100639510", }, { "source_file_abs": os.path.join(temp_dir, "input.xlsx"), "source_sheet": "Sheet1", "source_row": 3, "account_name": "shop", "alias": "alias", "item_id": "51100639511", }, ], path=db_path, ) tasks = db.list_tasks(batch_id=batch_id, path=db_path) old_cover = os.path.join(temp_dir, "old.jpg") new_cover = os.path.join(temp_dir, "new.jpg") batch_image_dir = os.path.join(temp_dir, "images", batch_id, "alias") os.makedirs(batch_image_dir, exist_ok=True) generated_cover = os.path.join(batch_image_dir, "1_51100639510_new.jpg") with open(generated_cover, "wb") as fh: fh.write(b"jpeg") db.set_collected(tasks[0].id, "旧标题", old_cover, path=db_path) db.set_generated(tasks[0].id, "新标题", new_cover, path=db_path) db.set_applied(tasks[0].id, True, path=db_path) result = db.delete_batch(batch_id, reason="导错文件", path=db_path) self.assertEqual(batch_id, result["batch_id"]) self.assertEqual(2, result["task_count"]) self.assertEqual(1, result["committed_count"]) self.assertEqual([old_cover, new_cover], result["image_paths"]) self.assertEqual([], db.list_batches(path=db_path)) self.assertEqual([], db.list_tasks(path=db_path)) self.assertIsNone(db.get_batch(batch_id, path=db_path)) self.assertIsNone(db.get_task(tasks[0].id, path=db_path)) deleted_task = db.get_task(tasks[0].id, path=db_path, include_deleted=True) self.assertEqual(tasks[0].id, deleted_task.id) deleted_batch = db.get_batch(batch_id, path=db_path, include_deleted=True) self.assertIsNotNone(deleted_batch.deleted_at) self.assertEqual("导错文件", deleted_batch.deleted_reason) deleted_tasks = db.list_tasks(batch_id=batch_id, path=db_path, include_deleted=True) self.assertEqual(2, len(deleted_tasks)) self.assertTrue(os.path.exists(batch_image_dir)) self.assertTrue(os.path.exists(generated_cover)) self.assert_removed(temp_dir) def test_duplicate_task_and_invalid_update_raise_clear_errors(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) batch_id = db.create_batch(["input.xlsx"], path=db_path) row = { "source_file": "input.xlsx", "source_file_abs": os.path.abspath("input.xlsx"), "source_sheet": "Sheet1", "source_row": 2, "alias": "alias", "item_id": "51100639510", } db.insert_tasks(batch_id, [row], path=db_path) with self.assertRaises(db.DbError): db.insert_tasks(batch_id, [row], path=db_path) with self.assertRaises(db.DbError): db.update_account("alias", path=db_path, unknown_field=True) self.assert_removed(temp_dir) def test_run_logs_and_events_are_persisted(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") db.init_db(db_path) run_id = db.create_run_log( "apply", dry_run=True, total=2, options={"api_key": "secret", "mode": "preview"}, path=db_path, ) db.add_run_log_event( run_id, "dry-run 预览任务", task_id=7, alias="alias", item_id="51100639510", path=db_path, ) db.finish_run_log( run_id, status="done", done=2, success_count=1, skipped_count=1, failed_count=0, summary_json={"password": "secret", "done": 2}, path=db_path, ) run = db.list_run_logs(path=db_path)[0] self.assertEqual(run_id, run.id) self.assertEqual("apply", run.run_type) self.assertEqual(1, run.dry_run) self.assertEqual("done", run.status) self.assertEqual(2, run.done) self.assertEqual("***", run.options["api_key"]) self.assertEqual("***", run.summary["password"]) events = db.list_run_log_events(run_id, path=db_path) self.assertEqual(1, len(events)) self.assertEqual("alias", events[0].alias) self.assertEqual("51100639510", events[0].item_id) self.assertIn("dry-run", events[0].message) self.assert_removed(temp_dir) if __name__ == "__main__": unittest.main()