m4
961 lines
32 KiB
Python
961 lines
32 KiB
Python
"""HTTP API contract tests for the model registry (design doc 9.1-9.3, 9.5).
|
|
|
|
Covers the Config API (GET/PUT registry, credential rotation, selector),
|
|
the eight section 9.2 atomic save-time checks, snapshot creation/binding/
|
|
deletion state machines, the unified error payload, and the OpenAPI export.
|
|
Authentication itself is covered exhaustively in test_delegation_auth.py;
|
|
these tests mint valid delegations per request.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import jwt
|
|
import pytest
|
|
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.model_registry.adapters import find_adapter_spec
|
|
from EvoScientist.model_registry.auth import BffAuthenticator
|
|
from EvoScientist.model_registry.endpoint_policy import EndpointPolicy
|
|
from EvoScientist.model_registry.hashing import configuration_hash
|
|
from EvoScientist.model_registry.http_api import (
|
|
ApiServices,
|
|
build_openapi_document,
|
|
model_registry_routes,
|
|
)
|
|
from EvoScientist.model_registry.platform import (
|
|
DelegationPublicKey,
|
|
PlatformConfigError,
|
|
)
|
|
from EvoScientist.model_registry.resolver import ModelRegistryResolver
|
|
from EvoScientist.model_registry.schemas import (
|
|
CredentialWrite,
|
|
DevelopmentEndpoint,
|
|
RegistryV4,
|
|
)
|
|
from EvoScientist.model_registry.snapshots import SnapshotService
|
|
from EvoScientist.model_registry.store import ModelRuntimeStore
|
|
|
|
SERVICE_TOKEN = "bff-service-token"
|
|
SECRET = "sk-live-9876abcd"
|
|
|
|
_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,
|
|
)
|
|
|
|
OPENAPI_PATH = (
|
|
Path(__file__).resolve().parent.parent
|
|
/ "EvoScientist"
|
|
/ "model_registry"
|
|
/ "openapi.json"
|
|
)
|
|
|
|
|
|
# --- registry fixtures (mirrors test_snapshots.py) ---------------------------
|
|
|
|
|
|
def _model_runtime(**overrides):
|
|
payload = {
|
|
"limit_mode": "combined",
|
|
"context_window_tokens": 1048576,
|
|
"max_input_tokens": None,
|
|
"max_output_tokens": 32768,
|
|
"min_effective_input_tokens": 8192,
|
|
"fixed_system_reserve_tokens": 4096,
|
|
"fixed_tools_reserve_tokens": 8192,
|
|
"fixed_attachments_reserve_tokens": 4096,
|
|
"limits_status": "confirmed",
|
|
"limits_source": "provider",
|
|
"temperature": None,
|
|
"top_p": None,
|
|
"reasoning_effort": "auto",
|
|
"declared_capabilities": {
|
|
"tools": True,
|
|
"vision": False,
|
|
"structured_output": True,
|
|
},
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _zhipu_provider(**overrides):
|
|
provider = {
|
|
"id": "zhipu-glm",
|
|
"name": "Zhipu GLM",
|
|
"adapter": "openai-compatible",
|
|
"base_url": "https://open.bigmodel.cn/api/paas/v4",
|
|
"auth": {"mode": "api_key", "credential_id": "zhipu-primary"},
|
|
"enabled": True,
|
|
"runtime": {
|
|
"timeout_seconds": 120,
|
|
"max_retries": 2,
|
|
"default_temperature": 0.7,
|
|
"default_top_p": 0.95,
|
|
"default_reasoning_effort": "auto",
|
|
},
|
|
"models": [
|
|
{
|
|
"key": "glm-5.2",
|
|
"name": "GLM-5.2",
|
|
"upstream_model_id": "glm-5.2",
|
|
"enabled": True,
|
|
"runtime": _model_runtime(),
|
|
}
|
|
],
|
|
}
|
|
provider.update(overrides)
|
|
return provider
|
|
|
|
|
|
def _ollama_provider(**overrides):
|
|
provider = {
|
|
"id": "local-ollama",
|
|
"name": "Local Ollama",
|
|
"adapter": "ollama",
|
|
"base_url": "http://localhost:11434",
|
|
"auth": {"mode": "none", "credential_id": None},
|
|
"enabled": True,
|
|
"runtime": {"timeout_seconds": 120, "max_retries": 2},
|
|
"models": [
|
|
{
|
|
"key": "qwen3",
|
|
"name": "Qwen3",
|
|
"upstream_model_id": "qwen3",
|
|
"enabled": True,
|
|
"runtime": _model_runtime(),
|
|
}
|
|
],
|
|
}
|
|
provider.update(overrides)
|
|
return provider
|
|
|
|
|
|
def _registry_payload():
|
|
return {
|
|
"version": 4,
|
|
"revision": 1,
|
|
"state": "bootstrap",
|
|
"defaults": {
|
|
"primary": {"provider_id": "zhipu-glm", "model_key": "glm-5.2"},
|
|
},
|
|
"providers": [_zhipu_provider(), _ollama_provider()],
|
|
}
|
|
|
|
|
|
def _verify(store, registry, provider_id, model_key):
|
|
provider = registry.find_provider(provider_id)
|
|
model = provider.find_model(model_key)
|
|
spec = find_adapter_spec(provider.adapter, model.upstream_model_id)
|
|
store.record_model_verification(
|
|
provider_id=provider_id,
|
|
model_key=model_key,
|
|
configuration_hash=configuration_hash(provider, model),
|
|
credential_revision=1 if provider.auth.credential_id else 0,
|
|
adapter_spec_revision=spec.spec_revision,
|
|
result="passed",
|
|
verified_capabilities={
|
|
"tools": True,
|
|
"vision": False,
|
|
"structured_output": True,
|
|
},
|
|
)
|
|
|
|
|
|
# --- app fixtures --------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
return ModelRuntimeStore(config_dir=tmp_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def services(store):
|
|
resolver = ModelRegistryResolver(store)
|
|
return ApiServices(
|
|
store=store,
|
|
resolver=resolver,
|
|
snapshot_service=SnapshotService(store, resolver),
|
|
endpoint_policy=EndpointPolicy(
|
|
[
|
|
DevelopmentEndpoint(
|
|
id="local-ollama",
|
|
url="http://localhost:11434",
|
|
label="Local Ollama",
|
|
)
|
|
]
|
|
),
|
|
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)
|
|
|
|
|
|
@pytest.fixture
|
|
def active_store(store):
|
|
registry = store.save_registry(
|
|
expected_revision=1,
|
|
registry=RegistryV4.model_validate(_registry_payload()),
|
|
credential_writes=[
|
|
CredentialWrite(credential_id="zhipu-primary", secret_value=SECRET)
|
|
],
|
|
)
|
|
assert registry.state == "active"
|
|
_verify(store, registry, "zhipu-glm", "glm-5.2")
|
|
_verify(store, registry, "local-ollama", "qwen3")
|
|
return store
|
|
|
|
|
|
@pytest.fixture
|
|
def active_client(client, active_store):
|
|
return client
|
|
|
|
|
|
# --- auth helpers --------------------------------------------------------------
|
|
|
|
|
|
def _headers(scopes, *, thread_id=None):
|
|
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,
|
|
}
|
|
if thread_id is not None:
|
|
claims["thread_id"] = thread_id
|
|
return {
|
|
"Authorization": f"Bearer {SERVICE_TOKEN}",
|
|
"X-Evo-Actor": jwt.encode(claims, _PRIVATE_PEM, algorithm="ES256"),
|
|
}
|
|
|
|
|
|
def _admin_headers(**kwargs):
|
|
return _headers(["model_config:read", "model_config:write"], **kwargs)
|
|
|
|
|
|
def _run_headers(thread_id="thread-1"):
|
|
return _headers(["run:create"], thread_id=thread_id)
|
|
|
|
|
|
# --- GET /api/model-registry ----------------------------------------------------
|
|
|
|
|
|
def test_get_registry_response_structure(active_client):
|
|
response = active_client.get("/api/model-registry", headers=_admin_headers())
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert set(body) == {
|
|
"revision",
|
|
"registry",
|
|
"adapter_specs",
|
|
"credential_status",
|
|
"model_status",
|
|
"endpoint_policy",
|
|
}
|
|
assert body["revision"] == 2
|
|
assert body["registry"]["state"] == "active"
|
|
assert len(body["adapter_specs"]) == 6
|
|
status = body["credential_status"]
|
|
assert [entry["credential_id"] for entry in status] == ["zhipu-primary"]
|
|
assert status[0]["configured"] is True
|
|
assert status[0]["hint"] == "...abcd"
|
|
assert status[0]["updated_at"]
|
|
assert SECRET not in response.text
|
|
states = {item["model_ref"]["model_key"]: item for item in body["model_status"]}
|
|
assert states["glm-5.2"]["state"] == "enabled"
|
|
assert states["glm-5.2"]["selectable"] is True
|
|
assert states["qwen3"]["state"] == "enabled"
|
|
policy = body["endpoint_policy"]
|
|
assert policy["public_https_allowed"] is True
|
|
assert policy["development_endpoints"] == [
|
|
{"id": "local-ollama", "url": "http://localhost:11434", "label": "Local Ollama"}
|
|
]
|
|
|
|
|
|
def test_get_registry_marks_model_stale_after_credential_rotation(active_client):
|
|
rotate = active_client.put(
|
|
"/api/model-registry/credentials/zhipu-primary",
|
|
headers=_admin_headers(),
|
|
json={"operation": "replace", "secret_value": "rotated-secret-9999"},
|
|
)
|
|
assert rotate.status_code == 200
|
|
|
|
body = active_client.get("/api/model-registry", headers=_admin_headers()).json()
|
|
states = {item["model_ref"]["model_key"]: item for item in body["model_status"]}
|
|
# The full credential_revisions mapping must reach compute_availability:
|
|
# after rotation the glm model compares against revision 2 and goes stale.
|
|
assert states["glm-5.2"]["state"] == "verification_stale"
|
|
assert states["glm-5.2"]["selectable"] is False
|
|
assert states["glm-5.2"]["verification"]["status"] == "stale"
|
|
# mode=none models are unaffected by credential rotation.
|
|
assert states["qwen3"]["state"] == "enabled"
|
|
|
|
|
|
# --- PUT /api/model-registry -----------------------------------------------------
|
|
|
|
|
|
def _current_registry(active_client):
|
|
return active_client.get("/api/model-registry", headers=_admin_headers()).json()
|
|
|
|
|
|
def _put(active_client, registry, *, expected_revision=2, credential_writes=None):
|
|
payload = {
|
|
"expected_revision": expected_revision,
|
|
"registry": registry,
|
|
"credential_writes": credential_writes or [],
|
|
}
|
|
return active_client.put(
|
|
"/api/model-registry", headers=_admin_headers(), json=payload
|
|
)
|
|
|
|
|
|
def test_put_registry_success_returns_full_response(active_client):
|
|
current = _current_registry(active_client)
|
|
registry = current["registry"]
|
|
registry["providers"][0]["models"][0]["name"] = "GLM-5.2 Turbo"
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["revision"] == current["revision"] + 1
|
|
assert body["registry"]["providers"][0]["models"][0]["name"] == "GLM-5.2 Turbo"
|
|
states = {item["model_ref"]["model_key"]: item for item in body["model_status"]}
|
|
# The model name is not part of the configuration hash: still enabled.
|
|
assert states["glm-5.2"]["state"] == "enabled"
|
|
|
|
|
|
def test_put_registry_stale_revision_is_ignored(active_client):
|
|
current = _current_registry(active_client)
|
|
response = _put(active_client, current["registry"], expected_revision=99)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["revision"] == current["revision"] + 1
|
|
|
|
|
|
def test_put_registry_without_expected_revision(active_client):
|
|
current = _current_registry(active_client)
|
|
payload = {"registry": current["registry"], "credential_writes": []}
|
|
response = active_client.put(
|
|
"/api/model-registry", headers=_admin_headers(), json=payload
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["revision"] == current["revision"] + 1
|
|
|
|
|
|
def test_put_registry_rejects_ssrf_endpoint(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["base_url"] = "http://127.0.0.1:9000/v1"
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["code"] == "ENDPOINT_NOT_ALLOWED"
|
|
assert body["details"][0]["path"].startswith("providers[0].base_url")
|
|
|
|
|
|
def test_put_registry_rejects_image_only_model(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["models"][0]["upstream_model_id"] = "gpt-image-2"
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["code"] == "IMAGE_MODEL_NOT_CHAT_MODEL"
|
|
assert body["details"][0]["path"].endswith("upstream_model_id")
|
|
|
|
|
|
def test_put_registry_rejects_unsupported_parameter(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
# The glm-5.2 contract caps temperature at 1.
|
|
registry["providers"][0]["models"][0]["runtime"]["temperature"] = 1.5
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["code"] == "UNSUPPORTED_RUNTIME_PARAMETER"
|
|
assert "providers[0].models[0]" in body["details"][0]["path"]
|
|
|
|
|
|
def test_put_registry_rejects_unconfirmed_limits(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["models"][0]["runtime"]["limits_status"] = (
|
|
"needs_confirmation"
|
|
)
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "MODEL_LIMITS_UNCONFIRMED"
|
|
|
|
|
|
def test_put_registry_rejects_enabled_model_without_verification(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["models"].append(
|
|
{
|
|
"key": "glm-5.3",
|
|
"name": "GLM-5.3",
|
|
"upstream_model_id": "glm-5.3",
|
|
"enabled": True,
|
|
"runtime": _model_runtime(),
|
|
}
|
|
)
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "MODEL_NOT_AVAILABLE"
|
|
|
|
|
|
def test_put_registry_rejects_unsatisfiable_budget(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
# A disabled draft model still must satisfy the four-mode budget rule.
|
|
registry["providers"][0]["models"].append(
|
|
{
|
|
"key": "glm-draft",
|
|
"name": "GLM Draft",
|
|
"upstream_model_id": "glm-draft",
|
|
"enabled": False,
|
|
"runtime": _model_runtime(min_effective_input_tokens=1048576),
|
|
}
|
|
)
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "CONTEXT_BUDGET_UNSATISFIABLE"
|
|
|
|
|
|
def test_put_registry_rejects_unsupported_auth_mode(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][1]["auth"] = {"mode": "api_key", "credential_id": "some-key"}
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "AUTH_MODE_UNSUPPORTED"
|
|
|
|
|
|
def test_put_registry_rejects_capability_unsupported_by_adapter(active_client):
|
|
# Rule 8: declared capabilities must not exceed the adapter contract's
|
|
# protocol capabilities — the glm-5.2 contract declares vision=False.
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["models"][0]["runtime"]["declared_capabilities"][
|
|
"vision"
|
|
] = True
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["code"] == "CAPABILITY_UNSUPPORTED_BY_ADAPTER"
|
|
assert "providers[0].models[0]" in body["details"][0]["path"]
|
|
|
|
|
|
def test_put_registry_rejects_missing_credential_reference(active_client):
|
|
# The api_key AuthSpec requires a credential: credential_id=None must
|
|
# fail with CREDENTIAL_NOT_CONFIGURED, not a verification-tuple miss.
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["auth"] = {"mode": "api_key", "credential_id": None}
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["code"] == "CREDENTIAL_NOT_CONFIGURED"
|
|
assert "auth.credential_id" in body["details"][0]["path"]
|
|
|
|
|
|
def test_put_registry_rejects_defaults_to_disabled_model(active_client):
|
|
# With a bootstrap-state payload the schema-level check does not fire, so
|
|
# the store's rule-7 guard (enforced on every save) is what rejects it.
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["state"] = "bootstrap"
|
|
registry["providers"][0]["models"][0]["enabled"] = False
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "MODEL_DISABLED"
|
|
|
|
|
|
def test_put_registry_active_state_validates_defaults_in_schema(active_client):
|
|
# The same rule fires earlier (Pydantic) when the payload claims active.
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["models"][0]["enabled"] = False
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "VALIDATION_FAILED"
|
|
|
|
|
|
def test_put_registry_atomic_rollback_with_credential_writes(active_client, store):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["base_url"] = "http://127.0.0.1:9000/v1"
|
|
|
|
response = _put(
|
|
active_client,
|
|
registry,
|
|
credential_writes=[
|
|
{
|
|
"credential_id": "zhipu-primary",
|
|
"operation": "replace",
|
|
"secret_value": "should-not-persist",
|
|
}
|
|
],
|
|
)
|
|
assert response.status_code == 422
|
|
# Neither the credential write nor the registry update may survive.
|
|
assert store.current_credential_revision("zhipu-primary") == 1
|
|
assert store.load_registry().revision == 2
|
|
|
|
|
|
def test_put_registry_schema_validation_details(active_client):
|
|
registry = _current_registry(active_client)["registry"]
|
|
registry["providers"][0]["id"] = "INVALID ID!"
|
|
|
|
response = _put(active_client, registry)
|
|
assert response.status_code == 422
|
|
body = response.json()
|
|
assert body["code"] == "VALIDATION_FAILED"
|
|
assert body["details"]
|
|
assert all("path" in detail and "code" in detail for detail in body["details"])
|
|
|
|
|
|
def test_put_registry_bootstrap_flow_configure_test_enable(client, store):
|
|
# Step 1: save a draft registry — models disabled, defaults null, and the
|
|
# credential written in the same transaction (section 10 steps 5-7).
|
|
draft = _registry_payload()
|
|
draft["defaults"] = {"primary": None}
|
|
for provider in draft["providers"]:
|
|
for model in provider["models"]:
|
|
model["enabled"] = False
|
|
response = client.put(
|
|
"/api/model-registry",
|
|
headers=_admin_headers(),
|
|
json={
|
|
"expected_revision": 1,
|
|
"registry": draft,
|
|
"credential_writes": [
|
|
{
|
|
"credential_id": "zhipu-primary",
|
|
"operation": "replace",
|
|
"secret_value": "bootstrap-secret",
|
|
}
|
|
],
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["registry"]["state"] == "bootstrap"
|
|
assert body["revision"] == 2
|
|
assert "bootstrap-secret" not in response.text
|
|
states = {item["model_ref"]["model_key"]: item for item in body["model_status"]}
|
|
assert states["glm-5.2"]["state"] == "configured"
|
|
|
|
# Step 2: provider tests pass (Task 5b writes these records via the test
|
|
# endpoint; here they are seeded directly).
|
|
registry = store.load_registry()
|
|
_verify(store, registry, "zhipu-glm", "glm-5.2")
|
|
_verify(store, registry, "local-ollama", "qwen3")
|
|
|
|
# Step 3: enabling the verified models with defaults turns the registry
|
|
# active in the same commit.
|
|
enabled = body["registry"]
|
|
enabled["defaults"] = {
|
|
"primary": {"provider_id": "zhipu-glm", "model_key": "glm-5.2"},
|
|
}
|
|
for provider in enabled["providers"]:
|
|
for model in provider["models"]:
|
|
model["enabled"] = True
|
|
response = client.put(
|
|
"/api/model-registry",
|
|
headers=_admin_headers(),
|
|
json={"expected_revision": 2, "registry": enabled, "credential_writes": []},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["registry"]["state"] == "active"
|
|
states = {item["model_ref"]["model_key"]: item for item in body["model_status"]}
|
|
assert states["glm-5.2"]["state"] == "enabled"
|
|
|
|
|
|
def test_put_registry_malformed_json_is_400(active_client):
|
|
response = active_client.put(
|
|
"/api/model-registry",
|
|
headers={**_admin_headers(), "Content-Type": "application/json"},
|
|
content="{not json",
|
|
)
|
|
assert response.status_code == 400
|
|
assert response.json()["code"] == "INVALID_REQUEST"
|
|
|
|
|
|
# --- PUT /api/model-registry/credentials/{credential_id} ------------------------
|
|
|
|
|
|
def test_credential_rotation_creates_new_masked_revision(active_client):
|
|
response = active_client.put(
|
|
"/api/model-registry/credentials/zhipu-primary",
|
|
headers=_admin_headers(),
|
|
json={"operation": "replace", "secret_value": "rotated-secret-9999"},
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["credential_id"] == "zhipu-primary"
|
|
assert body["revision"] == 2
|
|
assert body["configured"] is True
|
|
assert body["hint"] == "...9999"
|
|
assert body["updated_at"]
|
|
assert "rotated-secret-9999" not in response.text
|
|
|
|
|
|
def test_credential_rotation_rejects_invalid_credential_id(active_client):
|
|
response = active_client.put(
|
|
"/api/model-registry/credentials/INVALID!!",
|
|
headers=_admin_headers(),
|
|
json={"operation": "replace", "secret_value": "whatever"},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_credential_rotation_requires_write_scope(active_client):
|
|
response = active_client.put(
|
|
"/api/model-registry/credentials/zhipu-primary",
|
|
headers=_headers(["model_config:read"]),
|
|
json={"operation": "replace", "secret_value": "whatever"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
# --- GET /api/models -------------------------------------------------------------
|
|
|
|
|
|
def test_get_models_returns_only_selectable(active_client, store):
|
|
response = active_client.get("/api/models", headers=_headers(["model:select"]))
|
|
assert response.status_code == 200
|
|
models = response.json()["models"]
|
|
refs = {
|
|
(m["model_ref"]["provider_id"], m["model_ref"]["model_key"]) for m in models
|
|
}
|
|
assert refs == {("zhipu-glm", "glm-5.2"), ("local-ollama", "qwen3")}
|
|
glm = next(m for m in models if m["model_ref"]["model_key"] == "glm-5.2")
|
|
assert glm["name"] == "GLM-5.2"
|
|
assert glm["provider_name"] == "Zhipu GLM"
|
|
assert glm["effective_capabilities"] == {
|
|
"tools": True,
|
|
"vision": False,
|
|
"structured_output": True,
|
|
}
|
|
# The selector echoes the registry defaults so clients can label the
|
|
# `inherit` choice before the first turn.
|
|
assert response.json()["defaults"]["primary"] == {
|
|
"provider_id": "zhipu-glm",
|
|
"model_key": "glm-5.2",
|
|
}
|
|
|
|
# Disable qwen3: it leaves the selector.
|
|
registry = store.load_registry()
|
|
payload = registry.model_dump(mode="json")
|
|
payload["providers"][1]["models"][0]["enabled"] = False
|
|
store.save_registry(
|
|
expected_revision=registry.revision,
|
|
registry=RegistryV4.model_validate(payload),
|
|
)
|
|
models = active_client.get(
|
|
"/api/models", headers=_headers(["model:select"])
|
|
).json()["models"]
|
|
assert [m["model_ref"]["model_key"] for m in models] == ["glm-5.2"]
|
|
|
|
|
|
def test_get_models_requires_select_scope(active_client):
|
|
response = active_client.get("/api/models", headers=_headers(["run:create"]))
|
|
assert response.status_code == 403
|
|
|
|
|
|
# --- snapshot API -----------------------------------------------------------------
|
|
|
|
|
|
def _snapshot_body(**overrides):
|
|
body = {
|
|
"run_request_id": "req-1",
|
|
"thread_id": "thread-1",
|
|
"deployment_id": "webui-1",
|
|
"model_selection_revision": 4,
|
|
"primary": None,
|
|
}
|
|
body.update(overrides)
|
|
return body
|
|
|
|
|
|
def test_snapshot_create_201_and_idempotent_200(active_client):
|
|
first = active_client.post(
|
|
"/api/runtime-snapshots", headers=_run_headers(), json=_snapshot_body()
|
|
)
|
|
assert first.status_code == 201
|
|
body = first.json()
|
|
assert body["snapshot_id"]
|
|
assert body["registry_revision"] == 2
|
|
assert body["primary"]["provider_id"] == "zhipu-glm"
|
|
assert body["primary"]["adapter_spec_revision"] == 1
|
|
assert body["primary"]["runtime"]["max_output_tokens"] == 32768
|
|
assert SECRET not in first.text
|
|
|
|
replay = active_client.post(
|
|
"/api/runtime-snapshots", headers=_run_headers(), json=_snapshot_body()
|
|
)
|
|
assert replay.status_code == 200
|
|
assert replay.json()["snapshot_id"] == body["snapshot_id"]
|
|
|
|
|
|
def test_snapshot_create_conflicting_selection_is_409(active_client):
|
|
active_client.post(
|
|
"/api/runtime-snapshots", headers=_run_headers(), json=_snapshot_body()
|
|
)
|
|
conflict = active_client.post(
|
|
"/api/runtime-snapshots",
|
|
headers=_run_headers(),
|
|
json=_snapshot_body(
|
|
primary={"provider_id": "zhipu-glm", "model_key": "glm-5.2"}
|
|
),
|
|
)
|
|
assert conflict.status_code == 409
|
|
assert conflict.json()["code"] == "RUN_REQUEST_CONFLICT"
|
|
|
|
|
|
def test_snapshot_create_requires_active_registry(client):
|
|
response = client.post(
|
|
"/api/runtime-snapshots", headers=_run_headers(), json=_snapshot_body()
|
|
)
|
|
assert response.status_code == 422
|
|
assert response.json()["code"] == "MODEL_REGISTRY_NOT_READY"
|
|
|
|
|
|
def test_snapshot_create_thread_mismatch_is_403(active_client):
|
|
response = active_client.post(
|
|
"/api/runtime-snapshots",
|
|
headers=_run_headers("thread-1"),
|
|
json=_snapshot_body(thread_id="thread-2"),
|
|
)
|
|
assert response.status_code == 403
|
|
assert response.json()["code"] == "FORBIDDEN"
|
|
|
|
|
|
def test_snapshot_create_requires_thread_claim(active_client):
|
|
headers = _headers(["run:create"]) # no thread_id claim
|
|
response = active_client.post(
|
|
"/api/runtime-snapshots", headers=headers, json=_snapshot_body()
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_snapshot_create_unknown_model_is_404(active_client):
|
|
response = active_client.post(
|
|
"/api/runtime-snapshots",
|
|
headers=_run_headers(),
|
|
json=_snapshot_body(primary={"provider_id": "zhipu-glm", "model_key": "nope"}),
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.json()["code"] == "MODEL_NOT_FOUND"
|
|
|
|
|
|
def _create_snapshot(active_client, **overrides):
|
|
response = active_client.post(
|
|
"/api/runtime-snapshots",
|
|
headers=_run_headers(),
|
|
json=_snapshot_body(**overrides),
|
|
)
|
|
assert response.status_code == 201
|
|
return response.json()["snapshot_id"]
|
|
|
|
|
|
def test_snapshot_bind_state_machine(active_client):
|
|
snapshot_id = _create_snapshot(active_client)
|
|
bound = active_client.post(
|
|
f"/api/runtime-snapshots/{snapshot_id}/bind",
|
|
headers=_run_headers(),
|
|
json={"langgraph_run_id": "run-1"},
|
|
)
|
|
assert bound.status_code == 200
|
|
assert bound.json()["snapshot_id"] == snapshot_id
|
|
|
|
again = active_client.post(
|
|
f"/api/runtime-snapshots/{snapshot_id}/bind",
|
|
headers=_run_headers(),
|
|
json={"langgraph_run_id": "run-1"},
|
|
)
|
|
assert again.status_code == 200
|
|
|
|
conflict = active_client.post(
|
|
f"/api/runtime-snapshots/{snapshot_id}/bind",
|
|
headers=_run_headers(),
|
|
json={"langgraph_run_id": "run-2"},
|
|
)
|
|
assert conflict.status_code == 409
|
|
assert conflict.json()["code"] == "SNAPSHOT_ALREADY_BOUND"
|
|
|
|
|
|
def test_snapshot_bind_unknown_and_cross_thread_are_404(active_client):
|
|
snapshot_id = _create_snapshot(active_client)
|
|
missing = active_client.post(
|
|
"/api/runtime-snapshots/snap-nope/bind",
|
|
headers=_run_headers(),
|
|
json={"langgraph_run_id": "run-1"},
|
|
)
|
|
assert missing.status_code == 404
|
|
assert missing.json()["code"] == "SNAPSHOT_NOT_FOUND"
|
|
|
|
cross_thread = active_client.post(
|
|
f"/api/runtime-snapshots/{snapshot_id}/bind",
|
|
headers=_run_headers("thread-2"),
|
|
json={"langgraph_run_id": "run-1"},
|
|
)
|
|
assert cross_thread.status_code == 404
|
|
assert cross_thread.json()["code"] == "SNAPSHOT_NOT_FOUND"
|
|
|
|
|
|
def test_snapshot_delete_state_machine(active_client):
|
|
snapshot_id = _create_snapshot(active_client)
|
|
deleted = active_client.delete(
|
|
f"/api/runtime-snapshots/{snapshot_id}", headers=_run_headers()
|
|
)
|
|
assert deleted.status_code == 204
|
|
assert deleted.content == b""
|
|
|
|
bound_id = _create_snapshot(active_client, run_request_id="req-2")
|
|
active_client.post(
|
|
f"/api/runtime-snapshots/{bound_id}/bind",
|
|
headers=_run_headers(),
|
|
json={"langgraph_run_id": "run-1"},
|
|
)
|
|
delete_bound = active_client.delete(
|
|
f"/api/runtime-snapshots/{bound_id}", headers=_run_headers()
|
|
)
|
|
assert delete_bound.status_code == 409
|
|
assert delete_bound.json()["code"] == "SNAPSHOT_ALREADY_BOUND"
|
|
|
|
|
|
def test_snapshot_expired_rejects_bind_and_delete(active_client, services):
|
|
snapshot_id = _create_snapshot(active_client)
|
|
row = services.store.get_run_snapshot(snapshot_id)
|
|
services.store.insert_run_snapshot(
|
|
snapshot_id="snap-expired",
|
|
deployment_id="webui-1",
|
|
thread_id="thread-1",
|
|
run_request_id="req-expired",
|
|
selection_hash=row["selection_hash"],
|
|
payload=row["payload"],
|
|
expires_at=int(time.time()) - 1,
|
|
)
|
|
services.snapshot_service.cleanup_expired()
|
|
|
|
bind = active_client.post(
|
|
"/api/runtime-snapshots/snap-expired/bind",
|
|
headers=_run_headers(),
|
|
json={"langgraph_run_id": "run-1"},
|
|
)
|
|
assert bind.status_code == 409
|
|
assert bind.json()["code"] == "SNAPSHOT_EXPIRED"
|
|
|
|
delete = active_client.delete(
|
|
"/api/runtime-snapshots/snap-expired", headers=_run_headers()
|
|
)
|
|
assert delete.status_code == 409
|
|
assert delete.json()["code"] == "SNAPSHOT_EXPIRED"
|
|
|
|
# The expired triplet is free again: recreation returns 201.
|
|
recreated = active_client.post(
|
|
"/api/runtime-snapshots",
|
|
headers=_run_headers(),
|
|
json=_snapshot_body(run_request_id="req-expired"),
|
|
)
|
|
assert recreated.status_code == 201
|
|
|
|
|
|
def test_snapshot_error_payload_structure(active_client):
|
|
response = active_client.post(
|
|
"/api/runtime-snapshots",
|
|
headers=_run_headers(),
|
|
json=_snapshot_body(primary={"provider_id": "zhipu-glm", "model_key": "nope"}),
|
|
)
|
|
body = response.json()
|
|
assert set(body) == {"code", "message", "details", "request_id"}
|
|
assert isinstance(body["details"], list)
|
|
assert body["request_id"]
|
|
|
|
|
|
# --- platform configuration ------------------------------------------------
|
|
|
|
|
|
def test_missing_platform_config_refuses_service():
|
|
def _unconfigured():
|
|
raise PlatformConfigError("missing bff_service_token")
|
|
|
|
app = Starlette(routes=model_registry_routes(_unconfigured))
|
|
response = TestClient(app).get("/api/model-registry")
|
|
assert response.status_code == 500
|
|
body = response.json()
|
|
assert body["code"] == "PLATFORM_CONFIG_MISSING"
|
|
assert set(body) == {"code", "message", "details", "request_id"}
|
|
|
|
|
|
# --- OpenAPI export -----------------------------------------------------------------
|
|
|
|
|
|
def test_openapi_export_file_matches_pydantic_schemas():
|
|
assert OPENAPI_PATH.exists()
|
|
exported = json.loads(OPENAPI_PATH.read_text(encoding="utf-8"))
|
|
# The checked-in file is exactly what the export script regenerates.
|
|
assert exported == build_openapi_document()
|
|
|
|
for path in (
|
|
"/api/model-registry",
|
|
"/api/model-registry/credentials/{credential_id}",
|
|
"/api/models",
|
|
"/api/runtime-snapshots",
|
|
"/api/runtime-snapshots/{snapshot_id}/bind",
|
|
"/api/runtime-snapshots/{snapshot_id}",
|
|
):
|
|
assert path in exported["paths"]
|
|
|
|
schema = exported["components"]["schemas"]["GetModelRegistryResponse"]
|
|
assert set(schema["properties"]) == {
|
|
"revision",
|
|
"registry",
|
|
"adapter_specs",
|
|
"credential_status",
|
|
"model_status",
|
|
"endpoint_policy",
|
|
}
|
|
registry_schema = exported["components"]["schemas"]["RegistryV4"]
|
|
assert "providers" in registry_schema["properties"]
|
|
error_schema = exported["components"]["schemas"]["ErrorPayload"]
|
|
assert set(error_schema["properties"]) == {
|
|
"code",
|
|
"message",
|
|
"details",
|
|
"request_id",
|
|
}
|