115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
from cmvr_edge_ai.application import create_default_capability_registry
|
|
from cmvr_edge_ai.capabilities import CapabilityDeploymentStatus
|
|
from cmvr_edge_ai.config import load_config_data
|
|
from cmvr_edge_ai.core import HealthReport
|
|
from cmvr_edge_ai.server.catalog import build_deployed_catalog
|
|
|
|
|
|
def _config(): # type: ignore[no-untyped-def]
|
|
return load_config_data(
|
|
{
|
|
"api_version": "cmvr.edge.ai/v1",
|
|
"pipelines": {
|
|
"active_phone": {
|
|
"nodes": {
|
|
"source": {"uses": "test.source@1"},
|
|
"detector": {
|
|
"uses": "detection.model@1",
|
|
"with": {"model": "people-talking-yolov8x@1"},
|
|
},
|
|
}
|
|
},
|
|
"mobile_remote": {
|
|
"nodes": {
|
|
"source": {"uses": "server.request_source@1"},
|
|
"detector": {
|
|
"uses": "detection.model@1",
|
|
"with": {"model": "yolov8n-mobile-phone@1"},
|
|
},
|
|
}
|
|
},
|
|
},
|
|
"server": {
|
|
"enabled": True,
|
|
"routes": {
|
|
"detect.mobile_phone": {
|
|
"pipeline": "mobile_remote",
|
|
"model_id": "yolov8n-mobile-phone@1",
|
|
}
|
|
},
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def test_catalog_reports_only_capabilities_deployed_in_each_mode() -> None:
|
|
catalog = build_deployed_catalog(
|
|
_config(),
|
|
create_default_capability_registry(discover_entry_points=False),
|
|
)
|
|
|
|
assert [item.category for item in catalog.active_push] == ["detect.phone_use"]
|
|
assert [item.category for item in catalog.passive_invoke] == [
|
|
"detect.mobile_phone"
|
|
]
|
|
assert all(
|
|
item.deployment_status is CapabilityDeploymentStatus.STARTING
|
|
for item in (*catalog.active_push, *catalog.passive_invoke)
|
|
)
|
|
|
|
|
|
def test_catalog_uses_runtime_health_for_deployment_status() -> None:
|
|
catalog = build_deployed_catalog(
|
|
_config(),
|
|
create_default_capability_registry(discover_entry_points=False),
|
|
pipeline_health={
|
|
"active_phone": {
|
|
"source": HealthReport(True),
|
|
"detector": HealthReport(True),
|
|
},
|
|
"mobile_remote": {
|
|
"source": HealthReport(True),
|
|
"detector": HealthReport(False, "runtime task failed: model"),
|
|
},
|
|
},
|
|
)
|
|
|
|
assert (
|
|
catalog.active_push[0].deployment_status
|
|
is CapabilityDeploymentStatus.READY
|
|
)
|
|
assert (
|
|
catalog.passive_invoke[0].deployment_status
|
|
is CapabilityDeploymentStatus.FAILED
|
|
)
|
|
|
|
|
|
def test_catalog_rejects_category_to_model_mismatch() -> None:
|
|
config = _config()
|
|
config.server.routes["detect.mobile_phone"] # type: ignore[union-attr]
|
|
broken = config.model_copy(
|
|
update={
|
|
"server": config.server.model_copy( # type: ignore[union-attr]
|
|
update={
|
|
"routes": {
|
|
"detect.mobile_phone": config.server.routes[ # type: ignore[union-attr]
|
|
"detect.mobile_phone"
|
|
].model_copy(
|
|
update={"model_id": "people-talking-yolov8x@1"}
|
|
)
|
|
}
|
|
}
|
|
)
|
|
}
|
|
)
|
|
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError, match="does not match capability category"):
|
|
build_deployed_catalog(
|
|
broken,
|
|
create_default_capability_registry(discover_entry_points=False),
|
|
)
|