summaryrefslogtreecommitdiff
path: root/tests/test_service.py
diff options
context:
space:
mode:
authortanaka takashi <isazaka@gmail.com>2026-09-12 18:04:56 +0900
committertanaka takashi <isazaka@gmail.com>2026-09-12 18:04:56 +0900
commitcba15a62829a0fe57a03c8031f981e8f790e2efc (patch)
tree3bc0ef1a243f356f55c2fa97f63c3bfd5c77e8fc /tests/test_service.py
feat: Stable Diffusion 1.x MCPサーバを実装
仕様書・設計書に基づき、ローカルCUDA GPU上のStable Diffusion 1.x互換モデルをMCP経由で利用するサーバを新規実装。 実装内容: - generate_image、generate_guided_image、inpaint_imageの3ツールをMCP SDK v2で公開 - TOMLからcheckpoint、VAE、複数LoRAと重み、ControlNet、生成設定を読み込み - モデル種別ごとの固定ディレクトリ解決と起動時の設定・ファイル存在検証を追加 - Depth/MiDaS、OpenPose body/full、ControlNet Inpaintの前処理とパイプライン切替を実装 - FP16/CUDA、1枚固定生成、ControlNetオンデマンドロード、単一GPU直列実行を実装 - Base64 PNG/JPEG入力の形式・サイズ検証と任意のdata URI入力に対応 - seed、サイズ、モード、時刻、処理時間をPNG metadataへ記録 - 任意のローカルPNG保存、生成エラー分類、CUDA OOM後のキャッシュ回復を追加 - --configと--versionを備えたstdio CLIエントリポイントを追加 開発・運用: - Python 3.11およびuvによる依存・lock管理を構成 - config.example.tomlとセットアップ・運用手順をREADMEへ追加 - Ruff、Pyright、pytestの設定を追加 - 設定、パス解決、schema、画像検証、サービスルーティング、排他制御、MCP画像返却をテスト 検証結果: - pytest: 30 passed - Ruff lint/format: passed - Pyright: 0 errors, 0 warnings - CLI versionおよびMCPクライアント経由のPNG返却を確認 - 実GPU smoke testはCUDA GPUとモデルファイルがないため対象外
Diffstat (limited to 'tests/test_service.py')
-rw-r--r--tests/test_service.py118
1 files changed, 118 insertions, 0 deletions
diff --git a/tests/test_service.py b/tests/test_service.py
new file mode 100644
index 0000000..da82204
--- /dev/null
+++ b/tests/test_service.py
@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import asyncio
+import io
+import threading
+import time
+from dataclasses import replace
+
+import pytest
+from PIL import Image
+
+from mcp_sd15.config import ControlModelConfig, ControlNetConfig
+from mcp_sd15.errors import GenerationError, InvalidRequestError
+from mcp_sd15.schema import GuidedImageRequest, InpaintImageRequest, TextToImageRequest
+from mcp_sd15.service import ImageGenerationService
+
+
+class FakeBackend:
+ def __init__(self, delay: float = 0) -> None:
+ self.calls: list[str] = []
+ self.delay = delay
+ self.active = 0
+ self.max_active = 0
+ self.guard = threading.Lock()
+
+ def _result(self, name: str) -> Image.Image:
+ with self.guard:
+ self.active += 1
+ self.max_active = max(self.max_active, self.active)
+ time.sleep(self.delay)
+ with self.guard:
+ self.active -= 1
+ self.calls.append(name)
+ return Image.new("RGB", (4, 4), "red")
+
+ def text_to_image(self, request: object, seed: int) -> Image.Image:
+ return self._result("text")
+
+ def guided_image(self, request: object, image: Image.Image, seed: int) -> Image.Image:
+ assert image.mode == "RGB"
+ return self._result("guided")
+
+ def inpaint_image(
+ self, request: object, image: Image.Image, mask: Image.Image, seed: int
+ ) -> Image.Image:
+ assert mask.mode == "L"
+ return self._result("inpaint")
+
+
+async def test_routes_all_generation_types(app_config, image_base64: str) -> None:
+ backend = FakeBackend()
+ service = ImageGenerationService(app_config, backend)
+
+ text = await service.generate_image(TextToImageRequest.create("cat"))
+ guided = await service.generate_guided_image(
+ GuidedImageRequest.create("cat", "depth", image_base64)
+ )
+ inpaint = await service.inpaint_image(
+ InpaintImageRequest.create("cat", image_base64, image_base64)
+ )
+
+ assert backend.calls == ["text", "guided", "inpaint"]
+ for result in (text, guided, inpaint):
+ assert result.png_data.startswith(b"\x89PNG")
+ assert isinstance(result.metadata["seed"], int)
+
+
+async def test_generation_is_strictly_serialized(app_config) -> None:
+ backend = FakeBackend(delay=0.03)
+ service = ImageGenerationService(app_config, backend)
+ await asyncio.gather(
+ *(service.generate_image(TextToImageRequest.create(f"image {i}")) for i in range(4))
+ )
+ assert backend.max_active == 1
+
+
+async def test_disabled_controlnet_fails_before_backend(app_config, image_base64: str) -> None:
+ disabled = replace(
+ app_config,
+ controlnet=ControlNetConfig(
+ depth=ControlModelConfig(False, None, "midas"),
+ openpose=app_config.controlnet.openpose,
+ inpaint=app_config.controlnet.inpaint,
+ ),
+ )
+ service = ImageGenerationService(disabled, FakeBackend())
+ with pytest.raises(InvalidRequestError, match="disabled"):
+ await service.generate_guided_image(GuidedImageRequest.create("cat", "depth", image_base64))
+
+
+async def test_backend_error_is_sanitized(app_config) -> None:
+ class BrokenBackend(FakeBackend):
+ def text_to_image(self, request: object, seed: int) -> Image.Image:
+ raise RuntimeError("boom")
+
+ service = ImageGenerationService(app_config, BrokenBackend())
+ with pytest.raises(GenerationError, match="boom"):
+ await service.generate_image(TextToImageRequest.create("cat"))
+
+
+async def test_cuda_oom_is_classified(app_config) -> None:
+ class OomBackend(FakeBackend):
+ def text_to_image(self, request: object, seed: int) -> Image.Image:
+ raise RuntimeError("CUDA out of memory while allocating")
+
+ service = ImageGenerationService(app_config, OomBackend())
+ with pytest.raises(GenerationError, match="CUDA out of memory"):
+ await service.generate_image(TextToImageRequest.create("cat"))
+
+
+async def test_optional_output_is_saved(app_config) -> None:
+ enabled = replace(app_config, output=replace(app_config.output, enabled=True))
+ service = ImageGenerationService(enabled, FakeBackend())
+ await service.generate_image(TextToImageRequest.create("cat"))
+ files = list(enabled.output.directory.glob("*.png"))
+ assert len(files) == 1
+ with Image.open(io.BytesIO(files[0].read_bytes())) as image:
+ assert image.format == "PNG"