702 lines
30 KiB
Python
702 lines
30 KiB
Python
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"}.issubset(task_columns))
|
|
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)
|
|
|
|
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_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)
|
|
|
|
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)
|
|
|
|
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.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))
|
|
|
|
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"]))
|
|
|
|
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.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.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_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()
|