"""Contract tests for the /api/image-generation endpoints.""" from __future__ import annotations import json import time import uuid from pathlib import Path import jwt import pytest import yaml from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec from starlette.applications import Starlette from starlette.testclient import TestClient from EvoScientist.image_gen import config as image_config from EvoScientist.image_gen.config import ( ImageGenerationSettings, ImageModelEntry, save_image_generation_settings, ) from EvoScientist.model_registry import http_api from EvoScientist.model_registry.auth import BffAuthenticator from EvoScientist.model_registry.endpoint_policy import EndpointPolicy from EvoScientist.model_registry.hashing import configuration_hash # noqa: F401 (fixture parity) from EvoScientist.model_registry.http_api import ApiServices, model_registry_routes from EvoScientist.model_registry.platform import DelegationPublicKey from EvoScientist.model_registry.resolver import ModelRegistryResolver from EvoScientist.model_registry.snapshots import SnapshotService from EvoScientist.model_registry.store import ModelRuntimeStore SERVICE_TOKEN = "bff-service-token" _PRIVATE_KEY = ec.generate_private_key(ec.SECP256R1()) _PRIVATE_PEM = _PRIVATE_KEY.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ) _PUBLIC_PEM = _PRIVATE_KEY.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo, ) @pytest.fixture def config_path(tmp_path): path = tmp_path / "config.yaml" path.write_text("other_section: {keep: true}\n", encoding="utf-8") return path @pytest.fixture def services(tmp_path, config_path, monkeypatch): store = ModelRuntimeStore(config_dir=tmp_path / "runtime") monkeypatch.setattr( http_api, "load_image_generation_settings", lambda: image_config.load_image_generation_settings(config_path=config_path), ) monkeypatch.setattr( http_api, "save_image_generation_settings", lambda settings: image_config.save_image_generation_settings( settings, config_path=config_path ), ) resolver = ModelRegistryResolver(store) return ApiServices( store=store, resolver=resolver, snapshot_service=SnapshotService(store, resolver), endpoint_policy=EndpointPolicy([]), authenticator=BffAuthenticator( service_token=SERVICE_TOKEN, service_token_hash=None, delegation_keys=( DelegationPublicKey(deployment_id="webui-1", public_key=_PUBLIC_PEM), ), jti_store=store, ), ) @pytest.fixture def client(services): app = Starlette(routes=model_registry_routes(lambda: services)) return TestClient(app) def _headers(scopes): now = int(time.time()) claims = { "iss": "WebUI", "aud": "EvoScientist", "sub": "user-1", "scopes": list(scopes), "deployment_id": "webui-1", "iat": now, "exp": now + 30, "jti": uuid.uuid4().hex, } return { "Authorization": f"Bearer {SERVICE_TOKEN}", "X-Evo-Actor": jwt.encode(claims, _PRIVATE_PEM, algorithm="ES256"), } def _admin_headers(): return _headers(["model_config:read", "model_config:write"]) def _settings_payload(**overrides): payload = { "default_model": "gpt-image-2", "timeout_seconds": 120.0, "models": [ { "id": "gpt-image-2", "name": "GPT Image 2", "provider": "openai", "api_key": "sk-live-9876abcd", "base_url": "", "supports_generation": True, "supports_edit": True, "default_size": "1024x1024", "default_quality": "auto", "params": {}, } ], } payload.update(overrides) return payload def test_put_then_get_masks_api_key(client, config_path): put = client.put( "/api/image-generation", headers=_admin_headers(), json=_settings_payload() ) assert put.status_code == 200 body = put.json() model = body["models"][0] assert model["api_key"] == "" assert model["api_key_configured"] is True assert model["api_key_hint"] == "...abcd" assert "sk-live-9876abcd" not in json.dumps(body) get = client.get("/api/image-generation", headers=_admin_headers()) assert get.status_code == 200 assert get.json() == body # The plaintext key really landed in config.yaml, other sections intact. raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) assert raw["other_section"] == {"keep": True} assert raw["image_generation"]["models"][0]["api_key"] == "sk-live-9876abcd" def test_put_blank_api_key_keeps_existing(client): client.put( "/api/image-generation", headers=_admin_headers(), json=_settings_payload() ) update = _settings_payload() update["models"][0]["api_key"] = "" update["models"][0]["name"] = "Renamed" response = client.put( "/api/image-generation", headers=_admin_headers(), json=update ) assert response.status_code == 200 model = response.json()["models"][0] assert model["name"] == "Renamed" assert model["api_key_configured"] is True assert model["api_key_hint"] == "...abcd" def test_put_new_model_without_key_is_unconfigured(client): payload = _settings_payload(models=[_settings_payload()["models"][0] | {"api_key": ""}]) response = client.put( "/api/image-generation", headers=_admin_headers(), json=payload ) assert response.status_code == 200 model = response.json()["models"][0] assert model["api_key_configured"] is False assert model["api_key_hint"] is None def test_short_key_never_leaks_through_hint(client): payload = _settings_payload() payload["models"][0]["api_key"] = "abc" response = client.put( "/api/image-generation", headers=_admin_headers(), json=payload ) assert response.status_code == 200 model = response.json()["models"][0] assert model["api_key_configured"] is True assert model["api_key_hint"] is None def test_get_empty_when_unconfigured(client): response = client.get("/api/image-generation", headers=_admin_headers()) assert response.status_code == 200 assert response.json() == { "default_model": "", "timeout_seconds": 120.0, "models": [], } def test_put_rejects_invalid_payload(client): payload = _settings_payload() payload["models"][0]["provider"] = "midjourney" response = client.put( "/api/image-generation", headers=_admin_headers(), json=payload ) assert response.status_code == 422 body = response.json() assert body["code"] == "VALIDATION_FAILED" assert "midjourney" not in json.dumps(body) def test_read_scope_cannot_write(client): response = client.put( "/api/image-generation", headers=_headers(["model_config:read"]), json=_settings_payload(), ) assert response.status_code == 403 def test_get_corrupt_section_returns_422_envelope(client, config_path): config_path.write_text( "image_generation:\n models:\n - id: gpt-image-2\n provider: midjourney\n", encoding="utf-8", ) response = client.get("/api/image-generation", headers=_admin_headers()) assert response.status_code == 422 body = response.json() assert body["code"] == "VALIDATION_FAILED" assert "midjourney" not in json.dumps(body)