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

T-582 延迟导入失败通知弹窗

Browse Source
This commit is contained in:
chengma
2026-07-10 11:30:19 +08:00
parent 5fa3b921bb
commit f8c3597f47
3 changed files with 56 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui.py
+43 -1
View File
@@ -5907,10 +5907,14 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tab,
"_choose_excel_files",
return_value=[excel_path],
), mock.patch("app.gui.excel.import_tasks", side_effect=fake_import) as import_tasks:
), mock.patch("app.gui.excel.import_tasks", side_effect=fake_import) as import_tasks, mock.patch(
"app.gui.tabs.collect.QTimer.singleShot"
) as single_shot, mock.patch("app.gui.QMessageBox") as message_box:
tab.import_excel()
import_tasks.assert_called_once_with([excel_path], path=cfg["db_path"])
single_shot.assert_not_called()
message_box.warning.assert_not_called()
self.assertEqual(1, tab.model.rowCount())
self.assertEqual("alias-a", tab.model.index(0, 1).data())
self.assertIn("1 文件", tab.summary_label.text())
@@ -5925,6 +5929,44 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_collect_tab_import_error_dialog_is_delayed_one_event_loop(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
excel_path = os.path.join(temp_dir, "input.xlsx")
statuses = []
tab = CollectTab(config=cfg, status_callback=statuses.append)
self.addCleanup(tab.close)
with mock.patch.object(tab, "_choose_excel_files", return_value=[excel_path]), mock.patch(
"app.gui.excel.import_tasks",
side_effect=RuntimeError("Excel导入失败 token=SECRET"),
), mock.patch("app.gui.QMessageBox") as message_box, mock.patch(
"app.gui.tabs.collect.QTimer.singleShot"
) as single_shot:
tab.import_excel()
single_shot.assert_called_once()
self.assertEqual(0, single_shot.call_args[0][0])
delayed_callback = single_shot.call_args[0][1]
message_box.warning.assert_not_called()
self.assertIn("token=***", statuses[-1])
run_log = db.list_run_logs(limit=1, run_type="import", path=cfg["db_path"])[0]
self.assertEqual("failed", run_log.status)
self.assertEqual(1, run_log.failed_count)
events = db.list_run_log_events(run_log.id, path=cfg["db_path"])
messages = "\n".join(event.message for event in events)
self.assertIn("step=import result=failed", messages)
self.assertNotIn("SECRET", messages)
delayed_callback()
message_box.warning.assert_called_once()
self.assertEqual("导入采集", message_box.warning.call_args[0][1])
self.assertIn("Excel导入失败", message_box.warning.call_args[0][2])
self.assertIn("token=***", message_box.warning.call_args[0][2])
self.assert_removed(temp_dir)
def test_collect_worker_collects_success_and_skips_unmatched_or_logged_out(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)