Files
stonks-oracle/tests/test_inference_registry_api.py
Celes Renata a72f336ad1 feat: Intelligence Pipeline v3 — full implementation
Multi-stage evidence-grounded inference architecture replacing the
monolithic 9B model extraction pipeline. CPU-first specialist services
handle routine extraction while the 9B vLLM model is preserved for
semantic adjudication of ambiguous cases.

Key components:
- Capability-aware inference gateway (OpenAI-compatible + Ollama)
- Endpoint registry with DB migrations and REST API
- Sentence-aware document segmenter (property tests)
- Deterministic financial parsing with offset integrity
- Symbol resolution with ambiguity detection
- Specialist service (GLiNER2, dynamic batching, K8s deployment)
- Company-specific sentiment (FinBERT, calibration)
- Retrieval-based novelty and duplicate detection
- Confidence calibration pipeline
- Deterministic routing engine (property tests)
- 9B adjudication layer with VRAM gating
- Stock-specific impact model (features, labels, baseline, trained)
- Pipeline orchestrator (state machine, queues, leases, feature flags)
- Bounded parallelism (async workers, semaphore, load shedding)
- Observability (tracing, metrics, alerts)
- Compatibility adapter (v3→v2 golden mapping tests)
- Shadow/canary promotion framework
- Active learning and fine-tuning pipeline

Test results: 1,161 tests pass, ruff lint clean.
All 282 spec tasks completed.
2026-07-13 02:14:59 +00:00

671 lines
23 KiB
Python

"""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")