T-568 支持封面画廊单任务重生

tests/test_gui.py

@@ -43,6 +43,45 @@ from app.gui.tabs.generate import CoverGalleryDialog, OriginalImageDialog
 from app.gui.main_window import _fit_and_center_window
 
 
+class DummySignal:
+    def __init__(self):
+        self.callbacks = []
+
+    def connect(self, callback):
+        self.callbacks.append(callback)
+
+    def emit(self, *args):
+        for callback in list(self.callbacks):
+            callback(*args)
+
+
+class FakeThread:
+    def __init__(self):
+        self.finished = DummySignal()
+        self.started = False
+
+    def start(self):
+        self.started = True
+
+
+class FakeGenerateWorker:
+    instances = []
+
+    def __init__(self, tasks, prompt_values, db_path=None, config=None, diagnostic_log_dir=None):
+        self.tasks = list(tasks)
+        self.prompt_values = dict(prompt_values or {})
+        self.db_path = db_path
+        self.config = config
+        self.diagnostic_log_dir = diagnostic_log_dir
+        self.progress = DummySignal()
+        self.row_updated = DummySignal()
+        self.log = DummySignal()
+        self.failed = DummySignal()
+        self.finished = DummySignal()
+        self.cancelled = DummySignal()
+        FakeGenerateWorker.instances.append(self)
+
+
 class GuiTests(TempDirMixin, unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -89,6 +128,34 @@ class GuiTests(TempDirMixin, unittest.TestCase):
         self.assertTrue(image.save(path))
         return path
 
+    def _cover_gallery_task(self, temp_dir, generate_cover=True):
+        cfg = self.make_config(temp_dir)
+        cfg.setdefault("ai", {})["generate_cover"] = bool(generate_cover)
+        account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
+        db.init_db(cfg["db_path"])
+        batch_id = db.create_batch(["input.xlsx"], path=cfg["db_path"])
+        db.insert_tasks(
+            batch_id,
+            [
+                {
+                    "source_file_abs": os.path.join(temp_dir, "input.xlsx"),
+                    "source_sheet": "商品",
+                    "source_row": 2,
+                    "account_name": "Excel主店",
+                    "alias": "alias-a",
+                    "item_id": "51100639510",
+                }
+            ],
+            path=cfg["db_path"],
+        )
+        task = db.list_tasks(batch_id=batch_id, path=cfg["db_path"])[0]
+        old_cover = self.write_test_image(os.path.join(temp_dir, "old.jpg"))
+        canonical = image_paths.task_image_path(cfg["image_dir"], task, account, "new")
+        self.write_test_image(canonical)
+        db.set_collected(task.id, "旧标题", old_cover, path=cfg["db_path"])
+        db.set_generated(task.id, "新标题", canonical, path=cfg["db_path"])
+        return cfg, account, db.get_task(task.id, path=cfg["db_path"]), canonical
+
     def assert_foreground(self, model, row, column, color):
         value = model.data(model.index(row, column), gui.Qt.ForegroundRole)
         self.assertIsNotNone(value)
@@ -141,6 +208,13 @@ class GuiTests(TempDirMixin, unittest.TestCase):
             def clickedButton(self):
                 return self.buttons[selected_label]
 
+            @staticmethod
+            def warning(parent, title, text):
+                box = FakeMessageBox(parent)
+                box.setWindowTitle(title)
+                box.setText(text)
+                return 0
+
         return FakeMessageBox, boxes
 
     def test_collect_table_shows_product_unavailable_only_for_explicit_error(self):
@@ -2269,6 +2343,203 @@ class GuiTests(TempDirMixin, unittest.TestCase):
 
         self.assert_removed(temp_dir)
 
+    def test_cover_gallery_regenerate_cancel_has_no_side_effect(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg, account, task, canonical = self._cover_gallery_task(temp_dir)
+            dialog = CoverGalleryDialog(
+                task,
+                cfg["image_dir"],
+                cfg["db_path"],
+                account=account,
+                config=cfg,
+                prompt_values={"title": "标题", "cover": "封面"},
+            )
+            self.addCleanup(dialog.close)
+            message_box, _ = self.make_fake_message_box("取消")
+            FakeGenerateWorker.instances.clear()
+
+            with mock.patch("app.gui.tabs.generate.QMessageBox", message_box), \
+                    mock.patch("app.gui.tabs.generate.GenerateWorker", FakeGenerateWorker):
+                self.assertFalse(dialog.regenerate_cover())
+
+            self.assertEqual(canonical, db.get_task(task.id, path=cfg["db_path"]).new_cover_path)
+            self.assertTrue(os.path.exists(canonical))
+            self.assertEqual([], FakeGenerateWorker.instances)
+
+        self.assert_removed(temp_dir)
+
+    def test_cover_gallery_regenerate_uses_fresh_task_and_forces_cover_generation(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg, account, task, canonical = self._cover_gallery_task(temp_dir, generate_cover=False)
+            fake_thread = FakeThread()
+            running_states = []
+            dialog = CoverGalleryDialog(
+                task,
+                cfg["image_dir"],
+                cfg["db_path"],
+                account=account,
+                config=cfg,
+                prompt_values={"title": "标题提示", "cover": "封面提示 {新标题}"},
+                regenerate_running_callback=running_states.append,
+            )
+            self.addCleanup(dialog.close)
+            message_box, _ = self.make_fake_message_box("重新生成")
+            FakeGenerateWorker.instances.clear()
+
+            with mock.patch("app.gui.tabs.generate.QMessageBox", message_box), \
+                    mock.patch("app.gui.tabs.generate.GenerateWorker", FakeGenerateWorker), \
+                    mock.patch("app.gui.tabs.generate.run_worker", return_value=fake_thread):
+                self.assertTrue(dialog.regenerate_cover())
+
+            self.assertTrue(fake_thread.started)
+            self.assertEqual([True], running_states)
+            worker = FakeGenerateWorker.instances[0]
+            self.assertIsNone(worker.tasks[0].new_cover_path)
+            self.assertEqual("新标题", worker.tasks[0].new_title)
+            self.assertTrue(worker.config["ai"]["generate_cover"])
+            self.assertFalse(appconfig.ai_config(cfg).get("generate_cover", False))
+            self.assertFalse(os.path.exists(canonical))
+            self.assertTrue(any(path.endswith(".jpg") for path in dialog.candidates))
+            self.assertIn("开始重新生成商品", dialog.status_label.text())
+
+        self.assert_removed(temp_dir)
+
+    def test_cover_gallery_regenerate_success_refreshes_candidates_and_current_selection(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg, account, task, canonical = self._cover_gallery_task(temp_dir)
+            fake_thread = FakeThread()
+            refreshed = []
+            dialog = CoverGalleryDialog(
+                task,
+                cfg["image_dir"],
+                cfg["db_path"],
+                account=account,
+                config=cfg,
+                prompt_values={"title": "标题", "cover": "封面"},
+                regenerate_running_callback=lambda running: None,
+                gallery_updated_callback=lambda task: refreshed.append(task.id),
+            )
+            self.addCleanup(dialog.close)
+            message_box, _ = self.make_fake_message_box("重新生成")
+            FakeGenerateWorker.instances.clear()
+
+            with mock.patch("app.gui.tabs.generate.QMessageBox", message_box), \
+                    mock.patch("app.gui.tabs.generate.GenerateWorker", FakeGenerateWorker), \
+                    mock.patch("app.gui.tabs.generate.run_worker", return_value=fake_thread):
+                self.assertTrue(dialog.regenerate_cover())
+            worker = FakeGenerateWorker.instances[0]
+            self.write_test_image(canonical)
+            db.set_generated(task.id, "新标题", canonical, path=cfg["db_path"])
+
+            worker.finished.emit({"ok": True, "cover_done": 1, "failed": 0})
+
+            self.assertEqual(canonical, db.get_task(task.id, path=cfg["db_path"]).new_cover_path)
+            self.assertEqual(canonical, dialog.selected_path)
+            self.assertTrue(dialog.candidate_buttons[canonical].isChecked())
+            self.assertEqual("封面重新生成完成", dialog.status_label.text())
+            self.assertIn(task.id, refreshed)
+
+        self.assert_removed(temp_dir)
+
+    def test_cover_gallery_regenerate_failure_keeps_archived_candidate_selectable(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg, account, task, canonical = self._cover_gallery_task(temp_dir)
+            fake_thread = FakeThread()
+            dialog = CoverGalleryDialog(
+                task,
+                cfg["image_dir"],
+                cfg["db_path"],
+                account=account,
+                config=cfg,
+                prompt_values={"title": "标题", "cover": "封面"},
+            )
+            self.addCleanup(dialog.close)
+            message_box, _ = self.make_fake_message_box("重新生成")
+            FakeGenerateWorker.instances.clear()
+
+            with mock.patch("app.gui.tabs.generate.QMessageBox", message_box), \
+                    mock.patch("app.gui.tabs.generate.GenerateWorker", FakeGenerateWorker), \
+                    mock.patch("app.gui.tabs.generate.run_worker", return_value=fake_thread):
+                self.assertTrue(dialog.regenerate_cover())
+            worker = FakeGenerateWorker.instances[0]
+            worker.log.emit("[失败] 商品 51100639510 图片生成失败：测试失败")
+            worker.finished.emit({"ok": False, "cover_done": 0, "failed": 1})
+
+            self.assertFalse(os.path.exists(canonical))
+            self.assertIsNone(db.get_task(task.id, path=cfg["db_path"]).new_cover_path)
+            self.assertTrue(dialog.candidates)
+            archived = dialog.candidates[0]
+            dialog.candidate_buttons[archived].setChecked(True)
+            self.assertTrue(dialog.save_selection())
+            self.assertEqual(os.path.abspath(archived), db.get_task(task.id, path=cfg["db_path"]).new_cover_path)
+
+        self.assert_removed(temp_dir)
+
+    def test_cover_gallery_regenerate_archive_failure_does_not_start_worker(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg, account, task, canonical = self._cover_gallery_task(temp_dir)
+            dialog = CoverGalleryDialog(
+                task,
+                cfg["image_dir"],
+                cfg["db_path"],
+                account=account,
+                config=cfg,
+                prompt_values={"title": "标题", "cover": "封面"},
+            )
+            self.addCleanup(dialog.close)
+            message_box, boxes = self.make_fake_message_box("重新生成")
+            FakeGenerateWorker.instances.clear()
+
+            with mock.patch("app.gui.tabs.generate.QMessageBox", message_box), \
+                    mock.patch("app.db.os.rename", side_effect=PermissionError), \
+                    mock.patch("app.gui.tabs.generate.GenerateWorker", FakeGenerateWorker):
+                self.assertFalse(dialog.regenerate_cover())
+
+            self.assertTrue(os.path.exists(canonical))
+            self.assertEqual(canonical, db.get_task(task.id, path=cfg["db_path"]).new_cover_path)
+            self.assertEqual([], FakeGenerateWorker.instances)
+            self.assertIn("请先关闭正在查看的封面图片再重置", boxes[-1].text)
+
+        self.assert_removed(temp_dir)
+
+    def test_cover_gallery_regenerate_disabled_while_bulk_generation_runs(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg, account, task, canonical = self._cover_gallery_task(temp_dir)
+            dialog = CoverGalleryDialog(
+                task,
+                cfg["image_dir"],
+                cfg["db_path"],
+                account=account,
+                config=cfg,
+                prompt_values={"title": "标题", "cover": "封面"},
+                bulk_running_checker=lambda: True,
+            )
+            self.addCleanup(dialog.close)
+
+            self.assertFalse(dialog.regenerate_button.isEnabled())
+            self.assertIn("批量生成正在运行", dialog.regenerate_button.toolTip())
+            self.assertFalse(dialog.regenerate_cover())
+            self.assertEqual(canonical, db.get_task(task.id, path=cfg["db_path"]).new_cover_path)
+
+        self.assert_removed(temp_dir)
+
+    def test_generate_tab_blocks_batch_generate_while_gallery_regenerates(self):
+        with self.make_temp_dir() as temp_dir:
+            cfg = self.make_config(temp_dir)
+            statuses = []
+            tab = GenerateTab(config=cfg, status_callback=statuses.append)
+            self.addCleanup(tab.close)
+
+            tab._set_cover_gallery_regenerate_running(True)
+            tab.start_generate()
+
+            self.assertFalse(tab.generate_button.isEnabled())
+            self.assertIn("封面画廊正在重新生成", statuses[-1])
+            tab._set_cover_gallery_regenerate_running(False)
+            self.assertTrue(tab.generate_button.isEnabled())
+
+        self.assert_removed(temp_dir)
+
 
     def test_generate_tab_allows_editing_generated_title_locally(self):
         with self.make_temp_dir() as temp_dir: