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

fix: stabilize Shopee cover replacement

Browse Source
This commit is contained in:
chengma
2026-07-01 08:35:46 +08:00
parent f7ea9ec0fa
commit 483bec427c
18 changed files with 782 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_ai.py
+12 -1
View File
@@ -277,7 +277,18 @@ class AITests(TempDirMixin, unittest.TestCase):
updated = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
self.assertTrue(all(task.stage == "generated" for task in updated))
self.assertEqual({"新旧标题A", "新旧标题B"}, {task.new_title for task in updated})
self.assertTrue(all(task.new_cover_path.endswith("_new.jpg") for task in updated))
expected_cover_paths = {
os.path.abspath(
os.path.join(
cfg["image_dir"],
str(task.batch_id),
"main",
f"{task.id}_{task.item_id}_new.jpg",
)
)
for task in updated
}
self.assertEqual(expected_cover_paths, {task.new_cover_path for task in updated})
self.assertTrue(all(os.path.exists(task.new_cover_path) for task in updated))
self.assertIn("主店", "\n".join(cover_prompts))
self.assertIn("新旧标题A", "\n".join(cover_prompts))
tests/test_editor_login.py
+113 -6
View File
@@ -169,6 +169,8 @@ class FakeCoverCDP:
upload_input_ready=True,
restore_upload_after_delete=False,
page_error_toasts=None,
upload_tile_click=True,
require_upload_tile_click=False,
):
self.initial = cover_rects(count)
self.after_delete = cover_rects(max(0, count - 1), prefix="kept", start=1)
@@ -189,10 +191,14 @@ class FakeCoverCDP:
self.upload_input_ready = upload_input_ready
self.restore_upload_after_delete = restore_upload_after_delete
self.page_error_toasts = page_error_toasts or []
self.upload_tile_click = upload_tile_click
self.require_upload_tile_click = require_upload_tile_click
self.delete_clicked = False
self.confirm_clicked = False
self.upload_tile_clicked = False
self.uploaded = False
self.uploaded_after_stable = None
self.upload_tile_clicked_before_file_set = None
self.dragged = False
self.sent = []
self.drag_calls = []
@@ -244,6 +250,12 @@ class FakeCoverCDP:
return json.dumps(self._state(busy_count=1))
if self.uploaded and self.upload_mode == "error":
return json.dumps(self._state(errors=["圖片格式錯誤"]))
if self.uploaded and self.upload_mode == "duplicate":
state = self._state()
state["toasts"] = ["有1張重複的圖片"]
state["error_toasts"] = ["有1張重複的圖片"]
state["page_error_toasts"] = ["有1張重複的圖片"]
return json.dumps(state)
if self.confirm_clicked and not self.uploaded and self.post_delete_busy_polls > 0:
self.post_delete_busy_polls -= 1
return json.dumps(self._state(busy_count=1))
@@ -258,14 +270,22 @@ class FakeCoverCDP:
return json.dumps({"clicked": False, "reason": "NO_CONFIRM_BUTTON"})
self.confirm_clicked = True
return json.dumps({"clicked": True, "reason": None, "text": "刪除"})
if expr == editor.JS_CLICK_UPLOAD_TILE:
if not self.upload_tile_click:
return json.dumps({"clicked": False, "reason": "NO_UPLOAD_TILE"})
self.upload_tile_clicked = True
return json.dumps({"clicked": True, "reason": None})
return True
def object_id(self, expr):
if self.require_upload_tile_click and not self.upload_tile_clicked:
return None
return "upload-input" if self._upload_ready() else None
def send(self, method, params=None):
self.sent.append((method, params or {}))
if method == "DOM.setFileInputFiles":
self.upload_tile_clicked_before_file_set = self.upload_tile_clicked
self.uploaded_after_stable = self.post_delete_busy_polls == 0
self.uploaded = True
return {}
@@ -577,7 +597,9 @@ class EditorLoginTests(unittest.TestCase):
) as change_title, mock.patch("app.editor.replace_cover") as replace_cover, mock.patch(
"app.editor.click_update",
return_value={"clicked": True, "reason": None},
), mock.patch("app.editor.close_tab", return_value=True) as close_tab:
), mock.patch("app.editor.close_tab", return_value=True) as close_tab, mock.patch(
"app.editor.time.sleep"
) as sleep:
result = editor.apply_task(
{"debug_port": 9222},
{"item_id": "51100639510", "new_title": "新标题"},
@@ -588,6 +610,7 @@ class EditorLoginTests(unittest.TestCase):
change_title.assert_called_once_with(cdp, "新标题")
replace_cover.assert_not_called()
self.assertTrue(cdp.closed)
sleep.assert_called_once_with(2)
close_tab.assert_called_once_with("target-new", host="127.0.0.1:9222")
def test_apply_task_keeps_product_tab_after_failed_update(self):
@@ -602,7 +625,9 @@ class EditorLoginTests(unittest.TestCase):
), mock.patch(
"app.editor.click_update",
return_value={"clicked": False, "reason": "UPDATE_DISABLED"},
), mock.patch("app.editor.close_tab") as close_tab:
), mock.patch("app.editor.close_tab") as close_tab, mock.patch(
"app.editor.time.sleep"
) as sleep:
result = editor.apply_task(
{"debug_port": 9222},
{"item_id": "51100639510", "new_title": "新标题"},
@@ -613,6 +638,7 @@ class EditorLoginTests(unittest.TestCase):
self.assertEqual("UPDATE_DISABLED", result["error"])
self.assertTrue(cdp.closed)
close_tab.assert_not_called()
sleep.assert_not_called()
def test_apply_task_passes_old_cover_backup_to_replace_cover(self):
cdp = FakeProductCDP("ws-new")
@@ -640,8 +666,8 @@ class EditorLoginTests(unittest.TestCase):
old_cover_path="old.jpg",
)
def test_replace_cover_full_slots_requires_old_cover_backup(self):
cdp = FakeCoverCDP(count=9)
def test_replace_cover_requires_old_cover_backup_before_delete(self):
cdp = FakeCoverCDP(count=8)
missing_backup = os.path.join(os.path.dirname(__file__), "missing-old-cover.jpg")
with mock.patch("app.editor.time.sleep"):
@@ -678,6 +704,61 @@ class EditorLoginTests(unittest.TestCase):
self.assertEqual("https://susercontent.com/new-cover.jpg", result["new_src"])
def test_replace_cover_deletes_first_for_non_full_image_slots(self):
cdp = FakeCoverCDP(count=8)
with mock.patch("app.editor.time.sleep"):
result = editor.replace_cover(
cdp,
__file__,
old_cover_path=__file__,
timeout=1,
)
self.assertTrue(result["ok"])
self.assertTrue(cdp.delete_clicked)
self.assertTrue(cdp.confirm_clicked)
self.assertTrue(cdp.uploaded)
self.assertTrue(cdp.dragged)
self.assertEqual(8, result["count_before"])
self.assertEqual(8, result["count_after"])
self.assertEqual(7, result["delete"]["count_after"])
def test_replace_cover_clicks_upload_tile_before_setting_files(self):
cdp = FakeCoverCDP(count=9, require_upload_tile_click=True)
with mock.patch("app.editor.time.sleep"):
result = editor.replace_cover(
cdp,
__file__,
old_cover_path=__file__,
timeout=1,
)
self.assertTrue(result["ok"])
self.assertTrue(cdp.upload_tile_clicked)
self.assertTrue(cdp.uploaded)
self.assertIn("DOM.setFileInputFiles", [method for method, _params in cdp.sent])
self.assertTrue(cdp.upload_tile_clicked_before_file_set)
self.assertEqual({"clicked": True, "reason": None}, result["upload_click"])
def test_replace_cover_reports_upload_tile_not_ready(self):
cdp = FakeCoverCDP(count=8, upload_tile_click=False)
with mock.patch("app.editor.time.sleep"):
result = editor.replace_cover(
cdp,
__file__,
old_cover_path=__file__,
timeout=1,
)
self.assertFalse(result["ok"])
self.assertEqual("UPLOAD_TILE_NOT_READY", result["reason"])
self.assertFalse(cdp.uploaded)
self.assertEqual("NO_UPLOAD_TILE", result["upload_click"]["reason"])
def test_replace_cover_full_slots_deletes_before_upload_input_is_ready(self):
cdp = FakeCoverCDP(
count=9,
@@ -742,7 +823,12 @@ class EditorLoginTests(unittest.TestCase):
cdp = FakeCoverCDP(count=8, upload_mode="processing")
with mock.patch("app.editor.time.sleep"):
result = editor.replace_cover(cdp, __file__, timeout=0.01)
result = editor.replace_cover(
cdp,
__file__,
old_cover_path=__file__,
timeout=0.01,
)
self.assertFalse(result["ok"])
self.assertEqual("UPLOAD_STILL_PROCESSING", result["reason"])
@@ -754,12 +840,33 @@ class EditorLoginTests(unittest.TestCase):
cdp = FakeCoverCDP(count=8, upload_mode="error")
with mock.patch("app.editor.time.sleep"):
result = editor.replace_cover(cdp, __file__, timeout=1)
result = editor.replace_cover(
cdp,
__file__,
old_cover_path=__file__,
timeout=1,
)
self.assertFalse(result["ok"])
self.assertEqual("UPLOAD_PAGE_ERROR", result["reason"])
self.assertIn("圖片格式錯誤", result["upload_state"]["errors"])
def test_replace_cover_reports_duplicate_upload_toast(self):
cdp = FakeCoverCDP(count=8, upload_mode="duplicate")
with mock.patch("app.editor.time.sleep"):
result = editor.replace_cover(
cdp,
__file__,
old_cover_path=__file__,
timeout=1,
)
self.assertFalse(result["ok"])
self.assertEqual("UPLOAD_DUPLICATE_IMAGE", result["reason"])
self.assertIn("有1張重複的圖片", result["upload_state"]["error_toasts"])
self.assertIn("新封面与现有商品图片重复", editor._cover_upload_error_message(result))
def test_apply_task_cover_failure_returns_readable_upload_error(self):
cdp = FakeProductCDP("ws-new")
tests/test_gui.py
+82 -26
View File
@@ -152,14 +152,49 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual(11, tab.connect_timeout_spin.value())
self.assertFalse(tab.delete_model_button.isEnabled())
self.assertEqual("51100639510", tab.test_item_id_edit.text())
self.assertTrue(tab.test_item_id_edit.isHidden())
self.assertFalse(tab.allow_real_submit_checkbox.isChecked())
self.assertFalse(tab.allow_cover_update_checkbox.isChecked())
self.assertEqual(1, tab.max_items_per_run_spin.value())
self.assertFalse(tab.close_success_tab_checkbox.isChecked())
self.assertFalse(tab.dry_run_checkbox.isChecked())
self.assertTrue(tab.dry_run_checkbox.isHidden())
self.assertFalse(tab.parallel_accounts_checkbox.isChecked())
self.assertEqual(2, tab.max_parallel_accounts_spin.value())
def widget_position(layout, widget):
for index in range(layout.count()):
item = layout.itemAt(index)
if item is not None and item.widget() is widget:
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(
tab.shopee_update_form_layout,
tab.parallel_accounts_group,
)
cover_row, cover_col, _row_span, _col_span = widget_position(
tab.shopee_update_form_layout,
tab.allow_cover_update_checkbox,
)
self.assertEqual(cover_row, group_row)
self.assertGreater(group_col, cover_col)
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,
)
self.assert_removed(temp_dir)
def test_settings_tab_adds_saves_and_deletes_model(self):
@@ -357,12 +392,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tab.allow_cover_update_checkbox.setChecked(True)
tab.max_items_per_run_spin.setValue(2)
tab.close_success_tab_checkbox.setChecked(True)
tab.dry_run_checkbox.setChecked(True)
tab.parallel_accounts_checkbox.setChecked(True)
tab.max_parallel_accounts_spin.setValue(3)
tab.save_app_settings()
with mock.patch("app.gui.QMessageBox.information") as info:
tab.save_app_settings()
info.assert_called_once_with(tab, "保存设置", "设置已保存")
saved = appconfig.load_config(config_path)
self.assertEqual("Text B", saved["ai"]["default_text_model"])
self.assertEqual("Image B", saved["ai"]["default_image_model"])
@@ -385,7 +421,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
"allow_cover_update": True,
"max_items_per_run": 2,
"close_success_tab": True,
"dry_run": True,
"dry_run": False,
"parallel_accounts": True,
"max_parallel_accounts": 3,
},
@@ -414,18 +450,19 @@ class GuiTests(TempDirMixin, unittest.TestCase):
settings_tab.allow_cover_update_checkbox.setChecked(True)
settings_tab.max_items_per_run_spin.setValue(3)
settings_tab.close_success_tab_checkbox.setChecked(True)
settings_tab.dry_run_checkbox.setChecked(True)
settings_tab.parallel_accounts_checkbox.setChecked(True)
settings_tab.max_parallel_accounts_spin.setValue(4)
settings_tab.save_app_settings()
with mock.patch("app.gui.QMessageBox.information") as info:
settings_tab.save_app_settings()
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.assertTrue(safety_cfg["allow_cover_update"])
self.assertEqual(3, safety_cfg["max_items_per_run"])
self.assertTrue(safety_cfg["close_success_tab"])
self.assertTrue(safety_cfg["dry_run"])
self.assertFalse(safety_cfg["dry_run"])
self.assertTrue(safety_cfg["parallel_accounts"])
self.assertEqual(4, safety_cfg["max_parallel_accounts"])
@@ -897,6 +934,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertIsInstance(tab.task_table, QTableView)
self.assertEqual(["店铺", "商品ID", "新标题", "新封面", "阶段", "结果"], tab.model.HEADERS)
self.assertEqual("预览本轮更新", tab.preview_update_button.text())
self.assertEqual("开始更新", tab.start_update_button.text())
self.assertEqual("applyItemFilter", tab.item_filter.objectName())
self.assertEqual("停止", tab.stop_update_button.text())
@@ -1032,10 +1070,11 @@ class GuiTests(TempDirMixin, unittest.TestCase):
message = question.call_args[0][2]
self.assertIn("任务数：1", message)
self.assertIn("预计批次：1", message)
self.assertIn("提交线上", message)
self.assertIn("状态：已生成", message)
self.assertIn("商品ID：51100639510", message)
self.assertIn("测试商品ID=51100639510", message)
self.assertNotIn("测试商品ID", message)
self.assertEqual("已取消开始更新", statuses[-1])
apply_task.assert_not_called()
@@ -1098,9 +1137,11 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertTrue(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.batch_size)
self.assertFalse(tab.preview_update_button.isEnabled())
self.assertFalse(tab.start_update_button.isEnabled())
self.assertTrue(tab.stop_update_button.isEnabled())
self.assertEqual("开始更新：1 条", statuses[-1])
self.assertEqual("开始更新：1 条，按每批最多 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)
@@ -1108,7 +1149,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_dry_run_starts_without_real_submit_switch(self):
def test_apply_tab_preview_starts_without_real_submit_switch(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
cfg["shopee_update"] = {
@@ -1117,7 +1158,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
"allow_cover_update": False,
"max_items_per_run": 1,
"close_success_tab": False,
"dry_run": True,
"dry_run": False,
"parallel_accounts": True,
"max_parallel_accounts": 2,
}
@@ -1165,13 +1206,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
return_value=gui.QMessageBox.Yes,
), mock.patch("app.gui.QMessageBox.warning") as warning, \
mock.patch("app.gui.run_worker", return_value=fake_thread):
tab.start_update()
tab.preview_update()
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)
self.assertEqual("开始 dry-run 预览：1 条", statuses[-1])
self.assertEqual("开始预览本轮更新：1 条", statuses[-1])
self.assert_removed(temp_dir)
@@ -1252,7 +1293,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_apply_tab_safety_error_limits_max_count_and_test_item(self):
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)
@@ -1264,16 +1305,19 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.item_id = item_id
self.new_cover_path = new_cover_path
max_error = tab._update_safety_error(
count_error = tab._update_safety_error(
[
Task("51100639510"),
Task("51100639510"),
]
)
item_error = tab._update_safety_error([Task("51100639511")])
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.assertIn("超过单次最大更新条数", max_error)
self.assertIn("非测试商品ID", item_error)
self.assertIsNone(count_error)
self.assertIsNone(non_test_error)
self.assertIsNone(missing_test_id_error)
self.assert_removed(temp_dir)
@@ -1341,6 +1385,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
db_path=cfg["db_path"],
config=cfg,
close_success_tab=True,
batch_size=1,
)
worker.progress.connect(progress.append)
worker.row_updated.connect(lambda task_id, fields: rows.append((task_id, fields)))
@@ -1357,6 +1402,8 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual([batch_id], summary["batch_ids"])
self.assertFalse(summary["dry_run"])
self.assertFalse(summary["parallel_accounts"])
self.assertEqual(1, summary["batch_size"])
self.assertEqual(3, summary["batch_count"])
self.assertIsNotNone(summary["run_id"])
self.assertEqual(3, progress[-1]["done"])
self.assertEqual(3, progress[-1]["total"])
@@ -1364,6 +1411,8 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual(1, progress[-1]["skipped"])
self.assertEqual(1, progress[-1]["failed"])
self.assertFalse(progress[-1]["dry_run"])
self.assertEqual(1, progress[-1]["batch_size"])
self.assertEqual(3, progress[-1]["batch_count"])
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)
@@ -1380,10 +1429,9 @@ class GuiTests(TempDirMixin, unittest.TestCase):
run_logs = db.list_run_logs(run_type="apply", path=cfg["db_path"])
self.assertEqual(1, len(run_logs))
self.assertEqual("done", run_logs[0].status)
self.assertGreaterEqual(
len(db.list_run_log_events(run_logs[0].id, path=cfg["db_path"])),
3,
)
events = db.list_run_log_events(run_logs[0].id, path=cfg["db_path"])
self.assertGreaterEqual(len(events), 3)
self.assertTrue(any("更新批次 1/3" in event.message for event in events))
self.assert_removed(temp_dir)
@@ -1445,7 +1493,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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.assertTrue(any("dry-run" in line for line in logs))
self.assertTrue(any("预览" in line for line in logs))
run_log = db.list_run_logs(run_type="apply", path=cfg["db_path"])[0]
self.assertEqual(1, run_log.dry_run)
self.assertEqual("done", run_log.status)
@@ -1690,6 +1738,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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.accounts.launch_for_login") as launch_for_login, \
mock.patch("app.gui.editor.apply_task") as apply_task:
summary = ApplyWorker(tasks, db_path=cfg["db_path"], config=cfg).execute()
@@ -1698,6 +1747,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual("alias-a", summary["not_running"][0]["alias"])
is_running.assert_called_once_with(9222)
detect_login.assert_not_called()
launch_for_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)
@@ -1733,8 +1783,12 @@ class GuiTests(TempDirMixin, unittest.TestCase):
message = warning.call_args[0][2]
self.assertIn("Chrome 未启动", message)
self.assertIn("本轮更新已中止", message)
self.assertIn("不会自动打开账号 Chrome", message)
self.assertIn("不会提交任何商品", message)
self.assertIn("④ 账号管理", message)
self.assertEqual([True], opened)
self.assertIn("本轮更新已中止", statuses[-1])
self.assertIn("④ 账号管理", statuses[-1])
self.assert_removed(temp_dir)
@@ -2164,6 +2218,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
path=cfg["db_path"],
)
tasks = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])
expected_old_cover = os.path.join(str(batch_id), logged.slug, f"{tasks[0].id}_51100639510_old.jpg")
def fake_login(account, path=None, config=None):
self.assertEqual(cfg["db_path"], path)
@@ -2176,9 +2231,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
def fake_collect(account, task, on_step=None):
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")
))
self.assertTrue(task["old_cover_path"].endswith(expected_old_cover))
self.assertTrue(callable(on_step))
on_step("download_cover")
return {
@@ -2209,7 +2262,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
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.assertTrue(by_alias["alias-a"].old_cover_path.endswith(expected_old_cover))
self.assertEqual("imported", by_alias["alias-b"].stage)
self.assertEqual("skipped", by_alias["alias-b"].status)
self.assertIn("账号未登录", by_alias["alias-b"].last_error)
@@ -2528,8 +2581,11 @@ class GuiTests(TempDirMixin, unittest.TestCase):
message = warning.call_args[0][2]
self.assertIn("Chrome 未启动", message)
self.assertIn("本轮采集已中止", message)
self.assertIn("不会自动打开账号 Chrome", message)
self.assertIn("④ 账号管理", message)
self.assertEqual([True], opened)
self.assertIn("本轮采集已中止", statuses[-1])
self.assertIn("④ 账号管理", statuses[-1])
self.assert_removed(temp_dir)
tests/test_image_paths.py
+38
View File
@@ -0,0 +1,38 @@
import os
import sys
import unittest
from types import SimpleNamespace
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from app import image_paths
from app.config import make_slug
class ImagePathTests(unittest.TestCase):
def test_task_image_path_uses_batch_slug_task_and_item(self):
root = os.path.abspath(os.path.join("tmp", "images"))
task = SimpleNamespace(id=42, batch_id=7, item_id="51100639510", alias="alias-a")
account = SimpleNamespace(slug="main_shop")
path = image_paths.task_image_path(root, task, account, "new")
self.assertEqual(
os.path.join(root, "7", "main_shop", "42_51100639510_new.jpg"),
path,
)
def test_task_image_path_falls_back_to_account_alias_slug(self):
root = os.path.abspath(os.path.join("tmp", "images"))
task = {"id": 5, "batch_id": 3, "item_id": "ITEM/1", "alias": "alias-a"}
path = image_paths.task_image_path(root, task, None, "old")
self.assertEqual(
os.path.join(root, "3", make_slug("alias-a"), "5_ITEM_1_old.jpg"),
path,
)
if __name__ == "__main__":
unittest.main()