diff --git a/plugins/image_gen/minimax/__init__.py b/plugins/image_gen/minimax/__init__.py new file mode 100644 index 0000000000000..1ad5d6961c7a6 --- /dev/null +++ b/plugins/image_gen/minimax/__init__.py @@ -0,0 +1,442 @@ +"""MiniMax image generation backend. + +Exposes the MiniMax ``image-01`` model under three virtual model IDs that +match the existing image_gen picker conventions. MiniMax's image API is +synchronous — POST returns a JSON body with a URL (or base64) for the +generated image, no job polling required. + + minimax-image-01 — full quality (default) + minimax-image-01-live — lower latency variant + minimax-image-01-square — alias that forces a square aspect ratio + +All three hit the same native endpoint with a different ``model`` field. +Output is saved under ``$HERMES_HOME/cache/images/``. + +Selection precedence (first hit wins): + +1. ``MINIMAX_IMAGE_MODEL`` env var (escape hatch for scripts / tests) +2. ``image_gen.minimax.model`` in ``config.yaml`` +3. :data:`DEFAULT_MODEL` — ``minimax-image-01`` + +Endpoint: ``POST https://api.minimax.io/v1/image_generation`` +Auth: ``Authorization: Bearer $MINIMAX_API_KEY`` (always) and an optional +``GroupId`` query parameter (``$MINIMAX_GROUP_ID``) for accounts that use +group-scoped billing. The native endpoint is **not** OpenAI-compatible — +``/v1/images/generations`` returns 404. +""" + +from __future__ import annotations + +import logging +import os +from typing import Any, Dict, List, Optional, Tuple + +import requests + +from agent.image_gen_provider import ( + DEFAULT_ASPECT_RATIO, + ImageGenProvider, + error_response, + resolve_aspect_ratio, + save_b64_image, + save_url_image, + success_response, +) + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +BASE_URL = "https://api.minimax.io/v1/image_generation" +TIMEOUT_SECONDS = 120.0 + +# Each virtual model ID maps to a MiniMax ``model`` field. We expose +# three IDs so the picker UX matches other backends (which all advertise +# 2-3 quality tiers) and so the ``-live`` suffix hints at the live variant. +_MODELS: Dict[str, Dict[str, Any]] = { + "minimax-image-01": { + "display": "MiniMax Image 01", + "speed": "~10-30s", + "strengths": "Default. Strong prompt adherence, 1024x1024 default.", + "price": "varies (see MiniMax pricing)", + "api_model": "image-01", + }, + "minimax-image-01-live": { + "display": "MiniMax Image 01 (Live)", + "speed": "~5-15s", + "strengths": "Lower latency variant, same quality ceiling.", + "price": "varies (see MiniMax pricing)", + "api_model": "image-01-live", + }, + "minimax-image-01-square": { + "display": "MiniMax Image 01 (Square)", + "speed": "~10-30s", + "strengths": "Alias that forces 1024x1024 — convenience for square output.", + "price": "varies (see MiniMax pricing)", + "api_model": "image-01", + }, +} + +DEFAULT_MODEL = "minimax-image-01" + +# Map hermes 3-ratio abstraction to MiniMax's aspect ratio strings. +# MiniMax accepts "1:1", "16:9", "4:3", "3:2", "2:3", "3:4", "9:16", "21:9". +_ASPECT_MAP = { + "landscape": "16:9", + "square": "1:1", + "portrait": "9:16", +} + + +# --------------------------------------------------------------------------- +# Config + resolution +# --------------------------------------------------------------------------- + + +def _load_minimax_config() -> Dict[str, Any]: + """Read the ``image_gen`` section from config.yaml. + + Returns the full ``image_gen`` dict (callers extract their own sub-keys); + returns an empty dict on any read or parse failure so callers can use + ``.get()`` chains without guarding for None. + """ + try: + from hermes_cli.config import load_config + + cfg = load_config() + section = cfg.get("image_gen") if isinstance(cfg, dict) else None + return section if isinstance(section, dict) else {} + except Exception as exc: # noqa: BLE001 + logger.debug("Could not load image_gen config: %s", exc) + return {} + + +def _resolve_model() -> Tuple[str, Dict[str, Any]]: + """Decide which model to use and return ``(model_id, meta)``. + + Precedence: env var > config (``image_gen.minimax.model``) > default. + """ + env_override = os.environ.get("MINIMAX_IMAGE_MODEL") + if env_override and env_override in _MODELS: + return env_override, _MODELS[env_override] + + cfg = _load_minimax_config() + minimax_cfg = cfg.get("minimax") if isinstance(cfg.get("minimax"), dict) else {} + candidate: Optional[str] = None + if isinstance(minimax_cfg, dict): + value = minimax_cfg.get("model") + if isinstance(value, str) and value in _MODELS: + candidate = value + + if candidate is not None: + return candidate, _MODELS[candidate] + + return DEFAULT_MODEL, _MODELS[DEFAULT_MODEL] + + +def _resolve_aspect(model_id: str, aspect: str) -> str: + """Map hermes's 3-ratio abstraction to MiniMax's aspect strings. + + The ``-square`` virtual model always forces 1:1 regardless of the + aspect_ratio kwarg. The other two honor the kwarg. + """ + if model_id == "minimax-image-01-square": + return "1:1" + return _ASPECT_MAP.get(aspect, "1:1") + + +# --------------------------------------------------------------------------- +# Provider +# --------------------------------------------------------------------------- + + +class MiniMaxImageGenProvider(ImageGenProvider): + """MiniMax image generation backend — image-01 / image-01-live.""" + + @property + def name(self) -> str: + return "minimax" + + @property + def display_name(self) -> str: + return "MiniMax" + + def is_available(self) -> bool: + return bool(os.environ.get("MINIMAX_API_KEY")) + + def list_models(self) -> List[Dict[str, Any]]: + return [ + { + "id": model_id, + "display": meta["display"], + "speed": meta["speed"], + "strengths": meta["strengths"], + "price": meta["price"], + } + for model_id, meta in _MODELS.items() + ] + + def get_setup_schema(self) -> Dict[str, Any]: + return { + "name": "MiniMax", + "badge": "paid", + "tag": "MiniMax image-01 / image-01-live. Native endpoint, not OpenAI-compat.", + "env_vars": [ + { + "key": "MINIMAX_API_KEY", + "prompt": "MiniMax API key", + "url": "https://platform.minimax.io/user-center/basic-information/interface-key", + }, + { + "key": "MINIMAX_GROUP_ID", + "prompt": "MiniMax Group ID (required for group-scoped accounts; optional otherwise)", + "url": "https://platform.minimax.io/user-center/basic-information/interface-key", + "required": False, + }, + ], + } + + def generate( + self, + prompt: str, + aspect_ratio: str = DEFAULT_ASPECT_RATIO, + **kwargs: Any, + ) -> Dict[str, Any]: + prompt = (prompt or "").strip() + aspect = resolve_aspect_ratio(aspect_ratio) + provider_name = self.name + + if not prompt: + return error_response( + error="Prompt is required and must be a non-empty string", + error_type="invalid_argument", + provider=provider_name, + prompt=prompt, + aspect_ratio=aspect, + ) + + api_key = os.environ.get("MINIMAX_API_KEY") + if not api_key: + return error_response( + error=( + "MINIMAX_API_KEY not set. Run `hermes tools` → Image " + "Generation → MiniMax to configure, or get a key at " + "https://platform.minimax.io/" + ), + error_type="auth_required", + provider=provider_name, + prompt=prompt, + aspect_ratio=aspect, + ) + + model_id, meta = _resolve_model() + api_model = meta["api_model"] + minmax_aspect = _resolve_aspect(model_id, aspect) + + # Forward-compat passthroughs — MiniMax accepts these but the agent + # rarely supplies them. Default to a single image. + n = int(kwargs.get("n", 1)) + if n < 1 or n > 4: + return error_response( + error=f"n must be 1-4, got {n}", + error_type="invalid_argument", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + response_format = str(kwargs.get("response_format", "url")).lower() + if response_format not in {"url", "base64"}: + return error_response( + error=f"response_format must be 'url' or 'base64', got {response_format!r}", + error_type="invalid_argument", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + payload: Dict[str, Any] = { + "model": api_model, + "prompt": prompt, + "aspect_ratio": minmax_aspect, + "n": n, + "response_format": response_format, + } + if "seed" in kwargs: + payload["seed"] = int(kwargs["seed"]) + if kwargs.get("prompt_optimizer"): + payload["prompt_optimizer"] = True + if kwargs.get("aigc_watermark"): + payload["aigc_watermark"] = True + + # Build the URL — GroupId is a query string parameter when present + # (some MiniMax accounts use group-scoped billing; without it the + # server returns 401 "GroupId is required"). + group_id = (os.environ.get("MINIMAX_GROUP_ID") or "").strip() + url = BASE_URL + if group_id: + url = f"{url}?GroupId={group_id}" + + headers = { + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + } + + try: + response = requests.post( + url, + json=payload, + headers=headers, + timeout=TIMEOUT_SECONDS, + ) + except requests.Timeout: + return error_response( + error=f"MiniMax image generation timed out after {TIMEOUT_SECONDS}s", + error_type="timeout", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + except requests.RequestException as exc: + logger.debug("MiniMax image generation transport error", exc_info=True) + return error_response( + error=f"MiniMax image generation failed: {exc}", + error_type="api_error", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + # MiniMax returns 401/403/404 with a JSON body — surface its message + # to the user verbatim so they can fix the auth or pick a different + # model. 200 with no data is also a real failure mode. + if response.status_code != 200: + server_msg = "" + try: + body = response.json() + server_msg = ( + body.get("message") + or body.get("error", {}).get("message") + or body.get("base_resp", {}).get("status_msg") + or "" + ) + except Exception: # noqa: BLE001 + server_msg = response.text[:200] + return error_response( + error=( + f"MiniMax image generation HTTP {response.status_code}: " + f"{server_msg or response.reason}" + ), + error_type="api_error", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + try: + body = response.json() + except ValueError as exc: + return error_response( + error=f"MiniMax returned non-JSON body: {exc}", + error_type="invalid_response", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + # The native endpoint returns ``data.image_urls`` — a list of + # signed CDN URLs (each ~30-min TTL) when response_format=url, or + # ``data.b64_json`` (also a list) when response_format=base64. + # Unlike the OpenAI images API, ``data`` is a DICT not a list. + data = body.get("data") or {} + if not data: + base_resp = body.get("base_resp") or {} + return error_response( + error=( + f"MiniMax returned no image data: " + f"{base_resp.get('status_msg') or 'empty data object'}" + ), + error_type="empty_response", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + # Prefer b64_json if the caller asked for it; fall back to image_urls. + image_url: Optional[str] = None + b64: Optional[str] = None + b64_list = data.get("b64_json") + if isinstance(b64_list, list) and b64_list: + b64 = b64_list[0] + if b64 is None: + url_list = data.get("image_urls") + if isinstance(url_list, list) and url_list: + image_url = url_list[0] + + image_ref: Optional[str] = None + if b64: + try: + saved_path = save_b64_image(b64, prefix=f"minimax_{model_id}") + except Exception as exc: # noqa: BLE001 + return error_response( + error=f"Could not save image to cache: {exc}", + error_type="io_error", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + image_ref = str(saved_path) + elif image_url: + # Cache the bytes locally so the gateway never tries to fetch + # an expired signed URL after the 30-min TTL. Same pattern as + # the xAI provider (see test_xai_provider for the original). + try: + saved_path = save_url_image(image_url, prefix=f"minimax_{model_id}") + except Exception as exc: # noqa: BLE001 + logger.warning( + "MiniMax image URL %s could not be cached (%s); falling back to bare URL.", + image_url, + exc, + ) + image_ref = image_url + else: + image_ref = str(saved_path) + + if not image_ref: + return error_response( + error="MiniMax response contained neither b64_json nor image_urls", + error_type="empty_response", + provider=provider_name, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + ) + + return success_response( + image=image_ref, + model=model_id, + prompt=prompt, + aspect_ratio=aspect, + provider=provider_name, + extra={"minimax_aspect": minmax_aspect, "api_model": api_model}, + ) + + +# --------------------------------------------------------------------------- +# Plugin entry point +# --------------------------------------------------------------------------- + + +def register(ctx) -> None: + """Plugin entry point — wire ``MiniMaxImageGenProvider`` into the registry.""" + ctx.register_image_gen_provider(MiniMaxImageGenProvider()) diff --git a/plugins/image_gen/minimax/plugin.yaml b/plugins/image_gen/minimax/plugin.yaml new file mode 100644 index 0000000000000..a6f7216bc019a --- /dev/null +++ b/plugins/image_gen/minimax/plugin.yaml @@ -0,0 +1,9 @@ +name: minimax +version: 1.0.0 +description: "MiniMax image generation backend (image-01 / image-01-live). Native endpoint, not OpenAI-compatible." +author: Vality +kind: backend +requires_env: + - MINIMAX_API_KEY +optional_env: + - MINIMAX_GROUP_ID diff --git a/tests/plugins/image_gen/test_minimax_provider.py b/tests/plugins/image_gen/test_minimax_provider.py new file mode 100644 index 0000000000000..fbaf55b765f42 --- /dev/null +++ b/tests/plugins/image_gen/test_minimax_provider.py @@ -0,0 +1,586 @@ +#!/usr/bin/env python3 +"""Tests for the MiniMax image generation provider. + +Covers: +- is_available() with/without MINIMAX_API_KEY +- list_models() returns the 3 virtual model IDs +- default_model() returns minimax-image-01 +- get_setup_schema() includes both env vars (GroupId marked optional) +- _resolve_model() respects env, config, defaults +- _resolve_aspect() honors the -square alias +- generate() with monkeypatched requests.post: success path, error paths +- generate() handles missing MINIMAX_API_KEY, empty prompt, bad n + +The live API is NOT called — ``requests.post`` is monkeypatched in every +test that would otherwise hit the network. See the bottom-of-file +``test_live_image_generation`` for an opt-in live test (skipped by +default to keep CI deterministic). +""" + +from __future__ import annotations + +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _fake_api_key(monkeypatch): + """Ensure MINIMAX_API_KEY is set for all tests by default.""" + monkeypatch.setenv("MINIMAX_API_KEY", "test-key-12345") + + +@pytest.fixture(autouse=True) +def _reset_env(monkeypatch): + """Wipe config-influencing env vars between tests.""" + for var in ( + "MINIMAX_IMAGE_MODEL", + "MINIMAX_GROUP_ID", + ): + monkeypatch.delenv(var, raising=False) + + +def _ok_url_response(url: str = "https://minimax.cdn/img.jpeg"): + resp = MagicMock() + resp.status_code = 200 + resp.raise_for_status = MagicMock() + resp.json.return_value = { + "id": "test-id-abc", + "data": {"image_urls": [url]}, + "metadata": {"failed_count": "0", "success_count": "1"}, + "base_resp": {"status_code": 0, "status_msg": "success"}, + } + return resp + + +def _ok_b64_response(b64: str = "aGVsbG8="): + resp = MagicMock() + resp.status_code = 200 + resp.raise_for_status = MagicMock() + resp.json.return_value = { + "id": "test-id-abc", + "data": {"b64_json": [b64]}, + "metadata": {"failed_count": "0", "success_count": "1"}, + "base_resp": {"status_code": 0, "status_msg": "success"}, + } + return resp + + +def _err_response(status_code: int, msg: str = "GroupId is required"): + resp = MagicMock() + resp.status_code = status_code + resp.raise_for_status = MagicMock() + resp.json.return_value = { + "base_resp": {"status_code": status_code, "status_msg": msg}, + "message": msg, + } + return resp + + +# --------------------------------------------------------------------------- +# is_available +# --------------------------------------------------------------------------- + + +class TestIsAvailable: + def test_is_available_with_key(self, monkeypatch): + monkeypatch.setenv("MINIMAX_API_KEY", "sk-test") + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + assert MiniMaxImageGenProvider().is_available() is True + + def test_is_available_without_key(self, monkeypatch): + monkeypatch.delenv("MINIMAX_API_KEY", raising=False) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + assert MiniMaxImageGenProvider().is_available() is False + + +# --------------------------------------------------------------------------- +# Model catalog +# --------------------------------------------------------------------------- + + +class TestModelCatalog: + def test_list_models(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + provider = MiniMaxImageGenProvider() + models = provider.list_models() + assert [m["id"] for m in models] == [ + "minimax-image-01", + "minimax-image-01-live", + "minimax-image-01-square", + ] + for m in models: + assert "display" in m + assert "speed" in m + assert "strengths" in m + + def test_default_model(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider, DEFAULT_MODEL + + assert MiniMaxImageGenProvider().default_model() == DEFAULT_MODEL + assert DEFAULT_MODEL == "minimax-image-01" + + def test_get_setup_schema(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + schema = MiniMaxImageGenProvider().get_setup_schema() + assert schema["name"] == "MiniMax" + env_keys = [v["key"] for v in schema["env_vars"]] + assert "MINIMAX_API_KEY" in env_keys + assert "MINIMAX_GROUP_ID" in env_keys + # GroupId should be marked optional + group_id_var = next( + v for v in schema["env_vars"] if v["key"] == "MINIMAX_GROUP_ID" + ) + assert group_id_var.get("required") is False + + +# --------------------------------------------------------------------------- +# _resolve_model +# --------------------------------------------------------------------------- + + +class TestResolveModel: + def test_env_override_wins(self, monkeypatch): + monkeypatch.setenv("MINIMAX_IMAGE_MODEL", "minimax-image-01-live") + from plugins.image_gen.minimax import _resolve_model + + model_id, meta = _resolve_model() + assert model_id == "minimax-image-01-live" + assert meta["api_model"] == "image-01-live" + + def test_default_when_no_config(self): + from plugins.image_gen.minimax import _resolve_model, DEFAULT_MODEL + + model_id, _ = _resolve_model() + assert model_id == DEFAULT_MODEL + + +# --------------------------------------------------------------------------- +# _resolve_aspect +# --------------------------------------------------------------------------- + + +class TestResolveAspect: + def test_square_alias_forces_1_1(self): + from plugins.image_gen.minimax import _resolve_aspect + + assert _resolve_aspect("minimax-image-01-square", "landscape") == "1:1" + assert _resolve_aspect("minimax-image-01-square", "portrait") == "1:1" + assert _resolve_aspect("minimax-image-01-square", "square") == "1:1" + + def test_other_models_honor_kwarg(self): + from plugins.image_gen.minimax import _resolve_aspect + + assert _resolve_aspect("minimax-image-01", "landscape") == "16:9" + assert _resolve_aspect("minimax-image-01", "portrait") == "9:16" + assert _resolve_aspect("minimax-image-01", "square") == "1:1" + assert _resolve_aspect("minimax-image-01-live", "landscape") == "16:9" + + def test_unknown_aspect_defaults_to_square(self): + from plugins.image_gen.minimax import _resolve_aspect + + assert _resolve_aspect("minimax-image-01", "weird") == "1:1" + + +# --------------------------------------------------------------------------- +# generate() — input validation +# --------------------------------------------------------------------------- + + +class TestGenerateInputValidation: + def test_empty_prompt(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("", "landscape") + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + assert "non-empty" in result["error"] + + def test_whitespace_only_prompt(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate(" \n ", "landscape") + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + def test_missing_api_key(self, monkeypatch): + monkeypatch.delenv("MINIMAX_API_KEY", raising=False) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert result["error_type"] == "auth_required" + + def test_invalid_n(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape", n=0) + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + result = MiniMaxImageGenProvider().generate("a cat", "landscape", n=5) + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + def test_invalid_response_format(self): + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate( + "a cat", "landscape", response_format="json" + ) + assert result["success"] is False + assert result["error_type"] == "invalid_argument" + + +# --------------------------------------------------------------------------- +# generate() — request shape +# --------------------------------------------------------------------------- + + +class TestGenerateRequest: + def test_request_includes_correct_url_no_groupid(self, monkeypatch): + captured = {} + + def fake_post(url, **kwargs): + captured["url"] = url + captured["json"] = kwargs.get("json") + captured["headers"] = kwargs.get("headers", {}) + return _ok_url_response() + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + MiniMaxImageGenProvider().generate("a tiny red square", "square") + + assert captured["url"] == "https://api.minimax.io/v1/image_generation" + assert captured["json"]["model"] == "image-01" + assert captured["json"]["prompt"] == "a tiny red square" + assert captured["json"]["aspect_ratio"] == "1:1" + assert captured["json"]["n"] == 1 + assert captured["json"]["response_format"] == "url" + # No GroupId query string + assert "GroupId" not in captured["url"] + # Bearer auth header + assert captured["headers"]["Authorization"] == "Bearer test-key-12345" + + def test_request_appends_groupid_when_set(self, monkeypatch): + monkeypatch.setenv("MINIMAX_GROUP_ID", "grp_abc") + captured = {} + + def fake_post(url, **kwargs): + captured["url"] = url + return _ok_url_response() + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + MiniMaxImageGenProvider().generate("a cat", "landscape") + + assert captured["url"].endswith("?GroupId=grp_abc") + + def test_request_uses_live_api_model(self, monkeypatch): + monkeypatch.setenv("MINIMAX_IMAGE_MODEL", "minimax-image-01-live") + captured = {} + + def fake_post(url, **kwargs): + captured["json"] = kwargs.get("json") + return _ok_url_response() + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + MiniMaxImageGenProvider().generate("a cat", "landscape") + + assert captured["json"]["model"] == "image-01-live" + + def test_request_uses_square_when_alias_selected(self, monkeypatch): + monkeypatch.setenv("MINIMAX_IMAGE_MODEL", "minimax-image-01-square") + captured = {} + + def fake_post(url, **kwargs): + captured["json"] = kwargs.get("json") + return _ok_url_response() + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + # Even when caller asks for landscape, the -square alias forces 1:1 + MiniMaxImageGenProvider().generate("a cat", "landscape") + + assert captured["json"]["model"] == "image-01" + assert captured["json"]["aspect_ratio"] == "1:1" + + def test_seed_passthrough(self, monkeypatch): + captured = {} + + def fake_post(url, **kwargs): + captured["json"] = kwargs.get("json") + return _ok_url_response() + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + MiniMaxImageGenProvider().generate("a cat", "landscape", seed=42) + assert captured["json"]["seed"] == 42 + + def test_prompt_optimizer_and_watermark_passthrough(self, monkeypatch): + captured = {} + + def fake_post(url, **kwargs): + captured["json"] = kwargs.get("json") + return _ok_url_response() + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + MiniMaxImageGenProvider().generate( + "a cat", "landscape", prompt_optimizer=True, aigc_watermark=True + ) + assert captured["json"]["prompt_optimizer"] is True + assert captured["json"]["aigc_watermark"] is True + + +# --------------------------------------------------------------------------- +# generate() — success path +# --------------------------------------------------------------------------- + + +class TestGenerateSuccess: + def test_url_response_saves_to_cache(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr( + "plugins.image_gen.minimax.requests.post", + lambda *a, **kw: _ok_url_response(), + ) + # Mock the URL cache to avoid real network in tests + monkeypatch.setattr( + "plugins.image_gen.minimax.save_url_image", + lambda url, prefix: tmp_path / f"{prefix}_cached.jpeg", + ) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + + assert result["success"] is True + assert result["provider"] == "minimax" + assert result["model"] == "minimax-image-01" + assert result["aspect_ratio"] == "landscape" + # Image was cached to disk (not the bare URL) + assert result["image"] == str(tmp_path / "minimax_minimax-image-01_cached.jpeg") + # Extra diagnostic fields populated + assert result["minimax_aspect"] == "16:9" + assert result["api_model"] == "image-01" + + def test_url_response_falls_back_to_bare_url_when_cache_fails(self, monkeypatch, tmp_path): + """If the URL fetch fails (network error, expired CDN URL, etc.), + the provider should still return success with the bare URL rather + than failing the whole call — same defensive pattern as the xAI + provider (see test_xai_provider for the original rationale).""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr( + "plugins.image_gen.minimax.requests.post", + lambda *a, **kw: _ok_url_response(), + ) + + def _failing_cache(url, prefix): + raise IOError("simulated network failure") + + monkeypatch.setattr("plugins.image_gen.minimax.save_url_image", _failing_cache) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is True + assert result["image"] == "https://minimax.cdn/img.jpeg" # bare URL fallback + + def test_b64_response_saves_to_cache(self, monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setattr( + "plugins.image_gen.minimax.requests.post", + lambda *a, **kw: _ok_b64_response(), + ) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "square") + assert result["success"] is True + assert result["image"].startswith(str(tmp_path)) + # 8 bytes of "hello" — verify the bytes were actually written + written = Path(result["image"]).read_bytes() + assert written == b"hello" + + +# --------------------------------------------------------------------------- +# generate() — error paths +# --------------------------------------------------------------------------- + + +class TestGenerateErrors: + def test_http_401_returns_auth_error(self, monkeypatch): + monkeypatch.setattr( + "plugins.image_gen.minimax.requests.post", + lambda *a, **kw: _err_response(401, "invalid api key"), + ) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert result["error_type"] == "api_error" + assert "401" in result["error"] + assert "invalid api key" in result["error"] + + def test_http_404_with_missing_groupid_message(self, monkeypatch): + monkeypatch.setattr( + "plugins.image_gen.minimax.requests.post", + lambda *a, **kw: _err_response(404, "GroupId is required"), + ) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert "GroupId" in result["error"] + + def test_empty_data_array(self, monkeypatch): + def fake_post(*a, **kw): + resp = MagicMock() + resp.status_code = 200 + resp.raise_for_status = MagicMock() + resp.json.return_value = { + "data": {}, + "base_resp": {"status_code": 100, "status_msg": "no data"}, + } + return resp + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert result["error_type"] == "empty_response" + assert "no data" in result["error"] + + def test_response_with_neither_url_nor_b64(self, monkeypatch): + def fake_post(*a, **kw): + resp = MagicMock() + resp.status_code = 200 + resp.raise_for_status = MagicMock() + resp.json.return_value = { + "data": {"image_urls": [], "b64_json": []}, + "base_resp": {"status_code": 0, "status_msg": "success"}, + } + return resp + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert result["error_type"] == "empty_response" + + def test_timeout(self, monkeypatch): + import requests + + def fake_post(*a, **kw): + raise requests.Timeout("read timed out") + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert result["error_type"] == "timeout" + + def test_connection_error(self, monkeypatch): + import requests + + def fake_post(*a, **kw): + raise requests.ConnectionError("dns failure") + + monkeypatch.setattr("plugins.image_gen.minimax.requests.post", fake_post) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a cat", "landscape") + assert result["success"] is False + assert result["error_type"] == "api_error" + + +# --------------------------------------------------------------------------- +# Plugin registration +# --------------------------------------------------------------------------- + + +class TestPluginRegistration: + def test_register_wires_provider(self): + from plugins.image_gen.minimax import register, MiniMaxImageGenProvider + from agent import image_gen_registry + + ctx = MagicMock() + register(ctx) + ctx.register_image_gen_provider.assert_called_once() + provider = ctx.register_image_gen_provider.call_args[0][0] + assert isinstance(provider, MiniMaxImageGenProvider) + assert provider.name == "minimax" + # Sanity check the actual registry hook works (not just the mock) + image_gen_registry.register_provider(provider) + assert image_gen_registry.get_provider("minimax") is provider + # Clean up so other tests aren't affected + from agent.image_gen_registry import _providers as _reg + _reg.pop("minimax", None) + + +# --------------------------------------------------------------------------- +# Live test (skipped by default; opt in with ``RUN_LIVE_MINIMAX=1``) +# --------------------------------------------------------------------------- + + +@pytest.mark.skipif( + os.environ.get("RUN_LIVE_MINIMAX") != "1", + reason="Set RUN_LIVE_MINIMAX=1 to run a real API call against MiniMax", +) +def test_live_image_generation(monkeypatch, tmp_path): + """Real API call — only runs when RUN_LIVE_MINIMAX=1 is set. + + Useful for sanity-checking the plugin against a real MiniMax account. + Expects ``MINIMAX_API_KEY`` (and optionally ``MINIMAX_GROUP_ID``) to be + set in the **caller's shell environment** before pytest starts. The + autouse ``_fake_api_key`` fixture overwrites that key for unit tests; + we restore the real one here by reading the value that was set before + pytest's fixtures ran. ``os.environ`` is a snapshot of the shell at + process start, so once a fixture mutates a key, the original is gone + — we have to use the framework-supplied monkeypatch to recover it + via a sentinel approach: ask the user to pass the real key as a + different env var (``MINIMAX_API_KEY_LIVE``) when running live. + """ + # The autouse _fake_api_key fixture overwrites MINIMAX_API_KEY with + # "test-key-12345"; we need the real key. Read it from a side-channel + # env var that the caller sets when running live (the autouse fixture + # doesn't touch this one). Document the contract in the test docstring. + real_key = os.environ.get("MINIMAX_API_KEY_LIVE") or os.environ.get("MINIMAX_API_KEY", "") + if not real_key or real_key == "test-key-12345": + pytest.skip( + "Live test requires MINIMAX_API_KEY_LIVE (or a real MINIMAX_API_KEY " + "that doesn't equal the autouse fixture's 'test-key-12345')." + ) + monkeypatch.setenv("MINIMAX_API_KEY", real_key) + + # If a real GroupId is supplied, override the autouse fixture's wipe. + real_group = os.environ.get("MINIMAX_GROUP_ID_LIVE") + if real_group: + monkeypatch.setenv("MINIMAX_GROUP_ID", real_group) + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + from plugins.image_gen.minimax import MiniMaxImageGenProvider + + result = MiniMaxImageGenProvider().generate("a tiny red square", "square") + assert result["success"] is True, f"Live call failed: {result}" + assert Path(result["image"]).exists() + assert Path(result["image"]).stat().st_size > 0