forked from HAL9000/cleveragents-core
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""Steps for remaining stream_router uncovered lines."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import rx # type: ignore
|
|
from behave import then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.reactive.stream_router import StreamMessage, StreamType
|
|
|
|
|
|
@when("I convert a dict route config with only a name")
|
|
def step_convert_dict_config(context: Context) -> None:
|
|
context.stream_config = context.stream_router._to_stream_config({"name": "dict"})
|
|
context.expected_name = "dict"
|
|
|
|
|
|
@when("I convert a minimal object route config")
|
|
def step_convert_object_config(context: Context) -> None:
|
|
class _RouteConfig:
|
|
def __init__(self) -> None:
|
|
self.name = "object"
|
|
|
|
context.stream_config = context.stream_router._to_stream_config(_RouteConfig())
|
|
context.expected_name = "object"
|
|
|
|
|
|
@then("the stream config should use default values")
|
|
def step_verify_defaults(context: Context) -> None:
|
|
config = context.stream_config
|
|
assert config.name == context.expected_name
|
|
assert config.type == StreamType.COLD
|
|
assert config.buffer_size == 1
|
|
assert config.operators == []
|
|
assert config.subscriptions == []
|
|
assert config.publications == []
|
|
assert config.agents == []
|
|
|
|
|
|
@when("I apply a transform operator with type replace")
|
|
def step_transform_operator_type(context: Context) -> None:
|
|
operator = context.stream_router._create_operator(
|
|
{
|
|
"type": "transform",
|
|
"params": {"type": "replace", "target": "value", "value": 5},
|
|
}
|
|
)
|
|
captured: list[dict[str, int]] = []
|
|
rx.just({"value": 1}).pipe(operator).subscribe(captured.append)
|
|
context.transform_result = captured[0]
|
|
|
|
|
|
@then("the transform operator should update the field")
|
|
def step_verify_transform_operator(context: Context) -> None:
|
|
assert context.transform_result["value"] == 5
|
|
|
|
|
|
@when("I run a switch operator with no matches and no defaults")
|
|
def step_switch_no_matches(context: Context) -> None:
|
|
operator = context.stream_router._create_operator(
|
|
{
|
|
"type": "switch",
|
|
"params": {"cases": [{"condition": {"equals": "nope"}}]},
|
|
}
|
|
)
|
|
message = StreamMessage(content="ping")
|
|
captured: list[StreamMessage] = []
|
|
rx.just(message).pipe(operator).subscribe(captured.append)
|
|
context.switch_result = captured[0]
|
|
|
|
|
|
@then("the switch operator should emit the original message")
|
|
def step_verify_switch_fallback(context: Context) -> None:
|
|
assert isinstance(context.switch_result, StreamMessage)
|
|
assert context.switch_result.content == "ping"
|
|
|
|
|
|
@when("I evaluate an unsupported condition")
|
|
def step_eval_unsupported(context: Context) -> None:
|
|
context.condition_result = context.stream_router._evaluate_condition(
|
|
"value", {"unsupported": True}
|
|
)
|
|
|
|
|
|
@then("the unsupported condition result should be false")
|
|
def step_verify_condition_false(context: Context) -> None:
|
|
assert context.condition_result is False
|
|
|
|
|
|
@when("I map a message with an agent lacking processors")
|
|
def step_mapper_no_processor(context: Context) -> None:
|
|
class _BareAgent:
|
|
pass
|
|
|
|
mapper = context.stream_router._create_agent_mapper(_BareAgent())
|
|
message = StreamMessage(content="raw", metadata={"k": "v"})
|
|
context.mapper_result = mapper(message)
|
|
|
|
|
|
@then("the mapper should return the original content with metadata")
|
|
def step_verify_mapper_no_processor(context: Context) -> None:
|
|
result = context.mapper_result
|
|
assert result.content == "raw"
|
|
assert result.metadata.get("processed_by")
|
|
assert result.metadata.get("k") == "v"
|