feat: add batch progress overview

This commit is contained in:
chengma
2026-07-02 10:21:39 +08:00
parent d8c3c5d548
commit b622925d74
5 changed files with 165 additions and 9 deletions
+76
View File
@@ -166,6 +166,82 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertTrue(tab.empty_state_button.isHidden())
self.assert_removed(temp_dir)
def test_workflow_tabs_show_batch_progress_overview(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
db.init_db(cfg["db_path"])
accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
rows = []
for index in range(6):
rows.append(
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "待处理任务",
"source_row": index + 2,
"row_key": f"row-{index}",
"account_name": "主店",
"alias": "alias",
"item_id": str(1000 + index),
}
)
db.insert_tasks(batch_id, rows, path=cfg["db_path"])
tasks = {task.item_id: task for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"])}
db.set_collected(tasks["1001"].id, "旧标题", os.path.join(temp_dir, "old.jpg"), path=cfg["db_path"])
db.set_generated(tasks["1002"].id, "新标题", os.path.join(temp_dir, "new.jpg"), path=cfg["db_path"])
db.set_applied(tasks["1003"].id, committed=True, path=cfg["db_path"])
db.set_collected(tasks["1004"].id, "旧标题", os.path.join(temp_dir, "old-failed.jpg"), path=cfg["db_path"])
db.mark_failed(tasks["1004"].id, "generate", "生成失败", path=cfg["db_path"])
db.mark_skipped(tasks["1005"].id, "别名未匹配", path=cfg["db_path"])
active_other = db.create_batch(["other.xlsx"], path=cfg["db_path"])
db.insert_tasks(
active_other,
[
{
"source_file_abs": os.path.join(temp_dir, "other.xlsx"),
"source_sheet": "待处理任务",
"source_row": 2,
"row_key": "other-row",
"account_name": "主店",
"alias": "alias",
"item_id": "2000",
}
],
path=cfg["db_path"],
)
deleted_batch = db.create_batch(["deleted.xlsx"], path=cfg["db_path"])
db.insert_tasks(
deleted_batch,
[
{
"source_file_abs": os.path.join(temp_dir, "deleted.xlsx"),
"source_sheet": "待处理任务",
"source_row": 2,
"row_key": "deleted-row",
"account_name": "主店",
"alias": "alias",
"item_id": "3000",
}
],
path=cfg["db_path"],
)
db.delete_batch(deleted_batch, reason="测试软删除", path=cfg["db_path"])
expected = "批次进度:总数6 · 导入1 · 已采集1 · 已生成1 · 已更新1 · 失败1 · 略过1"
for tab_class in (CollectTab, GenerateTab, ApplyTab):
tab = tab_class(config=cfg)
self.addCleanup(tab.close)
self.assertEqual(-1, tab.batch_filter.findData(deleted_batch))
tab.batch_filter.setCurrentIndex(tab.batch_filter.findData(batch_id))
tab.refresh_tasks()
self.assertFalse(tab.batch_progress_label.isHidden())
self.assertEqual(expected, tab.batch_progress_label.text())
self.assert_removed(temp_dir)
def test_settings_tab_loads_ai_models_and_masks_key_field(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)