fix: 修改一下生图的 bug

server/tests/test_export.py

@@ -0,0 +1,32 @@
+"""单图导出响应头回归测试。"""
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+SERVER_DIR = Path(__file__).resolve().parents[1]
+if str(SERVER_DIR) not in sys.path:
+    sys.path.insert(0, str(SERVER_DIR))
+
+from api.export import ExportImageItem, export_image
+
+
+class ExportImageTests(unittest.IsolatedAsyncioTestCase):
+    async def test_chinese_filename_uses_rfc5987_header(self) -> None:
+        image = ExportImageItem(
+            url="https://example.com/image.png",
+            key="尺寸标注图",
+            variantName="蓝色",
+        )
+        with patch("api.export._download", new=AsyncMock(return_value=(b"image", "image/png"))):
+            response = await export_image(image)
+
+        disposition = response.headers["content-disposition"]
+        self.assertIn('filename="_____-__.png"', disposition)
+        self.assertIn("filename*=UTF-8''%E5%B0%BA%E5%AF%B8%E6%A0%87%E6%B3%A8%E5%9B%BE-%E8%93%9D%E8%89%B2.png", disposition)
+
+
+if __name__ == "__main__":
+    unittest.main()

server/tests/test_job_worker.py

@@ -0,0 +1,63 @@
+"""后台生图 Worker 的取消竞态回归测试。"""
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+SERVER_DIR = Path(__file__).resolve().parents[1]
+if str(SERVER_DIR) not in sys.path:
+    sys.path.insert(0, str(SERVER_DIR))
+
+from services.generator import GenerationAdvance
+from services.job_worker import process_item
+from services.jobs import ClaimedItem
+
+
+class WorkerCancellationTests(unittest.IsolatedAsyncioTestCase):
+    async def test_cancelled_task_is_not_returned_to_provider_waiting_queue(self) -> None:
+        claim = ClaimedItem(
+            job_id="00000000-0000-0000-0000-000000000001",
+            item_id="00000000-0000-0000-0000-000000000002",
+            job_kind="image_edit",
+            product_id=None,
+            provider="pcore",
+            model="gpt-image-2",
+            platform="ozon",
+            lang="ru",
+            ratio="3:4",
+            style_set=5,
+            request_payload={"context": {"title": "test"}, "ref_images": []},
+            item_payload={"kind": "custom", "title": "AI生图"},
+            provider_task_id="provider-task",
+            provider_task_kind="edits",
+        )
+        settings = SimpleNamespace(
+            image_provider="pcore",
+            dashscope_model="wan2.7-image-pro",
+            rightapi_image_model="gpt-image-2",
+            ark_image_model="doubao-seedream-4-5-251128",
+        )
+
+        with (
+            patch("services.job_worker.get_settings", return_value=settings),
+            patch("services.job_worker.build_context", return_value={}),
+            patch("services.job_worker.build_prompt", return_value="prompt"),
+            patch("services.job_worker._refs_for_job", return_value=[]),
+            patch("services.job_worker.advance_generation", new=AsyncMock(return_value=GenerationAdvance(
+                provider_task_id="provider-task", provider_task_kind="edits",
+            ))),
+            patch("services.job_worker.job_can_write", new=AsyncMock(return_value=False)),
+            patch("services.job_worker.mark_item_cancelled", new=AsyncMock()) as cancelled,
+            patch("services.job_worker.mark_waiting_provider", new=AsyncMock()) as waiting,
+        ):
+            await process_item(claim)
+
+        cancelled.assert_awaited_once_with(claim.item_id)
+        waiting.assert_not_awaited()
+
+
+if __name__ == "__main__":
+    unittest.main()

server/tests/test_planner.py

@@ -0,0 +1,69 @@
+"""AI 套图规划的均衡覆盖回归测试。"""
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+SERVER_DIR = Path(__file__).resolve().parents[1]
+if str(SERVER_DIR) not in sys.path:
+    sys.path.insert(0, str(SERVER_DIR))
+
+from services.planner import _balanced_items
+
+
+def item(kind: str, variant_name: str | None = None) -> dict:
+    return {
+        "kind": kind,
+        "title": kind,
+        "detail": "",
+        "prompt_hint": "",
+        "count": 1,
+        "variant_name": variant_name,
+    }
+
+
+class BalancedPlanTests(unittest.TestCase):
+    def test_many_skus_do_not_crowd_out_core_image_types(self) -> None:
+        variants = ["黑色", "白色", "蓝色", "粉色", "绿色"]
+        plan = _balanced_items(
+            [item("white_bg", variant) for variant in variants],
+            variants,
+            {"params": [{"key": "尺寸", "value": "30 × 20 × 10 cm"}]},
+            None,
+            budget=15,
+        )
+
+        self.assertLessEqual(sum(p["count"] for p in plan if p["kind"] == "white_bg"), 2)
+        self.assertTrue(any(p["kind"] == "lifestyle" for p in plan))
+        self.assertTrue(any(p["kind"] == "material" for p in plan))
+        self.assertTrue(any(p["kind"] == "size_chart" for p in plan))
+
+    def test_does_not_add_size_chart_without_dimensions(self) -> None:
+        plan = _balanced_items(
+            [item("white_bg", "黑色")],
+            ["黑色", "白色"],
+            {"params": [{"key": "规格", "value": "标准款"}, {"key": "材质", "value": "涤纶"}]},
+            None,
+            budget=15,
+        )
+
+        self.assertFalse(any(p["kind"] == "size_chart" for p in plan))
+        self.assertTrue(any(p["kind"] == "lifestyle" for p in plan))
+        self.assertTrue(any(p["kind"] == "material" for p in plan))
+
+    def test_explicit_every_sku_requirement_is_preserved(self) -> None:
+        variants = ["黑色", "白色", "蓝色"]
+        plan = _balanced_items(
+            [item("white_bg", variant) for variant in variants],
+            variants,
+            {},
+            "每个 SKU 都单独出一张白底图",
+            budget=15,
+        )
+
+        self.assertEqual(sum(p["count"] for p in plan if p["kind"] == "white_bg"), 3)
+
+
+if __name__ == "__main__":
+    unittest.main()