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

feat: add soft delete for import batches

Browse Source
This commit is contained in:
chengma
2026-07-01 09:41:10 +08:00
parent 5b044f2349
commit cc277366ed
10 changed files with 362 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
+55
View File
@@ -167,6 +167,61 @@ class DbTests(TempDirMixin, unittest.TestCase):
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")
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.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")
tests/test_gui.py
+82
View File
@@ -2084,6 +2084,88 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_collect_tab_soft_deletes_batch_and_refreshes_workflow_tabs(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
batch_a = db.create_batch(["input-a.xlsx"], path=cfg["db_path"])
batch_b = db.create_batch(["input-b.xlsx"], path=cfg["db_path"])
db.insert_tasks(
batch_a,
[
{
"source_file_abs": os.path.join(temp_dir, "input-a.xlsx"),
"source_sheet": "商品",
"source_row": 2,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639510",
}
],
path=cfg["db_path"],
)
db.insert_tasks(
batch_b,
[
{
"source_file_abs": os.path.join(temp_dir, "input-b.xlsx"),
"source_sheet": "商品",
"source_row": 2,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639511",
}
],
path=cfg["db_path"],
)
deleted_task = db.list_tasks(batch_id=batch_a, path=cfg["db_path"])[0]
db.set_collected(deleted_task.id, "旧标题", "old.jpg", path=cfg["db_path"])
db.set_generated(deleted_task.id, "新标题", "new.jpg", path=cfg["db_path"])
db.set_applied(deleted_task.id, True, path=cfg["db_path"])
statuses = []
refresh_calls = []
generate_tab = GenerateTab(config=cfg)
apply_tab = ApplyTab(config=cfg)
def refresh_workflow():
refresh_calls.append(True)
generate_tab.refresh_tasks()
apply_tab.refresh_tasks()
collect_tab = CollectTab(
config=cfg,
status_callback=statuses.append,
refresh_workflow_callback=refresh_workflow,
)
self.addCleanup(collect_tab.close)
self.addCleanup(generate_tab.close)
self.addCleanup(apply_tab.close)
collect_tab.batch_filter.setCurrentIndex(collect_tab.batch_filter.findData(batch_a))
with mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.Yes,
) as question, mock.patch("app.gui.QMessageBox.information") as info:
collect_tab.delete_current_batch()
deleted_batch = db.get_batch(batch_a, path=cfg["db_path"], include_deleted=True)
self.assertIsNotNone(deleted_batch.deleted_at)
self.assertIsNone(db.get_batch(batch_a, path=cfg["db_path"]))
self.assertEqual([], db.list_tasks(batch_id=batch_a, path=cfg["db_path"]))
self.assertEqual(1, len(db.list_tasks(batch_id=batch_a, path=cfg["db_path"], include_deleted=True)))
self.assertEqual(-1, collect_tab.batch_filter.findData(batch_a))
self.assertEqual(-1, generate_tab.batch_filter.findData(batch_a))
self.assertEqual(-1, apply_tab.batch_filter.findData(batch_a))
self.assertTrue(all(task.batch_id != batch_a for task in collect_tab.model.all_tasks))
self.assertTrue(all(task.batch_id != batch_a for task in generate_tab.model.tasks))
self.assertTrue(all(task.batch_id != batch_a for task in apply_tab.model.tasks))
self.assertEqual([True], refresh_calls)
self.assertIn("不会回滚 Shopee", question.call_args[0][2])
self.assertIn("已软删除批次", info.call_args[0][2])
self.assertIn("已软删除批次", statuses[-1])
self.assert_removed(temp_dir)
def test_collect_tab_can_filter_unmatched_tasks_from_summary_bar(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)