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

feat(apply): block abnormal product statuses

Browse Source
This commit is contained in:
chengma
2026-07-18 17:10:12 +08:00
parent 3a92119187
commit d64e3c7404
9 changed files with 499 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui.py
+241 -14
View File
@@ -6682,7 +6682,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
statuses = []
tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6730,7 +6736,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
statuses = []
tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6813,7 +6825,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
statuses = []
tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6877,7 +6895,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
statuses = []
tab = ApplyTab(config=cfg, status_callback=statuses.append)
@@ -6933,7 +6957,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(task.id, "新标题", None, path=cfg["db_path"])
tab = ApplyTab(config=cfg)
self.addCleanup(tab.close)
@@ -6983,6 +7013,149 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_blocks_all_abnormal_statuses_before_content_or_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)
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": index + 2,
"alias": "alias-a",
"item_id": str(51100639510 + index),
}
for index in range(3)
],
path=cfg["db_path"],
)
statuses = ["unlisted", "reviewing", None]
for task, status in zip(
db.list_tasks(batch_id=batch_id, path=cfg["db_path"]),
statuses,
):
db.set_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value=status,
path=cfg["db_path"],
)
db.set_generated(task.id, "新标题", "new.jpg", path=cfg["db_path"])
states = []
tab = ApplyTab(config=cfg, status_callback=states.append)
self.addCleanup(tab.close)
with mock.patch("app.gui.QMessageBox.warning") as warning, mock.patch(
"app.gui.QMessageBox.question"
) as question, mock.patch("app.gui.run_worker") as run_worker:
tab.preview_update()
self.assertEqual("商品状态不允许更新", warning.call_args[0][1])
message = warning.call_args[0][2]
self.assertIn("未上架：1 条", message)
self.assertIn("审核中：1 条", message)
self.assertIn("状态未知：1 条", message)
self.assertNotIn("缺少新标题", message)
question.assert_not_called()
run_worker.assert_not_called()
self.assertIn("状态正常且可更新", states[-1])
unchanged = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
self.assertTrue(all(task.stage == "generated" for task in unchanged))
self.assertTrue(all(task.status == "success" for task in unchanged))
self.assert_removed(temp_dir)
def test_apply_tab_confirmation_combines_status_and_content_exclusions(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": index + 2,
"alias": "alias-a",
"item_id": str(51100639510 + index),
}
for index in range(3)
],
path=cfg["db_path"],
)
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
for task, status in zip(tasks, ["normal", "reviewing", "normal"]):
db.set_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value=status,
path=cfg["db_path"],
)
db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
db.set_generated(tasks[1].id, "新标题", "new.jpg", path=cfg["db_path"])
db.set_generated(tasks[2].id, None, "new.jpg", path=cfg["db_path"])
tab = ApplyTab(config=cfg)
self.addCleanup(tab.close)
thread = FakeThread()
with mock.patch("app.gui.QMessageBox.question", return_value=gui.QMessageBox.Yes) as question, mock.patch(
"app.gui.run_worker",
return_value=thread,
):
tab.preview_update()
message = question.call_args[0][2]
self.assertIn("商品状态异常记录", message)
self.assertIn("审核中：1 条", message)
self.assertIn("缺少新标题：1 条", message)
self.assertEqual([tasks[0].id], [task.id for task in tab.apply_worker.tasks])
self.assertTrue(tab.apply_worker.dry_run)
self.assertTrue(thread.started)
self.assert_removed(temp_dir)
def test_apply_tab_discards_plan_when_candidates_change_during_confirmation(self):
with self.make_temp_dir() as temp_dir:
states = []
tab = ApplyTab(
config=self.make_config(temp_dir),
status_callback=states.append,
)
self.addCleanup(tab.close)
before = SimpleNamespace(
id=1,
item_id="51100639510",
updated_at="2026-07-18T10:00:00",
product_status="normal",
new_title="新标题",
new_cover_path="new.jpg",
)
after = SimpleNamespace(
id=1,
item_id="51100639510",
updated_at="2026-07-18T10:01:00",
product_status="normal",
new_title="新标题",
new_cover_path="new.jpg",
)
with mock.patch.object(tab, "_update_candidates", side_effect=[[before], [after]]), mock.patch(
"app.gui.QMessageBox.question",
return_value=gui.QMessageBox.Yes,
), mock.patch("app.gui.run_worker") as run_worker:
tab.preview_update()
run_worker.assert_not_called()
self.assertIn("当前任务数据已变化", states[-1])
self.assert_removed(temp_dir)
def test_apply_tab_skips_missing_content_but_starts_eligible_update(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
@@ -7020,7 +7193,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
)
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
for task in tasks:
db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
db.set_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
db.set_generated(tasks[1].id, "新标题", None, path=cfg["db_path"])
db.set_generated(tasks[2].id, None, "new.jpg", path=cfg["db_path"])
@@ -7080,7 +7259,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
)
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
for task in tasks:
db.set_collected(task.id, "旧标题", "old.jpg", path=cfg["db_path"])
db.set_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
path=cfg["db_path"],
)
db.set_generated(tasks[0].id, "新标题", "new.jpg", path=cfg["db_path"])
db.set_generated(tasks[1].id, None, "new.jpg", path=cfg["db_path"])
statuses = []
@@ -7176,7 +7361,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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 = []
@@ -7280,7 +7471,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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"])
foreground_flags = []
@@ -7335,7 +7532,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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"])
logs = []
@@ -7402,7 +7605,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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"])
started = {"alias-a": threading.Event(), "alias-b": threading.Event()}
@@ -7469,7 +7678,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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"])
@@ -7604,7 +7819,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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"])
@@ -9595,7 +9816,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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_collected(
task.id,
"旧标题",
"old.jpg",
product_status_value="normal",
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")
tests/test_product_status.py
+65
View File
@@ -132,3 +132,68 @@ class ProductStatusTests(unittest.TestCase):
)
self.assertEqual([1, 2, 3], [task.id if hasattr(task, "id") else task["id"] for task in all_statuses["execution_tasks"]])
self.assertNotEqual(normal_only["fingerprint"], changed["fingerprint"])
def test_apply_plan_excludes_abnormal_before_content_and_deduplicates_skips(self):
tasks = [
SimpleNamespace(
id=1,
updated_at="2026-07-18T10:00:00",
product_status="normal",
new_title="新标题",
new_cover_path="new.jpg",
),
SimpleNamespace(
id=2,
updated_at="2026-07-18T10:00:01",
product_status="unlisted",
new_title=None,
new_cover_path=None,
),
SimpleNamespace(
id=3,
updated_at="2026-07-18T10:00:02",
product_status="reviewing",
new_title=None,
new_cover_path="new.jpg",
),
SimpleNamespace(
id=4,
updated_at="2026-07-18T10:00:03",
product_status=None,
new_title="新标题",
new_cover_path=None,
),
SimpleNamespace(
id=5,
updated_at="2026-07-18T10:00:04",
product_status="normal",
new_title=None,
new_cover_path=None,
),
]
plan = product_status.build_apply_plan(tasks + [tasks[0]], "title_cover")
changed = product_status.build_apply_plan(
tasks[:4]
+ [
SimpleNamespace(
id=5,
updated_at="2026-07-18T10:01:00",
product_status="normal",
new_title=None,
new_cover_path=None,
)
],
"title_cover",
)
self.assertEqual([1], [task.id for task in plan["executable"]])
self.assertEqual([2], [task.id for task in plan["unlisted"]])
self.assertEqual([3], [task.id for task in plan["reviewing"]])
self.assertEqual([4], [task.id for task in plan["unknown"]])
self.assertEqual([5], [task.id for task in plan["missing_title"]])
self.assertEqual([5], [task.id for task in plan["missing_cover"]])
self.assertEqual(3, plan["status_scope_excluded"])
self.assertEqual(1, plan["content_scope_excluded"])
self.assertEqual(4, plan["scope_excluded"])
self.assertNotEqual(plan["fingerprint"], changed["fingerprint"])