414 lines
14 KiB
Python
414 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from cmvr_edge_ai.client.detect import (
|
|
DetectClient,
|
|
DetectClientClosedError,
|
|
DetectClientConfigurationError,
|
|
DetectHttpStatusError,
|
|
DetectProtocolError,
|
|
DetectTransportError,
|
|
)
|
|
from cmvr_edge_ai.application import create_default_registry, validate_application
|
|
from cmvr_edge_ai.config import load_config
|
|
from cmvr_edge_ai.contracts.catalog import ModelCatalog, ServingMode
|
|
from cmvr_edge_ai.contracts.inference import (
|
|
DetectionsOutput,
|
|
InferenceResponse,
|
|
InferenceStatus,
|
|
)
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(
|
|
self,
|
|
status_code: int,
|
|
payload: Any = None,
|
|
*,
|
|
json_error: BaseException | None = None,
|
|
text: str = "",
|
|
) -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
self._json_error = json_error
|
|
self.text = text
|
|
|
|
def json(self) -> Any:
|
|
if self._json_error is not None:
|
|
raise self._json_error
|
|
return deepcopy(self._payload)
|
|
|
|
|
|
class FakeHttpClient:
|
|
def __init__(self, *outcomes: Any) -> None:
|
|
self._outcomes = list(outcomes)
|
|
self.calls: list[dict[str, Any]] = []
|
|
self.close_count = 0
|
|
|
|
async def request(self, method: str, path: str, **kwargs: Any) -> FakeResponse:
|
|
self.calls.append({"method": method, "path": path, **deepcopy(kwargs)})
|
|
if not self._outcomes:
|
|
raise AssertionError("fake HTTP client has no configured outcome")
|
|
outcome = self._outcomes.pop(0)
|
|
if isinstance(outcome, BaseException):
|
|
raise outcome
|
|
if callable(outcome):
|
|
return outcome(self.calls[-1])
|
|
return outcome
|
|
|
|
async def aclose(self) -> None:
|
|
self.close_count += 1
|
|
|
|
|
|
def _catalog_payload() -> dict[str, Any]:
|
|
return {
|
|
"schema_version": "cmvr.model-catalog/v1",
|
|
"active_push": [],
|
|
"passive_invoke": [
|
|
{
|
|
"category": "detect.ppe",
|
|
"family": "detection",
|
|
"model_id": "construction-ppe-yolov8@1",
|
|
"serving_modes": ["passive_invoke"],
|
|
"registration_status": "registered",
|
|
"deployment_status": "ready",
|
|
"input_kinds": ["image"],
|
|
"output_kinds": ["detections"],
|
|
"artifact_roles": ["annotated"],
|
|
"parameters_schema": {},
|
|
"name": "Construction PPE",
|
|
"backend": "ultralytics",
|
|
"description": "PPE violations",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _inference_response(call: dict[str, Any]) -> FakeResponse:
|
|
request = call["json"]
|
|
return FakeResponse(
|
|
200,
|
|
{
|
|
"schema_version": "cmvr.inference-response/v1",
|
|
"request_id": request["request_id"],
|
|
"trace_id": "trace-1",
|
|
"category": request["category"],
|
|
"model": {
|
|
"model_id": "construction-ppe-yolov8@1",
|
|
"backend": "ultralytics",
|
|
"name": "Construction PPE",
|
|
},
|
|
"status": "succeeded",
|
|
"outputs": [
|
|
{
|
|
"kind": "detections",
|
|
"name": "detections",
|
|
"coordinate_space": "pixel_xyxy",
|
|
"items": [
|
|
{
|
|
"label": "No-Helmet",
|
|
"confidence": 0.94,
|
|
"box": {
|
|
"x_min": 10.0,
|
|
"y_min": 20.0,
|
|
"x_max": 100.0,
|
|
"y_max": 200.0,
|
|
},
|
|
"attributes": {},
|
|
}
|
|
],
|
|
}
|
|
],
|
|
"artifacts": [],
|
|
"timing": {"queue_ms": 1.0, "inference_ms": 12.5, "total_ms": 14.0},
|
|
"warnings": [],
|
|
},
|
|
)
|
|
|
|
|
|
def test_list_models_returns_strict_catalog_without_a_request_body() -> None:
|
|
async def scenario() -> tuple[ModelCatalog, FakeHttpClient]:
|
|
http = FakeHttpClient(FakeResponse(200, _catalog_payload()))
|
|
client = DetectClient("http://detect.invalid", http_client=http)
|
|
return await client.list_models(), http
|
|
|
|
catalog, http = asyncio.run(scenario())
|
|
|
|
assert isinstance(catalog, ModelCatalog)
|
|
assert catalog.active_push == ()
|
|
assert catalog.passive_invoke[0].category == "detect.ppe"
|
|
assert catalog.passive_invoke[0].serving_modes == (ServingMode.PASSIVE_INVOKE,)
|
|
assert http.calls == [{"method": "GET", "path": "/v1/models"}]
|
|
|
|
|
|
def test_infer_sends_versioned_base64_json_and_returns_typed_response() -> None:
|
|
image = b"\xff\xd8test-jpeg\xff\xd9"
|
|
|
|
async def scenario() -> tuple[InferenceResponse, FakeHttpClient]:
|
|
http = FakeHttpClient(_inference_response)
|
|
client = DetectClient("http://detect.invalid", http_client=http)
|
|
response = await client.infer(
|
|
"detect.ppe",
|
|
image,
|
|
media_type="IMAGE/JPEG",
|
|
parameters={"confidence": 0.5, "max_detections": 20},
|
|
image_roles=("annotated",),
|
|
)
|
|
return response, http
|
|
|
|
response, http = asyncio.run(scenario())
|
|
|
|
assert isinstance(response, InferenceResponse)
|
|
assert response.status is InferenceStatus.SUCCEEDED
|
|
assert isinstance(response.outputs[0], DetectionsOutput)
|
|
assert response.outputs[0].items[0].label == "No-Helmet"
|
|
assert len(http.calls) == 1
|
|
call = http.calls[0]
|
|
assert call["method"] == "POST"
|
|
assert call["path"] == "/v1/inference"
|
|
body = call["json"]
|
|
assert body["schema_version"] == "cmvr.inference-request/v1"
|
|
assert body["category"] == "detect.ppe"
|
|
assert body["parameters"] == {"confidence": 0.5, "max_detections": 20}
|
|
assert body["requested_artifact_roles"] == ["annotated"]
|
|
assert body["inputs"] == [
|
|
{
|
|
"kind": "image",
|
|
"name": "image",
|
|
"media_type": "image/jpeg",
|
|
"encoding": "base64",
|
|
"data": base64.b64encode(image).decode("ascii"),
|
|
"width": None,
|
|
"height": None,
|
|
"sha256": None,
|
|
}
|
|
]
|
|
assert "model_id" not in body
|
|
assert "weights" not in body
|
|
assert "pipeline" not in body
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kwargs", "message"),
|
|
[
|
|
({"category": "ppe"}, "category"),
|
|
({"category": "Detect.PPE"}, "category"),
|
|
({"image": b""}, "image"),
|
|
({"image": bytearray(b"image")}, "image"),
|
|
({"media_type": "application/octet-stream"}, "media_type"),
|
|
({"parameters": {"threshold": float("nan")}}, "parameters"),
|
|
({"parameters": {"payload": b"not-json"}}, "parameters"),
|
|
({"image_roles": "annotated"}, "image_roles"),
|
|
({"image_roles": ("annotated", "annotated")}, "image_roles"),
|
|
({"image_roles": ("Bad Role",)}, "image_roles"),
|
|
],
|
|
)
|
|
def test_infer_rejects_invalid_local_arguments_without_http(
|
|
kwargs: dict[str, Any],
|
|
message: str,
|
|
) -> None:
|
|
async def scenario() -> None:
|
|
http = FakeHttpClient()
|
|
arguments: dict[str, Any] = {
|
|
"category": "detect.ppe",
|
|
"image": b"image",
|
|
"media_type": "image/jpeg",
|
|
**kwargs,
|
|
}
|
|
with pytest.raises(DetectClientConfigurationError, match=message):
|
|
await DetectClient(
|
|
"http://detect.invalid", http_client=http
|
|
).infer(**arguments)
|
|
assert http.calls == []
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"base_url",
|
|
["", "detect.invalid", "ftp://detect.invalid", "http:///missing-host", "http://x/api"],
|
|
)
|
|
def test_client_rejects_invalid_base_url(base_url: str) -> None:
|
|
with pytest.raises(DetectClientConfigurationError, match="base_url"):
|
|
DetectClient(base_url, http_client=FakeHttpClient())
|
|
|
|
|
|
def test_http_status_error_preserves_status_and_safe_server_detail() -> None:
|
|
async def scenario() -> None:
|
|
client = DetectClient(
|
|
"http://detect.invalid",
|
|
http_client=FakeHttpClient(
|
|
FakeResponse(404, {"code": "unknown_category", "message": "not deployed"})
|
|
),
|
|
)
|
|
with pytest.raises(DetectHttpStatusError, match="not deployed") as captured:
|
|
await client.list_models()
|
|
assert captured.value.status_code == 404
|
|
assert captured.value.path == "/v1/models"
|
|
assert captured.value.error["code"] == "unknown_category"
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_transport_failure_is_wrapped_but_programming_errors_are_not() -> None:
|
|
async def scenario() -> None:
|
|
transport_client = DetectClient(
|
|
"http://detect.invalid",
|
|
http_client=FakeHttpClient(OSError("connection refused")),
|
|
)
|
|
with pytest.raises(DetectTransportError, match="connection refused"):
|
|
await transport_client.list_models()
|
|
|
|
broken_fake = DetectClient(
|
|
"http://detect.invalid",
|
|
http_client=FakeHttpClient(AssertionError("bad fake")),
|
|
)
|
|
with pytest.raises(AssertionError, match="bad fake"):
|
|
await broken_fake.list_models()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response",
|
|
[
|
|
FakeResponse(200, json_error=ValueError("invalid JSON")),
|
|
FakeResponse(200, []),
|
|
FakeResponse(200, {**_catalog_payload(), "unexpected": True}),
|
|
FakeResponse(
|
|
200,
|
|
{
|
|
**_catalog_payload(),
|
|
"active_push": _catalog_payload()["passive_invoke"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
def test_list_models_rejects_invalid_success_responses(response: FakeResponse) -> None:
|
|
async def scenario() -> None:
|
|
client = DetectClient(
|
|
"http://detect.invalid", http_client=FakeHttpClient(response)
|
|
)
|
|
with pytest.raises(DetectProtocolError):
|
|
await client.list_models()
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
@pytest.mark.parametrize("mismatch", ["request_id", "category"])
|
|
def test_infer_rejects_mismatched_response_identity(mismatch: str) -> None:
|
|
def mismatched_response(call: dict[str, Any]) -> FakeResponse:
|
|
response = _inference_response(call)
|
|
if mismatch == "request_id":
|
|
response._payload["request_id"] = "another-request"
|
|
else:
|
|
response._payload["category"] = "detect.mobile_phone"
|
|
return response
|
|
|
|
async def scenario() -> None:
|
|
client = DetectClient(
|
|
"http://detect.invalid",
|
|
http_client=FakeHttpClient(mismatched_response),
|
|
)
|
|
with pytest.raises(DetectProtocolError, match=mismatch):
|
|
await client.infer(
|
|
"detect.ppe", b"image", media_type="image/jpeg"
|
|
)
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_close_is_idempotent_does_not_close_injected_client_and_blocks_requests() -> None:
|
|
async def scenario() -> FakeHttpClient:
|
|
http = FakeHttpClient(FakeResponse(200, _catalog_payload()))
|
|
client = DetectClient("http://detect.invalid", http_client=http)
|
|
await client.aclose()
|
|
await client.aclose()
|
|
with pytest.raises(DetectClientClosedError):
|
|
await client.list_models()
|
|
return http
|
|
|
|
http = asyncio.run(scenario())
|
|
assert http.close_count == 0
|
|
assert http.calls == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"relative_path",
|
|
["configs/server_detect.yaml", "configs/server_gauge.yaml"],
|
|
)
|
|
def test_server_configs_use_the_root_app_config(relative_path: str) -> None:
|
|
config = load_config(PROJECT_ROOT / relative_path)
|
|
assert config.api_version == "cmvr.edge.ai/v1"
|
|
|
|
|
|
def test_server_detect_config_routes_stable_categories_to_server_owned_models() -> None:
|
|
config = load_config(PROJECT_ROOT / "configs/server_detect.yaml")
|
|
|
|
assert config.server is not None
|
|
assert config.server.enabled is True
|
|
assert set(config.server.routes) == {"detect.ppe", "detect.mobile_phone"}
|
|
assert config.server.routes["detect.ppe"].model_id == (
|
|
"construction-ppe-yolov8@2"
|
|
)
|
|
assert config.server.routes["detect.mobile_phone"].model_id == (
|
|
"yolov8n-mobile-phone@2"
|
|
)
|
|
for category, route in config.server.routes.items():
|
|
assert route.pipeline in config.pipelines
|
|
pipeline = config.pipelines[route.pipeline]
|
|
detector = pipeline.nodes["detector"]
|
|
assert detector.params["model"] == route.model_id
|
|
assert detector.params["attach_frame"] is True
|
|
assert category not in detector.params
|
|
assert pipeline.nodes["request_source"].params["category"] == category
|
|
assert pipeline.nodes["response"].params["category"] == category
|
|
assert [edge.source for edge in pipeline.edges] == [
|
|
"request_source.requests",
|
|
"image_decoder.frames",
|
|
"detector.detections",
|
|
"response.responses",
|
|
]
|
|
assert [edge.target for edge in pipeline.edges] == [
|
|
"image_decoder.requests",
|
|
"detector.frames",
|
|
"response.detections",
|
|
"response_sink.responses",
|
|
]
|
|
assert all(
|
|
edge.qos.profile == "request"
|
|
and edge.qos.capacity == route.queue_capacity
|
|
and edge.qos.overflow == "block"
|
|
for edge in pipeline.edges
|
|
)
|
|
|
|
|
|
def test_server_detect_config_compiles_both_passive_pipelines() -> None:
|
|
config = load_config(PROJECT_ROOT / "configs/server_detect.yaml")
|
|
registry = create_default_registry(discover_entry_points=False)
|
|
|
|
compiled = validate_application(config, registry)
|
|
|
|
assert {item.pipeline_id for item in compiled} == {
|
|
"detect_ppe",
|
|
"detect_mobile_phone",
|
|
}
|
|
for item in compiled:
|
|
assert set(item.plugin_specs) == {
|
|
"request_source",
|
|
"image_decoder",
|
|
"detector",
|
|
"response",
|
|
"response_sink",
|
|
}
|