feat: expand full-flow diagnostics

2026-07-01 15:32:27 +08:00
parent 84b35f1cdc
commit f80f00b5a2
7 changed files with 1085 additions and 53 deletions
@@ -21,6 +21,7 @@ from PySide6.QtWidgets import QApplication, QLineEdit, QPlainTextEdit, QTableVie

 from app.gui import (
    AccountDialog,
+    AccountLoginCheckWorker,
    AccountsTab,
    AIModelTestWorker,
    ApplyTab,
@@ -1434,7 +1435,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
            progress = []
            rows = []

-            def fake_apply(account, task, close_success_tab=False):
+            def fake_apply(account, task, close_success_tab=False, on_step=None):
                applied_aliases.append(account.alias)
                close_flags.append(close_success_tab)
                if account.alias == "alias-a":
@@ -1605,7 +1606,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
            started = {"alias-a": threading.Event(), "alias-b": threading.Event()}
            thread_names = set()

-            def fake_apply(account, task, close_success_tab=False):
+            def fake_apply(account, task, close_success_tab=False, on_step=None):
                thread_names.add(threading.current_thread().name)
                started[account.alias].set()
                other = "alias-b" if account.alias == "alias-a" else "alias-a"
@@ -2769,5 +2770,267 @@ class GuiTests(TempDirMixin, unittest.TestCase):
        )


+    def test_collect_tab_import_excel_writes_diagnostic_run_log(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)
+
+            def fake_import(file_paths, path=None):
+                self.assertEqual([excel_path], file_paths)
+                self.assertEqual(cfg["db_path"], path)
+                return {
+                    "batch_id": None,
+                    "rows": [],
+                    "stats": {
+                        "files": 1,
+                        "total": 2,
+                        "valid": 0,
+                        "invalid": 1,
+                        "inserted": 0,
+                        "file_errors": [
+                            {
+                                "file": excel_path,
+                                "sheet": "商品",
+                                "error": "缺少必需列",
+                                "missing_columns": ["别名"],
+                            }
+                        ],
+                        "row_errors": [
+                            {
+                                "file": excel_path,
+                                "sheet": "商品",
+                                "row": 3,
+                                "error": "商品id必须是数字 token=SECRET",
+                            }
+                        ],
+                    },
+                }
+
+            with mock.patch.object(tab, "_choose_excel_files", return_value=[excel_path]), \
+                    mock.patch("app.gui.excel.import_tasks", side_effect=fake_import):
+                tab.import_excel()
+
+            run_log = db.list_run_logs(limit=1, run_type="import", path=cfg["db_path"])[0]
+            self.assertEqual("done", run_log.status)
+            self.assertEqual(1, run_log.done)
+            self.assertEqual(2, 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=parse_file result=failed", messages)
+            self.assertIn("missing=别名", messages)
+            self.assertIn("step=row_validate result=failed", messages)
+            self.assertIn("step=db_insert result=success", messages)
+            self.assertNotIn("SECRET", messages)
+            self.assertIn("token=***", messages)
+            self.assertIn("入库0", statuses[-1])
+
+        self.assert_removed(temp_dir)
+
+    def test_write_back_worker_writes_run_log_and_diagnostic_on_failure(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            db.init_db(cfg["db_path"])
+            diagnostic_log_dir = os.path.join(temp_dir, "logs")
+
+            with mock.patch(
+                "app.gui.excel.write_back",
+                side_effect=RuntimeError("Excel 文件被占用 token=SECRET"),
+            ):
+                worker = WriteBackWorker(
+                    "batch-1",
+                    db_path=cfg["db_path"],
+                    diagnostic_log_dir=diagnostic_log_dir,
+                )
+                with self.assertRaises(RuntimeError):
+                    worker.execute()
+
+            run_log = db.list_run_logs(limit=1, run_type="write_back", 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=write_excel result=failed", messages)
+            self.assertNotIn("SECRET", messages)
+            self.assertIn("token=***", messages)
+
+            log_path = os.path.join(diagnostic_log_dir, "cmshopee.log")
+            self.assertTrue(os.path.exists(log_path))
+            with open(log_path, "r", encoding="utf-8") as fh:
+                raw_log = fh.read()
+            self.assertNotIn("SECRET", raw_log)
+            entry = json.loads(raw_log.strip().splitlines()[-1])
+            self.assertEqual("write_excel", entry["step"])
+            self.assertEqual("RuntimeError", entry["exception"])
+
+        self.assert_removed(temp_dir)
+
+    def test_apply_worker_writes_step_run_log_and_diagnostic_on_failure(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"])
+            task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+            diagnostic_log_dir = os.path.join(temp_dir, "logs")
+
+            def fake_apply(account, task, close_success_tab=False, on_step=None):
+                on_step({"step": "open_product", "result": "start"})
+                on_step({"step": "replace_cover", "result": "failed", "detail": "token=SECRET"})
+                return {
+                    "committed": False,
+                    "error": "新封面上传失败 token=SECRET",
+                    "cover": {"ok": False, "error": "token=SECRET"},
+                }
+
+            with mock.patch("app.gui.editor.apply_task", side_effect=fake_apply):
+                summary = ApplyWorker(
+                    [task],
+                    db_path=cfg["db_path"],
+                    config=cfg,
+                    preflight=False,
+                    diagnostic_log_dir=diagnostic_log_dir,
+                ).execute()
+
+            self.assertFalse(summary["ok"])
+            self.assertEqual(1, summary["failed"])
+            run_log = db.list_run_logs(limit=1, run_type="apply", path=cfg["db_path"])[0]
+            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=open_product result=start", messages)
+            self.assertIn("step=replace_cover result=failed", messages)
+            self.assertIn("step=db_write result=success", messages)
+            self.assertNotIn("SECRET", messages)
+            self.assertIn("token=***", messages)
+
+            log_path = os.path.join(diagnostic_log_dir, "cmshopee.log")
+            self.assertTrue(os.path.exists(log_path))
+            with open(log_path, "r", encoding="utf-8") as fh:
+                raw_log = fh.read()
+            self.assertNotIn("SECRET", raw_log)
+            entry = json.loads(raw_log.strip().splitlines()[-1])
+            self.assertEqual("replace_cover", entry["step"])
+            self.assertEqual("alias-a", entry["alias"])
+            self.assertEqual("51100639510", entry["item_id"])
+
+        self.assert_removed(temp_dir)
+    def test_account_login_check_worker_writes_run_log(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
+            diagnostic_log_dir = os.path.join(temp_dir, "logs")
+
+            with mock.patch(
+                "app.gui.accounts.detect_login",
+                return_value={"logged_in": False, "reason": "LOGIN_PAGE token=SECRET"},
+            ) as detect_login:
+                result = AccountLoginCheckWorker(
+                    account,
+                    db_path=cfg["db_path"],
+                    config=cfg,
+                    diagnostic_log_dir=diagnostic_log_dir,
+                ).execute()
+
+            detect_login.assert_called_once_with(
+                account,
+                timeout=8,
+                path=cfg["db_path"],
+                config=cfg,
+            )
+            self.assertEqual("alias-a", result["alias"])
+            self.assertFalse(result["status"]["logged_in"])
+            run_log = db.list_run_logs(limit=1, run_type="login_check", path=cfg["db_path"])[0]
+            self.assertEqual("done", 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=detect_login result=start", messages)
+            self.assertIn("step=detect_login result=failed", messages)
+            self.assertNotIn("SECRET", messages)
+            self.assertIn("token=***", messages)
+
+        self.assert_removed(temp_dir)
+
+    def test_ai_model_test_worker_writes_run_log(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            db.init_db(cfg["db_path"])
+            models_path = os.path.join(temp_dir, "ai_models.json")
+            worker = AIModelTestWorker(
+                "Text A",
+                ai_models_path=models_path,
+                db_path=cfg["db_path"],
+                diagnostic_log_dir=os.path.join(temp_dir, "logs"),
+            )
+
+            with mock.patch(
+                "app.gui.appconfig.test_ai_model",
+                return_value={"ok": False, "status": 401, "error": "api_key=SECRET"},
+            ):
+                result = worker.execute()
+
+            self.assertFalse(result["ok"])
+            self.assertEqual("Text A", result["name"])
+            run_log = db.list_run_logs(limit=1, run_type="ai_model_test", path=cfg["db_path"])[0]
+            self.assertEqual("done", 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=test_connection result=start", messages)
+            self.assertIn("step=test_connection result=failed", messages)
+            self.assertNotIn("SECRET", messages)
+            self.assertIn("api_key=***", messages)
+
+        self.assert_removed(temp_dir)
+
+    def test_accounts_tab_launch_login_writes_chrome_launch_run_log(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
+            statuses = []
+            tab = AccountsTab(config=cfg, status_callback=statuses.append)
+            self.addCleanup(tab.close)
+            tab.table.selectRow(0)
+
+            class FakeProcess:
+                pid = 1234
+
+            with mock.patch(
+                "app.gui.accounts.launch_for_login",
+                return_value=FakeProcess(),
+            ) as launch_for_login:
+                tab.launch_login()
+
+            launch_for_login.assert_called_once_with(account, config=cfg)
+            self.assertEqual("已启动", tab.login_statuses["alias-a"])
+            self.assertIn("Chrome 已启动", statuses[-1])
+            run_log = db.list_run_logs(limit=1, run_type="chrome_launch", path=cfg["db_path"])[0]
+            self.assertEqual("done", run_log.status)
+            self.assertEqual(1, run_log.success_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=launch_chrome result=start", messages)
+            self.assertIn("step=launch_chrome result=success", messages)
+            self.assertIn("pid=1234", messages)
+
+        self.assert_removed(temp_dir)
 if __name__ == "__main__":
    unittest.main()