83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from cmvr_edge_ai.application import EdgeAIApplication
|
|
from cmvr_edge_ai.config import load_config_data
|
|
from cmvr_edge_ai.core import ComponentContext, Source
|
|
from cmvr_edge_ai.plugins import PluginKind, PluginRegistry, PluginSpec
|
|
|
|
|
|
class _MetadataSource(Source):
|
|
def __init__(self, node_id: str, params: Mapping[str, Any]) -> None:
|
|
del node_id, params
|
|
self.seen: object | None = None
|
|
|
|
async def setup(self, context: ComponentContext) -> None:
|
|
self.seen = context.metadata.get("invocation_broker")
|
|
|
|
async def messages(self): # type: ignore[no-untyped-def]
|
|
if False:
|
|
yield None
|
|
|
|
|
|
def test_application_passes_embedding_metadata_to_pipeline_components() -> None:
|
|
source = _MetadataSource("source", {})
|
|
registry = PluginRegistry()
|
|
registry.register(
|
|
PluginSpec(
|
|
plugin_id="test.metadata_source@1",
|
|
kind=PluginKind.SOURCE,
|
|
factory=lambda _node_id, _params: source,
|
|
outputs={"output": "*"},
|
|
)
|
|
)
|
|
config = load_config_data(
|
|
{
|
|
"api_version": "cmvr.edge.ai/v1",
|
|
"pipelines": {
|
|
"metadata": {
|
|
"nodes": {"source": {"uses": "test.metadata_source@1"}}
|
|
}
|
|
},
|
|
}
|
|
)
|
|
broker = object()
|
|
|
|
async def exercise() -> None:
|
|
application = EdgeAIApplication(
|
|
config,
|
|
registry,
|
|
metadata={"invocation_broker": broker},
|
|
)
|
|
await application.start()
|
|
await application.wait()
|
|
await application.stop()
|
|
|
|
asyncio.run(exercise())
|
|
assert source.seen is broker
|
|
|
|
|
|
def test_application_rejects_metadata_that_overrides_owned_resources() -> None:
|
|
config = load_config_data(
|
|
{
|
|
"api_version": "cmvr.edge.ai/v1",
|
|
"pipelines": {
|
|
"empty": {
|
|
"nodes": {"source": {"uses": "test.source@1"}}
|
|
}
|
|
},
|
|
}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="must not override owned resource"):
|
|
EdgeAIApplication(
|
|
config,
|
|
PluginRegistry(),
|
|
metadata={"thread_executor": object()},
|
|
)
|