4422 lines
185 KiB
Python
4422 lines
185 KiB
Python
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 _configure_direct_image_gateway(self, config):
|
||
models_path = appconfig.ai_models_config_path(config)
|
||
ai_config = appconfig.ai_config(config)
|
||
ai_config["backend"] = "direct"
|
||
config["ai"] = ai_config
|
||
config["ai_models_path"] = models_path
|
||
models_config = appconfig.default_ai_models_config()
|
||
for model in models_config["models"]:
|
||
if model["category"] == "image":
|
||
model["api_key"] = "test-image-key"
|
||
appconfig.save_ai_models_config(models_config, path=models_path)
|
||
|
||
def _configure_cmhub_vision(self, config):
|
||
ai_config = appconfig.ai_config(config)
|
||
ai_config["cmhub"].update(
|
||
{
|
||
"base_url": "https://cmhub.example.com",
|
||
"vision_alias": "vision-standard",
|
||
}
|
||
)
|
||
config["ai"] = ai_config
|
||
appconfig.save_cmhub_config(
|
||
{"api_key": "test-key"},
|
||
path=config["cmhub_config_path"],
|
||
)
|
||
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}],
|
||
}
|
||
],
|
||
)
|
||
self.addCleanup(cmhub_models.clear_model_catalog_cache)
|
||
|
||
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_direct_gateway_enables_new_suite_actions_but_keeps_default_job_resume_available(self):
|
||
with self.make_temp_dir() as temp_dir:
|
||
config = self._config(temp_dir)
|
||
config["ai"] = appconfig.ai_config(config)
|
||
config["ai"]["backend"] = "direct"
|
||
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="已提交图片",
|
||
generation_source="cmhub",
|
||
provider="cmhub",
|
||
path=config["db_path"],
|
||
)
|
||
image_studio.set_job_submitted(job.id, "cmhub-task-1", path=config["db_path"])
|
||
direct_job = image_studio.create_job(
|
||
project.id,
|
||
source_asset_id=assets[0].id,
|
||
job_type="场景图",
|
||
prompt="直连失败图片",
|
||
generation_source=image_studio.GENERATION_SOURCE_DIRECT,
|
||
provider=image_studio.PROVIDER_OPENAI_IMAGES_EDITS,
|
||
path=config["db_path"],
|
||
)
|
||
image_studio.update_job_status(
|
||
direct_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
|
||
tab._load_state(state)
|
||
tab.refresh_gateway_state()
|
||
|
||
self.assertTrue(tab.generate_button.isEnabled())
|
||
self.assertFalse(tab.ai_write_button.isEnabled())
|
||
self.assertIn("默认网关", tab.ai_write_button.toolTip())
|
||
self.assertFalse(tab.resume_submitted_button.isHidden())
|
||
self.assertTrue(tab.resume_submitted_button.isEnabled())
|
||
self.assertEqual(
|
||
[job.id],
|
||
[resumable.id for resumable in tab._resumable_default_gateway_jobs(state)],
|
||
)
|
||
self.assertIn("不计点数", tab.generate_button.toolTip())
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_direct_generation_gateway_keeps_ai_write_available_with_cmhub_vision(self):
|
||
with self.make_temp_dir() as temp_dir:
|
||
config = self._config(temp_dir)
|
||
config["ai"] = appconfig.ai_config(config)
|
||
config["ai"]["backend"] = "direct"
|
||
config["ai"]["cmhub"].update(
|
||
{
|
||
"base_url": "https://cmhub.example.com",
|
||
"vision_alias": "vision-standard",
|
||
}
|
||
)
|
||
appconfig.save_cmhub_config(
|
||
{"api_key": "test-key"},
|
||
path=config["cmhub_config_path"],
|
||
)
|
||
self._configure_direct_image_gateway(config)
|
||
project, assets = self._create_project_with_assets(temp_dir, config, 1)
|
||
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}],
|
||
}
|
||
],
|
||
)
|
||
|
||
self.assertTrue(tab.generate_button.isEnabled())
|
||
self.assertTrue(tab.ai_write_button.isEnabled())
|
||
self.assertEqual("", tab.ai_write_button.toolTip())
|
||
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((assets[0].id,), asset_ids)
|
||
self.assertEqual("2", cmhub_models.format_points_cost(points_cost))
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_direct_generation_and_ai_write_configuration_errors_are_independent(self):
|
||
with self.make_temp_dir() as temp_dir:
|
||
config = self._config(temp_dir)
|
||
config["ai"] = appconfig.ai_config(config)
|
||
config["ai"]["backend"] = "direct"
|
||
config["ai"]["cmhub"].update(
|
||
{
|
||
"base_url": "https://cmhub.example.com",
|
||
"vision_alias": "vision-standard",
|
||
}
|
||
)
|
||
self._configure_direct_image_gateway(config)
|
||
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)
|
||
|
||
cases = (
|
||
("vision_alias", "", "尚未选择图片理解模型", "test-key"),
|
||
("vision_alias", "vision-standard", "尚未配置默认网关 API Key", ""),
|
||
)
|
||
for key, value, expected_error, api_key in cases:
|
||
with self.subTest(key=key, expected_error=expected_error):
|
||
tab.config["ai"]["cmhub"].update(
|
||
{
|
||
"base_url": "https://cmhub.example.com",
|
||
"vision_alias": "vision-standard",
|
||
key: value,
|
||
}
|
||
)
|
||
appconfig.save_cmhub_config(
|
||
{"api_key": api_key},
|
||
path=tab.cmhub_config_path,
|
||
)
|
||
self.assertEqual(expected_error, tab._ai_write_config_error())
|
||
tab._apply_running_state(state)
|
||
self.assertTrue(tab.generate_button.isEnabled())
|
||
self.assertFalse(tab.ai_write_button.isEnabled())
|
||
self.assertIn(expected_error, tab.ai_write_button.toolTip())
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_direct_gateway_preflight_and_confirmation_do_not_call_cmhub_catalog(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)
|
||
self._configure_direct_image_gateway(config)
|
||
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"] = False
|
||
state.settings["ratio"] = "3:4"
|
||
tab._load_state(state)
|
||
confirmations = []
|
||
tab._confirm = lambda title, message, **kwargs: confirmations.append(
|
||
(title, message, kwargs)
|
||
) and False
|
||
|
||
with mock.patch.object(
|
||
tab,
|
||
"_cmhub_catalog_params",
|
||
side_effect=AssertionError("自定义网关不应读取 cmhub 目录"),
|
||
):
|
||
self.assertFalse(tab.start_generation(state))
|
||
|
||
self.assertEqual(1, len(confirmations))
|
||
message = confirmations[0][1]
|
||
self.assertIn("生成来源：自定义网关", message)
|
||
self.assertIn("自定义网关不计点数", message)
|
||
self.assertIn("主体一致性可能弱于默认网关", message)
|
||
self.assertIn("输出尺寸：1024x1536（接近比例生成）", message)
|
||
self.assertNotIn("cmhub 点数", message)
|
||
self.assertEqual([], image_studio.list_jobs(project.id, path=config["db_path"]))
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_direct_gateway_invalid_image_model_blocks_before_confirmation(self):
|
||
with self.make_temp_dir() as temp_dir:
|
||
config = self._config(temp_dir)
|
||
config["ai"] = appconfig.ai_config(config)
|
||
config["ai"]["backend"] = "direct"
|
||
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))
|
||
|
||
self.assertFalse(tab.start_generation(state))
|
||
self.assertEqual("自定义网关配置不完整", messages[0][0])
|
||
self.assertIn("请到⑤设置", messages[0][1])
|
||
self.assertEqual([], image_studio.list_jobs(project.id, path=config["db_path"]))
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_direct_gateway_stop_waits_for_current_image_result(self):
|
||
with self.make_temp_dir() as temp_dir:
|
||
config = self._config(temp_dir)
|
||
config["ai"] = appconfig.ai_config(config)
|
||
config["ai"]["backend"] = "direct"
|
||
tab = ProductSuiteTab(config=config, db_path=config["db_path"])
|
||
self.addCleanup(tab.close)
|
||
state = tab._displayed_state
|
||
state.worker = mock.Mock()
|
||
state.generation_source = image_studio.GENERATION_SOURCE_DIRECT
|
||
statuses = []
|
||
tab._confirm = lambda *args, **kwargs: True
|
||
tab._status = lambda message, level=None: statuses.append((message, level))
|
||
|
||
tab.toggle_generation()
|
||
|
||
state.worker.cancel.assert_called_once()
|
||
self.assertTrue(state.generation_stop_requested)
|
||
self.assertEqual("正在停止...", tab.generate_button.text())
|
||
self.assertEqual(("正在停止，等待当前图片返回后结束", "warning"), statuses[-1])
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_direct_failed_job_retry_requires_repeat_billing_confirmation(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)
|
||
self._configure_direct_image_gateway(config)
|
||
failed_job = image_studio.create_job(
|
||
project.id,
|
||
source_asset_id=assets[0].id,
|
||
job_type="白底图",
|
||
prompt="重新生成测试",
|
||
generation_source=image_studio.GENERATION_SOURCE_DIRECT,
|
||
provider=image_studio.PROVIDER_OPENAI_IMAGES_EDITS,
|
||
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
|
||
tab._load_state(state)
|
||
|
||
confirmations = []
|
||
tab._confirm = lambda title, message, **kwargs: confirmations.append(
|
||
(title, message, kwargs)
|
||
) and False
|
||
with mock.patch.object(tab, "_start_generation_worker") as start_worker:
|
||
tab.retry_job(failed_job)
|
||
self.assertEqual(1, len(confirmations))
|
||
self.assertEqual("确认重新生成商品套图", confirmations[0][0])
|
||
self.assertIn("可能对上次未确认请求已计费", confirmations[0][1])
|
||
self.assertIn("本次重新生成可能再次收费", confirmations[0][1])
|
||
self.assertIn("生成来源：自定义网关", confirmations[0][1])
|
||
self.assertTrue(confirmations[0][2]["default_cancel"])
|
||
start_worker.assert_not_called()
|
||
|
||
tab._confirm = lambda *args, **kwargs: True
|
||
with mock.patch.object(tab, "_start_generation_worker", return_value=True) as start_worker:
|
||
tab.retry_job(failed_job)
|
||
self.assertTrue(start_worker.called)
|
||
self.assertTrue(start_worker.call_args.kwargs["retrying"])
|
||
self.assertEqual(failed_job.id, start_worker.call_args.kwargs["retry_job_id"])
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_generation_source_labels_are_persisted_for_results_and_history(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)
|
||
cmhub_job = image_studio.create_job(
|
||
project.id,
|
||
source_asset_id=assets[0].id,
|
||
job_type="白底图",
|
||
prompt="默认网关图片",
|
||
generation_source=image_studio.GENERATION_SOURCE_CMHUB,
|
||
provider=image_studio.PROVIDER_CMHUB,
|
||
path=config["db_path"],
|
||
)
|
||
cmhub_job = image_studio.update_job_status(
|
||
cmhub_job.id,
|
||
"failed",
|
||
error="默认网关失败",
|
||
path=config["db_path"],
|
||
)
|
||
direct_job = image_studio.create_job(
|
||
project.id,
|
||
source_asset_id=assets[0].id,
|
||
job_type="场景图",
|
||
prompt="自定义网关图片",
|
||
generation_source=image_studio.GENERATION_SOURCE_DIRECT,
|
||
provider=image_studio.PROVIDER_OPENAI_IMAGES_EDITS,
|
||
path=config["db_path"],
|
||
)
|
||
direct_job = image_studio.update_job_status(
|
||
direct_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 = [cmhub_job.id, direct_job.id]
|
||
tab._load_state(state)
|
||
tab._refresh_results(state)
|
||
|
||
result_sources = {
|
||
card.job.id: card.findChild(QLabel, "suiteResultCardSource").text()
|
||
for card in tab.findChildren(SuiteResultCard)
|
||
}
|
||
self.assertEqual("默认网关", result_sources[cmhub_job.id])
|
||
self.assertEqual("自定义网关", result_sources[direct_job.id])
|
||
|
||
self._configure_direct_image_gateway(config)
|
||
tab.refresh_gateway_state()
|
||
result_sources_after_switch = {
|
||
card.job.id: card.findChild(QLabel, "suiteResultCardSource").text()
|
||
for card in tab.findChildren(SuiteResultCard)
|
||
}
|
||
self.assertEqual(result_sources, result_sources_after_switch)
|
||
|
||
history = ProductSuiteHistoryDialog(project.id, db_path=config["db_path"], parent=tab)
|
||
self.addCleanup(history.close)
|
||
self.assertTrue(
|
||
any(
|
||
label.text() == "来源：默认网关、自定义网关"
|
||
for label in history.findChildren(QLabel, "suiteHistoryRoundSource")
|
||
)
|
||
)
|
||
history_cards = {
|
||
card.job.id: card.toolTip()
|
||
for card in history.findChildren(SuiteHistoryImageCard)
|
||
}
|
||
self.assertIn("生成来源：默认网关", history_cards[cmhub_job.id])
|
||
self.assertIn("生成来源：自定义网关", history_cards[direct_job.id])
|
||
|
||
global_history = ProductSuiteGlobalHistoryDialog(db_path=config["db_path"], parent=tab)
|
||
self.addCleanup(global_history.close)
|
||
self.assertTrue(
|
||
any(
|
||
label.text() == "来源：默认网关、自定义网关"
|
||
for label in global_history.findChildren(QLabel, "suiteGlobalHistorySource")
|
||
)
|
||
)
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
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(
|
||
"参考图规则:使用第一張上傳圖作為主商品圖，其餘圖片只作為參考。",
|
||
expected,
|
||
)
|
||
self.assertNotIn("第2至", 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)
|
||
self._configure_cmhub_vision(config)
|
||
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 = []
|
||
reordered = []
|
||
original_list.filesDropped.connect(dropped.append)
|
||
original_list.clipboardImage.connect(pasted.append)
|
||
original_list.orderChanged.connect(reordered.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])),
|
||
)
|
||
self.assertEqual([], reordered)
|
||
|
||
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()):
|
||
item = tab.original_list.item(row)
|
||
self.assertIsNotNone(item.data(ORIGINAL_CHECK_STATE_ROLE))
|
||
self.assertTrue(item.flags() & Qt.ItemIsDragEnabled)
|
||
self.assertFalse(item.flags() & Qt.ItemIsDropEnabled)
|
||
self.assertEqual(Qt.MoveAction, tab.original_list.defaultDropAction())
|
||
|
||
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())
|
||
self.assertEqual("主图", tab.original_list.item(0).text())
|
||
self.assertTrue(tab.original_list.item(0).font().bold())
|
||
self.assertIn("拖动到第一位即可设为主图", tab.original_list.item(0).toolTip())
|
||
statuses = []
|
||
tab._status = lambda message, level=None: statuses.append((message, level))
|
||
tab.reorder_originals([assets[1].id, assets[0].id])
|
||
self.assertEqual([assets[1].id], tab.original_list.checked_asset_ids())
|
||
self.assertEqual([assets[1].id, assets[0].id], tab.original_list.asset_ids())
|
||
self.assertEqual(("已更新商品原图顺序，第一张图片为主图", "success"), statuses[-1])
|
||
|
||
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_original_list_moves_dragged_item_without_duplicate_asset_ids(self):
|
||
original_list = ProductOriginalList()
|
||
self.addCleanup(original_list.close)
|
||
original_list.setFixedSize(360, 120)
|
||
for asset_id in (11, 12, 13):
|
||
item = QListWidgetItem("原图%d" % asset_id)
|
||
item.setData(Qt.UserRole, asset_id)
|
||
original_list.addItem(item)
|
||
original_list.show()
|
||
self.app.processEvents()
|
||
changes = []
|
||
original_list.orderChanged.connect(changes.append)
|
||
|
||
self.assertTrue(original_list._begin_internal_drag(original_list.item(1)))
|
||
first_rect = original_list.visualItemRect(original_list.item(0))
|
||
self.assertTrue(original_list._move_dragged_item_to(first_rect.center()))
|
||
self.assertEqual([12, 11, 13], original_list.asset_ids())
|
||
self.assertEqual([], changes)
|
||
self.assertTrue(original_list._finish_internal_drag())
|
||
self.assertEqual([[12, 11, 13]], changes)
|
||
self.assertEqual(3, len(set(original_list.asset_ids())))
|
||
|
||
self.assertTrue(original_list._begin_internal_drag(original_list.item(0)))
|
||
append_position = original_list.viewport().rect().bottomRight() - QPoint(2, 2)
|
||
self.assertTrue(original_list._move_dragged_item_to(append_position))
|
||
self.assertEqual([11, 13, 12], original_list.asset_ids())
|
||
self.assertEqual([[12, 11, 13]], changes)
|
||
self.assertTrue(original_list._finish_internal_drag())
|
||
self.assertEqual([[12, 11, 13], [11, 13, 12]], changes)
|
||
|
||
self.assertTrue(original_list._begin_internal_drag(original_list.item(1)))
|
||
same_rect = original_list.visualItemRect(original_list.item(1))
|
||
self.assertFalse(original_list._move_dragged_item_to(same_rect.center()))
|
||
self.assertFalse(original_list._finish_internal_drag())
|
||
self.assertEqual([11, 13, 12], original_list.asset_ids())
|
||
self.assertEqual([[12, 11, 13], [11, 13, 12]], changes)
|
||
|
||
self.assertFalse(original_list._move_dragged_item_to(first_rect.center()))
|
||
self.assertEqual([11, 13, 12], original_list.asset_ids())
|
||
|
||
duplicate = QListWidgetItem("重复原图")
|
||
duplicate.setData(Qt.UserRole, 12)
|
||
original_list.addItem(duplicate)
|
||
self.assertFalse(original_list._begin_internal_drag(original_list.item(0)))
|
||
self.assertFalse(original_list._move_dragged_item_to(first_rect.center()))
|
||
self.assertEqual(4, original_list.count())
|
||
self.assertEqual([[12, 11, 13], [11, 13, 12]], changes)
|
||
|
||
original_list.takeItem(3)
|
||
original_list.set_reordering_enabled(False)
|
||
self.assertFalse(original_list._begin_internal_drag(original_list.item(0)))
|
||
|
||
def test_original_list_start_drag_owns_move_without_qt_source_cleanup(self):
|
||
original_list = ProductOriginalList()
|
||
self.addCleanup(original_list.close)
|
||
original_list.setFixedSize(360, 120)
|
||
for asset_id in (11, 12, 13):
|
||
item = QListWidgetItem("原图%d" % asset_id)
|
||
item.setData(Qt.UserRole, asset_id)
|
||
item.setIcon(QIcon(QPixmap(24, 18)))
|
||
original_list.addItem(item)
|
||
original_list.show()
|
||
self.app.processEvents()
|
||
original_list.setCurrentItem(original_list.item(1))
|
||
changes = []
|
||
restores = []
|
||
original_list.orderChanged.connect(changes.append)
|
||
original_list.orderRestoreRequested.connect(lambda: restores.append(True))
|
||
|
||
with mock.patch("app.gui.tabs.product_suite.QDrag") as drag_type:
|
||
drag = drag_type.return_value
|
||
|
||
def execute_drag(*args):
|
||
first_rect = original_list.visualItemRect(original_list.item(0))
|
||
self.assertTrue(original_list._move_dragged_item_to(first_rect.center()))
|
||
self.assertEqual([], changes)
|
||
return Qt.MoveAction
|
||
|
||
drag.exec.side_effect = execute_drag
|
||
original_list.startDrag(Qt.MoveAction)
|
||
|
||
drag_type.assert_called_once_with(original_list)
|
||
mime_data = drag.setMimeData.call_args.args[0]
|
||
self.assertTrue(mime_data.hasFormat(original_list.INTERNAL_MIME_TYPE))
|
||
self.assertEqual([12, 11, 13], original_list.asset_ids())
|
||
self.assertEqual(3, original_list.count())
|
||
self.assertEqual([[12, 11, 13]], changes)
|
||
self.assertEqual([], restores)
|
||
|
||
def test_original_list_start_drag_requests_restore_after_unexpected_item_loss(self):
|
||
original_list = ProductOriginalList()
|
||
self.addCleanup(original_list.close)
|
||
original_list.setFixedSize(360, 120)
|
||
for asset_id in (11, 12, 13):
|
||
item = QListWidgetItem("原图%d" % asset_id)
|
||
item.setData(Qt.UserRole, asset_id)
|
||
original_list.addItem(item)
|
||
original_list.show()
|
||
self.app.processEvents()
|
||
original_list.setCurrentItem(original_list.item(1))
|
||
changes = []
|
||
restores = []
|
||
original_list.orderChanged.connect(changes.append)
|
||
original_list.orderRestoreRequested.connect(lambda: restores.append(True))
|
||
|
||
with mock.patch("app.gui.tabs.product_suite.QDrag") as drag_type:
|
||
drag = drag_type.return_value
|
||
|
||
def execute_drag(*args):
|
||
first_rect = original_list.visualItemRect(original_list.item(0))
|
||
self.assertTrue(original_list._move_dragged_item_to(first_rect.center()))
|
||
original_list.takeItem(original_list.count() - 1)
|
||
return Qt.MoveAction
|
||
|
||
drag.exec.side_effect = execute_drag
|
||
original_list.startDrag(Qt.MoveAction)
|
||
|
||
self.assertEqual([], changes)
|
||
self.assertEqual([True], restores)
|
||
|
||
def test_original_list_repeated_primary_moves_keep_every_asset_and_label(self):
|
||
with self.make_temp_dir() as temp_dir:
|
||
config = self._config(temp_dir)
|
||
project, assets = self._create_project_with_assets(temp_dir, config, 5)
|
||
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)
|
||
tab.resize(1180, 760)
|
||
tab.show()
|
||
self.app.processEvents()
|
||
|
||
original_ids = {asset.id for asset in assets}
|
||
expected_labels = ["主图", "参考1", "参考2", "参考3", "参考4"]
|
||
for asset in assets[1:]:
|
||
row = tab.original_list.asset_ids().index(asset.id)
|
||
tab.original_list.setCurrentItem(tab.original_list.item(row))
|
||
with mock.patch("app.gui.tabs.product_suite.QDrag") as drag_type:
|
||
drag = drag_type.return_value
|
||
|
||
def execute_drag(*args):
|
||
first_rect = tab.original_list.visualItemRect(
|
||
tab.original_list.item(0)
|
||
)
|
||
self.assertTrue(
|
||
tab.original_list._move_dragged_item_to(first_rect.center())
|
||
)
|
||
database_ids_before_finish = [
|
||
stored.id
|
||
for stored in image_studio.list_assets(
|
||
project.id,
|
||
kind=image_studio.ASSET_KIND_ORIGINAL,
|
||
path=config["db_path"],
|
||
)
|
||
]
|
||
self.assertNotEqual(
|
||
tab.original_list.asset_ids(),
|
||
database_ids_before_finish,
|
||
)
|
||
return Qt.MoveAction
|
||
|
||
drag.exec.side_effect = execute_drag
|
||
tab.original_list.startDrag(Qt.MoveAction)
|
||
visible_ids = tab.original_list.asset_ids()
|
||
database_ids = [
|
||
stored.id
|
||
for stored in image_studio.list_assets(
|
||
project.id,
|
||
kind=image_studio.ASSET_KIND_ORIGINAL,
|
||
path=config["db_path"],
|
||
)
|
||
]
|
||
self.assertEqual(5, tab.original_list.count())
|
||
self.assertEqual(original_ids, set(visible_ids))
|
||
self.assertEqual(5, len(set(visible_ids)))
|
||
self.assertEqual(asset.id, visible_ids[0])
|
||
self.assertEqual(visible_ids, database_ids)
|
||
self.assertEqual(
|
||
expected_labels,
|
||
[tab.original_list.item(row).text() for row in range(5)],
|
||
)
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_original_list_restores_database_order_after_drag_item_loss(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
|
||
tab._load_state(state)
|
||
tab.resize(1180, 760)
|
||
tab.show()
|
||
self.app.processEvents()
|
||
statuses = []
|
||
tab._status = lambda message, level=None: statuses.append((message, level))
|
||
tab.original_list.setCurrentItem(tab.original_list.item(1))
|
||
|
||
with mock.patch("app.gui.tabs.product_suite.QDrag") as drag_type:
|
||
drag = drag_type.return_value
|
||
|
||
def execute_drag(*args):
|
||
first_rect = tab.original_list.visualItemRect(
|
||
tab.original_list.item(0)
|
||
)
|
||
self.assertTrue(
|
||
tab.original_list._move_dragged_item_to(first_rect.center())
|
||
)
|
||
tab.original_list.takeItem(tab.original_list.count() - 1)
|
||
return Qt.MoveAction
|
||
|
||
drag.exec.side_effect = execute_drag
|
||
tab.original_list.startDrag(Qt.MoveAction)
|
||
|
||
expected_ids = [asset.id for asset in assets]
|
||
self.assertEqual(expected_ids, tab.original_list.asset_ids())
|
||
self.assertEqual(3, tab.original_list.count())
|
||
self.assertEqual(
|
||
("原图排序未完成，已恢复原顺序，请重试", "warning"),
|
||
statuses[-1],
|
||
)
|
||
|
||
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)
|
||
self._configure_cmhub_vision(config)
|
||
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_item_id_commit_normalizes_and_blocks_generation_before_state_changes(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 = project.account_alias
|
||
state.item_id = project.item_id
|
||
state.project_id = project.id
|
||
state.project_binding_state = project.binding_state
|
||
tab._load_state(state)
|
||
|
||
tab.item_id_edit.setText(project.item_id + chr(0x200B))
|
||
with mock.patch.object(tab, "_ensure_generation_gateway", return_value=False) as gateway:
|
||
self.assertFalse(tab.start_generation(state))
|
||
gateway.assert_called_once()
|
||
self.assertEqual(project.item_id, tab.item_id_edit.text())
|
||
self.assertEqual(project.item_id, state.item_id)
|
||
self.assertEqual(project.id, state.project_id)
|
||
|
||
tab.item_id_edit.setText(project.item_id + "-1")
|
||
with mock.patch.object(tab, "_ensure_generation_gateway", return_value=False) as gateway:
|
||
self.assertFalse(tab.start_generation(state))
|
||
gateway.assert_not_called()
|
||
self.assertEqual(project.item_id, state.item_id)
|
||
self.assertEqual(project.id, state.project_id)
|
||
self.assertEqual(project.item_id + "-1", tab.item_id_edit.text())
|
||
|
||
tab.item_id_edit.setText("51100639511")
|
||
with mock.patch.object(tab, "_confirm", return_value=False):
|
||
self.assertFalse(tab._commit_item_id(state))
|
||
self.assertEqual(project.item_id, state.item_id)
|
||
self.assertEqual(project.id, state.project_id)
|
||
self.assertEqual(project.item_id, tab.item_id_edit.text())
|
||
|
||
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())
|
||
self.assertTrue(tab.original_list.isEnabled())
|
||
self.assertFalse(tab.original_list.reordering_enabled())
|
||
self.assertFalse(tab.original_list.dragEnabled())
|
||
self.assertIn("完成后可调整图片顺序", tab.original_list.toolTip())
|
||
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_pull_locks_original_reordering_without_changing_saved_order(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
|
||
tab._load_state(state)
|
||
statuses = []
|
||
tab._status = lambda message, level=None: statuses.append((message, level))
|
||
|
||
state.pull_run_token = "pull-in-progress"
|
||
tab._apply_running_state(state)
|
||
self.assertTrue(tab.original_list.isEnabled())
|
||
self.assertFalse(tab.original_list.reordering_enabled())
|
||
tab.reorder_originals([assets[1].id, assets[0].id])
|
||
self.assertEqual([assets[0].id, assets[1].id], tab.original_list.asset_ids())
|
||
self.assertEqual(
|
||
("正在拉取蝦皮主图，完成后可调整图片顺序", "warning"),
|
||
statuses[-1],
|
||
)
|
||
|
||
state.pull_run_token = ""
|
||
tab._apply_running_state(state)
|
||
self.assertTrue(tab.original_list.reordering_enabled())
|
||
tab.reorder_originals([assets[1].id, assets[0].id])
|
||
self.assertEqual([assets[1].id, assets[0].id], tab.original_list.asset_ids())
|
||
|
||
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_accepts_business_prompt_with_braces(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)
|
||
captured = []
|
||
tab._start_generation_price_confirmation = (
|
||
lambda _state, _assets, specs, *_args: captured.extend(specs) or True
|
||
)
|
||
|
||
self.assertTrue(tab.start_generation(state))
|
||
|
||
self.assertTrue(captured)
|
||
self.assertIn("适合 {儿童} 使用", captured[0]["prompt"])
|
||
self.assertEqual([], image_studio.list_jobs(project.id, path=config["db_path"]))
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_generation_renders_error_as_chinese_feedback_without_side_effects(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 = []
|
||
statuses = []
|
||
tab._message = lambda title, message, **_kwargs: messages.append((title, message))
|
||
tab._status = lambda message, level=None: statuses.append((message, level))
|
||
|
||
with mock.patch(
|
||
"app.gui.tabs.product_suite.product_suite.build_job_specs",
|
||
side_effect=product_suite.ProductSuitePromptError("内部渲染失败"),
|
||
), mock.patch.object(tab, "_start_generation_price_confirmation") as confirmation:
|
||
self.assertFalse(tab.start_generation(state))
|
||
|
||
self.assertEqual(
|
||
[("无法生成商品套图", "提示词渲染失败，请检查提示词设置中的变量格式。")],
|
||
messages,
|
||
)
|
||
self.assertEqual(
|
||
[("提示词渲染失败，请检查提示词设置中的变量格式。", "danger")],
|
||
statuses,
|
||
)
|
||
confirmation.assert_not_called()
|
||
self.assertIsNone(state.worker)
|
||
self.assertEqual([], image_studio.list_jobs(project.id, path=config["db_path"]))
|
||
|
||
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("第1张可用原图作为主图，另有1张参考图会随每次请求一同提交", message)
|
||
self.assertIn("本次生成总数：5张", message)
|
||
|
||
self.assert_removed(temp_dir)
|
||
|
||
def test_per_image_helper_and_confirmation_explain_reference_limit(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)
|
||
tab.resize(1180, 760)
|
||
tab.show()
|
||
self.app.processEvents()
|
||
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
|
||
tab._load_state(state)
|
||
self.app.processEvents()
|
||
specs = product_suite.build_job_specs(
|
||
assets,
|
||
"商品卖点",
|
||
state.settings,
|
||
state.item_id,
|
||
template_text=prompts.load_product_suite_prompt(tab.product_suite_prompt_path),
|
||
)
|
||
|
||
self.assertTrue(tab.per_image_helper_label.wordWrap())
|
||
self.assertGreater(tab.per_image_helper_label.width(), 0)
|
||
self.assertIn("多SKU图请勾选", tab.per_image_helper_label.text())
|
||
self.assertGreater(
|
||
tab.per_image_helper_label.geometry().left(),
|
||
tab.per_image_checkbox.geometry().left(),
|
||
)
|
||
|
||
message = tab._generation_confirmation_message(state, assets, specs)
|
||
self.assertIn("另有7张参考图会随每次请求一同提交", message)
|
||
self.assertIn("参考图仅取前7张，其余1张不参与本轮生成", 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()
|