chengma
6e690cc02a
新增GenerateTab和GenerateTaskTableModel，将② AI生成从占位页替换为左右布局：左侧标题/封面提示词多行输入，右侧批次、店铺、状态筛选和任务列表。 任务列表展示店铺、商品ID、旧标题、新标题、状态；筛选只读SQLite，不调用AI、不写库，生成执行留给T-303，提示词管理留给T-302p。 补充GUI测试覆盖Tab②挂载、提示词输入区、任务列表展示与批次/店铺/状态筛选；同步任务看板、API、routes、current-state和progress。
727 lines
30 KiB
Python
727 lines
30 KiB
Python
import unittest
|
|
import os
|
|
import sys
|
|
from unittest import mock
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from _helpers import TempDirMixin
|
|
|
|
from app import gui
|
|
from app import accounts, db
|
|
|
|
if gui.QT_IMPORT_ERROR is not None:
|
|
raise unittest.SkipTest("PySide6 未安装")
|
|
|
|
from PySide6.QtWidgets import QApplication, QLineEdit, QPlainTextEdit, QTableView
|
|
|
|
from app.gui import (
|
|
AccountDialog,
|
|
AccountsTab,
|
|
CollectWorker,
|
|
CollectTab,
|
|
GenerateTab,
|
|
MainWindow,
|
|
TAB_STYLE,
|
|
TAB_TITLES,
|
|
WriteBackWorker,
|
|
)
|
|
|
|
|
|
class GuiTests(TempDirMixin, unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
|
|
def tearDown(self):
|
|
for widget in QApplication.topLevelWidgets():
|
|
widget.close()
|
|
widget.deleteLater()
|
|
self.app.processEvents()
|
|
|
|
def make_config(self, temp_dir):
|
|
return {
|
|
"chrome_path": "chrome.exe",
|
|
"user_data_root": os.path.join(temp_dir, "chrome_user_data_dir"),
|
|
"image_dir": os.path.join(temp_dir, "images"),
|
|
"db_path": os.path.join(temp_dir, "cmshopee.db"),
|
|
"debug_port_range": [9222, 9260],
|
|
}
|
|
|
|
def test_main_window_has_five_tabs_in_workflow_order(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
window = MainWindow(config=self.make_config(temp_dir))
|
|
self.addCleanup(window.close)
|
|
|
|
self.assertEqual(5, window.tabs.count())
|
|
self.assertEqual(
|
|
TAB_TITLES,
|
|
[window.tabs.tabText(index) for index in range(window.tabs.count())],
|
|
)
|
|
self.assertEqual("就绪", window.statusBar().currentMessage())
|
|
self.assertEqual(TAB_STYLE, window.tabs.styleSheet())
|
|
self.assertIn("min-width: 128px", window.tabs.styleSheet())
|
|
self.assertIn("padding: 8px 18px", window.tabs.styleSheet())
|
|
self.assertIn("margin-right: 8px", window.tabs.styleSheet())
|
|
self.assertIsInstance(window.tabs.widget(0), CollectTab)
|
|
self.assertIsInstance(window.tabs.widget(1), GenerateTab)
|
|
self.assertEqual(
|
|
"回写旧数据到 Excel",
|
|
window.tabs.widget(0).write_back_button.text(),
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_generate_tab_has_prompt_editors_and_task_table(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
tab = GenerateTab(config=self.make_config(temp_dir))
|
|
self.addCleanup(tab.close)
|
|
|
|
self.assertIsInstance(tab.title_prompt_edit, QPlainTextEdit)
|
|
self.assertIsInstance(tab.cover_prompt_edit, QPlainTextEdit)
|
|
self.assertIsInstance(tab.task_table, QTableView)
|
|
self.assertEqual("标题提示词", tab.title_prompt_edit.placeholderText())
|
|
self.assertEqual("封面提示词", tab.cover_prompt_edit.placeholderText())
|
|
self.assertEqual(["店铺", "商品ID", "旧标题", "新标题", "状态"], tab.model.HEADERS)
|
|
self.assertEqual("任务 0/0 条", tab.summary_label.text())
|
|
|
|
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)
|
|
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",
|
|
},
|
|
],
|
|
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": "51100639513",
|
|
}
|
|
],
|
|
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"])
|
|
db.set_collected(tasks_a[0].id, "旧标题A", "old-a.jpg", path=cfg["db_path"])
|
|
db.set_collected(tasks_a[1].id, "旧标题B", "old-b.jpg", path=cfg["db_path"])
|
|
db.set_collected(tasks_a[2].id, "旧标题C", "old-c.jpg", path=cfg["db_path"])
|
|
db.set_generated(tasks_a[2].id, "新标题C", "new-c.jpg", path=cfg["db_path"])
|
|
db.mark_failed(tasks_b[0].id, "generate", "生成失败", path=cfg["db_path"])
|
|
|
|
tab = GenerateTab(config=cfg)
|
|
self.addCleanup(tab.close)
|
|
|
|
self.assertEqual(4, tab.model.rowCount())
|
|
self.assertEqual("主店", tab.model.index(0, 0).data())
|
|
self.assertEqual("51100639510", tab.model.index(0, 1).data())
|
|
self.assertEqual("旧标题A", tab.model.index(0, 2).data())
|
|
self.assertEqual("", tab.model.index(0, 3).data())
|
|
self.assertEqual("待生成", tab.model.index(0, 4).data())
|
|
self.assertEqual("已生成", tab.model.index(2, 4).data())
|
|
|
|
tab.status_filter.setCurrentIndex(tab.status_filter.findData("to_generate"))
|
|
self.assertEqual(2, tab.model.rowCount())
|
|
self.assertTrue(all(tab.model.index(row, 4).data() == "待生成" for row in range(2)))
|
|
self.assertEqual("任务 2/4 条", tab.summary_label.text())
|
|
|
|
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.status_filter.setCurrentIndex(tab.status_filter.findData("all"))
|
|
tab.batch_filter.setCurrentIndex(tab.batch_filter.findData(batch_b))
|
|
self.assertEqual(1, tab.model.rowCount())
|
|
self.assertEqual("失败", tab.model.index(0, 4).data())
|
|
self.assertEqual(
|
|
"生成失败",
|
|
tab.model.data(tab.model.index(0, 0), gui.Qt.ToolTipRole),
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_tab_switch_updates_status_bar(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
window = MainWindow(config=self.make_config(temp_dir))
|
|
self.addCleanup(window.close)
|
|
|
|
window.tabs.setCurrentIndex(2)
|
|
|
|
self.assertEqual("③ 更新shopee", window.tabs.tabText(window.tabs.currentIndex()))
|
|
self.assertEqual("当前：③ 更新shopee", window.statusBar().currentMessage())
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_accounts_tab_lists_accounts_without_showing_password(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account(
|
|
"主店",
|
|
"alias",
|
|
"seller.shopee.tw",
|
|
9222,
|
|
password="secret",
|
|
config=cfg,
|
|
)
|
|
|
|
window = MainWindow(config=cfg)
|
|
self.addCleanup(window.close)
|
|
tab = window.tabs.widget(TAB_TITLES.index("④ 账号管理"))
|
|
|
|
self.assertIsInstance(tab, AccountsTab)
|
|
self.assertEqual(1, tab.table.rowCount())
|
|
self.assertEqual("主店", tab.table.item(0, 0).text())
|
|
self.assertEqual("alias", tab.table.item(0, 1).text())
|
|
self.assertEqual("未知", tab.table.item(0, 4).text())
|
|
visible_values = [
|
|
tab.table.item(0, column).text()
|
|
for column in range(tab.table.columnCount())
|
|
if tab.table.item(0, column) is not None
|
|
]
|
|
self.assertNotIn("secret", visible_values)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_account_dialog_masks_password_field(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
account = accounts.create_account(
|
|
"主店",
|
|
"alias",
|
|
debug_port=9222,
|
|
password="secret",
|
|
config=cfg,
|
|
)
|
|
|
|
dialog = AccountDialog(account=account, config=cfg)
|
|
self.addCleanup(dialog.close)
|
|
|
|
self.assertEqual(QLineEdit.Password, dialog.password_edit.echoMode())
|
|
self.assertEqual("secret", dialog.password_edit.text())
|
|
self.assertIn(account.slug, dialog.user_data_dir_edit.text())
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_accounts_tab_updates_login_status_cell(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
tab = AccountsTab(config=cfg)
|
|
self.addCleanup(tab.close)
|
|
|
|
tab._on_login_check_finished(
|
|
{"alias": "alias", "status": {"logged_in": True, "reason": None}}
|
|
)
|
|
|
|
self.assertEqual("已登录", tab.table.item(0, 4).text())
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_accounts_tab_create_shortcut_for_selected_account(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
account = accounts.create_account("主店", "alias", debug_port=9222, config=cfg)
|
|
statuses = []
|
|
tab = AccountsTab(config=cfg, status_callback=statuses.append)
|
|
self.addCleanup(tab.close)
|
|
tab.table.selectRow(0)
|
|
shortcut_path = os.path.join(temp_dir, "Desktop", "主店.lnk")
|
|
|
|
with mock.patch(
|
|
"app.gui.accounts.create_shortcut",
|
|
return_value=shortcut_path,
|
|
) as create_shortcut, mock.patch("app.gui.QMessageBox.information") as info:
|
|
tab.create_shortcut()
|
|
|
|
create_shortcut.assert_called_once_with(account, config=cfg)
|
|
info.assert_called_once()
|
|
self.assertIn(shortcut_path, statuses[-1])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_tab_lists_tasks_with_unmatched_alias_as_skipped(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",
|
|
},
|
|
{
|
|
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
|
|
"source_sheet": "商品",
|
|
"source_row": 3,
|
|
"account_name": "Excel副店",
|
|
"alias": "missing",
|
|
"item_id": "51100639511",
|
|
},
|
|
],
|
|
path=cfg["db_path"],
|
|
)
|
|
|
|
tab = CollectTab(config=cfg)
|
|
self.addCleanup(tab.close)
|
|
|
|
self.assertIsInstance(tab.table, QTableView)
|
|
self.assertEqual(2, tab.model.rowCount())
|
|
self.assertEqual("主店", tab.model.index(0, 0).data())
|
|
self.assertEqual("alias-a", tab.model.index(0, 1).data())
|
|
self.assertEqual("51100639510", tab.model.index(0, 2).data())
|
|
self.assertEqual("待采集", tab.model.index(0, 3).data())
|
|
self.assertEqual("Excel副店", tab.model.index(1, 0).data())
|
|
self.assertEqual("missing", tab.model.index(1, 1).data())
|
|
self.assertEqual("略过", tab.model.index(1, 3).data())
|
|
self.assertIn("2 行", tab.summary_label.text())
|
|
self.assertIn("有效2/无效0", tab.summary_label.text())
|
|
self.assertIn("匹配1", tab.summary_label.text())
|
|
self.assertIn("未匹配1", tab.summary_label.text())
|
|
self.assertIn("主店1", tab.match_detail_label.text())
|
|
self.assertIn("未匹配", tab.empty_label.text())
|
|
|
|
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)
|
|
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",
|
|
},
|
|
{
|
|
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
|
|
"source_sheet": "商品",
|
|
"source_row": 3,
|
|
"account_name": "Excel副店",
|
|
"alias": "missing",
|
|
"item_id": "51100639511",
|
|
},
|
|
],
|
|
path=cfg["db_path"],
|
|
)
|
|
tab = CollectTab(config=cfg)
|
|
self.addCleanup(tab.close)
|
|
|
|
tab.show_unmatched_tasks()
|
|
|
|
self.assertEqual(1, tab.model.rowCount())
|
|
self.assertEqual("missing", tab.model.index(0, 1).data())
|
|
self.assertEqual("略过", tab.model.index(0, 3).data())
|
|
|
|
tab.show_all_tasks()
|
|
|
|
self.assertEqual(2, tab.model.rowCount())
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_tab_import_button_imports_excel_and_refreshes_tasks(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)
|
|
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):
|
|
batch_id = db.create_batch(file_paths, path=path)
|
|
db.insert_tasks(
|
|
batch_id,
|
|
[
|
|
{
|
|
"source_file_abs": excel_path,
|
|
"source_sheet": "商品",
|
|
"source_row": 2,
|
|
"account_name": "Excel主店",
|
|
"alias": "alias-a",
|
|
"item_id": "51100639510",
|
|
}
|
|
],
|
|
path=path,
|
|
)
|
|
return {
|
|
"batch_id": batch_id,
|
|
"rows": [],
|
|
"stats": {
|
|
"files": 1,
|
|
"total": 2,
|
|
"valid": 1,
|
|
"invalid": 1,
|
|
"inserted": 1,
|
|
},
|
|
}
|
|
|
|
with mock.patch.object(
|
|
tab,
|
|
"_choose_excel_files",
|
|
return_value=[excel_path],
|
|
), mock.patch("app.gui.excel.import_tasks", side_effect=fake_import) as import_tasks:
|
|
tab.import_excel()
|
|
|
|
import_tasks.assert_called_once_with([excel_path], path=cfg["db_path"])
|
|
self.assertEqual(1, tab.model.rowCount())
|
|
self.assertEqual("alias-a", tab.model.index(0, 1).data())
|
|
self.assertIn("1 文件", tab.summary_label.text())
|
|
self.assertIn("2 行", tab.summary_label.text())
|
|
self.assertIn("有效1/无效1", tab.summary_label.text())
|
|
self.assertIn("匹配1", tab.summary_label.text())
|
|
self.assertIn("入库1", statuses[-1])
|
|
|
|
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)
|
|
logged = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
|
|
logged_out = accounts.create_account("副店", "alias-b", debug_port=9223, 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",
|
|
},
|
|
{
|
|
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
|
|
"source_sheet": "商品",
|
|
"source_row": 3,
|
|
"account_name": "Excel副店",
|
|
"alias": "alias-b",
|
|
"item_id": "51100639511",
|
|
},
|
|
{
|
|
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
|
|
"source_sheet": "商品",
|
|
"source_row": 4,
|
|
"account_name": "Excel未知",
|
|
"alias": "missing",
|
|
"item_id": "51100639512",
|
|
},
|
|
],
|
|
path=cfg["db_path"],
|
|
)
|
|
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
|
|
|
|
def fake_login(account, path=None, config=None):
|
|
self.assertEqual(cfg["db_path"], path)
|
|
if account.alias == logged.alias:
|
|
return {"logged_in": True, "reason": None}
|
|
if account.alias == logged_out.alias:
|
|
return {"logged_in": False, "reason": "LOGIN_PAGE"}
|
|
raise AssertionError(account.alias)
|
|
|
|
def fake_collect(account, task):
|
|
self.assertEqual(logged.alias, account.alias)
|
|
self.assertEqual("51100639510", task["item_id"])
|
|
self.assertTrue(task["old_cover_path"].endswith(
|
|
os.path.join(logged.slug, "51100639510_old.jpg")
|
|
))
|
|
return {
|
|
"old_title": "旧标题",
|
|
"old_cover_path": task["old_cover_path"],
|
|
}
|
|
|
|
with mock.patch("app.gui.accounts.detect_login", side_effect=fake_login), \
|
|
mock.patch("app.gui.editor.collect", side_effect=fake_collect) as collect:
|
|
summary = CollectWorker(
|
|
tasks,
|
|
db_path=cfg["db_path"],
|
|
config=cfg,
|
|
preflight=False,
|
|
).execute()
|
|
|
|
self.assertEqual({"ok": True, "total": 3, "done": 3, "collected": 1, "skipped": 2, "failed": 0}, summary)
|
|
collect.assert_called_once()
|
|
updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
|
|
by_alias = {task.alias: task for task in updated}
|
|
self.assertEqual("collected", by_alias["alias-a"].stage)
|
|
self.assertEqual("success", by_alias["alias-a"].status)
|
|
self.assertEqual("旧标题", by_alias["alias-a"].old_title)
|
|
self.assertTrue(by_alias["alias-a"].old_cover_path.endswith("51100639510_old.jpg"))
|
|
self.assertEqual("imported", by_alias["alias-b"].stage)
|
|
self.assertEqual("skipped", by_alias["alias-b"].status)
|
|
self.assertIn("账号未登录", by_alias["alias-b"].last_error)
|
|
self.assertEqual("skipped", by_alias["missing"].status)
|
|
self.assertEqual("别名未匹配账号", by_alias["missing"].last_error)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_tab_auto_starts_write_back_after_collect_success(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
db.init_db(cfg["db_path"])
|
|
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"],
|
|
)
|
|
statuses = []
|
|
tab = CollectTab(config=cfg, status_callback=statuses.append)
|
|
self.addCleanup(tab.close)
|
|
|
|
with mock.patch.object(tab, "_start_write_back", return_value=True) as start_write_back:
|
|
tab._on_collect_finished({"collected": 1, "skipped": 0, "failed": 0})
|
|
|
|
start_write_back.assert_called_once_with(batch_id, auto=True)
|
|
self.assertIn("正在自动回写 Excel", statuses[-1])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_tab_does_not_auto_write_back_when_nothing_collected(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
statuses = []
|
|
tab = CollectTab(config=cfg, status_callback=statuses.append)
|
|
self.addCleanup(tab.close)
|
|
|
|
with mock.patch.object(tab, "_start_write_back") as start_write_back:
|
|
tab._on_collect_finished({"collected": 0, "skipped": 1, "failed": 0})
|
|
|
|
start_write_back.assert_not_called()
|
|
self.assertEqual("采集完成：成功0，略过1，失败0", statuses[-1])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_auto_write_back_locked_file_message_points_to_manual_retry(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
statuses = []
|
|
tab = CollectTab(config=self.make_config(temp_dir), status_callback=statuses.append)
|
|
self.addCleanup(tab.close)
|
|
|
|
with mock.patch("app.gui.QMessageBox.warning") as warning:
|
|
tab._on_write_back_failed(
|
|
-1,
|
|
"Excel 文件被占用，请关闭后重试: input.xlsx",
|
|
auto=True,
|
|
)
|
|
|
|
message = warning.call_args[0][2]
|
|
self.assertIn("Excel 自动回写失败", message)
|
|
self.assertIn("点击「回写旧数据到 Excel」手动重试", message)
|
|
self.assertIn("手动重试", statuses[-1])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_worker_preflight_blocks_when_no_accounts(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self.make_config(temp_dir)
|
|
db.init_db(cfg["db_path"])
|
|
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"],
|
|
)
|
|
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
|
|
|
|
summary = CollectWorker(tasks, db_path=cfg["db_path"], config=cfg).execute()
|
|
|
|
self.assertTrue(summary["blocked"])
|
|
self.assertTrue(summary["no_accounts"])
|
|
self.assertEqual("NO_ACCOUNTS", summary["reason"])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_worker_preflight_blocks_when_chrome_not_running(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"],
|
|
)
|
|
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
|
|
|
|
with mock.patch("app.gui.chrome.is_running", return_value=False) as is_running, \
|
|
mock.patch("app.gui.accounts.detect_login") as detect_login, \
|
|
mock.patch("app.gui.editor.collect") as collect:
|
|
summary = CollectWorker(tasks, db_path=cfg["db_path"], config=cfg).execute()
|
|
|
|
self.assertTrue(summary["blocked"])
|
|
self.assertEqual("ACCOUNT_NOT_READY", summary["reason"])
|
|
self.assertEqual("alias-a", summary["not_running"][0]["alias"])
|
|
is_running.assert_called_once_with(9222)
|
|
detect_login.assert_not_called()
|
|
collect.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_worker_preflight_blocks_when_account_not_logged_in(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"],
|
|
)
|
|
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
|
|
|
|
with mock.patch("app.gui.chrome.is_running", return_value=True), \
|
|
mock.patch(
|
|
"app.gui.accounts.detect_login",
|
|
return_value={"logged_in": False, "reason": "LOGIN_PAGE"},
|
|
), mock.patch("app.gui.editor.collect") as collect:
|
|
summary = CollectWorker(tasks, db_path=cfg["db_path"], config=cfg).execute()
|
|
|
|
self.assertTrue(summary["blocked"])
|
|
self.assertEqual("alias-a", summary["logged_out"][0]["alias"])
|
|
self.assertIn("账号未登录", summary["logged_out"][0]["reason"])
|
|
collect.assert_not_called()
|
|
blocked_task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
|
|
self.assertEqual("imported", blocked_task.stage)
|
|
self.assertEqual("pending", blocked_task.status)
|
|
self.assertIsNone(blocked_task.last_error)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_collect_tab_blocked_preflight_guides_to_accounts_tab(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
opened = []
|
|
statuses = []
|
|
tab = CollectTab(
|
|
config=self.make_config(temp_dir),
|
|
status_callback=statuses.append,
|
|
open_accounts_callback=lambda: opened.append(True),
|
|
)
|
|
self.addCleanup(tab.close)
|
|
payload = {
|
|
"blocked": True,
|
|
"not_running": [
|
|
{
|
|
"account_name": "主店",
|
|
"alias": "alias-a",
|
|
"reason": "CDP 端口未响应",
|
|
}
|
|
],
|
|
"logged_out": [],
|
|
}
|
|
|
|
with mock.patch("app.gui.QMessageBox.warning") as warning:
|
|
tab._on_collect_finished(payload)
|
|
|
|
message = warning.call_args[0][2]
|
|
self.assertIn("Chrome 未启动", message)
|
|
self.assertIn("④ 账号管理", message)
|
|
self.assertEqual([True], opened)
|
|
self.assertIn("④ 账号管理", statuses[-1])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_write_back_worker_calls_excel_write_back(self):
|
|
with mock.patch(
|
|
"app.gui.excel.write_back",
|
|
return_value={"ok": True, "batch_id": "batch-1", "files": 1, "rows": 2},
|
|
) as write_back:
|
|
summary = WriteBackWorker("batch-1", db_path="db.sqlite").execute()
|
|
|
|
self.assertEqual({"ok": True, "batch_id": "batch-1", "files": 1, "rows": 2}, summary)
|
|
write_back.assert_called_once_with(
|
|
"batch-1",
|
|
excel_path=None,
|
|
path="db.sqlite",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|