feat: 完成T-401更新确认页

- 新增 ApplyTab 和 ApplyTaskTableModel,③ Tab 展示已生成更新候选任务

- 支持按批次、店铺、状态筛选,默认列出待更新任务并展示失败/已更新状态

- 开始更新弹窗确认当前筛选条件、任务数和线上提交风险,取消不执行不改库

- 补充 GUI 测试覆盖筛选、确认弹窗和不调用 apply_task 的 T-401 边界

- 同步任务看板、API、路由、当前状态与 progress 文档
This commit is contained in:
chengma
2026-06-27 16:59:46 +08:00
parent e6d7de3d7c
commit 5a63af4620
7 changed files with 510 additions and 10 deletions
+166
View File
@@ -20,6 +20,7 @@ from PySide6.QtWidgets import QApplication, QLineEdit, QPlainTextEdit, QTableVie
from app.gui import (
AccountDialog,
AccountsTab,
ApplyTab,
CollectWorker,
CollectTab,
GenerateWorker,
@@ -68,6 +69,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIn("margin-right: 8px", window.tabs.styleSheet())
self.assertIsInstance(window.tabs.widget(0), CollectTab)
self.assertIsInstance(window.tabs.widget(1), GenerateTab)
self.assertIsInstance(window.tabs.widget(2), ApplyTab)
self.assertEqual(
"回写旧数据到 Excel",
window.tabs.widget(0).write_back_button.text(),
@@ -233,6 +235,170 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_lists_generated_tasks_and_filters_by_batch_shop_status(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)
accounts.create_account("副店", "alias-b", debug_port=9223, 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",
},
{
"source_file_abs": os.path.join(temp_dir, "input-a.xlsx"),
"source_sheet": "商品",
"source_row": 3,
"account_name": "Excel副店",
"alias": "alias-b",
"item_id": "51100639511",
},
{
"source_file_abs": os.path.join(temp_dir, "input-a.xlsx"),
"source_sheet": "商品",
"source_row": 4,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639512",
},
{
"source_file_abs": os.path.join(temp_dir, "input-a.xlsx"),
"source_sheet": "商品",
"source_row": 5,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639513",
},
],
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": "51100639514",
}
],
path=cfg["db_path"],
)
tasks_a = db.list_tasks(batch_id=batch_a, path=cfg["db_path"])
tasks_b = db.list_tasks(batch_id=batch_b, path=cfg["db_path"])
for task in tasks_a[:3] + tasks_b:
db.set_collected(task.id, "旧标题" + task.item_id[-2:], "old.jpg", path=cfg["db_path"])
db.set_generated(
task.id,
"新标题" + task.item_id[-2:],
os.path.join(temp_dir, task.item_id + "_new.jpg"),
path=cfg["db_path"],
)
db.mark_failed(tasks_a[2].id, "apply", "更新失败", path=cfg["db_path"])
db.set_applied(tasks_b[0].id, True, path=cfg["db_path"])
tab = ApplyTab(config=cfg)
self.addCleanup(tab.close)
self.assertIsInstance(tab.task_table, QTableView)
self.assertEqual(["店铺", "商品ID", "新标题", "新封面", "阶段", "结果"], tab.model.HEADERS)
self.assertEqual("开始更新", tab.start_update_button.text())
self.assertEqual("停止", tab.stop_update_button.text())
self.assertFalse(tab.stop_update_button.isEnabled())
self.assertEqual(2, tab.model.rowCount())
self.assertEqual("主店", tab.model.index(0, 0).data())
self.assertEqual("51100639510", tab.model.index(0, 1).data())
self.assertEqual("新标题10", tab.model.index(0, 2).data())
self.assertEqual("51100639510_new.jpg", tab.model.index(0, 3).data())
self.assertEqual("待更新", tab.model.index(0, 4).data())
self.assertEqual("待更新", tab.model.index(0, 5).data())
self.assertEqual("任务 2/4 条", tab.summary_label.text())
tab.status_filter.setCurrentIndex(tab.status_filter.findData("failed"))
self.assertEqual(1, tab.model.rowCount())
self.assertEqual("失败", tab.model.index(0, 5).data())
self.assertEqual(
"更新失败",
tab.model.data(tab.model.index(0, 0), gui.Qt.ToolTipRole),
)
tab.status_filter.setCurrentIndex(tab.status_filter.findData("generated"))
tab.shop_filter.setCurrentIndex(tab.shop_filter.findData("alias-b"))
self.assertEqual(1, tab.model.rowCount())
self.assertEqual("副店", tab.model.index(0, 0).data())
tab.shop_filter.setCurrentIndex(tab.shop_filter.findData(None))
tab.batch_filter.setCurrentIndex(tab.batch_filter.findData(batch_b))
tab.status_filter.setCurrentIndex(tab.status_filter.findData("applied"))
self.assertEqual(1, tab.model.rowCount())
self.assertEqual("已更新", tab.model.index(0, 4).data())
self.assertEqual("成功", tab.model.index(0, 5).data())
self.assert_removed(temp_dir)
def test_apply_tab_start_update_requires_confirmation_without_running_apply(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_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
db.insert_tasks(
batch_id,
[
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "商品",
"source_row": 2,
"account_name": "Excel主店",
"alias": "alias-a",
"item_id": "51100639510",
}
],
path=cfg["db_path"],
)
task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
statuses = []
tab = ApplyTab(config=cfg, status_callback=statuses.append)
self.addCleanup(tab.close)
with mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.No,
) as question, mock.patch("app.gui.editor.apply_task") as apply_task:
tab.start_update()
message = question.call_args[0][2]
self.assertIn("任务数:1", message)
self.assertIn("提交线上", message)
self.assertIn("状态:已生成", message)
self.assertEqual("已取消开始更新", statuses[-1])
apply_task.assert_not_called()
with mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.Yes,
), mock.patch("app.gui.editor.apply_task") as apply_task:
tab.start_update()
self.assertIn("已确认更新范围:1 条", statuses[-1])
apply_task.assert_not_called()
unchanged = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
self.assertEqual("generated", unchanged.stage)
self.assertEqual("success", unchanged.status)
self.assert_removed(temp_dir)
def test_generate_tab_lists_tasks_and_filters_by_shop_status_and_batch(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)