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

T-580 精简设置页更新执行配置

Browse Source
This commit is contained in:
chengma
2026-07-10 10:12:15 +08:00
parent c078e68e98
commit e70e831afa
12 changed files with 259 additions and 287 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui.py
+58 -120
View File
@@ -111,13 +111,14 @@ class GuiTests(TempDirMixin, unittest.TestCase):
allow_cover=True,
max_items=1,
close_success_tab=False,
max_parallel_accounts=1,
):
cfg["shopee_update"] = {
"test_item_id": item_id,
"allow_real_submit": True,
"allow_cover_update": allow_cover,
"update_mode": "title_cover" if allow_cover else "title",
"max_items_per_run": max_items,
"close_success_tab": close_success_tab,
"dry_run": False,
"max_parallel_accounts": max_parallel_accounts,
}
return cfg
@@ -840,7 +841,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertFalse(hasattr(tab, "test_item_id_edit"))
self.assertFalse(hasattr(tab, "dry_run_checkbox"))
self.assertEqual(
"蝦皮更新安全 / 执行模式",
"蝦皮更新执行",
tab.shopee_update_section_title.text(),
)
self.assertEqual("基础设施（路径与端口）", tab.infrastructure_section_title.text())
@@ -848,14 +849,16 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tab.settings_panel_layout.indexOf(tab.shopee_update_section_title),
tab.settings_panel_layout.indexOf(tab.infrastructure_section_title),
)
self.assertFalse(tab.allow_real_submit_checkbox.isChecked())
self.assertFalse(tab.allow_cover_update_checkbox.isChecked())
self.assertTrue(tab.allow_cover_update_checkbox.isHidden())
self.assertEqual(-1, tab.shopee_update_form_layout.indexOf(tab.allow_cover_update_checkbox))
self.assertFalse(hasattr(tab, "allow_real_submit_checkbox"))
self.assertFalse(hasattr(tab, "allow_cover_update_checkbox"))
self.assertFalse(hasattr(tab, "close_success_tab_checkbox"))
self.assertFalse(hasattr(tab, "parallel_accounts_checkbox"))
self.assertFalse(hasattr(tab, "parallel_accounts_group"))
self.assertFalse(hasattr(tab, "jpg_quality_spin"))
self.assertEqual(1, tab.max_items_per_run_spin.value())
self.assertFalse(tab.close_success_tab_checkbox.isChecked())
self.assertFalse(tab.parallel_accounts_checkbox.isChecked())
self.assertEqual(2, tab.max_parallel_accounts_spin.value())
self.assertEqual(1, tab.max_parallel_accounts_spin.value())
self.assertEqual(1, tab.max_parallel_accounts_spin.minimum())
self.assertEqual(5, tab.max_parallel_accounts_spin.maximum())
def widget_position(layout, widget):
for index in range(layout.count()):
@@ -864,27 +867,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
return layout.getItemPosition(index)
self.fail(f"Widget not found in layout: {widget.objectName()}")
group_row, group_col, _row_span, group_col_span = widget_position(
parallel_row, parallel_col, _row_span, parallel_col_span = widget_position(
tab.shopee_update_form_layout,
tab.parallel_accounts_group,
)
self.assertGreaterEqual(group_row, 0)
self.assertGreaterEqual(group_col, 0)
self.assertEqual(4, group_col_span)
group_widgets = [
tab.parallel_accounts_group.layout().itemAt(index).widget()
for index in range(tab.parallel_accounts_group.layout().count())
if tab.parallel_accounts_group.layout().itemAt(index).widget() is not None
]
self.assertEqual(
[
tab.parallel_accounts_checkbox,
tab.max_parallel_accounts_label,
tab.max_parallel_accounts_spin,
],
group_widgets,
tab.max_parallel_accounts_spin,
)
self.assertGreaterEqual(parallel_row, 0)
self.assertGreaterEqual(parallel_col, 0)
self.assertEqual(1, parallel_col_span)
for hidden_widget in (
tab.user_data_root_edit,
tab.image_dir_edit,
@@ -1144,16 +1133,12 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tab.retry_spin.setValue(1)
tab.resolution_combo.setCurrentIndex(tab.resolution_combo.findData("2k"))
self.assertEqual("标题 600 秒 / 图片 900 秒", tab.response_timeout_label.text())
tab.jpg_quality_spin.setValue(86)
tab.chrome_path_edit.setText("D:\\Chrome\\chrome.exe")
tab.default_debug_port_spin.setValue(9300)
tab.debug_port_start_spin.setValue(9300)
tab.debug_port_end_spin.setValue(9350)
tab.cdp_ready_timeout_spin.setValue(45)
tab.allow_real_submit_checkbox.setChecked(True)
tab.max_items_per_run_spin.setValue(2)
tab.close_success_tab_checkbox.setChecked(True)
tab.parallel_accounts_checkbox.setChecked(True)
tab.max_parallel_accounts_spin.setValue(3)
with mock.patch("app.gui.QMessageBox.information") as info:
@@ -1168,7 +1153,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual(2, saved["ai"]["image_concurrency"])
self.assertEqual(1, saved["ai"]["retry"])
self.assertEqual("2k", saved["ai"]["resolution"])
self.assertEqual(86, saved["ai"]["jpg_quality"])
self.assertEqual(90, saved["ai"]["jpg_quality"])
self.assertEqual("D:\\Chrome\\chrome.exe", saved["chrome_path"])
self.assertEqual("manual_profiles", saved["user_data_root"])
self.assertEqual("manual_images", saved["image_dir"])
@@ -1179,13 +1164,9 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual(
{
"test_item_id": "51100639510",
"allow_real_submit": True,
"allow_cover_update": False,
"update_mode": "title",
"max_items_per_run": 2,
"close_success_tab": True,
"dry_run": False,
"parallel_accounts": True,
"max_parallel_accounts": 3,
},
saved["shopee_update"],
@@ -1218,7 +1199,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_settings_save_updates_apply_tab_shared_safety_config(self):
def test_settings_save_updates_apply_tab_shared_update_config(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
cfg["shopee_update"] = dict(appconfig.default_config()["shopee_update"])
@@ -1232,10 +1213,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
settings_tab = window.tabs.widget(TAB_TITLES.index("⑤ 设置"))
apply_tab = window.tabs.widget(TAB_TITLES.index("③ 更新蝦皮"))
settings_tab.allow_real_submit_checkbox.setChecked(True)
settings_tab.max_items_per_run_spin.setValue(3)
settings_tab.close_success_tab_checkbox.setChecked(True)
settings_tab.parallel_accounts_checkbox.setChecked(True)
settings_tab.max_parallel_accounts_spin.setValue(4)
with mock.patch("app.gui.QMessageBox.information") as info:
settings_tab.save_app_settings()
@@ -1243,13 +1221,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
info.assert_called_once_with(settings_tab, "保存设置", "设置已保存")
safety_cfg = apply_tab._shopee_update_config()
self.assertEqual("123456789", safety_cfg["test_item_id"])
self.assertTrue(safety_cfg["allow_real_submit"])
self.assertFalse(safety_cfg["allow_cover_update"])
self.assertNotIn("allow_real_submit", safety_cfg)
self.assertNotIn("allow_cover_update", safety_cfg)
self.assertEqual("title", safety_cfg["update_mode"])
self.assertEqual(3, safety_cfg["max_items_per_run"])
self.assertTrue(safety_cfg["close_success_tab"])
self.assertNotIn("close_success_tab", safety_cfg)
self.assertFalse(safety_cfg["dry_run"])
self.assertTrue(safety_cfg["parallel_accounts"])
self.assertNotIn("parallel_accounts", safety_cfg)
self.assertEqual(4, safety_cfg["max_parallel_accounts"])
self.assert_removed(temp_dir)
@@ -4307,7 +4285,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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)
self.allow_shopee_update(cfg, close_success_tab=True)
self.allow_shopee_update(cfg)
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
db.insert_tasks(
@@ -4359,9 +4337,9 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIsInstance(tab.apply_worker, ApplyWorker)
self.assertIs(tab.apply_thread, fake_thread)
self.assertTrue(fake_thread.started)
self.assertTrue(tab.apply_worker.close_success_tab)
self.assertFalse(hasattr(tab.apply_worker, "close_success_tab"))
self.assertFalse(tab.apply_worker.dry_run)
self.assertFalse(tab.apply_worker.parallel_accounts)
self.assertEqual(1, tab.apply_worker.max_parallel_accounts)
self.assertEqual(1, tab.apply_worker.batch_size)
self.assertEqual("title_cover", tab.apply_worker.update_mode)
log_text = tab.run_log_view.toPlainText()
@@ -4440,7 +4418,6 @@ class GuiTests(TempDirMixin, unittest.TestCase):
warning.assert_not_called()
self.assertTrue(tab.apply_worker.dry_run)
self.assertTrue(tab.apply_worker.parallel_accounts)
self.assertEqual(2, tab.apply_worker.max_parallel_accounts)
log_text = tab.run_log_view.toPlainText()
self.assertNotIn("上一轮检查失败", log_text)
@@ -4449,9 +4426,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_blocks_update_when_real_submit_switch_is_off(self):
def test_apply_tab_old_real_submit_switch_does_not_block_final_confirmation(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
cfg["shopee_update"] = {
"allow_real_submit": False,
"max_items_per_run": 1,
}
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
db.insert_tasks(
@@ -4476,17 +4457,31 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.addCleanup(tab.close)
tab.item_filter.setText("51100639510")
with mock.patch.object(tab, "_show_update_safety_error") as safety_error, \
mock.patch("app.gui.QMessageBox.question") as question, \
mock.patch("app.gui.run_worker") as run_worker:
class FakeSignal:
def connect(self, callback):
self.callback = 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,
) as question, mock.patch("app.gui.run_worker", return_value=fake_thread) as run_worker:
tab.start_update()
message = safety_error.call_args[0][0]
self.assertIn("允许真实提交线上商品", message)
self.assertIn("⑤设置", message)
question.assert_not_called()
run_worker.assert_not_called()
self.assertIn("已阻止本次更新", statuses[-1])
message = question.call_args[0][2]
self.assertIn("即将按当前筛选结果分批更新蝦皮线上商品", message)
self.assertIn("任务数：1", message)
run_worker.assert_called_once()
self.assertTrue(fake_thread.started)
self.assertEqual("开始更新：1 条，按每批最多 1 条执行", statuses[-1])
self.assert_removed(temp_dir)
@@ -4533,56 +4528,6 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_safety_error_allows_more_than_batch_size_and_non_test_items(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
self.allow_shopee_update(cfg, item_id="51100639510", max_items=1)
tab = ApplyTab(config=cfg)
self.addCleanup(tab.close)
class Task:
def __init__(self, item_id, new_cover_path=None):
self.item_id = item_id
self.new_cover_path = new_cover_path
count_error = tab._update_safety_error(
[
Task("51100639510"),
Task("51100639510"),
]
)
non_test_error = tab._update_safety_error([Task("51100639511")])
cfg["shopee_update"]["test_item_id"] = ""
missing_test_id_error = tab._update_safety_error([Task("26887160467")])
self.assertIsNone(count_error)
self.assertIsNone(non_test_error)
self.assertIsNone(missing_test_id_error)
self.assert_removed(temp_dir)
def test_apply_tab_safety_error_popup_can_open_settings_tab(self):
with self.make_temp_dir() as temp_dir:
opened = []
tab = ApplyTab(
config=self.make_config(temp_dir),
open_settings_callback=lambda: opened.append(True),
)
self.addCleanup(tab.close)
settings_button = object()
box = mock.Mock()
box.addButton.side_effect = [settings_button, object()]
box.clickedButton.return_value = settings_button
with mock.patch("app.gui.QMessageBox", return_value=box) as message_box:
tab._show_update_safety_error("设置未开启")
message_box.assert_called_once_with(tab)
box.setWindowTitle.assert_called_once_with("更新安全开关")
box.setText.assert_called_once_with("设置未开启")
self.assertEqual([True], opened)
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)
@@ -4624,14 +4569,12 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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 = []
close_flags = []
foreground_flags = []
progress = []
rows = []
def fake_apply(account, task, close_success_tab=False, on_step=None, bring_to_front=True, update_mode=None):
applied_aliases.append(account.alias)
close_flags.append(close_success_tab)
foreground_flags.append(bring_to_front)
if account.alias == "alias-a":
return {"committed": True, "error": None}
@@ -4648,7 +4591,6 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tasks,
db_path=cfg["db_path"],
config=cfg,
close_success_tab=True,
batch_size=1,
)
worker.progress.connect(progress.append)
@@ -4656,7 +4598,6 @@ class GuiTests(TempDirMixin, unittest.TestCase):
summary = worker.execute()
self.assertEqual(["alias-a", "alias-b"], applied_aliases)
self.assertEqual([True, True], close_flags)
self.assertEqual([True, True], foreground_flags)
self.assertFalse(summary["ok"])
self.assertEqual(3, summary["total"])
@@ -4666,7 +4607,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual(1, summary["failed"])
self.assertEqual([batch_id], summary["batch_ids"])
self.assertFalse(summary["dry_run"])
self.assertFalse(summary["parallel_accounts"])
self.assertFalse(summary["account_parallel"])
self.assertEqual(1, summary["batch_size"])
self.assertEqual(3, summary["batch_count"])
self.assertIsNotNone(summary["run_id"])
@@ -4796,7 +4737,6 @@ class GuiTests(TempDirMixin, unittest.TestCase):
db_path=cfg["db_path"],
config=cfg,
dry_run=True,
parallel_accounts=True,
max_parallel_accounts=2,
)
worker.log.connect(logs.append)
@@ -4875,12 +4815,11 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tasks,
db_path=cfg["db_path"],
config=cfg,
parallel_accounts=True,
max_parallel_accounts=2,
).execute()
self.assertTrue(summary["ok"])
self.assertTrue(summary["parallel_accounts"])
self.assertTrue(summary["account_parallel"])
self.assertEqual(2, summary["applied"])
self.assertGreaterEqual(len(thread_names), 2)
self.assertEqual({"alias-a": True, "alias-b": True}, foreground_by_alias)
@@ -4929,7 +4868,6 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tasks,
db_path=cfg["db_path"],
config=cfg,
parallel_accounts=True,
max_parallel_accounts=2,
).execute()