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

feat: 完成T-402确认后串行更新

Browse Source 
③更新shopee在开始更新确认后创建ApplyWorker，按当前筛选结果串行调用editor.apply_task并逐条set_applied落库。

新增更新前账号就绪预检：无账号、Chrome未启动或未登录时整体阻断并引导到④账号管理，未确认或预检失败均不调用apply。

补充GUI开始/停止/进度刷新以及成功、失败、未匹配账号继续处理的单元测试，同步任务看板、API、流程和当前状态文档。
This commit is contained in:
chengma
2026-06-27 17:13:18 +08:00
parent 5a63af4620
commit 04ff06a2e3
7 changed files with 560 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui.py
+217 -3
View File
@@ -21,6 +21,7 @@ from app.gui import (
AccountDialog,
AccountsTab,
ApplyTab,
ApplyWorker,
CollectWorker,
CollectTab,
GenerateWorker,
@@ -346,7 +347,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_start_update_requires_confirmation_without_running_apply(self):
def test_apply_tab_start_update_requires_confirmation_before_starting_worker(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)
@@ -385,13 +386,64 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual("已取消开始更新", statuses[-1])
apply_task.assert_not_called()
self.assert_removed(temp_dir)
def test_apply_tab_start_update_starts_apply_worker_after_confirmation(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)
class FakeSignal:
def __init__(self):
self.callbacks = []
def connect(self, callback):
self.callbacks.append(callback)
class FakeThread:
def __init__(self):
self.finished = FakeSignal()
self.started = False
def start(self):
self.started = True
fake_thread = FakeThread()
with mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.Yes,
), mock.patch("app.gui.editor.apply_task") as apply_task:
), mock.patch("app.gui.run_worker", return_value=fake_thread) as run_worker, \
mock.patch("app.gui.editor.apply_task") as apply_task:
tab.start_update()
self.assertIn("已确认更新范围：1 条", statuses[-1])
run_worker.assert_called_once()
self.assertIsInstance(tab.apply_worker, ApplyWorker)
self.assertIs(tab.apply_thread, fake_thread)
self.assertTrue(fake_thread.started)
self.assertFalse(tab.start_update_button.isEnabled())
self.assertTrue(tab.stop_update_button.isEnabled())
self.assertEqual("开始更新：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)
@@ -399,6 +451,168 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_worker_applies_success_failure_and_unmatched_serially(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_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"],
)
for task in db.list_tasks(batch_id=batch_id, path=cfg["db_path"]):
db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
applied_aliases = []
progress = []
rows = []
def fake_apply(account, task):
applied_aliases.append(account.alias)
if account.alias == "alias-a":
return {"committed": True, "error": None}
if account.alias == "alias-b":
return {"committed": False, "error": "UPDATE_DISABLED"}
raise AssertionError(account.alias)
with mock.patch("app.gui.chrome.is_running", return_value=True), \
mock.patch(
"app.gui.accounts.detect_login",
return_value={"logged_in": True, "reason": None},
), mock.patch("app.gui.editor.apply_task", side_effect=fake_apply):
worker = ApplyWorker(tasks, db_path=cfg["db_path"], config=cfg)
worker.progress.connect(progress.append)
worker.row_updated.connect(lambda task_id, fields: rows.append((task_id, fields)))
summary = worker.execute()
self.assertEqual(["alias-a", "alias-b"], applied_aliases)
self.assertEqual(
{"ok": False, "total": 3, "done": 3, "applied": 1, "skipped": 1, "failed": 1},
summary,
)
self.assertEqual(
{"done": 3, "total": 3, "applied": 1, "skipped": 1, "failed": 1},
progress[-1],
)
updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
by_alias = {task.alias: task for task in updated}
self.assertEqual("applied", by_alias["alias-a"].stage)
self.assertEqual("success", by_alias["alias-a"].status)
self.assertEqual(1, by_alias["alias-a"].committed)
self.assertEqual("generated", by_alias["alias-b"].stage)
self.assertEqual("failed", by_alias["alias-b"].status)
self.assertEqual("UPDATE_DISABLED", by_alias["alias-b"].last_error)
self.assertEqual(0, by_alias["alias-b"].committed)
self.assertEqual("skipped", by_alias["missing"].status)
self.assertEqual("别名未匹配账号", by_alias["missing"].last_error)
self.assertTrue(any(fields.get("stage") == "applied" for _task_id, fields in rows))
self.assertTrue(any(fields.get("status") == "failed" for _task_id, fields in rows))
self.assert_removed(temp_dir)
def test_apply_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"],
)
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"])
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.apply_task") as apply_task:
summary = ApplyWorker(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()
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.assertIsNone(unchanged.last_error)
self.assert_removed(temp_dir)
def test_apply_tab_blocked_preflight_guides_to_accounts_tab(self):
with self.make_temp_dir() as temp_dir:
opened = []
statuses = []
tab = ApplyTab(
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_apply_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_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)