chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: include failed step in task errors

Browse Source
This commit is contained in:
chengma
2026-07-08 11:08:01 +08:00
parent ffbad8e64d
commit 75d545f182
7 changed files with 212 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
+70
View File
@@ -88,6 +88,7 @@ class DbTests(TempDirMixin, unittest.TestCase):
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]
@@ -125,6 +126,75 @@ class DbTests(TempDirMixin, unittest.TestCase):
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: