"""Tests for the inference registry API. Covers: - CRUD operations for endpoints, deployments, bindings - auth_secret_ref NEVER appears in any response body - Probe action returns structured results - Enable/disable toggles - External egress requires confirmation - Protocol validation rejects unknown protocols - Endpoint creation validates URL format Requirements: 3.6, 3.7 """ from __future__ import annotations import uuid from datetime import datetime, timezone from typing import Any import pytest import pytest_asyncio from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from services.inference_registry.router import ( InferenceRegistryDB, router, set_db, ) # --------------------------------------------------------------------------- # Mock DB implementation # --------------------------------------------------------------------------- class MockInferenceDB(InferenceRegistryDB): """In-memory mock implementation of the registry database.""" def __init__(self) -> None: self.endpoints: dict[uuid.UUID, dict[str, Any]] = {} self.deployments: dict[uuid.UUID, dict[str, Any]] = {} self.bindings: dict[uuid.UUID, dict[str, Any]] = {} self.probes: dict[uuid.UUID, dict[str, Any]] = {} self.egress_confirmations: set[uuid.UUID] = set() async def list_endpoints(self) -> list[dict[str, Any]]: return list(self.endpoints.values()) async def get_endpoint(self, endpoint_id: uuid.UUID) -> dict[str, Any] | None: return self.endpoints.get(endpoint_id) async def create_endpoint(self, data: dict[str, Any]) -> dict[str, Any]: self.endpoints[data["id"]] = data return data async def update_endpoint( self, endpoint_id: uuid.UUID, data: dict[str, Any] ) -> dict[str, Any] | None: if endpoint_id not in self.endpoints: return None self.endpoints[endpoint_id].update(data) return self.endpoints[endpoint_id] async def disable_endpoint(self, endpoint_id: uuid.UUID) -> dict[str, Any] | None: if endpoint_id not in self.endpoints: return None self.endpoints[endpoint_id]["enabled"] = False self.endpoints[endpoint_id]["updated_at"] = datetime.now(timezone.utc) return self.endpoints[endpoint_id] async def list_deployments( self, endpoint_id: uuid.UUID | None = None ) -> list[dict[str, Any]]: if endpoint_id: return [ d for d in self.deployments.values() if d["endpoint_id"] == endpoint_id ] return list(self.deployments.values()) async def get_deployment(self, deployment_id: uuid.UUID) -> dict[str, Any] | None: return self.deployments.get(deployment_id) async def create_deployment(self, data: dict[str, Any]) -> dict[str, Any]: self.deployments[data["id"]] = data return data async def list_bindings( self, agent_id: uuid.UUID | None = None, endpoint_id: uuid.UUID | None = None, ) -> list[dict[str, Any]]: result = list(self.bindings.values()) if agent_id: result = [b for b in result if b["agent_id"] == agent_id] return result async def create_binding(self, data: dict[str, Any]) -> dict[str, Any]: self.bindings[data["id"]] = data return data async def get_bindings_for_endpoint( self, endpoint_id: uuid.UUID ) -> list[dict[str, Any]]: # Find bindings whose deployment is on this endpoint dep_ids = { d["id"] for d in self.deployments.values() if d["endpoint_id"] == endpoint_id } return [ b for b in self.bindings.values() if b.get("model_deployment_id") in dep_ids ] async def get_last_probe( self, endpoint_id: uuid.UUID ) -> dict[str, Any] | None: return self.probes.get(endpoint_id) async def store_probe_result( self, endpoint_id: uuid.UUID, result: dict[str, Any] ) -> None: self.probes[endpoint_id] = result async def get_egress_confirmation(self, endpoint_id: uuid.UUID) -> bool: return endpoint_id in self.egress_confirmations async def store_egress_confirmation(self, endpoint_id: uuid.UUID) -> None: self.egress_confirmations.add(endpoint_id) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def mock_db() -> MockInferenceDB: return MockInferenceDB() @pytest.fixture def app(mock_db: MockInferenceDB) -> FastAPI: test_app = FastAPI() test_app.include_router(router) set_db(mock_db) return test_app @pytest_asyncio.fixture async def client(app: FastAPI) -> AsyncClient: transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as c: yield c def _make_endpoint_payload( name: str = "test-endpoint", protocol: str = "openai_chat", base_url: str = "http://localhost:8000", auth_secret_ref: str | None = "VLLM_API_KEY", ) -> dict[str, Any]: """Helper to build a valid endpoint creation payload.""" return { "name": name, "protocol": protocol, "base_url": base_url, "auth_secret_ref": auth_secret_ref, } # --------------------------------------------------------------------------- # Test: CRUD endpoints (19.1) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_create_endpoint(client: AsyncClient): """Creating an endpoint returns 201 with redacted secrets.""" payload = _make_endpoint_payload() resp = await client.post("/api/inference/endpoints", json=payload) assert resp.status_code == 201 data = resp.json() assert data["name"] == "test-endpoint" assert data["protocol"] == "openai_chat" assert data["base_url"] == "http://localhost:8000" assert data["auth_secret_status"] == "configured" # CRITICAL: auth_secret_ref must NEVER appear in response assert "auth_secret_ref" not in data assert "VLLM_API_KEY" not in str(data) @pytest.mark.asyncio async def test_create_endpoint_no_secret(client: AsyncClient): """Creating an endpoint without a secret shows not_configured.""" payload = _make_endpoint_payload(auth_secret_ref=None) resp = await client.post("/api/inference/endpoints", json=payload) assert resp.status_code == 201 data = resp.json() assert data["auth_secret_status"] == "not_configured" @pytest.mark.asyncio async def test_list_endpoints(client: AsyncClient): """Listing endpoints returns all with redacted secrets.""" await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(name="ep-1"), ) await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(name="ep-2", auth_secret_ref="SECRET_KEY"), ) resp = await client.get("/api/inference/endpoints") assert resp.status_code == 200 data = resp.json() assert len(data) == 2 for ep in data: assert "auth_secret_ref" not in ep assert "SECRET_KEY" not in str(ep) assert "VLLM_API_KEY" not in str(ep) @pytest.mark.asyncio async def test_get_endpoint_detail(client: AsyncClient): """Getting an endpoint by ID returns detail with redacted secrets.""" create_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(auth_secret_ref="MY_SECRET"), ) ep_id = create_resp.json()["id"] resp = await client.get(f"/api/inference/endpoints/{ep_id}") assert resp.status_code == 200 data = resp.json() assert data["auth_secret_status"] == "configured" assert "MY_SECRET" not in str(data) assert "auth_secret_ref" not in data @pytest.mark.asyncio async def test_get_endpoint_not_found(client: AsyncClient): """Getting a nonexistent endpoint returns 404.""" fake_id = str(uuid.uuid4()) resp = await client.get(f"/api/inference/endpoints/{fake_id}") assert resp.status_code == 404 @pytest.mark.asyncio async def test_update_endpoint(client: AsyncClient): """Updating an endpoint works and still redacts secrets.""" create_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = create_resp.json()["id"] resp = await client.put( f"/api/inference/endpoints/{ep_id}", json={"name": "updated-endpoint", "base_url": "http://new-host:9000"}, ) assert resp.status_code == 200 data = resp.json() assert data["name"] == "updated-endpoint" assert data["base_url"] == "http://new-host:9000" assert data["revision"] == 2 assert "auth_secret_ref" not in data @pytest.mark.asyncio async def test_delete_endpoint_soft_disables(client: AsyncClient): """Deleting an endpoint soft-disables it.""" create_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = create_resp.json()["id"] resp = await client.delete(f"/api/inference/endpoints/{ep_id}") assert resp.status_code == 200 data = resp.json() assert data["enabled"] is False # --------------------------------------------------------------------------- # Test: Probe, enable, disable actions (19.2) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_enable_endpoint(client: AsyncClient, mock_db: MockInferenceDB): """Enable action sets enabled=True for local endpoints.""" create_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(base_url="http://localhost:8000"), ) ep_id = create_resp.json()["id"] # First disable it await client.post(f"/api/inference/endpoints/{ep_id}/disable") # Then enable resp = await client.post(f"/api/inference/endpoints/{ep_id}/enable") assert resp.status_code == 200 assert resp.json()["enabled"] is True @pytest.mark.asyncio async def test_disable_endpoint_action(client: AsyncClient): """Disable action sets enabled=False.""" create_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = create_resp.json()["id"] resp = await client.post(f"/api/inference/endpoints/{ep_id}/disable") assert resp.status_code == 200 assert resp.json()["enabled"] is False # --------------------------------------------------------------------------- # Test: External egress requires confirmation (19.5) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_external_endpoint_disabled_without_egress(client: AsyncClient): """External endpoint is created disabled until egress confirmed.""" payload = _make_endpoint_payload( base_url="https://api.openai.com", name="openai-prod", ) payload["enabled"] = True # Request enabled, but external resp = await client.post("/api/inference/endpoints", json=payload) assert resp.status_code == 201 data = resp.json() # External endpoints are forced disabled until egress is confirmed assert data["enabled"] is False @pytest.mark.asyncio async def test_enable_external_requires_confirmation(client: AsyncClient): """Enabling an external endpoint without confirmation returns 403.""" payload = _make_endpoint_payload( base_url="https://api.openai.com", name="openai-prod", ) resp = await client.post("/api/inference/endpoints", json=payload) ep_id = resp.json()["id"] # Try to enable without confirming egress resp = await client.post(f"/api/inference/endpoints/{ep_id}/enable") assert resp.status_code == 403 assert "egress confirmation" in resp.json()["detail"].lower() @pytest.mark.asyncio async def test_confirm_egress_enables_external(client: AsyncClient): """Confirming egress enables the external endpoint.""" payload = _make_endpoint_payload( base_url="https://api.openai.com", name="openai-prod", ) resp = await client.post("/api/inference/endpoints", json=payload) ep_id = resp.json()["id"] # Confirm egress resp = await client.post( f"/api/inference/endpoints/{ep_id}/confirm-egress", json={"confirmed": True}, ) assert resp.status_code == 200 assert resp.json()["enabled"] is True @pytest.mark.asyncio async def test_confirm_egress_rejects_false(client: AsyncClient): """Egress confirmation with confirmed=false is rejected.""" payload = _make_endpoint_payload( base_url="https://api.openai.com", name="openai-prod", ) resp = await client.post("/api/inference/endpoints", json=payload) ep_id = resp.json()["id"] resp = await client.post( f"/api/inference/endpoints/{ep_id}/confirm-egress", json={"confirmed": False}, ) assert resp.status_code == 422 # Pydantic validation error @pytest.mark.asyncio async def test_confirm_egress_local_endpoint_rejected(client: AsyncClient): """Confirming egress on a local endpoint returns 400.""" payload = _make_endpoint_payload(base_url="http://localhost:8000") resp = await client.post("/api/inference/endpoints", json=payload) ep_id = resp.json()["id"] resp = await client.post( f"/api/inference/endpoints/{ep_id}/confirm-egress", json={"confirmed": True}, ) assert resp.status_code == 400 # --------------------------------------------------------------------------- # Test: Protocol validation (19.1, 19.3) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_invalid_protocol_rejected(client: AsyncClient): """Unknown protocol values are rejected during creation.""" payload = _make_endpoint_payload(protocol="unknown_provider") resp = await client.post("/api/inference/endpoints", json=payload) assert resp.status_code == 422 @pytest.mark.asyncio async def test_invalid_url_rejected(client: AsyncClient): """URLs not starting with http:// or https:// are rejected.""" payload = _make_endpoint_payload(base_url="ftp://bad-url.com") resp = await client.post("/api/inference/endpoints", json=payload) assert resp.status_code == 422 @pytest.mark.asyncio async def test_empty_url_rejected(client: AsyncClient): """Empty base_url is rejected.""" payload = _make_endpoint_payload(base_url="") resp = await client.post("/api/inference/endpoints", json=payload) assert resp.status_code == 422 # --------------------------------------------------------------------------- # Test: Deployments and bindings (19.3, 19.4) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_create_and_list_deployments(client: AsyncClient): """Create a deployment and list it.""" # First create an endpoint ep_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = ep_resp.json()["id"] dep_payload = { "endpoint_id": ep_id, "served_model_name": "stonks-adjudicator-9b", "display_name": "Qwen 9B Adjudicator", "capabilities": {"json_schema": True, "usage": True}, "context_window": 8192, "max_output_tokens": 4096, } resp = await client.post("/api/inference/deployments", json=dep_payload) assert resp.status_code == 201 data = resp.json() assert data["served_model_name"] == "stonks-adjudicator-9b" assert data["context_window"] == 8192 # List resp = await client.get("/api/inference/deployments") assert resp.status_code == 200 assert len(resp.json()) == 1 @pytest.mark.asyncio async def test_create_deployment_invalid_endpoint(client: AsyncClient): """Creating a deployment with non-existent endpoint returns 404.""" dep_payload = { "endpoint_id": str(uuid.uuid4()), "served_model_name": "model", "display_name": "Model", } resp = await client.post("/api/inference/deployments", json=dep_payload) assert resp.status_code == 404 @pytest.mark.asyncio async def test_get_deployment_detail(client: AsyncClient): """Get a deployment by ID with capabilities and limits.""" ep_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = ep_resp.json()["id"] dep_payload = { "endpoint_id": ep_id, "served_model_name": "test-model", "display_name": "Test Model", "capabilities": {"json_schema": True, "seed": True}, "context_window": 16384, "max_output_tokens": 8192, "quantization": "NVFP4", } create_resp = await client.post("/api/inference/deployments", json=dep_payload) dep_id = create_resp.json()["id"] resp = await client.get(f"/api/inference/deployments/{dep_id}") assert resp.status_code == 200 data = resp.json() assert data["capabilities"] == {"json_schema": True, "seed": True} assert data["context_window"] == 16384 assert data["max_output_tokens"] == 8192 assert data["quantization"] == "NVFP4" @pytest.mark.asyncio async def test_create_and_list_bindings(client: AsyncClient): """Create a binding and list it.""" ep_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = ep_resp.json()["id"] dep_payload = { "endpoint_id": ep_id, "served_model_name": "test-model", "display_name": "Test Model", } dep_resp = await client.post("/api/inference/deployments", json=dep_payload) dep_id = dep_resp.json()["id"] agent_id = str(uuid.uuid4()) binding_payload = { "agent_id": agent_id, "stage": "extraction", "model_deployment_id": dep_id, "route_order": 0, } resp = await client.post("/api/inference/bindings", json=binding_payload) assert resp.status_code == 201 data = resp.json() assert data["stage"] == "extraction" assert data["agent_id"] == agent_id # List resp = await client.get("/api/inference/bindings") assert resp.status_code == 200 assert len(resp.json()) == 1 # --------------------------------------------------------------------------- # Test: Secrets NEVER leak in any response (comprehensive) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_secret_never_in_any_response(client: AsyncClient): """Verify auth_secret_ref NEVER appears in any endpoint response.""" secret_ref = "super-secret-api-key-ref-12345" payload = _make_endpoint_payload(auth_secret_ref=secret_ref) # Create resp = await client.post("/api/inference/endpoints", json=payload) assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text ep_id = resp.json()["id"] # List resp = await client.get("/api/inference/endpoints") assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text # Get detail resp = await client.get(f"/api/inference/endpoints/{ep_id}") assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text # Update resp = await client.put( f"/api/inference/endpoints/{ep_id}", json={"name": "renamed"}, ) assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text # Disable resp = await client.post(f"/api/inference/endpoints/{ep_id}/disable") assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text # Enable resp = await client.post(f"/api/inference/endpoints/{ep_id}/enable") assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text # Delete resp = await client.delete(f"/api/inference/endpoints/{ep_id}") assert secret_ref not in resp.text assert "auth_secret_ref" not in resp.text # --------------------------------------------------------------------------- # Test: Protocol selectors (19.3) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_list_protocols(client: AsyncClient): """Protocol selector returns valid options.""" resp = await client.get("/api/inference/protocols") assert resp.status_code == 200 data = resp.json() values = [p["value"] for p in data["protocols"]] assert "ollama_native" in values assert "openai_chat" in values assert "specialist_http" in values # --------------------------------------------------------------------------- # Test: Endpoint detail with bindings and capabilities (19.4) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_endpoint_detail_includes_bindings(client: AsyncClient): """GET endpoint detail includes active stage bindings.""" # Create endpoint + deployment + binding ep_resp = await client.post( "/api/inference/endpoints", json=_make_endpoint_payload(), ) ep_id = ep_resp.json()["id"] dep_payload = { "endpoint_id": ep_id, "served_model_name": "model-a", "display_name": "Model A", "capabilities": {"json_schema": True}, } dep_resp = await client.post("/api/inference/deployments", json=dep_payload) dep_id = dep_resp.json()["id"] binding_payload = { "agent_id": str(uuid.uuid4()), "stage": "adjudication", "model_deployment_id": dep_id, } await client.post("/api/inference/bindings", json=binding_payload) # Get endpoint detail resp = await client.get(f"/api/inference/endpoints/{ep_id}") assert resp.status_code == 200 data = resp.json() assert data["capabilities"] == {"json_schema": True} assert data["active_bindings"] is not None assert len(data["active_bindings"]) == 1 assert data["active_bindings"][0]["stage"] == "adjudication" # --------------------------------------------------------------------------- # Test: is_external_endpoint helper # --------------------------------------------------------------------------- def test_is_external_detection(): """Verify external endpoint detection logic.""" from services.inference_registry.security import is_external_endpoint # Local/cluster endpoints assert not is_external_endpoint("http://localhost:8000") assert not is_external_endpoint("http://127.0.0.1:11434") assert not is_external_endpoint("http://ollama.ollama-service.svc.cluster.local:11434") assert not is_external_endpoint("http://10.1.1.12:2701") assert not is_external_endpoint("http://192.168.1.100:8080") assert not is_external_endpoint("http://172.16.0.1:9000") # External endpoints assert is_external_endpoint("https://api.openai.com") assert is_external_endpoint("https://generativelanguage.googleapis.com") assert is_external_endpoint("https://api.anthropic.com") assert is_external_endpoint("https://some-cloud-provider.example.com")