373 lines
10 KiB
Python
373 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from cmvr_edge_ai.compiler import PipelineCompileError, compile_pipeline
|
|
from cmvr_edge_ai.config import load_config_data
|
|
from cmvr_edge_ai.core import Envelope, Operator, Sink, Source
|
|
from cmvr_edge_ai.plugins import PluginKind, PluginRegistry, PluginSpec
|
|
|
|
|
|
class EmptySource(Source):
|
|
async def messages(self): # type: ignore[no-untyped-def]
|
|
if False:
|
|
yield Envelope("unused")
|
|
|
|
|
|
class IdentityOperator(Operator):
|
|
async def process(
|
|
self, envelope: Envelope[Any], input_port: str = "input"
|
|
) -> Envelope[Any]:
|
|
del input_port
|
|
return envelope
|
|
|
|
|
|
class NullSink(Sink):
|
|
async def consume(self, envelope: Envelope[Any], input_port: str = "input") -> None:
|
|
del envelope, input_port
|
|
|
|
|
|
def _source_factory(node_id: str, params: Mapping[str, Any]) -> EmptySource:
|
|
del node_id, params
|
|
return EmptySource()
|
|
|
|
|
|
def _operator_factory(node_id: str, params: Mapping[str, Any]) -> IdentityOperator:
|
|
del node_id, params
|
|
return IdentityOperator()
|
|
|
|
|
|
def _sink_factory(node_id: str, params: Mapping[str, Any]) -> NullSink:
|
|
del node_id, params
|
|
return NullSink()
|
|
|
|
|
|
def _registry() -> PluginRegistry:
|
|
registry = PluginRegistry()
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.text_source@1",
|
|
PluginKind.SOURCE,
|
|
_source_factory,
|
|
outputs={"events": "TextEvent/v1"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.text_operator@1",
|
|
PluginKind.OPERATOR,
|
|
_operator_factory,
|
|
inputs={"input": "TextEvent/v1"},
|
|
outputs={"events": "TextEvent/v1"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.audio_sink@1",
|
|
PluginKind.SINK,
|
|
_sink_factory,
|
|
inputs={"input": "AudioChunk/v1"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.any_sink@1",
|
|
PluginKind.SINK,
|
|
_sink_factory,
|
|
inputs={"input": "*"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.command_source@1",
|
|
PluginKind.SOURCE,
|
|
_source_factory,
|
|
outputs={"commands": "RobotCommand/v1"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.approved_source@1",
|
|
PluginKind.SOURCE,
|
|
_source_factory,
|
|
outputs={"commands": "ApprovedRobotCommand/v1"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.safety_gate@1",
|
|
PluginKind.OPERATOR,
|
|
_operator_factory,
|
|
inputs={"input": "RobotCommand/v1"},
|
|
outputs={"commands": "ApprovedRobotCommand/v1"},
|
|
tags=frozenset({"safety_gate"}),
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.approved_operator@1",
|
|
PluginKind.OPERATOR,
|
|
_operator_factory,
|
|
inputs={"input": "ApprovedRobotCommand/v1"},
|
|
outputs={"commands": "ApprovedRobotCommand/v1"},
|
|
)
|
|
)
|
|
registry.register(
|
|
PluginSpec(
|
|
"test.actuator@1",
|
|
PluginKind.SINK,
|
|
_sink_factory,
|
|
inputs={"commands": "ApprovedRobotCommand/v1"},
|
|
tags=frozenset({"actuator"}),
|
|
)
|
|
)
|
|
return registry
|
|
|
|
|
|
def _config(nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]]):
|
|
return load_config_data(
|
|
{
|
|
"api_version": "cmvr.edge.ai/v1",
|
|
"pipelines": {
|
|
"test": {
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("edge", "message"),
|
|
[
|
|
(
|
|
{"from": "source.missing", "to": "sink.input"},
|
|
"unknown output port",
|
|
),
|
|
(
|
|
{"from": "source.events", "to": "sink.missing"},
|
|
"unknown input port",
|
|
),
|
|
],
|
|
)
|
|
def test_compile_rejects_unknown_plugin_ports(
|
|
edge: dict[str, str], message: str
|
|
) -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.text_source@1"},
|
|
"sink": {"uses": "test.any_sink@1"},
|
|
},
|
|
[edge],
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match=message):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
def test_compile_rejects_incompatible_schemas() -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.text_source@1"},
|
|
"sink": {"uses": "test.audio_sink@1"},
|
|
},
|
|
[{"from": "source.events", "to": "sink.input"}],
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match="schema mismatch"):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
def test_compile_accepts_wildcard_schema() -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.text_source@1"},
|
|
"sink": {"uses": "test.any_sink@1"},
|
|
},
|
|
[{"from": "source.events", "to": "sink.input"}],
|
|
)
|
|
|
|
compiled = compile_pipeline(config, "test", _registry())
|
|
|
|
assert compiled.pipeline_id == "test"
|
|
assert compiled.runtime.edges[0].source_port == "events"
|
|
|
|
|
|
def test_video_contiguous_qos_requires_lossless_overflow() -> None:
|
|
nodes = {
|
|
"source": {"uses": "test.text_source@1"},
|
|
"operator": {"uses": "test.text_operator@1"},
|
|
}
|
|
accepted = _config(
|
|
nodes,
|
|
[
|
|
{
|
|
"from": "source.events",
|
|
"to": "operator.input",
|
|
"qos": {
|
|
"profile": "video_contiguous",
|
|
"capacity": 8,
|
|
"overflow": "block",
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
compiled = compile_pipeline(accepted, "test", _registry())
|
|
assert compiled.runtime.edges[0].capacity == 8
|
|
|
|
rejected = _config(
|
|
nodes,
|
|
[
|
|
{
|
|
"from": "source.events",
|
|
"to": "operator.input",
|
|
"qos": {
|
|
"profile": "video_contiguous",
|
|
"capacity": 2,
|
|
"overflow": "drop_oldest",
|
|
},
|
|
}
|
|
],
|
|
)
|
|
with pytest.raises(PipelineCompileError, match="does not allow"):
|
|
compile_pipeline(rejected, "test", _registry())
|
|
|
|
|
|
def test_compile_rejects_direct_path_into_actuator() -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.approved_source@1"},
|
|
"actuator": {"uses": "test.actuator@1"},
|
|
},
|
|
[{"from": "source.commands", "to": "actuator.commands"}],
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match="must pass through a safety gate"):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
def test_compile_accepts_actuator_only_after_safety_gate() -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.command_source@1"},
|
|
"gate": {"uses": "test.safety_gate@1"},
|
|
"actuator": {"uses": "test.actuator@1"},
|
|
},
|
|
[
|
|
{"from": "source.commands", "to": "gate.input"},
|
|
{"from": "gate.commands", "to": "actuator.commands"},
|
|
],
|
|
)
|
|
|
|
compiled = compile_pipeline(config, "test", _registry())
|
|
|
|
assert set(compiled.plugin_specs) == {"source", "gate", "actuator"}
|
|
|
|
|
|
def test_compile_rejects_mixed_safe_and_unsafe_actuator_paths() -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.command_source@1"},
|
|
"bypass": {"uses": "test.approved_source@1"},
|
|
"gate": {"uses": "test.safety_gate@1"},
|
|
"actuator": {"uses": "test.actuator@1"},
|
|
},
|
|
[
|
|
{"from": "source.commands", "to": "gate.input"},
|
|
{"from": "gate.commands", "to": "actuator.commands"},
|
|
{"from": "bypass.commands", "to": "actuator.commands"},
|
|
],
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match="unsafe actuator"):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
def test_compile_rejects_operator_between_gate_and_actuator() -> None:
|
|
config = _config(
|
|
{
|
|
"source": {"uses": "test.command_source@1"},
|
|
"gate": {"uses": "test.safety_gate@1"},
|
|
"transform": {"uses": "test.approved_operator@1"},
|
|
"actuator": {"uses": "test.actuator@1"},
|
|
},
|
|
[
|
|
{"from": "source.commands", "to": "gate.input"},
|
|
{"from": "gate.commands", "to": "transform.input"},
|
|
{"from": "transform.commands", "to": "actuator.commands"},
|
|
],
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match="safety gate directly"):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("node_options", "message"),
|
|
[
|
|
({"execution": {"mode": "thread"}}, "execution settings not implemented"),
|
|
({"resources": {"memory_mb": 256}}, "resource limits"),
|
|
],
|
|
)
|
|
def test_compile_rejects_unenforced_node_settings(
|
|
node_options: dict[str, Any], message: str
|
|
) -> None:
|
|
config = _config(
|
|
{
|
|
"source": {
|
|
"uses": "test.text_source@1",
|
|
**node_options,
|
|
}
|
|
},
|
|
[],
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match=message):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("max_processes", 2),
|
|
("process_start_method", "forkserver"),
|
|
("health_bind", "127.0.0.1:8080"),
|
|
("reserved_memory_mb", 256),
|
|
],
|
|
)
|
|
def test_compile_rejects_reserved_runtime_fields(field: str, value: Any) -> None:
|
|
config = load_config_data(
|
|
{
|
|
"api_version": "cmvr.edge.ai/v1",
|
|
"runtime": {field: value},
|
|
"pipelines": {
|
|
"test": {"nodes": {"source": {"uses": "test.text_source@1"}}}
|
|
},
|
|
}
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match="reserved but not enforced"):
|
|
compile_pipeline(config, "test", _registry())
|
|
|
|
|
|
def test_compile_rejects_pipeline_priority_without_scheduler() -> None:
|
|
config = load_config_data(
|
|
{
|
|
"api_version": "cmvr.edge.ai/v1",
|
|
"pipelines": {
|
|
"test": {
|
|
"priority": "high",
|
|
"nodes": {"source": {"uses": "test.text_source@1"}},
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
with pytest.raises(PipelineCompileError, match="no priority scheduler"):
|
|
compile_pipeline(config, "test", _registry())
|