Files
cmshoppe/tests/test_product_suite_gui.py
T

3572 lines
146 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import os
import sys
import time
import unittest
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 (
accounts,
appconfig,
cmhub_models,
image_studio,
image_studio_images,
product_suite,
prompts,
)
from app import gui
if gui.QT_IMPORT_ERROR is not None:
raise unittest.SkipTest("PySide6 未安装")
from PySide6.QtCore import QMimeData, QPoint, QPointF, Qt, QUrl
from PySide6.QtGui import QIcon, QImage, QPixmap, QTextCursor, QWheelEvent
from PySide6.QtTest import QTest
from PySide6.QtWidgets import QApplication, QLabel, QListWidgetItem, QPushButton, QScrollArea
from app.gui.tabs.product_suite import (
AutoHeightPlainTextEdit,
ORIGINAL_CHECK_STATE_ROLE,
ProductOriginalDelegate,
ProductOriginalList,
ProductSuiteGlobalHistoryDialog,
ProductSuiteHistoryDialog,
ProductSuitePreviewDialog,
ProductSuiteRoundPreviewDialog,
ProductSuiteTab,
SuiteGlobalHistoryRoundRow,
SuiteGlobalHistoryThumbnail,
SuiteHistoryImageCard,
SuiteResultCard,
)
from app.gui.product_suite_prompt_dialog import ProductSuitePromptDialog
class ProductSuiteGuiTests(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 _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],
"config_path": os.path.join(temp_dir, "config.json"),
"cmhub_config_path": os.path.join(temp_dir, "cmhub.json"),
}
def _write_image(self, path):
image = QImage(40, 30, QImage.Format_RGB32)
image.fill(0xFF336699)
self.assertTrue(image.save(path))
def _send_wheel(self, widget, delta=-120):
local_position = widget.rect().center()
global_position = widget.mapToGlobal(local_position)
event = QWheelEvent(
QPointF(local_position),
QPointF(global_position),
QPoint(),
QPoint(0, int(delta)),
Qt.NoButton,
Qt.NoModifier,
Qt.ScrollUpdate,
False,
)
QApplication.sendEvent(widget, event)
def _create_project_with_assets(self, temp_dir, config, count, item_id="51100639510"):
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id=item_id,
path=config["db_path"],
)
assets = []
for index in range(count):
source_path = os.path.join(temp_dir, "%s-%02d.png" % (item_id, index + 1))
self._write_image(source_path)
assets.append(
image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
source_order=index + 1,
path=config["db_path"],
)
)
return project, assets
def _create_history_job(
self,
project,
source,
db_path,
*,
round_key=None,
slot_index=None,
status="succeeded",
local_path=None,
job_type="白底图",
):
asset = None
if local_path is not None:
asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=local_path,
parent_asset_id=source.id,
path=db_path,
)
job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type=job_type,
prompt="历史图片",
generation_round_key=round_key,
generation_slot_index=slot_index,
path=db_path,
)
return image_studio.update_job_status(
job.id,
status,
error="测试失败" if status == "failed" else None,
output_asset_id=asset.id if asset is not None else None,
path=db_path,
)
def test_tab_builds_suite_controls_without_old_detail_workspace(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
tab.resize(1180, 760)
tab.show()
self.app.processEvents()
self.assertEqual("productSuiteTab", tab.objectName())
self.assertEqual(1, tab.task_tabs.count())
self.assertEqual("套图任务 1", tab.task_tabs.tabText(0))
self.assertEqual("alias-a", tab.account_combo.currentData())
self.assertEqual("Shopee", tab.platform_combo.currentData())
self.assertEqual("中国台湾", tab.country_combo.currentData())
self.assertEqual("繁体中文", tab.language_combo.currentData())
self.assertEqual("1:1", tab.ratio_combo.currentData())
self.assertEqual("Shopee", tab.platform_combo.currentText())
self.assertEqual("中国台湾", tab.country_combo.currentText())
self.assertEqual("繁体中文", tab.language_combo.currentText())
self.assertEqual("1:1", tab.ratio_combo.currentText())
self.assertEqual("平台", tab.platform_label.text())
self.assertEqual("站点", tab.country_label.text())
self.assertEqual("语言", tab.language_label.text())
self.assertEqual("比例", tab.ratio_label.text())
for column, combo in enumerate(
(
tab.platform_combo,
tab.country_combo,
tab.language_combo,
tab.ratio_combo,
)
):
self.assertIs(combo, tab.settings_grid.itemAtPosition(1, column).widget())
self.assertEqual(
1,
len(
{
tab.platform_combo.width(),
tab.country_combo.width(),
tab.language_combo.width(),
tab.ratio_combo.width(),
}
),
)
self.assertLessEqual(tab.task_tabs.maximumHeight(), 36)
self.assertEqual(120, tab.account_combo.minimumWidth())
self.assertEqual(160, tab.account_combo.maximumWidth())
self.assertGreaterEqual(tab.item_id_edit.minimumWidth(), 120)
self.assertLessEqual(tab.item_id_edit.maximumWidth(), 140)
self.assertLess(
tab.context_bar_layout.indexOf(tab.history_button),
tab.context_bar_layout.indexOf(tab.account_combo),
)
self.assertLess(
tab.context_bar_layout.indexOf(tab.add_images_button),
tab.context_bar_layout.indexOf(tab.account_combo),
)
self.assertEqual(-1, tab.results_toolbar_layout.indexOf(tab.history_button))
self.assertFalse(hasattr(tab, "open_folder_button"))
self.assertEqual("合计 5 张", tab.category_total_label.text())
self.assertEqual("生成套图(5)", tab.generate_button.text())
self.assertLess(
tab.prompt_title_layout.indexOf(tab.prompt_title_label),
tab.prompt_title_layout.indexOf(tab.ai_write_button),
)
self.assertLess(
tab.prompt_title_layout.indexOf(tab.ai_write_button),
tab.prompt_title_layout.indexOf(tab.ai_cancel_button),
)
self.assertLess(
tab.prompt_title_layout.indexOf(tab.ai_cancel_button),
tab.prompt_title_layout.indexOf(tab.prompt_settings_button),
)
self.assertEqual("提示词设置", tab.prompt_settings_button.text())
visible_text = " ".join(
[widget.text() for widget in tab.findChildren(QLabel)]
+ [widget.text() for widget in tab.findChildren(QPushButton)]
)
self.assertNotIn("详情图", visible_text)
self.assertNotIn("AI工场", visible_text)
self.assertNotIn("打开结果文件夹", visible_text)
self.assertIn("白底图", visible_text)
self.assertIn("场景图", visible_text)
self.assertIn("卖点图", visible_text)
tab.add_custom_category()
self.assertFalse(tab.custom_category_edit.isHidden())
tab.custom_category_edit.setText("尺寸图")
tab._commit_custom_category()
self.assertTrue(tab.custom_category_edit.isHidden())
self.assertIn("尺寸图", tab._displayed_state.settings["categories"])
self.assertEqual(1, tab.category_rows["尺寸图"].count())
self.assert_removed(temp_dir)
def test_suite_setting_combos_forward_closed_wheel_to_config_scroll(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
tab.resize(620, 280)
tab.show()
self.app.processEvents()
scroll = tab.findChild(QScrollArea, "suiteConfigScroll")
self.assertIsNotNone(scroll)
scroll.widget().setMinimumHeight(scroll.viewport().height() + 400)
self.app.processEvents()
scroll.verticalScrollBar().setValue(0)
original_values = {
combo.objectName(): combo.currentData()
for combo in (
tab.platform_combo,
tab.country_combo,
tab.language_combo,
tab.ratio_combo,
)
}
for combo in (
tab.platform_combo,
tab.country_combo,
tab.language_combo,
tab.ratio_combo,
):
self._send_wheel(combo)
self.app.processEvents()
self.assertEqual(original_values[combo.objectName()], combo.currentData())
self.assertGreater(scroll.verticalScrollBar().value(), 0)
combo = tab.country_combo
combo.setCurrentIndex(0)
combo.showPopup()
self.app.processEvents()
self._send_wheel(combo.view())
self.app.processEvents()
self.assertEqual(0, combo.currentIndex())
combo.hidePopup()
combo.showPopup()
self.app.processEvents()
target_index = combo.model().index(1, 0)
target_rect = combo.view().visualRect(target_index)
QTest.mouseClick(combo.view().viewport(), Qt.LeftButton, Qt.NoModifier, target_rect.center())
self.app.processEvents()
self.assertEqual("新加坡", combo.currentData())
ratio_before = tab.ratio_combo.currentIndex()
tab.ratio_combo.setFocus()
QTest.keyClick(tab.ratio_combo, Qt.Key_Down)
self.app.processEvents()
self.assertNotEqual(ratio_before, tab.ratio_combo.currentIndex())
self.assertEqual(tab.ratio_combo.currentData(), tab._displayed_state.settings["ratio"])
self.assert_removed(temp_dir)
def test_prompt_settings_dialog_previews_validates_saves_and_restores(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
prompt_path = appconfig.product_suite_prompt_path(config)
prompts.ensure_default_product_suite_prompt(prompt_path)
dialog = ProductSuitePromptDialog(
prompt_path=prompt_path,
base_prompt="40小时续航,适合通勤",
settings=product_suite.default_suite_settings(),
item_id="51100639510",
)
self.addCleanup(dialog.close)
dialog.show()
self.app.processEvents()
self.assertEqual("白底图", dialog.category_combo.currentData())
self.assertTrue(dialog.preview_edit.isReadOnly())
self.assertIn(
"生成目标:白底图,生成 Shopee 台灣商品白底主圖,純白背景,商品清晰置中,不添加多餘文字。",
dialog.preview_edit.toPlainText(),
)
self.assertIn("40小时续航", dialog.preview_edit.toPlainText())
sizes = dialog.splitter.sizes()
self.assertLessEqual(abs(sizes[0] - sizes[1]), 12)
cursor = dialog.template_edit.textCursor()
cursor.movePosition(QTextCursor.End)
dialog.template_edit.setTextCursor(cursor)
dialog.insert_variable("商品ID")
self.assertTrue(dialog.template_edit.toPlainText().endswith("{商品ID}"))
default_text = prompts.load_default_product_suite_prompt()
invalid_text = default_text.replace("{图片比例}", "")
dialog.template_edit.setPlainText(invalid_text)
QTest.qWait(230)
self.assertFalse(dialog.save_button.isEnabled())
self.assertIn("缺少必需变量", dialog.validation_label.text())
custom_text = "自定义规则\n" + default_text
dialog.template_edit.setPlainText(custom_text)
QTest.qWait(230)
self.assertTrue(dialog.save_button.isEnabled())
self.assertTrue(dialog.save_template())
self.assertEqual(custom_text, prompts.load_product_suite_prompt(prompt_path))
self.assertFalse(dialog.is_dirty())
dialog.template_edit.setPlainText("临时未保存\n" + custom_text)
with mock.patch.object(dialog, "_confirm_restore", return_value=True):
self.assertTrue(dialog.restore_default())
self.assertEqual(default_text, dialog.template_edit.toPlainText())
self.assertEqual(default_text, prompts.load_product_suite_prompt(prompt_path))
self.assertFalse(dialog.is_dirty())
self.assert_removed(temp_dir)
def test_prompt_settings_dialog_unsaved_close_uses_chinese_three_way_action(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
prompt_path = appconfig.product_suite_prompt_path(config)
prompts.ensure_default_product_suite_prompt(prompt_path)
dialog = ProductSuitePromptDialog(
prompt_path=prompt_path,
base_prompt="卖点",
settings=product_suite.default_suite_settings(),
item_id="",
)
dialog.show()
self.app.processEvents()
dialog.template_edit.setPlainText(
"未保存修改\n" + prompts.load_default_product_suite_prompt()
)
self.assertTrue(dialog.is_dirty())
with mock.patch.object(dialog, "_unsaved_action", return_value="cancel"):
dialog.reject()
self.assertTrue(dialog.isVisible())
with mock.patch.object(dialog, "_unsaved_action", return_value="discard"):
dialog.reject()
self.assertFalse(dialog.isVisible())
self.assert_removed(temp_dir)
def test_invalid_product_suite_template_blocks_before_project_or_worker(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
with open(tab.product_suite_prompt_path, "w", encoding="utf-8") as handle:
handle.write("无效模板{未知变量}")
message = mock.Mock()
with mock.patch.object(tab, "_message", message), mock.patch.object(
tab,
"_start_thread",
) as start_thread:
result = tab.start_generation(tab._displayed_state)
self.assertFalse(result)
self.assertIsNone(tab._displayed_state.project_id)
self.assertEqual([], image_studio.list_projects(path=config["db_path"]))
start_thread.assert_not_called()
self.assertEqual("套图提示词模板无效", message.call_args.args[0])
self.assert_removed(temp_dir)
def test_prompt_preview_matches_frozen_generation_job_prompt(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 3)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
state.prompt = "40小时续航,适合通勤"
tab._load_state(state)
dialog = ProductSuitePromptDialog(
prompt_path=tab.product_suite_prompt_path,
base_prompt=state.prompt,
settings=state.settings,
item_id=state.item_id,
source_image_count=len(assets),
)
self.addCleanup(dialog.close)
expected = dialog.preview_edit.toPlainText()
with mock.patch.object(
tab,
"_confirm",
return_value=True,
), mock.patch.object(tab, "_start_thread", return_value=object()):
self.assertTrue(tab.start_generation(state))
self.assertEqual(expected, state.worker.job_specs[0]["prompt"])
self.assertIn("第2至3张仅作为风格、构图、场景或排版参考", expected)
state.worker = None
state.thread = None
self.assert_removed(temp_dir)
def test_category_rows_are_vertical_with_helpers_and_independent_counters(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
tab.resize(1180, 760)
tab.show()
self.app.processEvents()
expected = {
"白底图": (1, "白底主图,多角度呈现商品细节"),
"场景图": (2, "生活化场景展示商品使用方式"),
"模特场景图": (0, "模特或手持场景展示商品真实使用"),
"细节说明图": (0, "突出材质、做工和关键细节"),
"卖点图": (2, "突出核心卖点和差异化优势"),
}
self.assertEqual(list(expected), list(tab.category_rows))
for row_index, (name, (count, helper)) in enumerate(expected.items()):
row = tab.category_rows[name]
layout_index = tab.category_grid.indexOf(row)
self.assertEqual(
(row_index, 0, 1, 1),
tab.category_grid.getItemPosition(layout_index),
)
self.assertEqual(name, row.name_label.text())
self.assertEqual(count, row.count())
self.assertEqual(helper, row.helper_label.text())
self.assertIsNone(row.rename_button)
self.assertIsNone(row.delete_button)
scene_row = tab.category_rows["场景图"]
scene_row.plus_button.click()
self.assertIs(scene_row, tab.category_rows["场景图"])
self.assertEqual(3, scene_row.count())
self.assertEqual(1, tab.category_rows["白底图"].count())
self.assertEqual(2, tab.category_rows["卖点图"].count())
self.assertEqual("合计 6 张", tab.category_total_label.text())
self.assertEqual("生成套图(6)", tab.generate_button.text())
white_row = tab.category_rows["白底图"]
white_row.minus_button.click()
self.assertEqual(0, white_row.count())
self.assertFalse(white_row.minus_button.isEnabled())
white_row.minus_button.click()
self.assertEqual(0, tab._displayed_state.settings["categories"]["白底图"])
self.assertEqual("合计 5 张", tab.category_total_label.text())
tab._displayed_state.worker = object()
tab._apply_running_state(tab._displayed_state)
self.assertTrue(
all(
not row.plus_button.isEnabled()
for row in tab.category_rows.values()
)
)
tab._displayed_state.worker = None
tab._apply_running_state(tab._displayed_state)
self.assert_removed(temp_dir)
def test_custom_category_row_preserves_count_and_order_when_renamed(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
fixed_counts = {
name: tab.category_rows[name].count()
for name in ("白底图", "场景图", "卖点图")
}
tab.add_custom_category()
tab.custom_category_edit.setText("尺寸图")
tab._commit_custom_category()
custom_row = tab.category_rows["尺寸图"]
self.assertEqual(1, custom_row.count())
self.assertIsNone(custom_row.helper_label)
self.assertIsNotNone(custom_row.rename_button)
self.assertIsNotNone(custom_row.delete_button)
custom_row.plus_button.click()
self.assertEqual(2, custom_row.count())
with mock.patch(
"app.gui.tabs.product_suite.QInputDialog.getText",
return_value=("规格图", True),
):
custom_row.rename_button.click()
self.assertNotIn("尺寸图", tab.category_rows)
self.assertEqual(2, tab.category_rows["规格图"].count())
self.assertEqual(
["规格图"],
tab._displayed_state.settings["custom_category_order"],
)
tab.category_rows["规格图"].delete_button.click()
self.assertNotIn("规格图", tab.category_rows)
self.assertEqual([], tab._displayed_state.settings["custom_category_order"])
self.assertEqual(
fixed_counts,
{
name: tab.category_rows[name].count()
for name in ("白底图", "场景图", "卖点图")
},
)
self.assert_removed(temp_dir)
def test_category_counts_restore_after_task_switch_and_project_reload(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
first = ProductSuiteTab(config=config, db_path=config["db_path"])
state = first._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
first.change_category_count("场景图", 1)
first.add_custom_category()
first.custom_category_edit.setText("细节图")
first._commit_custom_category()
first.category_rows["细节图"].plus_button.click()
first.add_task(inherit=False)
self.assertEqual(2, first.category_rows["场景图"].count())
first.task_tabs.setCurrentIndex(0)
self.app.processEvents()
self.assertEqual(3, first.category_rows["场景图"].count())
self.assertEqual(2, first.category_rows["细节图"].count())
first.close()
first.deleteLater()
self.app.processEvents()
second = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(second.close)
reloaded = second._displayed_state
reloaded.account_alias = "alias-a"
reloaded.item_id = "51100639510"
second._bind_project(reloaded, load_existing=True)
self.assertEqual(3, second.category_rows["场景图"].count())
self.assertEqual(2, second.category_rows["细节图"].count())
self.assertEqual(
["细节图"],
reloaded.settings["custom_category_order"],
)
self.assert_removed(temp_dir)
def test_recent_dropdown_settings_restore_after_restart_and_project_wins(self):
with self.make_temp_dir() as temp_dir:
config_path = os.path.join(temp_dir, "config.json")
config = appconfig.save_config(self._config(temp_dir), path=config_path)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
project_settings = image_studio.project_suite_settings(project)
project_settings["ratio"] = "4:3"
image_studio.update_project_suite_settings(
project.id,
project_settings,
path=config["db_path"],
)
first = ProductSuiteTab(
config=config,
config_path=config_path,
db_path=config["db_path"],
)
first.ratio_combo.setCurrentIndex(first.ratio_combo.findData("16:9"))
self.assertEqual(
"16:9",
appconfig.product_suite_last_settings(
appconfig.load_config(config_path)
)["ratio"],
)
first.close()
first.deleteLater()
self.app.processEvents()
reloaded = appconfig.load_config(config_path)
second = ProductSuiteTab(
config=reloaded,
config_path=config_path,
db_path=reloaded["db_path"],
)
self.addCleanup(second.close)
self.assertEqual("16:9", second.ratio_combo.currentData())
state = second._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
second._bind_project(state, load_existing=True)
self.assertEqual("4:3", second.ratio_combo.currentData())
self.assertEqual("4:3", state.settings["ratio"])
self.assert_removed(temp_dir)
def test_recent_account_restores_and_missing_account_falls_back(self):
with self.make_temp_dir() as temp_dir:
config_path = os.path.join(temp_dir, "config.json")
config = appconfig.save_config(self._config(temp_dir), path=config_path)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
accounts.create_account("副店", "alias-b", debug_port=9223, config=config)
first = ProductSuiteTab(
config=config,
config_path=config_path,
db_path=config["db_path"],
)
first.account_combo.setCurrentIndex(
first.account_combo.findData("alias-b")
)
self.app.processEvents()
self.assertEqual(
"alias-b",
appconfig.product_suite_last_account_alias(
appconfig.load_config(config_path)
),
)
first.close()
first.deleteLater()
self.app.processEvents()
second_config = appconfig.load_config(config_path)
second = ProductSuiteTab(
config=second_config,
config_path=config_path,
db_path=second_config["db_path"],
)
self.assertEqual("alias-b", second.account_combo.currentData())
second.close()
second.deleteLater()
self.app.processEvents()
accounts.delete_account("alias-b", config=second_config)
third_config = appconfig.load_config(config_path)
third = ProductSuiteTab(
config=third_config,
config_path=config_path,
db_path=third_config["db_path"],
)
self.addCleanup(third.close)
self.assertEqual("alias-a", third.account_combo.currentData())
self.assertEqual(
"alias-a",
appconfig.product_suite_last_account_alias(
appconfig.load_config(config_path)
),
)
self.assert_removed(temp_dir)
def test_task_tabs_keep_independent_prompt_and_settings(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
tab.prompt_edit.setPlainText("任务一卖点")
first_state = tab._displayed_state
tab.ratio_combo.setCurrentIndex(tab.ratio_combo.findData("3:4"))
second_state = tab.add_task(inherit=True)
self.assertEqual(2, tab.task_tabs.count())
self.assertEqual("任务一卖点", second_state.prompt)
self.assertEqual("3:4", second_state.settings["ratio"])
tab.prompt_edit.setPlainText("任务二卖点")
tab.ratio_combo.setCurrentIndex(tab.ratio_combo.findData("16:9"))
tab.task_tabs.setCurrentIndex(0)
self.assertIs(first_state, tab._displayed_state)
self.assertEqual("任务一卖点", tab.prompt_edit.toPlainText())
self.assertEqual("3:4", tab.ratio_combo.currentData())
self.assert_removed(temp_dir)
def test_prompt_autosaves_after_debounce_and_ignores_unchanged_text(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
draft_prompt="原卖点",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = project.draft_prompt
state.last_saved_prompt = project.draft_prompt
tab._load_state(state)
original_update = image_studio.update_project_prompt
with mock.patch(
"app.gui.tabs.product_suite.image_studio.update_project_prompt",
wraps=original_update,
) as update_prompt:
tab.prompt_edit.setPlainText("第一版卖点")
tab.prompt_edit.setPlainText("最终卖点")
QTest.qWait(650)
self.app.processEvents()
stored = image_studio.get_project(project.id, path=config["db_path"])
self.assertEqual("最终卖点", stored.draft_prompt)
self.assertEqual(1, update_prompt.call_count)
QTest.qWait(600)
self.app.processEvents()
self.assertEqual(1, update_prompt.call_count)
self.assert_removed(temp_dir)
def test_prompt_switch_and_close_flush_to_the_correct_projects(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
first_project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
second_project = image_studio.create_or_get_project(
account,
item_id="51100639511",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
first_state = tab._displayed_state
first_state.account_alias = "alias-a"
first_state.item_id = first_project.item_id
first_state.project_id = first_project.id
first_state.project_binding_state = first_project.binding_state
tab._load_state(first_state)
tab.prompt_edit.setPlainText("商品一卖点")
second_state = tab.add_task(inherit=False)
second_state.account_alias = "alias-a"
second_state.item_id = second_project.item_id
second_state.project_id = second_project.id
second_state.project_binding_state = second_project.binding_state
tab._load_state(second_state)
tab.prompt_edit.setPlainText("商品二卖点")
tab.task_tabs.setCurrentIndex(0)
self.app.processEvents()
self.assertEqual(
"商品一卖点",
image_studio.get_project(
first_project.id,
path=config["db_path"],
).draft_prompt,
)
self.assertEqual(
"商品二卖点",
image_studio.get_project(
second_project.id,
path=config["db_path"],
).draft_prompt,
)
tab.prompt_edit.setPlainText("商品一关闭任务前卖点")
tab.close_task(0)
self.app.processEvents()
self.assertEqual(
"商品一关闭任务前卖点",
image_studio.get_project(
first_project.id,
path=config["db_path"],
).draft_prompt,
)
self.assertIs(second_state, tab._displayed_state)
tab.prompt_edit.setPlainText("商品二关闭程序前卖点")
tab.close()
self.app.processEvents()
self.assertEqual(
"商品二关闭程序前卖点",
image_studio.get_project(
second_project.id,
path=config["db_path"],
).draft_prompt,
)
self.assert_removed(temp_dir)
def test_prompt_without_project_creates_and_restores_draft(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
state = tab._displayed_state
tab.prompt_edit.setPlainText("尚未建立项目的卖点")
QTest.qWait(650)
self.app.processEvents()
projects = image_studio.list_projects(path=config["db_path"])
self.assertEqual(1, len(projects))
draft = projects[0]
self.assertIsNotNone(draft)
stored = image_studio.get_project(draft.id, path=config["db_path"])
self.assertEqual("尚未建立项目的卖点", stored.draft_prompt)
self.assertEqual("尚未建立项目的卖点", state.last_saved_prompt)
self.assertEqual(
[draft.id],
[
project.id
for project in image_studio.list_recoverable_draft_projects(
path=config["db_path"]
)
],
)
tab.close()
tab.deleteLater()
self.app.processEvents()
restored = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(restored.close)
restored_state = next(
value
for value in restored._states.values()
if value.project_id == draft.id
)
self.assertEqual("尚未建立项目的卖点", restored_state.prompt)
new_state = restored.add_task(inherit=False)
self.assertEqual("", new_state.prompt)
self.assert_removed(temp_dir)
def test_blank_prompt_without_project_does_not_create_draft(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
tab.prompt_edit.setPlainText(" ")
QTest.qWait(650)
self.app.processEvents()
self.assertEqual([], image_studio.list_projects(path=config["db_path"]))
self.assert_removed(temp_dir)
def test_prompt_autosave_failure_keeps_memory_text_and_reports_chinese_error(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
status = mock.Mock()
tab = ProductSuiteTab(
config=config,
db_path=config["db_path"],
status_callback=status,
)
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
tab._load_state(state)
with mock.patch(
"app.gui.tabs.product_suite.image_studio.update_project_prompt",
side_effect=OSError("disk unavailable"),
):
tab.prompt_edit.setPlainText("保存失败仍保留")
QTest.qWait(650)
self.app.processEvents()
self.assertEqual("保存失败仍保留", state.prompt)
self.assertEqual("", state.last_saved_prompt)
self.assertTrue(
any(
"商品卖点自动保存失败" in str(call.args[0])
for call in status.call_args_list
)
)
self.assertTrue(
any(call.kwargs.get("level") == "danger" for call in status.call_args_list)
)
self.assert_removed(temp_dir)
def test_ai_write_result_saves_prompt_without_an_extra_user_action(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
draft_prompt="原卖点",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = project.draft_prompt
state.last_saved_prompt = project.draft_prompt
state.ai_prompt_snapshot = project.draft_prompt
tab._load_state(state)
tab._on_ai_write_finished(
state,
{"ok": True, "cancelled": False, "text": "AI生成的新卖点"},
)
stored = image_studio.get_project(project.id, path=config["db_path"])
self.assertEqual("AI生成的新卖点", stored.draft_prompt)
self.assertEqual("AI生成的新卖点", state.last_saved_prompt)
self.assertEqual("AI生成的新卖点", tab.prompt_edit.toPlainText())
self.assert_removed(temp_dir)
def test_ai_write_uses_first_eight_originals_in_source_order_regardless_of_checks(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 9)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
tab._load_state(state)
tab.original_list.set_checked_asset_ids([assets[-1].id])
captured = {}
class _Signal:
def connect(self, callback):
self.callback = callback
class _AiWriteWorker:
def __init__(self, instruction, context, **kwargs):
captured["instruction"] = instruction
captured["context"] = context
captured["image_paths"] = list(kwargs.get("image_paths") or [])
self.finished = _Signal()
self.cancelled = _Signal()
self.failed = _Signal()
def cancel(self):
pass
with mock.patch(
"app.gui.tabs.product_suite.ProductSuiteAiWriteWorker",
_AiWriteWorker,
), mock.patch.object(tab, "_start_thread", return_value=object()), mock.patch.object(
tab, "_status"
) as status, mock.patch.object(
tab,
"_confirm_ai_write_request",
side_effect=lambda target, asset_ids, points_cost: tab._start_confirmed_ai_write(
target,
asset_ids,
),
):
tab.start_ai_write()
self.assertEqual(
[asset.local_path for asset in assets[:8]],
captured["image_paths"],
)
self.assertTrue(
any(
"已使用前8张商品原图进行理解" in str(call.args[0])
for call in status.call_args_list
)
)
state.ai_worker = None
state.ai_thread = None
state.ai_started_at = None
tab._apply_running_state(state)
self.assert_removed(temp_dir)
def test_ai_write_uses_cached_price_before_confirming_the_first_eight_images(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
config["ai"] = {
"cmhub": {
"base_url": "https://cmhub.example.com",
"vision_alias": "vision-standard",
}
}
appconfig.save_cmhub_config({"api_key": "test-key"}, path=config["cmhub_config_path"])
project, assets = self._create_project_with_assets(temp_dir, config, 9)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
self.addCleanup(cmhub_models.clear_model_catalog_cache)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
tab._load_state(state)
cmhub_models.cache_model_catalog(
"https://cmhub.example.com",
[
{
"alias": "vision-standard",
"operation_type": "vision",
"requires_image": True,
"pricing_status": "priced",
"prices": [{"points_cost": 2}],
}
],
)
with mock.patch.object(tab, "_confirm_ai_write_request") as confirm:
tab.start_ai_write()
state_arg, asset_ids, points_cost = confirm.call_args.args
self.assertIs(state_arg, state)
self.assertEqual(tuple(asset.id for asset in assets[:8]), asset_ids)
self.assertEqual("2", cmhub_models.format_points_cost(points_cost))
self.assertIsNone(state.ai_worker)
self.assertIsNone(state.ai_price_worker)
self.assert_removed(temp_dir)
def test_ai_write_cancelled_confirmation_does_not_start_worker(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 1)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
tab._load_state(state)
class _Button:
pass
class _MessageBox:
Question = 1
AcceptRole = 2
RejectRole = 3
def __init__(self, *args):
self.start_button = _Button()
self.cancel_button = _Button()
self.clicked = self.cancel_button
def setIcon(self, value):
pass
def setWindowTitle(self, value):
pass
def setText(self, value):
pass
def addButton(self, text, role):
return self.start_button if role == self.AcceptRole else self.cancel_button
def setDefaultButton(self, button):
pass
def setEscapeButton(self, button):
pass
def exec(self):
return 0
def clickedButton(self):
return self.clicked
with mock.patch("app.gui.tabs.product_suite.QMessageBox", _MessageBox), mock.patch.object(
tab,
"_start_confirmed_ai_write",
) as start_confirmed:
tab._confirm_ai_write_request(state, (assets[0].id,), None)
start_confirmed.assert_not_called()
self.assertIsNone(state.ai_worker)
self.assertFalse(state.ai_confirmation_open)
self.assert_removed(temp_dir)
def test_prompt_edit_expands_shrinks_and_reflows_without_internal_scrollbars(self):
edit = AutoHeightPlainTextEdit()
self.addCleanup(edit.close)
edit.resize(420, 96)
edit.show()
self.app.processEvents()
self.assertEqual(Qt.ScrollBarAlwaysOff, edit.horizontalScrollBarPolicy())
self.assertEqual(Qt.ScrollBarAlwaysOff, edit.verticalScrollBarPolicy())
self.assertGreaterEqual(edit.height(), 96)
minimum_height = edit.height()
edit.setPlainText("\n".join("第%d行商品卖点" % index for index in range(1, 13)))
QTest.qWait(50)
self.app.processEvents()
expanded_height = edit.height()
self.assertGreater(expanded_height, minimum_height)
edit.clear()
QTest.qWait(50)
self.app.processEvents()
self.assertEqual(minimum_height, edit.height())
edit.setPlainText("这是一段用于测试窗口变窄后自动换行的商品卖点内容。" * 16)
edit.setFixedWidth(420)
QTest.qWait(50)
self.app.processEvents()
wide_height = edit.height()
edit.setFixedWidth(180)
QTest.qWait(50)
self.app.processEvents()
self.assertGreater(edit.height(), wide_height)
def test_generation_terminal_watchdog_finalizes_once_and_restores_button(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 1)
jobs = [
image_studio.create_job(
project.id,
source_asset_id=assets[0].id,
job_type="白底图",
prompt="终态看门狗测试",
path=config["db_path"],
)
for _ in range(2)
]
for job in jobs:
image_studio.update_job_status(
job.id,
"succeeded",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.worker = mock.Mock()
state.thread = mock.Mock()
state.generation_run_token = "watchdog-run"
state.current_job_ids = [job.id for job in jobs]
state.generation_job_ids = [job.id for job in jobs]
state.total = len(jobs)
state.started_at = time.monotonic()
tab._generation_run_states["watchdog-run"] = state.key
tab._load_state(state)
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
tab._check_generation_watchdogs()
self.assertIsNotNone(state.worker)
tab._check_generation_watchdogs()
self.assertIsNone(state.worker)
self.assertIsNone(state.thread)
self.assertEqual("", state.generation_run_token)
self.assertTrue(tab.generate_button.text().startswith("生成套图"))
self.assertEqual(1, len(messages))
self.assertEqual("商品套图生成完成", messages[0][0])
self.assertFalse(
tab._finalize_generation(
state,
"watchdog-run",
{"total": 2, "success": 2},
source="worker",
)
)
self.assertEqual(1, len(messages))
self.assert_removed(temp_dir)
def test_generation_real_qthread_completion_restores_gui_state(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 1)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = "真实线程完成测试"
tab._load_state(state)
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
def fake_run_jobs(jobs, **kwargs):
job_list = list(jobs)
for job in job_list:
image_studio.update_job_status(
job.id,
"succeeded",
path=config["db_path"],
)
return {
"total": len(job_list),
"success": len(job_list),
"failed": 0,
"cancelled": 0,
"jobs": [],
}
with mock.patch(
"app.gui.workers.image_studio_generation.run_jobs",
side_effect=fake_run_jobs,
):
self.assertTrue(
tab.start_generation(
state,
specs=[
{
"source_asset_id": assets[0].id,
"job_type": "白底图",
"prompt": "真实线程完成测试",
}
],
)
)
deadline = time.monotonic() + 3
while state.worker is not None and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
self.assertIsNone(state.worker)
self.assertIsNone(state.thread)
self.assertTrue(tab.generate_button.text().startswith("生成套图"))
self.assertEqual(1, len(messages))
self.assertEqual("商品套图生成完成", messages[0][0])
self.assert_removed(temp_dir)
def test_generation_thread_finished_reconciles_nonterminal_job(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 1)
job = image_studio.create_job(
project.id,
source_asset_id=assets[0].id,
job_type="场景图",
prompt="线程结束兜底测试",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.worker = mock.Mock()
state.thread = mock.Mock()
state.generation_run_token = "thread-fallback"
state.current_job_ids = [job.id]
state.generation_job_ids = [job.id]
state.total = 1
state.started_at = time.monotonic()
tab._generation_run_states["thread-fallback"] = state.key
tab._load_state(state)
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
tab._handle_generation_thread_finished("thread-fallback")
stored = image_studio.get_job(job.id, path=config["db_path"])
self.assertEqual("cancelled", stored.status)
self.assertEqual(
image_studio.JOB_RECOVERY_REGENERATE,
stored.recovery_action,
)
self.assertIsNone(state.worker)
self.assertEqual("商品套图生成未完整结束", messages[0][0])
self.assertIn("稍后继续查询", messages[0][1])
self.assert_removed(temp_dir)
def test_generation_old_run_token_and_repeated_stop_are_ignored(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
statuses = []
tab = ProductSuiteTab(
config=config,
db_path=config["db_path"],
status_callback=lambda message, level=None: statuses.append(
(message, level)
),
)
self.addCleanup(tab.close)
state = tab._displayed_state
state.worker = mock.Mock()
state.thread = mock.Mock()
state.generation_run_token = "current-run"
state.generation_stop_requested = True
state.total = 2
tab._generation_run_states["current-run"] = state.key
original_worker = state.worker
self.assertFalse(
tab._finalize_generation(
state,
"old-run",
{"total": 2, "success": 2},
source="worker",
)
)
self.assertIs(original_worker, state.worker)
confirm = mock.Mock(return_value=True)
with mock.patch.object(tab, "_confirm", confirm):
tab.toggle_generation()
confirm.assert_not_called()
original_worker.cancel.assert_not_called()
self.assertEqual(("正在停止当前套图任务", "warning"), statuses[-1])
self.assert_removed(temp_dir)
def test_generation_immediate_stop_before_job_creation_finishes_cleanly(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.worker = mock.Mock()
state.thread = mock.Mock()
state.generation_run_token = "immediate-stop"
state.generation_stop_requested = True
state.total = 3
state.started_at = time.monotonic()
tab._generation_run_states["immediate-stop"] = state.key
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
tab._on_generation_finished_signal(
{
"run_token": "immediate-stop",
"cancelled": True,
}
)
self.assertIsNone(state.worker)
self.assertEqual(3, state.done)
self.assertEqual("商品套图生成已停止", messages[0][0])
self.assertIn("停止3张", messages[0][1])
self.assert_removed(temp_dir)
def test_project_settings_and_result_history_use_existing_backend(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
source_path = os.path.join(temp_dir, "source.png")
self._write_image(source_path)
source = image_studio_images.import_original_files(
project.id,
[source_path],
path=config["db_path"],
config=config,
)["assets"][0]
job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="场景卖点",
path=config["db_path"],
)
image_studio.update_job_status(
job.id,
"failed",
error="上游超时 https://example.invalid/private",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
state.current_job_ids = [job.id]
tab._load_state(state)
self.assertEqual([source.id], tab.original_list.asset_ids())
self.assertEqual("共 1 张 · 成功 0 张", tab.result_summary_label.text())
cards = tab.findChildren(SuiteResultCard)
self.assertEqual(1, len(cards))
self.assertNotIn(
"https://",
" ".join(label.text() for label in cards[0].findChildren(QLabel)),
)
state.settings["ratio"] = "4:3"
state.prompt = "持久化卖点"
tab._persist_state(state)
stored = image_studio.get_project(project.id, path=config["db_path"])
self.assertEqual("持久化卖点", stored.draft_prompt)
self.assertEqual("4:3", image_studio.project_suite_settings(stored)["ratio"])
self.assert_removed(temp_dir)
def test_failed_job_retry_replaces_current_slot_and_keeps_history(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
success_path = os.path.join(temp_dir, "success.jpg")
retry_path = os.path.join(temp_dir, "retry.jpg")
self._write_image(success_path)
self._write_image(retry_path)
success_asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=success_path,
parent_asset_id=source.id,
path=config["db_path"],
)
success_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="白底图",
prompt="成功图",
path=config["db_path"],
)
success_job = image_studio.update_job_status(
success_job.id,
"succeeded",
output_asset_id=success_asset.id,
path=config["db_path"],
)
failed_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="失败图",
path=config["db_path"],
)
failed_job = image_studio.update_job_status(
failed_job.id,
"failed",
error="上游生成失败",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.current_job_ids = [success_job.id, failed_job.id]
tab._load_state(state)
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
def fake_run_jobs(jobs, **kwargs):
job = list(jobs)[0]
retry_asset = image_studio.add_asset(
project.id,
"generated_main",
local_path=retry_path,
parent_asset_id=source.id,
path=config["db_path"],
)
image_studio.update_job_status(
job.id,
"succeeded",
output_asset_id=retry_asset.id,
path=config["db_path"],
)
return {
"total": 1,
"success": 1,
"failed": 0,
"cancelled": 0,
"jobs": [],
}
with mock.patch(
"app.gui.workers.image_studio_generation.run_jobs",
side_effect=fake_run_jobs,
):
tab.retry_job(failed_job)
generation_thread = state.thread
deadline = time.monotonic() + 3
while state.worker is not None and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
while generation_thread is not None and time.monotonic() < deadline:
try:
running = generation_thread.isRunning()
except RuntimeError:
generation_thread = None
break
if not running:
break
QTest.qWait(20)
self.app.processEvents()
if generation_thread is not None:
self.assertFalse(generation_thread.isRunning())
all_jobs = image_studio.list_jobs(project.id, path=config["db_path"])
retry_jobs = [
job
for job in all_jobs
if job.id not in {success_job.id, failed_job.id}
]
self.assertEqual(1, len(retry_jobs))
retry_job = retry_jobs[0]
self.assertEqual(
[success_job.id, retry_job.id],
state.current_job_ids,
)
self.assertEqual([], state.generation_job_ids)
self.assertEqual(
[success_job.id, retry_job.id],
[job.id for job in tab._jobs_for_state(state)],
)
self.assertEqual("图片重试成功", messages[-1][0])
self.assert_removed(temp_dir)
def test_generation_round_restores_after_project_rebind_and_retry_keeps_slot(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
round_key = "persisted-round"
first = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="白底图",
prompt="第一张",
generation_round_key=round_key,
generation_slot_index=0,
path=config["db_path"],
)
first = image_studio.update_job_status(
first.id,
"succeeded",
path=config["db_path"],
)
failed = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="第二张",
generation_round_key=round_key,
generation_slot_index=1,
path=config["db_path"],
)
failed = image_studio.update_job_status(
failed.id,
"failed",
path=config["db_path"],
)
image_studio.set_current_generation_round(
project.id,
round_key,
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
tab._bind_project(state, load_existing=True)
self.assertEqual(round_key, state.current_generation_round_key)
self.assertEqual([first.id, failed.id], state.current_job_ids)
retry = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type=failed.job_type,
prompt=failed.prompt,
generation_round_key=failed.generation_round_key,
generation_slot_index=failed.generation_slot_index,
path=config["db_path"],
)
image_studio.update_job_status(retry.id, "succeeded", path=config["db_path"])
reopened = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(reopened.close)
reopened_state = reopened._displayed_state
reopened_state.account_alias = "alias-a"
reopened_state.item_id = project.item_id
reopened._bind_project(reopened_state, load_existing=True)
self.assertEqual(round_key, reopened_state.current_generation_round_key)
self.assertEqual([first.id, retry.id], reopened_state.current_job_ids)
self.assertEqual(
[first.id, retry.id],
[job.id for job in reopened._jobs_for_state(reopened_state)],
)
self.assert_removed(temp_dir)
def test_new_generation_round_promotes_partial_success_and_keeps_previous_on_failure(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
old_round = "old-current-round"
old_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="白底图",
prompt="旧结果",
generation_round_key=old_round,
generation_slot_index=0,
path=config["db_path"],
)
old_job = image_studio.update_job_status(
old_job.id,
"succeeded",
path=config["db_path"],
)
image_studio.set_current_generation_round(
project.id,
old_round,
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
tab._bind_project(state, load_existing=True)
messages = []
tab._message = lambda title, message, **kwargs: messages.append(title)
first_run_jobs = []
def partial_success(jobs, **kwargs):
first_run_jobs[:] = list(jobs)
image_studio.update_job_status(
first_run_jobs[0].id,
"succeeded",
path=config["db_path"],
)
image_studio.update_job_status(
first_run_jobs[1].id,
"failed",
path=config["db_path"],
)
return {"total": 2, "success": 1, "failed": 1, "cancelled": 0}
specs = [
{"source_asset_id": source.id, "job_type": "白底图", "prompt": "新图1"},
{"source_asset_id": source.id, "job_type": "场景图", "prompt": "新图2"},
]
with mock.patch(
"app.gui.workers.image_studio_generation.run_jobs",
side_effect=partial_success,
):
self.assertTrue(tab.start_generation(state, specs=specs))
first_thread = state.thread
deadline = time.monotonic() + 3
while state.worker is not None and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
if first_thread is not None:
try:
deadline = time.monotonic() + 3
while first_thread.isRunning() and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
self.assertFalse(first_thread.isRunning())
except RuntimeError:
pass
new_round = image_studio.get_current_generation_round(
project.id,
path=config["db_path"],
)
self.assertNotEqual(old_round, new_round)
self.assertEqual(
[(new_round, 0), (new_round, 1)],
[
(job.generation_round_key, job.generation_slot_index)
for job in first_run_jobs
],
)
self.assertEqual(
[job.id for job in first_run_jobs],
state.current_job_ids,
)
def all_failed(jobs, **kwargs):
for job in jobs:
image_studio.update_job_status(
job.id,
"failed",
path=config["db_path"],
)
return {"total": 2, "success": 0, "failed": 2, "cancelled": 0}
with mock.patch(
"app.gui.workers.image_studio_generation.run_jobs",
side_effect=all_failed,
):
self.assertTrue(tab.start_generation(state, specs=specs))
second_thread = state.thread
deadline = time.monotonic() + 3
while state.worker is not None and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
if second_thread is not None:
try:
deadline = time.monotonic() + 3
while second_thread.isRunning() and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
self.assertFalse(second_thread.isRunning())
except RuntimeError:
pass
self.assertEqual(
new_round,
image_studio.get_current_generation_round(
project.id,
path=config["db_path"],
),
)
self.assertEqual(
[job.id for job in first_run_jobs],
state.current_job_ids,
)
self.assertEqual(
["商品套图生成完成", "商品套图生成完成"],
messages,
)
self.assert_removed(temp_dir)
def test_retry_tracks_only_new_job_and_keeps_current_results(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
success_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="白底图",
prompt="成功图",
path=config["db_path"],
)
success_job = image_studio.update_job_status(
success_job.id,
"succeeded",
path=config["db_path"],
)
failed_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="失败图",
path=config["db_path"],
)
failed_job = image_studio.update_job_status(
failed_job.id,
"failed",
path=config["db_path"],
)
retry_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="重试图",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.current_job_ids = [success_job.id, failed_job.id]
state.generation_mode = "retry"
state.generation_retry_job_id = failed_job.id
state.total = 1
tab._set_generation_job_ids(state, [retry_job.id])
self.assertEqual(
[success_job.id, retry_job.id],
state.current_job_ids,
)
self.assertEqual([retry_job.id], state.generation_job_ids)
snapshot = tab._generation_job_snapshot(state)
self.assertEqual(1, snapshot["job_ids"])
self.assertEqual(1, snapshot["active"])
image_studio.update_job_status(
retry_job.id,
"failed",
error="重试仍失败",
path=config["db_path"],
)
state.worker = mock.Mock()
state.thread = mock.Mock()
state.generation_run_token = "retry-failed"
state.started_at = time.monotonic()
tab._generation_run_states["retry-failed"] = state.key
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
self.assertTrue(
tab._finalize_generation(
state,
"retry-failed",
{"total": 1, "success": 0, "failed": 1},
source="worker",
)
)
self.assertEqual("图片重试失败", messages[-1][0])
retry_cards = [
card
for card in tab.findChildren(SuiteResultCard)
if card.job.id == retry_job.id
]
self.assertEqual(1, len(retry_cards))
self.assertTrue(
any(
button.text() == "重试"
for button in retry_cards[0].findChildren(QPushButton)
)
)
tab._load_state(state)
with mock.patch.object(
tab,
"_start_thread",
return_value=mock.Mock(),
):
self.assertTrue(
tab.start_generation(
state,
specs=[
{
"source_asset_id": source.id,
"job_type": failed_job.job_type,
"prompt": failed_job.prompt,
}
],
retry_job_id=failed_job.id,
)
)
self.assertFalse(tab.history_button.isCheckable())
self.assertEqual("历史生成", tab.history_button.text())
state.worker = None
state.thread = None
state.generation_run_token = ""
self.assert_removed(temp_dir)
def test_history_dialog_groups_current_round_retries_and_paginates(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
db_path = config["db_path"]
other_project = image_studio.create_or_get_project(
account_alias="其他店",
account_slug="other-shop",
item_id="51100639511",
path=db_path,
)
other_source = image_studio.add_asset(
other_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=db_path,
)
other_job = self._create_history_job(
other_project,
other_source,
db_path,
round_key="other-round",
slot_index=0,
)
legacy_job = self._create_history_job(project, source, db_path)
for index in range(20):
self._create_history_job(
project,
source,
db_path,
round_key="old-round-%02d" % index,
slot_index=0,
job_type="场景图",
)
current_key = "current-round"
self._create_history_job(
project,
source,
db_path,
round_key=current_key,
slot_index=0,
status="failed",
)
missing_path = os.path.join(temp_dir, "missing-history-image.png")
retry_job = self._create_history_job(
project,
source,
db_path,
round_key=current_key,
slot_index=0,
local_path=missing_path,
)
usable_path = os.path.join(temp_dir, "usable-history-image.png")
self._write_image(usable_path)
usable_job = self._create_history_job(
project,
source,
db_path,
round_key=current_key,
slot_index=1,
local_path=usable_path,
job_type="卖点图",
)
image_studio.set_current_generation_round(
project.id,
current_key,
path=db_path,
)
dialog = ProductSuiteHistoryDialog(project.id, db_path=db_path)
dialog.show()
self.app.processEvents()
self.assertIn("店铺:主店", dialog.context_label.text())
self.assertIn("商品ID:51100639510", dialog.context_label.text())
self.assertEqual(20, dialog._round_count)
self.assertEqual(1, dialog._available_image_count)
self.assertTrue(dialog.load_more_button.isVisible())
self.assertTrue(
any(label.text() == "当前" for label in dialog.findChildren(QLabel))
)
cards = dialog.findChildren(SuiteHistoryImageCard)
cards_by_job = {card.job.id: card for card in cards}
self.assertIn(retry_job.id, cards_by_job)
self.assertIn(usable_job.id, cards_by_job)
self.assertNotIn(other_job.id, cards_by_job)
self.assertNotIn(legacy_job.id, cards_by_job)
self.assertIn("本槽位已重试1次", cards_by_job[retry_job.id].toolTip())
self.assertIn("本地图片文件不可用", cards_by_job[retry_job.id].toolTip())
dialog.load_more()
self.app.processEvents()
self.assertEqual(22, dialog._round_count)
self.assertFalse(dialog.load_more_button.isVisible())
self.assertTrue(
any(
label.text() == "旧版历史记录"
for label in dialog.findChildren(QLabel)
)
)
cards_by_job = {
card.job.id: card
for card in dialog.findChildren(SuiteHistoryImageCard)
}
self.assertIn(legacy_job.id, cards_by_job)
self.assert_removed(temp_dir)
def test_history_dialog_actions_are_read_only(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
image_path = os.path.join(temp_dir, "history-image.png")
self._write_image(image_path)
job = self._create_history_job(
project,
sources[0],
config["db_path"],
round_key="current-round",
slot_index=0,
local_path=image_path,
)
image_studio.set_current_generation_round(
project.id,
"current-round",
path=config["db_path"],
)
asset = image_studio.get_asset(job.output_asset_id, path=config["db_path"])
dialog = ProductSuiteHistoryDialog(project.id, db_path=config["db_path"])
with mock.patch.object(ProductSuitePreviewDialog, "exec", return_value=0) as preview:
dialog._preview_job(job, asset)
preview.assert_called_once_with()
class MenuAction:
def __init__(self, text):
self._text = text
def text(self):
return self._text
class MenuStub:
selected_text = ""
observed_actions = []
def __init__(self, parent=None):
self._actions = []
def addAction(self, text):
action = MenuAction(text)
self._actions.append(action)
return action
def exec(self, _position):
type(self).observed_actions.extend(
action.text() for action in self._actions
)
return next(
(
action
for action in self._actions
if action.text() == type(self).selected_text
),
None,
)
with mock.patch("app.gui.tabs.product_suite.QMenu", MenuStub):
dialog._show_job_menu(job, asset, None)
self.assertEqual(
["预览", "复制路径", "打开所在文件夹"],
MenuStub.observed_actions,
)
MenuStub.selected_text = "复制路径"
with mock.patch("app.gui.tabs.product_suite.QMenu", MenuStub):
dialog._show_job_menu(job, asset, None)
self.assertEqual(image_path, QApplication.clipboard().text())
MenuStub.selected_text = "打开所在文件夹"
with mock.patch("app.gui.tabs.product_suite.QMenu", MenuStub), mock.patch(
"app.gui.tabs.product_suite.file_manager.open_in_file_manager"
) as open_folder:
dialog._show_job_menu(job, asset, None)
open_folder.assert_called_once_with(os.path.dirname(image_path))
self.assert_removed(temp_dir)
def test_global_history_dialog_lists_filters_and_previews_round_images(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
source = sources[0]
other_account = accounts.create_account(
"副店",
"other-shop",
debug_port=9223,
config=config,
)
accounts.create_account(
"空店",
"empty-shop",
debug_port=9224,
config=config,
)
other_project = image_studio.create_or_get_project(
other_account,
item_id="51100639511",
path=config["db_path"],
)
former_project = image_studio.create_or_get_project(
account_alias="former-shop",
account_name="旧店",
account_slug="former-shop",
item_id="51100639512",
path=config["db_path"],
)
other_source = image_studio.add_asset(
other_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=config["db_path"],
)
former_source = image_studio.add_asset(
former_project.id,
image_studio.ASSET_KIND_ORIGINAL,
path=config["db_path"],
)
for index in range(7):
image_path = os.path.join(temp_dir, "global-%d.png" % index)
self._write_image(image_path)
self._create_history_job(
project,
source,
config["db_path"],
round_key="main-round",
slot_index=index,
local_path=image_path,
job_type="场景图",
)
other_path = os.path.join(temp_dir, "other-global.png")
self._write_image(other_path)
self._create_history_job(
other_project,
other_source,
config["db_path"],
round_key="other-round",
slot_index=0,
local_path=other_path,
job_type="卖点图",
)
former_path = os.path.join(temp_dir, "former-global.png")
self._write_image(former_path)
self._create_history_job(
former_project,
former_source,
config["db_path"],
round_key="former-round",
slot_index=0,
local_path=former_path,
job_type="白底图",
)
image_studio.set_current_generation_round(
project.id,
"main-round",
path=config["db_path"],
)
dialog = ProductSuiteGlobalHistoryDialog(
current_project_id=project.id,
db_path=config["db_path"],
)
dialog.show()
self.app.processEvents()
self.assertEqual(3, dialog._round_count)
self.assertEqual("全部店铺", dialog.account_filter_combo.itemText(0))
main_account_index = dialog.account_filter_combo.findData("alias-a")
other_account_index = dialog.account_filter_combo.findData("other-shop")
empty_account_index = dialog.account_filter_combo.findData("empty-shop")
former_account_index = dialog.account_filter_combo.findData("former-shop")
self.assertGreaterEqual(main_account_index, 0)
self.assertGreaterEqual(other_account_index, 0)
self.assertGreaterEqual(empty_account_index, 0)
self.assertGreaterEqual(former_account_index, 0)
self.assertEqual(
"主店(alias-a)",
dialog.account_filter_combo.itemText(main_account_index),
)
self.assertEqual(
"副店(other-shop)",
dialog.account_filter_combo.itemText(other_account_index),
)
self.assertEqual(
"历史店铺:former-shop(账号已删除)",
dialog.account_filter_combo.itemText(former_account_index),
)
rows = dialog.findChildren(SuiteGlobalHistoryRoundRow)
main_row = next(
row for row in rows if row.round_info.project_id == project.id
)
self.assertEqual(5, len(main_row.findChildren(SuiteGlobalHistoryThumbnail)))
self.assertTrue(
any(
label.text() == "+2"
for label in main_row.findChildren(QLabel)
)
)
self.assertTrue(
any(label.text() == "当前" for label in main_row.findChildren(QLabel))
)
with mock.patch.object(
ProductSuiteRoundPreviewDialog,
"exec",
return_value=0,
) as preview:
dialog._preview_round(main_row, 1)
preview.assert_called_once_with()
dialog.account_filter_combo.setCurrentIndex(other_account_index)
self.app.processEvents()
self.assertEqual(1, dialog._round_count)
self.assertEqual(
[other_project.id],
[row.round_info.project_id for row in dialog._history_rows],
)
dialog.item_filter_edit.setText("51100639511")
dialog.refresh_history()
self.app.processEvents()
self.assertEqual(1, dialog._round_count)
self.assertEqual("other-shop", dialog.account_filter_combo.currentData())
dialog.item_filter_edit.clear()
dialog.account_filter_combo.setCurrentIndex(0)
dialog.current_project_checkbox.setChecked(True)
self.app.processEvents()
self.assertEqual(1, dialog._round_count)
self.assertEqual(
[project.id],
[row.round_info.project_id for row in dialog._history_rows],
)
self.assert_removed(temp_dir)
def test_history_button_reuses_global_dialog_and_keeps_it_when_task_closes(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, sources = self._create_project_with_assets(temp_dir, config, 1)
image_path = os.path.join(temp_dir, "current-image.png")
self._write_image(image_path)
job = self._create_history_job(
project,
sources[0],
config["db_path"],
round_key="current-round",
slot_index=0,
local_path=image_path,
)
image_studio.set_current_generation_round(
project.id,
"current-round",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.current_generation_round_key = "current-round"
state.current_job_ids = [job.id]
tab._load_state(state)
tab.show()
self.app.processEvents()
self.assertFalse(tab.history_button.isCheckable())
self.assertEqual([job.id], [entry.id for entry in tab._jobs_for_state(state)])
tab.open_history_dialog()
self.app.processEvents()
dialog = tab._history_dialog
self.assertIsInstance(dialog, ProductSuiteGlobalHistoryDialog)
self.assertEqual(project.id, dialog.current_project_id)
self.assertTrue(dialog.isVisible())
self.assertFalse(dialog.current_project_checkbox.isChecked())
tab.open_history_dialog(
current_project_only=True,
current_project_id=project.id,
)
self.app.processEvents()
self.assertTrue(dialog.current_project_checkbox.isChecked())
tab.open_history_dialog()
self.assertIs(dialog, tab._history_dialog)
self.assertFalse(dialog.current_project_checkbox.isChecked())
tab.close_task(0)
self.app.processEvents()
self.assertIs(dialog, tab._history_dialog)
self.assertTrue(dialog.isVisible())
self.assert_removed(temp_dir)
def test_history_button_opens_global_dialog_for_empty_current_project(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, _ = self._create_project_with_assets(temp_dir, config, 0)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
tab.open_history_dialog()
self.app.processEvents()
dialog = tab._history_dialog
self.assertIsInstance(dialog, ProductSuiteGlobalHistoryDialog)
self.assertTrue(dialog.isVisible())
self.assertTrue(
any(
label.text() == "暂无套图历史生成记录,完成套图生成后会自动出现在这里"
for label in dialog.findChildren(QLabel)
)
)
self.assert_removed(temp_dir)
def test_original_list_expands_without_internal_scrollbars(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 0)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab.resize(1180, 760)
tab.show()
tab._load_state(state)
self.app.processEvents()
self.assertEqual(Qt.ScrollBarAlwaysOff, tab.original_list.horizontalScrollBarPolicy())
self.assertEqual(Qt.ScrollBarAlwaysOff, tab.original_list.verticalScrollBarPolicy())
self.assertEqual(0, tab.original_list.count())
self.assertEqual([], tab.original_list.asset_ids())
self.assertEqual("0/16", tab.original_count_label.text())
self.assertEqual("已选 0 张", tab.original_selected_label.text())
self.assertFalse(tab.select_all_originals_button.isEnabled())
self.assertFalse(tab.invert_originals_button.isEnabled())
self.assertEqual(1, tab.original_list.content_row_count())
empty_height = tab.original_list.height()
for target_count in (1, 2, 5, 6, 7, 16):
for index in range(len(assets), target_count):
source_path = os.path.join(temp_dir, "added-%02d.png" % (index + 1))
self._write_image(source_path)
assets.append(
image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
source_order=index + 1,
path=config["db_path"],
)
)
tab._refresh_originals(state)
self.app.processEvents()
self.assertEqual(target_count, tab.original_list.count())
self.assertEqual(target_count, len(tab.original_list.asset_ids()))
columns = tab.original_list.content_column_count()
expected_rows = (target_count + columns - 1) // columns
self.assertEqual(expected_rows, tab.original_list.content_row_count())
self.assertEqual("%d/16" % target_count, tab.original_count_label.text())
if target_count == 1:
self.assertEqual(empty_height, tab.original_list.height())
self.assertTrue(tab.add_images_button.isEnabled())
wide_height = tab.original_list.height()
tab.original_list.setFixedWidth(250)
self.app.processEvents()
self.assertEqual(2, tab.original_list.content_column_count())
self.assertEqual(8, tab.original_list.content_row_count())
self.assertGreater(tab.original_list.height(), wide_height)
self.assertFalse(tab.add_images_button.isEnabled())
self.assertEqual("已达到16张商品原图上限", tab.add_images_button.toolTip())
self.assert_removed(temp_dir)
def test_empty_original_list_accepts_file_drop_and_clipboard_image(self):
original_list = ProductOriginalList()
self.addCleanup(original_list.close)
original_list.setFixedSize(360, 110)
original_list.show()
self.app.processEvents()
self.assertEqual(0, original_list.count())
self.assertEqual("暂无商品原图", original_list.EMPTY_STATE_TEXT)
self.assertEqual(1, original_list.content_row_count())
dropped = []
pasted = []
original_list.filesDropped.connect(dropped.append)
original_list.clipboardImage.connect(pasted.append)
class DropEvent:
def __init__(self, mime_data):
self._mime_data = mime_data
self.accepted = False
def mimeData(self):
return self._mime_data
def acceptProposedAction(self):
self.accepted = True
mime_data = QMimeData()
expected_path = os.path.abspath("待导入图片.png")
mime_data.setUrls([QUrl.fromLocalFile(expected_path)])
drop_event = DropEvent(mime_data)
original_list.dropEvent(drop_event)
self.assertTrue(drop_event.accepted)
self.assertEqual(
os.path.normcase(os.path.normpath(expected_path)),
os.path.normcase(os.path.normpath(dropped[0][0])),
)
clipboard_image = QImage(24, 18, QImage.Format_RGB32)
clipboard_image.fill(0xFF336699)
QApplication.clipboard().setImage(clipboard_image)
original_list.setFocus()
QTest.keyClick(original_list, Qt.Key_V, Qt.ControlModifier)
self.app.processEvents()
QApplication.clipboard().clear()
self.assertEqual(1, len(pasted))
self.assertTrue(pasted[0].startswith(b"\x89PNG\r\n\x1a\n"))
def test_add_images_button_tracks_capacity_import_and_generation(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, _ = self._create_project_with_assets(temp_dir, config, 1)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab._load_state(state)
style = tab.add_images_button.styleSheet()
for color in ("#0969da", "#eef4ff", "#dbeafe", "#c7ddff", "#f6f8fa"):
self.assertIn(color, style)
self.assertTrue(tab.add_images_button.isEnabled())
self.assertEqual("添加本地商品原图", tab.add_images_button.toolTip())
state.import_worker = object()
tab._refresh_add_images_action(state)
self.assertFalse(tab.add_images_button.isEnabled())
self.assertEqual("正在添加商品原图", tab.add_images_button.toolTip())
state.import_thread = object()
tab._on_import_failed(state, "测试导入失败")
self.assertIsNone(state.import_worker)
self.assertIsNone(state.import_thread)
self.assertTrue(tab.add_images_button.isEnabled())
state.worker = object()
tab._refresh_add_images_action(state)
self.assertFalse(tab.add_images_button.isEnabled())
self.assertEqual("生成中不能添加商品原图", tab.add_images_button.toolTip())
state.worker = None
tab._refresh_add_images_action(state)
self.assertTrue(tab.add_images_button.isEnabled())
def test_original_checks_preserve_on_refresh_and_clear_on_task_switch(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 2)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab._load_state(state)
self.assertEqual(2, tab.original_list.count())
self.assertTrue(tab.select_all_originals_button.isEnabled())
self.assertTrue(tab.invert_originals_button.isEnabled())
for row in range(tab.original_list.count()):
self.assertIsNotNone(tab.original_list.item(row).data(ORIGINAL_CHECK_STATE_ROLE))
tab.select_all_originals_button.click()
self.assertEqual([asset.id for asset in assets], tab.original_list.checked_asset_ids())
self.assertEqual("已选 2 张", tab.original_selected_label.text())
tab.invert_originals_button.click()
self.assertEqual([], tab.original_list.checked_asset_ids())
tab.original_list.set_checked_asset_ids([assets[1].id])
tab._refresh_originals(state)
self.assertEqual([assets[1].id], tab.original_list.checked_asset_ids())
tab.reorder_originals([assets[1].id, assets[0].id])
self.assertEqual([assets[1].id], tab.original_list.checked_asset_ids())
tab.add_task(inherit=True)
self.assertEqual([], tab.original_list.checked_asset_ids())
self.assertFalse(tab.select_all_originals_button.isEnabled())
tab.task_tabs.setCurrentIndex(0)
self.assertEqual([], tab.original_list.checked_asset_ids())
self.assertEqual("已选 0 张", tab.original_selected_label.text())
self.assert_removed(temp_dir)
def test_temporary_draft_allows_local_work_but_blocks_shopee_pull_and_recovers(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
self.assertTrue(tab.add_images_button.isEnabled())
self.assertFalse(tab.item_id_hint_label.isHidden())
self.assertEqual("请输入正确的商品ID", tab.item_id_hint_label.text())
with mock.patch(
"app.gui.tabs.product_suite.QFileDialog.getOpenFileNames",
return_value=([], ""),
):
tab.choose_images()
self.assertEqual([], image_studio.list_projects(path=config["db_path"]))
source_path = os.path.join(temp_dir, "draft-source.png")
self._write_image(source_path)
with mock.patch.object(tab, "_start_thread", return_value=object()):
tab._start_import(file_paths=[source_path])
worker = state.import_worker
draft = image_studio.get_project(state.project_id, path=config["db_path"])
self.assertTrue(image_studio.is_draft_project(draft))
self.assertEqual("", state.item_id)
self.assertIn("临时草稿", tab.task_tabs.tabText(tab.task_tabs.currentIndex()))
self.assertFalse(tab.item_id_hint_label.isHidden())
self.assertTrue(tab.pull_button.isEnabled())
self.assertEqual("需要先绑定正式商品ID", tab.pull_button.toolTip())
tab._on_import_finished(state, worker.execute())
self.assertEqual(1, len(image_studio.list_assets(draft.id, path=config["db_path"])))
captured = {}
class _Signal:
def connect(self, callback):
self.callback = callback
class _AiWriteWorker:
def __init__(self, instruction, context, **kwargs):
captured["instruction"] = instruction
captured["context"] = context
captured["image_paths"] = list(kwargs.get("image_paths") or [])
self.finished = _Signal()
self.cancelled = _Signal()
self.failed = _Signal()
def cancel(self):
pass
with mock.patch(
"app.gui.tabs.product_suite.ProductSuiteAiWriteWorker",
_AiWriteWorker,
), mock.patch.object(tab, "_start_thread", return_value=object()), mock.patch.object(
tab,
"_confirm_ai_write_request",
side_effect=lambda target, asset_ids, points_cost: tab._start_confirmed_ai_write(
target,
asset_ids,
),
):
tab.start_ai_write()
self.assertIn("未绑定商品", captured["context"])
self.assertNotIn("draft_", captured["context"])
self.assertEqual(
[
image_studio.list_assets(
draft.id,
kind=image_studio.ASSET_KIND_ORIGINAL,
path=config["db_path"],
)[0].local_path
],
captured["image_paths"],
)
state.ai_worker = None
state.ai_thread = None
state.ai_started_at = None
tab._apply_running_state(state)
pull_message = mock.Mock()
with mock.patch.object(tab, "_message", pull_message):
tab.pull_main_images()
self.assertIsNone(state.pull_worker)
self.assertEqual("无法拉取蝦皮主图", pull_message.call_args.args[0])
self.assertIn("当前为临时项目", pull_message.call_args.args[1])
tab.item_id_edit.setText("51100639510")
with mock.patch.object(tab, "_confirm", return_value=True):
tab._on_item_finished()
bound = image_studio.get_project(draft.id, path=config["db_path"])
self.assertEqual("51100639510", bound.item_id)
self.assertEqual(image_studio.PROJECT_BINDING_BOUND, bound.binding_state)
self.assertTrue(tab.item_id_hint_label.isHidden())
self.assertIn("套图任务", tab.task_tabs.tabText(tab.task_tabs.currentIndex()))
draft_state = tab.add_task(inherit=False)
draft = tab._create_draft_project(draft_state)
image_studio.add_asset(
draft.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
path=config["db_path"],
)
draft_state.ai_worker = mock.Mock()
with mock.patch.object(tab, "_draft_close_action", return_value="cancel"):
tab.close_task(tab.task_tabs.currentIndex())
draft_state.ai_worker.cancel.assert_not_called()
self.assertIn(draft_state.key, tab._states)
draft_state.ai_worker = None
with mock.patch.object(tab, "_draft_close_action", return_value="keep"):
tab.close_task(tab.task_tabs.currentIndex())
self.assertIsNotNone(image_studio.get_project(draft.id, path=config["db_path"]))
tab.close()
restored = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(restored.close)
restored_state = next(
state for state in restored._states.values() if state.project_id == draft.id
)
self.assertEqual("", restored_state.item_id)
self.assertTrue(restored._is_draft_state(restored_state))
restored_index = next(
index
for index in range(restored.task_tabs.count())
if restored.task_tabs.tabData(index) == restored_state.key
)
self.assertIn("临时草稿", restored.task_tabs.tabText(restored_index))
self.assert_removed(temp_dir)
def test_first_failed_import_discards_new_empty_temporary_draft(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
bad_path = os.path.join(temp_dir, "not-an-image.png")
with open(bad_path, "wb") as handle:
handle.write(b"not an image")
state = tab._displayed_state
with mock.patch.object(tab, "_start_thread", return_value=object()):
tab._start_import(file_paths=[bad_path])
worker = state.import_worker
draft_id = state.project_id
with mock.patch.object(tab, "_message"):
tab._on_import_finished(state, worker.execute())
discarded = image_studio.get_project(
draft_id,
path=config["db_path"],
include_deleted=True,
)
self.assertIsNotNone(discarded.deleted_at)
self.assertIsNone(state.project_id)
self.assertEqual([], image_studio.list_recoverable_draft_projects(path=config["db_path"]))
self.assert_removed(temp_dir)
def test_pull_confirmation_always_shows_account_item_and_existing_count(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
tab.item_id_edit.setText("51100639510")
confirmations = []
tab._confirm = lambda title, message, **kwargs: confirmations.append(
(title, message, kwargs)
) and False
tab.pull_main_images()
self.assertIsNone(state.pull_worker)
self.assertEqual("确认拉取蝦皮主图", confirmations[0][0])
self.assertIn("主店(alias-a)", confirmations[0][1])
self.assertIn("商品ID:51100639510", confirmations[0][1])
self.assertIn("当前可用商品原图:0张", confirmations[0][1])
self.assertIn("不会修改蝦皮线上商品", confirmations[0][1])
self.assertEqual("确认拉取", confirmations[0][2]["confirm_text"])
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
source_path = os.path.join(temp_dir, "existing.png")
self._write_image(source_path)
image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
path=config["db_path"],
)
state.project_id = project.id
state.project_binding_state = project.binding_state
state.item_id = project.item_id
tab.item_id_edit.setText(project.item_id)
tab._refresh_originals(state)
tab.pull_main_images()
self.assertIsNone(state.pull_worker)
self.assertIn("当前可用商品原图:1张", confirmations[1][1])
self.assertIn("本地手动添加图片会保留", confirmations[1][1])
self.assert_removed(temp_dir)
def test_pull_button_stops_with_confirmation_and_repeated_click_is_ignored(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
tab.item_id_edit.setText("51100639510")
with mock.patch.object(tab, "_confirm", return_value=True), \
mock.patch.object(tab, "_start_thread", return_value=mock.Mock()):
tab.pull_main_images()
self.assertTrue(state.pull_running())
self.assertEqual("停止拉取蝦皮", tab.pull_button.text())
self.assertTrue(tab.pull_button.isEnabled())
worker = state.pull_worker
with mock.patch.object(tab, "_pull_stop_action", return_value="continue"):
tab.pull_main_images()
self.assertFalse(worker.is_cancelled())
self.assertFalse(state.pull_stop_requested)
with mock.patch.object(tab, "_pull_stop_action", return_value="keep"):
tab.pull_main_images()
self.assertTrue(worker.is_cancelled())
self.assertTrue(state.pull_stop_requested)
self.assertEqual("正在停止...", tab.pull_button.text())
stop_action = mock.Mock(return_value="clear_current")
with mock.patch.object(tab, "_pull_stop_action", stop_action):
tab.pull_main_images()
stop_action.assert_not_called()
state.pull_worker = None
state.pull_thread = None
tab._pull_run_states.clear()
state.pull_run_token = ""
self.assert_removed(temp_dir)
def test_stopping_pull_cancels_only_current_pull_downloads(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
pull_worker = mock.Mock()
manual_worker = mock.Mock()
state.download_queue = [11, 12]
state.downloads = {
13: (pull_worker, mock.Mock()),
14: (manual_worker, mock.Mock()),
}
state.download_tokens = {
11: "pull-token",
12: "",
13: "pull-token",
14: "",
}
state.pull_download_asset_ids = {11, 13}
tab._cancel_pull_downloads(state)
self.assertEqual([12], state.download_queue)
self.assertEqual({13}, state.pull_download_asset_ids)
self.assertNotIn(11, state.download_tokens)
pull_worker.cancel.assert_called_once()
manual_worker.cancel.assert_not_called()
self.assert_removed(temp_dir)
def test_clear_stopped_pull_removes_only_new_unreferenced_remote_assets(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
existing_remote = image_studio.sync_original_asset_urls(
project.id,
[{"index": 1, "src": "https://susercontent.com/existing.jpg"}],
path=config["db_path"],
)[0]
local_path = os.path.join(temp_dir, "local.png")
self._write_image(local_path)
local_asset = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=local_path,
source_order=2,
path=config["db_path"],
)
before_assets = image_studio.list_assets(
project.id,
kind=image_studio.ASSET_KIND_ORIGINAL,
path=config["db_path"],
)
synced = image_studio.sync_original_asset_urls(
project.id,
[
{"index": 1, "src": "https://susercontent.com/new-one.jpg"},
{"index": 2, "src": "https://susercontent.com/new-two.jpg"},
],
path=config["db_path"],
)
new_assets = [
asset
for asset in synced
if asset.remote_url
and asset.remote_url.endswith(("new-one.jpg", "new-two.jpg"))
]
referenced = next(
asset for asset in new_assets if asset.remote_url.endswith("new-two.jpg")
)
image_studio.create_job(
project.id,
source_asset_id=referenced.id,
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.pull_run_token = "clear-pull"
state.pull_stop_requested = True
state.pull_cleanup_mode = "clear_current"
state.pull_before_asset_ids = {int(asset.id) for asset in before_assets}
state.pull_before_asset_states = [
{
"id": int(asset.id),
"status": asset.status,
"source_order": int(asset.source_order),
}
for asset in before_assets
]
state.pull_asset_ids = {int(asset.id) for asset in new_assets}
state.pull_started_at = time.monotonic()
tab._pull_run_states["clear-pull"] = state.key
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
self.assertTrue(tab._finalize_pull(state, "clear-pull"))
remaining = {
asset.id: asset
for asset in image_studio.list_assets(
project.id,
kind=image_studio.ASSET_KIND_ORIGINAL,
path=config["db_path"],
)
}
removed = next(
asset for asset in new_assets if asset.id != referenced.id
)
self.assertNotIn(removed.id, remaining)
self.assertIn(referenced.id, remaining)
self.assertEqual(
image_studio.ASSET_STATUS_AVAILABLE,
remaining[existing_remote.id].status,
)
self.assertEqual(1, remaining[existing_remote.id].source_order)
self.assertIn(local_asset.id, remaining)
self.assertGreater(
remaining[referenced.id].source_order,
remaining[local_asset.id].source_order,
)
self.assertEqual("拉取蝦皮主图已停止", messages[-1][0])
self.assertIn("清理1张", messages[-1][1])
self.assertIn("因引用保留1张", messages[-1][1])
self.assert_removed(temp_dir)
def test_old_pull_result_does_not_override_current_run(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.pull_run_token = "current-pull"
state.pull_worker = mock.Mock()
tab._pull_run_states["current-pull"] = state.key
tab._on_pull_finished(
"old-pull",
{"project": None, "assets": [], "count": 0},
)
self.assertEqual("current-pull", state.pull_run_token)
self.assertIsNotNone(state.pull_worker)
state.pull_worker = None
state.pull_run_token = ""
tab._pull_run_states.clear()
self.assert_removed(temp_dir)
def test_pull_thread_finished_fallback_finalizes_requested_stop(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
accounts.create_account("主店", "alias-a", debug_port=9222, config=config)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.pull_run_token = "pull-fallback"
state.pull_stop_requested = True
state.pull_worker = mock.Mock()
state.pull_thread = mock.Mock()
state.pull_started_at = time.monotonic()
tab._pull_run_states["pull-fallback"] = state.key
messages = []
tab._message = lambda title, message, **kwargs: messages.append(
(title, message)
)
tab._handle_pull_thread_finished("pull-fallback")
self.assertEqual("", state.pull_run_token)
self.assertIsNone(state.pull_worker)
self.assertEqual("拉取蝦皮主图已停止", messages[-1][0])
self.assertEqual("拉取蝦皮主图", tab.pull_button.text())
self.assert_removed(temp_dir)
def test_pull_real_qthread_completion_restores_button_once(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
tab.item_id_edit.setText(project.item_id)
with mock.patch.object(tab, "_confirm", return_value=True), \
mock.patch(
"app.gui.workers.image_studio.pull_remote_main_image_urls",
return_value={
"project": project,
"assets": [],
"account": account,
},
):
tab.pull_main_images()
pull_thread = state.pull_thread
deadline = time.monotonic() + 3
while state.pull_running() and time.monotonic() < deadline:
QTest.qWait(20)
self.app.processEvents()
while pull_thread is not None and time.monotonic() < deadline:
try:
running = pull_thread.isRunning()
except RuntimeError:
pull_thread = None
break
if not running:
break
QTest.qWait(20)
self.app.processEvents()
self.assertFalse(state.pull_running())
self.assertIsNone(state.pull_worker)
self.assertEqual("拉取蝦皮主图", tab.pull_button.text())
if pull_thread is not None:
self.assertFalse(pull_thread.isRunning())
self.assert_removed(temp_dir)
def test_generation_confirmation_prevents_job_creation_and_retry_bypasses_it(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
source_path = os.path.join(temp_dir, "source.png")
self._write_image(source_path)
source = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = "轻便耐用,适合日常使用"
state.settings["per_image_primary"] = True
tab._load_state(state)
confirmations = []
tab._confirm = lambda title, message, **kwargs: confirmations.append(
(title, message, kwargs)
) and False
self.assertFalse(tab.start_generation(state))
self.assertEqual([], image_studio.list_jobs(project.id, path=config["db_path"]))
self.assertEqual("确认生成商品套图", confirmations[0][0])
self.assertIn("主店(alias-a)", confirmations[0][1])
self.assertIn("商品ID:51100639510", confirmations[0][1])
self.assertIn("可用商品原图:1张", confirmations[0][1])
self.assertIn("逐图主图:已开启", confirmations[0][1])
self.assertIn("白底图只使用第一张原图", confirmations[0][1])
self.assertIn("图片比例:1:1", confirmations[0][1])
self.assertIn("本次生成总数:", confirmations[0][1])
self.assertIn("暂时无法取得预计扣点", confirmations[0][1])
self.assertEqual("确认生成", confirmations[0][2]["confirm_text"])
self.assertEqual("返回修改", confirmations[0][2]["cancel_text"])
self.assertTrue(confirmations[0][2]["default_cancel"])
failed_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="失败重试",
path=config["db_path"],
)
failed_job = image_studio.update_job_status(
failed_job.id,
"failed",
path=config["db_path"],
)
confirmations.clear()
with mock.patch.object(tab, "_start_thread", return_value=mock.Mock()):
self.assertTrue(
tab.start_generation(
state,
specs=[
{
"source_asset_id": source.id,
"job_type": failed_job.job_type,
"prompt": failed_job.prompt,
}
],
retry_job_id=failed_job.id,
)
)
self.assertEqual([], confirmations)
state.worker = None
state.thread = None
state.generation_run_token = ""
tab._generation_run_states.clear()
self.assert_removed(temp_dir)
def test_generation_confirmation_uses_planned_specs_for_cached_image_price(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
config["ai"] = {
"cmhub": {
"base_url": "https://cmhub.example.com",
"image_alias": "image-standard",
}
}
appconfig.save_cmhub_config({"api_key": "test-key"}, path=config["cmhub_config_path"])
project, _assets = self._create_project_with_assets(temp_dir, config, 3)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
self.addCleanup(cmhub_models.clear_model_catalog_cache)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = "轻便耐用,适合日常使用"
state.settings["per_image_primary"] = True
tab._load_state(state)
cmhub_models.cache_model_catalog(
"https://cmhub.example.com",
[
{
"alias": "image-standard",
"operation_type": "image",
"requires_image": True,
"pricing_status": "priced",
"prices": [{"points_cost": 2}],
}
],
)
confirmations = []
tab._confirm = lambda title, message, **kwargs: confirmations.append(
(title, message, kwargs)
) and False
self.assertFalse(tab.start_generation(state))
self.assertEqual(1, len(confirmations))
message = confirmations[0][1]
self.assertIn("逐图主图:已开启", message)
self.assertIn("白底图:1张", message)
self.assertIn("场景图:6张", message)
self.assertIn("卖点图:6张", message)
self.assertIn("本次生成总数:13张", message)
self.assertIn("预计单张扣点:2 点", message)
self.assertIn("预计总扣点:26 点", message)
self.assertEqual([], image_studio.list_jobs(project.id, path=config["db_path"]))
self.assert_removed(temp_dir)
def test_generation_confirmation_explains_non_per_image_primary(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 2)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.settings["per_image_primary"] = False
specs = product_suite.build_job_specs(
assets,
"商品卖点",
state.settings,
state.item_id,
template_text=prompts.load_product_suite_prompt(tab.product_suite_prompt_path),
)
message = tab._generation_confirmation_message(
state,
assets,
specs,
)
self.assertIn("逐图主图:未开启", message)
self.assertIn("所有分类都只使用第一张可用原图生成", message)
self.assertIn("本次生成总数:5张", message)
self.assert_removed(temp_dir)
def test_generation_confirmation_rejects_plan_changed_while_reading_price(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 1)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = "原卖点"
tab._load_state(state)
template_text = prompts.load_product_suite_prompt(tab.product_suite_prompt_path)
specs = product_suite.build_job_specs(
assets,
state.prompt,
state.settings,
state.item_id,
template_text=template_text,
)
snapshot = tab._generation_plan_snapshot(state, assets, specs, template_text)
state.prompt = "已修改卖点"
with mock.patch.object(tab, "_confirm") as confirm:
started = tab._confirm_generation_price_request(
state,
assets,
specs,
"round-key",
snapshot,
None,
)
self.assertFalse(started)
confirm.assert_not_called()
self.assertIsNone(state.worker)
self.assert_removed(temp_dir)
def test_successful_history_requires_decision_before_new_generation(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
account = accounts.create_account(
"主店",
"alias-a",
debug_port=9222,
config=config,
)
project = image_studio.create_or_get_project(
account,
item_id="51100639510",
path=config["db_path"],
)
source_path = os.path.join(temp_dir, "source.png")
self._write_image(source_path)
source = image_studio.add_asset(
project.id,
image_studio.ASSET_KIND_ORIGINAL,
local_path=source_path,
path=config["db_path"],
)
history_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="已有成功历史",
generation_round_key="history-round",
generation_slot_index=0,
path=config["db_path"],
)
image_studio.update_job_status(
history_job.id,
"succeeded",
path=config["db_path"],
)
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = project.item_id
state.project_id = project.id
state.project_binding_state = project.binding_state
state.prompt = "轻便耐用,适合日常使用"
tab._load_state(state)
history_confirmations = []
cost_confirmations = []
tab._confirm_new_generation_history = (
lambda _state, summary: history_confirmations.append(summary) or "cancel"
)
tab._confirm = lambda title, message, **kwargs: cost_confirmations.append(
(title, message, kwargs)
) and False
self.assertFalse(tab.start_generation(state))
self.assertEqual(1, len(history_confirmations))
self.assertEqual(1, history_confirmations[0].successful_round_count)
self.assertEqual(1, history_confirmations[0].successful_image_count)
self.assertEqual([], cost_confirmations)
self.assertEqual(
[history_job.id],
[job.id for job in image_studio.list_jobs(project.id, path=config["db_path"])],
)
tab._confirm_new_generation_history = lambda _state, _summary: "history"
tab.open_history_dialog = mock.Mock()
self.assertFalse(tab.start_generation(state))
tab.open_history_dialog.assert_called_once_with(
current_project_only=True,
current_project_id=project.id,
)
self.assertEqual([], cost_confirmations)
tab._confirm_new_generation_history = lambda _state, _summary: "continue"
tab._confirm = lambda title, message, **kwargs: cost_confirmations.append(
(title, message, kwargs)
) or True
with mock.patch.object(tab, "_start_thread", return_value=mock.Mock()):
self.assertTrue(tab.start_generation(state))
self.assertEqual(1, len(cost_confirmations))
self.assertEqual("确认生成商品套图", cost_confirmations[0][0])
self.assertNotEqual("history-round", state.generation_round_key)
self.assertTrue(state.generation_round_key)
state.worker = None
state.thread = None
state.generation_run_token = ""
tab._generation_run_states.clear()
failed_job = image_studio.create_job(
project.id,
source_asset_id=source.id,
job_type="场景图",
prompt="失败重试",
generation_round_key="retry-round",
generation_slot_index=0,
path=config["db_path"],
)
failed_job = image_studio.update_job_status(
failed_job.id,
"failed",
path=config["db_path"],
)
history_decision = mock.Mock(return_value="cancel")
tab._confirm_new_generation_history = history_decision
tab._confirm = mock.Mock(return_value=True)
with mock.patch.object(tab, "_start_thread", return_value=mock.Mock()):
self.assertTrue(
tab.start_generation(
state,
specs=[
{
"source_asset_id": source.id,
"job_type": failed_job.job_type,
"prompt": failed_job.prompt,
}
],
retry_job_id=failed_job.id,
)
)
history_decision.assert_not_called()
tab._confirm.assert_not_called()
state.worker = None
state.thread = None
state.generation_run_token = ""
tab._generation_run_states.clear()
self.assert_removed(temp_dir)
def test_original_checkbox_click_and_keyboard_delete_keep_actions_separate(self):
original_list = ProductOriginalList()
self.addCleanup(original_list.close)
original_list.setFixedSize(240, 210)
for asset_id in (11, 12):
item = QListWidgetItem("原图%d" % asset_id)
item.setData(Qt.UserRole, asset_id)
item.setData(ORIGINAL_CHECK_STATE_ROLE, Qt.Unchecked)
pixmap = QPixmap(82, 64)
pixmap.fill(0xFF336699)
item.setIcon(QIcon(pixmap))
original_list.addItem(item)
original_list.show()
self.app.processEvents()
clicked = []
double_clicked = []
batch_deleted = []
single_deleted = []
original_list.itemClicked.connect(lambda item: clicked.append(item.data(Qt.UserRole)))
original_list.itemDoubleClicked.connect(
lambda item: double_clicked.append(item.data(Qt.UserRole))
)
original_list.deleteAssetsRequested.connect(lambda ids: batch_deleted.append(ids))
original_list.deleteRequested.connect(lambda asset_id: single_deleted.append(asset_id))
first_rect = original_list.visualItemRect(original_list.item(0))
check_point = ProductOriginalDelegate.checkbox_hit_rect(first_rect).center()
QTest.mouseClick(original_list.viewport(), Qt.LeftButton, pos=check_point)
self.assertEqual([11], original_list.checked_asset_ids())
self.assertEqual([], clicked)
QTest.mouseDClick(original_list.viewport(), Qt.LeftButton, pos=check_point)
self.assertEqual([], double_clicked)
QTest.mouseClick(original_list.viewport(), Qt.LeftButton, pos=first_rect.center())
self.assertEqual([11], clicked)
original_list.set_checked_asset_ids([11, 12])
self.assertEqual(
[("删除选中的2张图片…", [11, 12])],
original_list.context_delete_options(11),
)
original_list.set_checked_asset_ids([12])
self.assertEqual(
[("删除这张图片…", [11]), ("删除选中图片…", [12])],
original_list.context_delete_options(11),
)
original_list.setFocus()
original_list.set_checked_asset_ids([11, 12])
QTest.keyClick(original_list, Qt.Key_Delete)
self.assertEqual([[11, 12]], batch_deleted)
original_list.clear_checks()
original_list.setCurrentRow(1)
QTest.keyClick(original_list, Qt.Key_Backspace)
self.assertEqual([12], single_deleted)
def test_batch_delete_confirms_main_image_and_blocks_running_or_downloading(self):
with self.make_temp_dir() as temp_dir:
config = self._config(temp_dir)
project, assets = self._create_project_with_assets(temp_dir, config, 3)
statuses = []
tab = ProductSuiteTab(
config=config,
db_path=config["db_path"],
status_callback=lambda message, level=None: statuses.append((message, level)),
)
self.addCleanup(tab.close)
state = tab._displayed_state
state.account_alias = "alias-a"
state.item_id = "51100639510"
state.project_id = project.id
tab._load_state(state)
confirmations = []
tab._confirm = lambda title, message, **kwargs: confirmations.append(
(title, message, kwargs)
) or True
tab.original_list.set_checked_asset_ids([assets[0].id, assets[1].id])
tab.delete_originals(tab.original_list.checked_asset_ids())
self.assertEqual([assets[2].id], tab.original_list.asset_ids())
self.assertEqual([], tab.original_list.checked_asset_ids())
self.assertIn("选中的2张", confirmations[0][1])
self.assertIn("下一张图片将成为主图", confirmations[0][1])
self.assertIn("不会删除蝦皮线上图片", confirmations[0][1])
self.assertEqual(("已移除2张商品原图", "success"), statuses[-1])
messages = []
tab._message = lambda title, message, **kwargs: messages.append((title, message))
state.download_queue = [assets[2].id]
tab.delete_originals([assets[2].id])
self.assertIn("仍在后台下载", messages[-1][1])
self.assertIsNotNone(image_studio.get_asset(assets[2].id, path=config["db_path"]))
class RunningWorker:
def cancel(self):
return None
state.download_queue = []
state.worker = RunningWorker()
tab.delete_originals([assets[2].id])
self.assertEqual(("生成中不能删除当前任务的商品原图", "warning"), statuses[-1])
self.assertIsNotNone(image_studio.get_asset(assets[2].id, path=config["db_path"]))
state.worker = None
self.assert_removed(temp_dir)
if __name__ == "__main__":
unittest.main()