748 lines
26 KiB
Python
748 lines
26 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import base64
|
||
|
|
import logging
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from cmvr_edge_ai.application import create_default_capability_registry
|
||
|
|
from cmvr_edge_ai.config import load_config_data
|
||
|
|
from cmvr_edge_ai.contracts import (
|
||
|
|
DetectionsOutput,
|
||
|
|
InferenceResponse,
|
||
|
|
InferenceStatus,
|
||
|
|
ModelReference,
|
||
|
|
ScalarOutput,
|
||
|
|
)
|
||
|
|
from cmvr_edge_ai.core import HealthReport
|
||
|
|
from cmvr_edge_ai.server import InvocationBroker
|
||
|
|
from cmvr_edge_ai.server.api import (
|
||
|
|
InferenceHttpApi,
|
||
|
|
_InvalidContentLength,
|
||
|
|
_read_limited_body,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _config(
|
||
|
|
*,
|
||
|
|
max_request_bytes: int = 4096,
|
||
|
|
request_timeout_s: float = 0.2,
|
||
|
|
model_id: str = "yolov8n-mobile-phone@1",
|
||
|
|
bearer_token: str | None = None,
|
||
|
|
): # type: ignore[no-untyped-def]
|
||
|
|
return load_config_data(
|
||
|
|
{
|
||
|
|
"api_version": "cmvr.edge.ai/v1",
|
||
|
|
"pipelines": {
|
||
|
|
"mobile_remote": {
|
||
|
|
"nodes": {
|
||
|
|
"source": {"uses": "server.request_source@1"},
|
||
|
|
}
|
||
|
|
}
|
||
|
|
},
|
||
|
|
"server": {
|
||
|
|
"enabled": True,
|
||
|
|
"http": {
|
||
|
|
"request_timeout_s": request_timeout_s,
|
||
|
|
"max_request_bytes": max_request_bytes,
|
||
|
|
"max_image_bytes": min(max_request_bytes, 1024),
|
||
|
|
**(
|
||
|
|
{}
|
||
|
|
if bearer_token is None
|
||
|
|
else {"bearer_token": bearer_token}
|
||
|
|
),
|
||
|
|
},
|
||
|
|
"routes": {
|
||
|
|
"detect.mobile_phone": {
|
||
|
|
"pipeline": "mobile_remote",
|
||
|
|
"model_id": model_id,
|
||
|
|
"queue_capacity": 1,
|
||
|
|
}
|
||
|
|
},
|
||
|
|
},
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _request_payload(*, category: str = "detect.mobile_phone") -> dict[str, object]:
|
||
|
|
return {
|
||
|
|
"schema_version": "cmvr.inference-request/v1",
|
||
|
|
"request_id": "request-1",
|
||
|
|
"category": category,
|
||
|
|
"inputs": [
|
||
|
|
{
|
||
|
|
"kind": "image",
|
||
|
|
"name": "image",
|
||
|
|
"media_type": "image/jpeg",
|
||
|
|
"encoding": "base64",
|
||
|
|
"data": base64.b64encode(b"jpeg-bytes").decode("ascii"),
|
||
|
|
}
|
||
|
|
],
|
||
|
|
"parameters": {},
|
||
|
|
"requested_artifact_roles": [],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _response(
|
||
|
|
request_id: str,
|
||
|
|
*,
|
||
|
|
category: str = "detect.mobile_phone",
|
||
|
|
model_id: str = "yolov8n-mobile-phone@1",
|
||
|
|
outputs: tuple[object, ...] | None = None,
|
||
|
|
) -> InferenceResponse:
|
||
|
|
return InferenceResponse(
|
||
|
|
request_id=request_id,
|
||
|
|
trace_id=request_id,
|
||
|
|
category=category,
|
||
|
|
model=ModelReference(model_id=model_id, backend="test"),
|
||
|
|
status=InferenceStatus.NO_RESULT,
|
||
|
|
outputs=(
|
||
|
|
(DetectionsOutput(items=()),)
|
||
|
|
if outputs is None
|
||
|
|
else outputs
|
||
|
|
),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_routes_one_request_through_the_broker() -> None:
|
||
|
|
async def exercise() -> None:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {
|
||
|
|
"state": "running",
|
||
|
|
"pipelines": {
|
||
|
|
"mobile_remote": {"source": HealthReport(True)}
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def pipeline() -> None:
|
||
|
|
item = await broker.receive("detect.mobile_phone")
|
||
|
|
await broker.complete(
|
||
|
|
item.request_id,
|
||
|
|
InferenceResponse(
|
||
|
|
request_id=item.request_id,
|
||
|
|
trace_id=item.request_id,
|
||
|
|
category=item.category,
|
||
|
|
model=ModelReference(
|
||
|
|
model_id="yolov8n-mobile-phone@1",
|
||
|
|
backend="test",
|
||
|
|
),
|
||
|
|
status=InferenceStatus.NO_RESULT,
|
||
|
|
outputs=(DetectionsOutput(items=()),),
|
||
|
|
),
|
||
|
|
invocation_token=item.invocation_token,
|
||
|
|
)
|
||
|
|
|
||
|
|
worker = asyncio.create_task(pipeline())
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
response = await client.post("/v1/inference", json=_request_payload())
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json()["request_id"] == "request-1"
|
||
|
|
assert response.json()["status"] == "no_result"
|
||
|
|
|
||
|
|
models = await client.get("/v1/models")
|
||
|
|
assert models.status_code == 200
|
||
|
|
assert [
|
||
|
|
item["category"] for item in models.json()["passive_invoke"]
|
||
|
|
] == ["detect.mobile_phone"]
|
||
|
|
|
||
|
|
ready = await client.get("/health/ready")
|
||
|
|
assert ready.status_code == 200
|
||
|
|
assert ready.json()["status"] == "ready"
|
||
|
|
await worker
|
||
|
|
await broker.close()
|
||
|
|
|
||
|
|
asyncio.run(exercise())
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_returns_structured_errors_for_invalid_requests() -> None:
|
||
|
|
async def exercise() -> None:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "starting", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
unsupported = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
json=_request_payload(category="detect.unknown"),
|
||
|
|
)
|
||
|
|
assert unsupported.status_code == 404
|
||
|
|
assert unsupported.json()["error"]["code"] == "UNSUPPORTED_CATEGORY"
|
||
|
|
|
||
|
|
malformed = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
content=b"not-json",
|
||
|
|
headers={"content-type": "application/json"},
|
||
|
|
)
|
||
|
|
assert malformed.status_code == 422
|
||
|
|
assert malformed.json()["schema_version"] == "cmvr.inference-error/v1"
|
||
|
|
|
||
|
|
wrong_type = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
content=b"{}",
|
||
|
|
headers={"content-type": "text/plain"},
|
||
|
|
)
|
||
|
|
assert wrong_type.status_code == 415
|
||
|
|
|
||
|
|
ready = await client.get("/health/ready")
|
||
|
|
assert ready.status_code == 503
|
||
|
|
await broker.close()
|
||
|
|
|
||
|
|
asyncio.run(exercise())
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_rejects_body_before_pydantic_parsing_when_too_large() -> None:
|
||
|
|
async def exercise() -> None:
|
||
|
|
config = _config(max_request_bytes=1024)
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
response = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
content=b"x" * 1025,
|
||
|
|
headers={"content-type": "application/json"},
|
||
|
|
)
|
||
|
|
assert response.status_code == 413
|
||
|
|
assert response.json()["error"]["code"] == "REQUEST_TOO_LARGE"
|
||
|
|
await broker.close()
|
||
|
|
|
||
|
|
asyncio.run(exercise())
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_requires_configured_bearer_token_except_for_health() -> None:
|
||
|
|
async def exercise() -> None:
|
||
|
|
config = _config(bearer_token="deployment-secret")
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {
|
||
|
|
"state": "running",
|
||
|
|
"pipelines": {
|
||
|
|
"mobile_remote": {"source": HealthReport(True)}
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
live = await client.get("/health/live")
|
||
|
|
ready = await client.get("/health/ready")
|
||
|
|
missing = await client.get("/v1/models")
|
||
|
|
wrong = await client.get(
|
||
|
|
"/v1/models",
|
||
|
|
headers={"Authorization": "Bearer wrong-secret"},
|
||
|
|
)
|
||
|
|
openapi_without_token = await client.get("/openapi.json")
|
||
|
|
malformed_without_token = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
json={},
|
||
|
|
)
|
||
|
|
authorized = await client.get(
|
||
|
|
"/v1/models",
|
||
|
|
headers={"Authorization": "bearer deployment-secret"},
|
||
|
|
)
|
||
|
|
openapi_with_token = await client.get(
|
||
|
|
"/openapi.json",
|
||
|
|
headers={"Authorization": "Bearer deployment-secret"},
|
||
|
|
)
|
||
|
|
malformed_with_token = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
json={},
|
||
|
|
headers={"Authorization": "Bearer deployment-secret"},
|
||
|
|
)
|
||
|
|
|
||
|
|
await broker.close()
|
||
|
|
assert live.status_code == 200
|
||
|
|
assert ready.status_code == 200
|
||
|
|
for response in (
|
||
|
|
missing,
|
||
|
|
wrong,
|
||
|
|
openapi_without_token,
|
||
|
|
malformed_without_token,
|
||
|
|
):
|
||
|
|
assert response.status_code == 401
|
||
|
|
assert response.headers["www-authenticate"] == "Bearer"
|
||
|
|
assert response.json()["error"]["code"] == "UNAUTHORIZED"
|
||
|
|
assert authorized.status_code == 200
|
||
|
|
assert openapi_with_token.status_code == 200
|
||
|
|
assert malformed_with_token.status_code == 422
|
||
|
|
|
||
|
|
asyncio.run(exercise())
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("content_length", ["invalid", "-1", "1, 2"])
|
||
|
|
def test_limited_body_rejects_invalid_content_length(
|
||
|
|
content_length: str,
|
||
|
|
) -> None:
|
||
|
|
class Request:
|
||
|
|
headers = {"content-length": content_length}
|
||
|
|
|
||
|
|
async def stream(self): # type: ignore[no-untyped-def]
|
||
|
|
yield b"{}"
|
||
|
|
|
||
|
|
with pytest.raises(_InvalidContentLength):
|
||
|
|
asyncio.run(_read_limited_body(Request(), 1024))
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_generates_and_echoes_request_id_when_client_omits_it() -> None:
|
||
|
|
async def exercise() -> str:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def pipeline() -> None:
|
||
|
|
item = await broker.receive("detect.mobile_phone")
|
||
|
|
assert item.payload.request_id == item.request_id
|
||
|
|
await broker.complete(
|
||
|
|
item.request_id,
|
||
|
|
_response(item.request_id),
|
||
|
|
invocation_token=item.invocation_token,
|
||
|
|
)
|
||
|
|
|
||
|
|
payload = _request_payload()
|
||
|
|
del payload["request_id"]
|
||
|
|
worker = asyncio.create_task(pipeline())
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
response = await client.post("/v1/inference", json=payload)
|
||
|
|
await worker
|
||
|
|
await broker.close()
|
||
|
|
assert response.status_code == 200
|
||
|
|
return response.json()["request_id"]
|
||
|
|
|
||
|
|
request_id = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert len(request_id) == 32
|
||
|
|
assert all(character in "0123456789abcdef" for character in request_id)
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_rejects_pipeline_responses_for_another_request_or_model(
|
||
|
|
caplog: pytest.LogCaptureFixture,
|
||
|
|
) -> None:
|
||
|
|
async def exercise() -> list[dict[str, object]]:
|
||
|
|
cases = (
|
||
|
|
lambda request_id: _response("another-request"),
|
||
|
|
lambda request_id: _response(
|
||
|
|
request_id,
|
||
|
|
category="detect.analog_gauge",
|
||
|
|
),
|
||
|
|
lambda request_id: _response(
|
||
|
|
request_id,
|
||
|
|
model_id="another-model@1",
|
||
|
|
),
|
||
|
|
lambda request_id: _response(
|
||
|
|
request_id,
|
||
|
|
outputs=(ScalarOutput(name="reading", value=1.0),),
|
||
|
|
),
|
||
|
|
lambda request_id: {"request_id": request_id},
|
||
|
|
)
|
||
|
|
errors: list[dict[str, object]] = []
|
||
|
|
for index, build_response in enumerate(cases):
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def pipeline() -> None:
|
||
|
|
item = await broker.receive("detect.mobile_phone")
|
||
|
|
await broker.complete(
|
||
|
|
item.request_id,
|
||
|
|
build_response(item.request_id),
|
||
|
|
invocation_token=item.invocation_token,
|
||
|
|
)
|
||
|
|
|
||
|
|
payload = _request_payload()
|
||
|
|
payload["request_id"] = f"mismatch-{index}"
|
||
|
|
worker = asyncio.create_task(pipeline())
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
response = await client.post("/v1/inference", json=payload)
|
||
|
|
await worker
|
||
|
|
await broker.close()
|
||
|
|
assert response.status_code == 500
|
||
|
|
body = response.json()
|
||
|
|
assert body["request_id"] == f"mismatch-{index}"
|
||
|
|
assert body["category"] == "detect.mobile_phone"
|
||
|
|
assert body["error"]["code"] == "INVALID_PIPELINE_RESPONSE"
|
||
|
|
errors.append(body)
|
||
|
|
return errors
|
||
|
|
|
||
|
|
caplog.set_level(logging.ERROR, logger="cmvr_edge_ai.server.api")
|
||
|
|
asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert "response request_id does not match" in caplog.text
|
||
|
|
assert "response category does not match" in caplog.text
|
||
|
|
assert "response model_id does not match" in caplog.text
|
||
|
|
assert "unsupported output kind" in caplog.text
|
||
|
|
assert "unexpected response type dict" in caplog.text
|
||
|
|
|
||
|
|
|
||
|
|
def test_http_api_maps_broker_overload_duplicate_timeout_and_close() -> None:
|
||
|
|
async def exercise() -> tuple[dict[str, object], ...]:
|
||
|
|
config = _config(request_timeout_s=0.01)
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
await broker.submit("detect.mobile_phone", "duplicate", object())
|
||
|
|
duplicate_payload = _request_payload()
|
||
|
|
duplicate_payload["request_id"] = "duplicate"
|
||
|
|
duplicate = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
json=duplicate_payload,
|
||
|
|
)
|
||
|
|
await broker.cancel("duplicate")
|
||
|
|
|
||
|
|
await broker.submit("detect.mobile_phone", "blocker", object())
|
||
|
|
busy_payload = _request_payload()
|
||
|
|
busy_payload["request_id"] = "busy"
|
||
|
|
busy = await client.post("/v1/inference", json=busy_payload)
|
||
|
|
await broker.cancel("blocker")
|
||
|
|
|
||
|
|
timeout_payload = _request_payload()
|
||
|
|
timeout_payload["request_id"] = "timeout"
|
||
|
|
timeout = await client.post("/v1/inference", json=timeout_payload)
|
||
|
|
|
||
|
|
await broker.close()
|
||
|
|
closed_payload = _request_payload()
|
||
|
|
closed_payload["request_id"] = "closed"
|
||
|
|
closed = await client.post("/v1/inference", json=closed_payload)
|
||
|
|
|
||
|
|
assert duplicate.status_code == 409
|
||
|
|
assert busy.status_code == 429
|
||
|
|
assert timeout.status_code == 504
|
||
|
|
assert closed.status_code == 503
|
||
|
|
return tuple(
|
||
|
|
response.json()
|
||
|
|
for response in (duplicate, busy, timeout, closed)
|
||
|
|
)
|
||
|
|
|
||
|
|
duplicate, busy, timeout, closed = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert duplicate["error"]["code"] == "DUPLICATE_REQUEST_ID"
|
||
|
|
assert busy["error"]["code"] == "BUSY"
|
||
|
|
assert busy["error"]["retryable"] is True
|
||
|
|
assert timeout["error"]["code"] == "TIMEOUT"
|
||
|
|
assert timeout["error"]["retryable"] is True
|
||
|
|
assert closed["error"]["code"] == "MODEL_UNAVAILABLE"
|
||
|
|
|
||
|
|
|
||
|
|
def test_configured_route_missing_from_broker_is_server_misconfiguration() -> None:
|
||
|
|
async def exercise() -> tuple[int, dict[str, object]]:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
response = await client.post("/v1/inference", json=_request_payload())
|
||
|
|
await broker.close()
|
||
|
|
return response.status_code, response.json()
|
||
|
|
|
||
|
|
status_code, body = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert status_code == 503
|
||
|
|
assert body["error"]["code"] == "ROUTE_MISCONFIGURED"
|
||
|
|
assert body["request_id"] == "request-1"
|
||
|
|
|
||
|
|
|
||
|
|
def test_pipeline_exception_is_logged_but_not_exposed_to_client(
|
||
|
|
caplog: pytest.LogCaptureFixture,
|
||
|
|
) -> None:
|
||
|
|
async def exercise() -> tuple[int, dict[str, object]]:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
|
||
|
|
async def pipeline() -> None:
|
||
|
|
item = await broker.receive("detect.mobile_phone")
|
||
|
|
await broker.fail(
|
||
|
|
item.request_id,
|
||
|
|
RuntimeError("private adapter path /tmp/private-model.pt"),
|
||
|
|
invocation_token=item.invocation_token,
|
||
|
|
)
|
||
|
|
|
||
|
|
worker = asyncio.create_task(pipeline())
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
response = await client.post("/v1/inference", json=_request_payload())
|
||
|
|
await worker
|
||
|
|
await broker.close()
|
||
|
|
return response.status_code, response.json()
|
||
|
|
|
||
|
|
caplog.set_level(logging.ERROR, logger="cmvr_edge_ai.server.api")
|
||
|
|
status_code, body = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert status_code == 500
|
||
|
|
assert body["error"]["code"] == "INFERENCE_FAILED"
|
||
|
|
assert "private-model.pt" not in str(body)
|
||
|
|
assert "request_id=request-1" in caplog.text
|
||
|
|
assert "private-model.pt" in caplog.text
|
||
|
|
|
||
|
|
|
||
|
|
def test_health_failure_keeps_liveness_and_catalog_available_with_unavailable_status(
|
||
|
|
caplog: pytest.LogCaptureFixture,
|
||
|
|
) -> None:
|
||
|
|
async def exercise() -> tuple[object, object, object]:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
raise RuntimeError("health backend failed")
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
live = await client.get("/health/live")
|
||
|
|
ready = await client.get("/health/ready")
|
||
|
|
models = await client.get("/v1/models")
|
||
|
|
await broker.close()
|
||
|
|
return live, ready, models
|
||
|
|
|
||
|
|
caplog.set_level(logging.ERROR, logger="cmvr_edge_ai.server.api")
|
||
|
|
live, ready, models = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert live.status_code == 200
|
||
|
|
assert ready.status_code == 503
|
||
|
|
assert ready.json() == {
|
||
|
|
"status": "not_ready",
|
||
|
|
"application_state": "unknown",
|
||
|
|
}
|
||
|
|
assert models.status_code == 200
|
||
|
|
assert (
|
||
|
|
models.json()["passive_invoke"][0]["deployment_status"]
|
||
|
|
== "unavailable"
|
||
|
|
)
|
||
|
|
assert caplog.text.count("health provider failed") == 2
|
||
|
|
|
||
|
|
|
||
|
|
def test_ready_and_catalog_accept_wire_safe_health_reports() -> None:
|
||
|
|
async def exercise() -> tuple[object, object]:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {
|
||
|
|
"state": "running",
|
||
|
|
"pipelines": {
|
||
|
|
"mobile_remote": {
|
||
|
|
"source": {"healthy": True, "detail": "ready"}
|
||
|
|
}
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
ready = await client.get("/health/ready")
|
||
|
|
models = await client.get("/v1/models")
|
||
|
|
await broker.close()
|
||
|
|
return ready, models
|
||
|
|
|
||
|
|
ready, models = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert ready.status_code == 200
|
||
|
|
assert models.status_code == 200
|
||
|
|
assert models.json()["passive_invoke"][0]["deployment_status"] == "ready"
|
||
|
|
|
||
|
|
|
||
|
|
def test_closed_broker_makes_readiness_and_passive_catalog_unavailable() -> None:
|
||
|
|
async def exercise() -> tuple[object, object]:
|
||
|
|
config = _config()
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
await broker.close()
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {
|
||
|
|
"state": "running",
|
||
|
|
"pipelines": {
|
||
|
|
"mobile_remote": {"source": HealthReport(True)}
|
||
|
|
},
|
||
|
|
}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
ready = await client.get("/health/ready")
|
||
|
|
models = await client.get("/v1/models")
|
||
|
|
return ready, models
|
||
|
|
|
||
|
|
ready, models = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert ready.status_code == 503
|
||
|
|
assert models.status_code == 200
|
||
|
|
assert (
|
||
|
|
models.json()["passive_invoke"][0]["deployment_status"]
|
||
|
|
== "unavailable"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_unknown_capability_returns_structured_catalog_and_route_errors(
|
||
|
|
caplog: pytest.LogCaptureFixture,
|
||
|
|
) -> None:
|
||
|
|
async def exercise() -> tuple[object, object]:
|
||
|
|
config = _config(model_id="not-installed@1")
|
||
|
|
broker = InvocationBroker()
|
||
|
|
broker.register_route("detect.mobile_phone", capacity=1)
|
||
|
|
|
||
|
|
async def health(): # type: ignore[no-untyped-def]
|
||
|
|
return {"state": "running", "pipelines": {}}
|
||
|
|
|
||
|
|
api = InferenceHttpApi(
|
||
|
|
config,
|
||
|
|
create_default_capability_registry(discover_entry_points=False),
|
||
|
|
broker,
|
||
|
|
health_provider=health,
|
||
|
|
)
|
||
|
|
async with httpx.AsyncClient(
|
||
|
|
transport=httpx.ASGITransport(app=api.app),
|
||
|
|
base_url="http://test",
|
||
|
|
) as client:
|
||
|
|
models = await client.get("/v1/models")
|
||
|
|
inference = await client.post(
|
||
|
|
"/v1/inference",
|
||
|
|
json=_request_payload(),
|
||
|
|
)
|
||
|
|
await broker.close()
|
||
|
|
return models, inference
|
||
|
|
|
||
|
|
caplog.set_level(logging.ERROR, logger="cmvr_edge_ai.server.api")
|
||
|
|
models, inference = asyncio.run(exercise())
|
||
|
|
|
||
|
|
assert models.status_code == 503
|
||
|
|
assert models.json()["error"]["code"] == "CATALOG_UNAVAILABLE"
|
||
|
|
assert models.json()["schema_version"] == "cmvr.inference-error/v1"
|
||
|
|
assert inference.status_code == 503
|
||
|
|
assert inference.json()["error"]["code"] == "ROUTE_MISCONFIGURED"
|
||
|
|
assert "not-installed@1" not in str(inference.json())
|
||
|
|
assert "failed to build deployed model catalog" in caplog.text
|