219 lines
7.4 KiB
Python
219 lines
7.4 KiB
Python
"""Steps for stream router coverage (actor-first).
|
|
|
|
These steps focus on creation/error cases and lightweight operator helpers
|
|
compatible with the current ReactiveStreamRouter implementation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.core.exceptions import StreamRoutingError
|
|
from cleveragents.reactive.stream_router import (
|
|
ReactiveStreamRouter,
|
|
StreamConfig,
|
|
StreamMessage,
|
|
StreamType,
|
|
)
|
|
|
|
|
|
class DummyAgent:
|
|
def __init__(self, name: str):
|
|
self.name = name
|
|
|
|
def process_message_sync(self, message, metadata=None): # type: ignore[override]
|
|
return f"echo:{message}"
|
|
|
|
def get_capabilities(self): # pragma: no cover - not used in assertions
|
|
return []
|
|
|
|
|
|
@given("I have a stream message for router testing")
|
|
def step_stream_message_testing(context):
|
|
context.stream_message = StreamMessage(
|
|
content="orig", metadata={"a": 1}, source_stream="s1", timestamp=1.0
|
|
)
|
|
context.original_msg = context.stream_message
|
|
|
|
|
|
@given("a reactive stream router harness")
|
|
def step_router_harness(context):
|
|
context.router = ReactiveStreamRouter()
|
|
context.errors = {}
|
|
context.last_messages = []
|
|
|
|
|
|
@given("I have a replay stream configuration:")
|
|
@given("I have a hot stream configuration:")
|
|
@given("I have a stream configuration:")
|
|
@given("I have a stream configuration with agent mapper:")
|
|
@given("I have a stream configuration with missing agent:")
|
|
@given("I have a stream configuration with extract transform:")
|
|
@given("I have a stream configuration with unknown operator:")
|
|
@given("I have a stream configuration with LangGraph operator:")
|
|
@given("I have a stream configuration with equals filter:")
|
|
@given("I have a stream configuration with field condition:")
|
|
@given("I have a stream configuration with buffer and window:")
|
|
@given("I have a stream configuration with take skip merge:")
|
|
@given("I have a stream configuration with accumulate:")
|
|
@given("I have a stream configuration with switch operator:")
|
|
def step_load_stream_config(context):
|
|
cfg_text = context.text.strip()
|
|
cfg_dict = json.loads(cfg_text)
|
|
cfg_dict.setdefault("operators", [])
|
|
if "type" in cfg_dict:
|
|
cfg_dict["type"] = StreamType(cfg_dict["type"].lower())
|
|
context.stream_config = StreamConfig(**cfg_dict)
|
|
|
|
|
|
@given("I have a stream for router testing")
|
|
def step_stream_for_testing(context):
|
|
context.stream_config = StreamConfig(name="test_stream")
|
|
context.router.create_stream(context.stream_config)
|
|
|
|
|
|
@given('I have a tool agent named "{name}" with echo tool')
|
|
def step_tool_agent(context, name):
|
|
agent = DummyAgent(name)
|
|
context.router.register_agent(name, agent)
|
|
|
|
|
|
@when("I create the reactive stream")
|
|
@given("I create the reactive stream")
|
|
def step_create_stream(context):
|
|
try:
|
|
context.router.create_stream(context.stream_config)
|
|
context.errors["create_stream"] = None
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.errors["create_stream"] = exc
|
|
|
|
|
|
@when("I try to create the reactive stream")
|
|
@when("I try to create another stream with the same name")
|
|
def step_create_stream_error(context):
|
|
try:
|
|
context.router.create_stream(context.stream_config)
|
|
context.errors["create_stream"] = None
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.errors["create_stream"] = exc
|
|
|
|
|
|
@when('I send router message "{message}" to stream "{name}"')
|
|
def step_send_message(context, message, name):
|
|
if name not in context.router.streams:
|
|
context.router.create_stream(StreamConfig(name=name))
|
|
collected = []
|
|
context.router.streams[name].subscribe(collected.append)
|
|
context.router.send_message(name, message)
|
|
context.last_messages = collected
|
|
|
|
|
|
@when('I send router None message to stream "{name}"')
|
|
def step_send_none_message(context, name):
|
|
step_send_message(context, None, name)
|
|
|
|
|
|
@when('I send a dictionary message with field "{field}"')
|
|
def step_send_dict_message(context, field):
|
|
name = context.stream_config.name
|
|
if name not in context.router.streams:
|
|
context.router.create_stream(context.stream_config)
|
|
collected = []
|
|
context.router.streams[name].subscribe(collected.append)
|
|
context.router.send_message(name, {field: "extracted"})
|
|
context.last_messages = collected
|
|
|
|
|
|
@when("I try to send message to missing stream")
|
|
def step_send_missing_stream(context):
|
|
try:
|
|
context.router.send_message("missing_stream", "data")
|
|
context.errors["send"] = None
|
|
except Exception as exc: # pylint: disable=broad-except
|
|
context.errors["send"] = exc
|
|
|
|
|
|
@when("I copy the message with modifications")
|
|
def step_copy_message(context):
|
|
msg = StreamMessage(
|
|
content="orig", metadata={"a": 1}, source_stream="s1", timestamp=1.0
|
|
)
|
|
context.original_msg = msg
|
|
context.copied_msg = msg.copy_with(content="new", metadata={"b": 2})
|
|
|
|
|
|
@when("I dispose of the stream router")
|
|
def step_dispose_router(context):
|
|
context.router.dispose()
|
|
|
|
|
|
@when("I send a message to check timestamp")
|
|
def step_send_timestamp(context):
|
|
name = "ts_stream"
|
|
if name not in context.router.streams:
|
|
context.router.create_stream(StreamConfig(name=name))
|
|
collected = []
|
|
context.router.streams[name].subscribe(collected.append)
|
|
context.router.send_message(name, "payload")
|
|
context.last_messages = collected
|
|
|
|
|
|
@then("the stream should be created successfully")
|
|
def step_stream_created(context):
|
|
assert context.stream_config.name in context.router.streams
|
|
assert context.errors.get("create_stream") is None
|
|
|
|
|
|
@then("I should get a stream router coverage error")
|
|
@then("I should get a stream router coverage error about missing agent")
|
|
@then("I should get a stream router coverage error about unknown operator")
|
|
@then("I should get a stream router coverage error about LangGraph bridge")
|
|
def step_stream_error(context):
|
|
err = context.errors.get("create_stream")
|
|
assert isinstance(err, StreamRoutingError)
|
|
|
|
|
|
@then("the field value should be extracted")
|
|
def step_field_extracted(context):
|
|
# Directly apply transform to a dict message
|
|
transform = context.stream_config.operators[0]["params"]["transform"]
|
|
result = context.router._apply_transform({"data": "value"}, transform) # pylint: disable=protected-access
|
|
assert result == "value"
|
|
|
|
|
|
@then("the new message should have the modifications")
|
|
def step_new_message_has_mods(context):
|
|
assert context.copied_msg.content == "new"
|
|
assert context.copied_msg.metadata == {"b": 2}
|
|
|
|
|
|
@then("the original message should be unchanged")
|
|
def step_original_message_unchanged(context):
|
|
assert context.original_msg.content == "orig"
|
|
assert context.original_msg.metadata == {"a": 1}
|
|
|
|
|
|
@then("the agent should handle None gracefully")
|
|
def step_agent_handles_none(context):
|
|
# Agent mapper returns string even on None
|
|
mapper = context.router._create_agent_mapper(DummyAgent("none_agent")) # pylint: disable=protected-access
|
|
result = mapper(None)
|
|
assert isinstance(result, StreamMessage)
|
|
assert "echo:" in result.content
|
|
|
|
|
|
@then("the stream router should be disposed properly")
|
|
def step_disposed(context):
|
|
assert context.router.streams == {}
|
|
assert context.router.observables == {}
|
|
assert context.router.stream_configs == {}
|
|
|
|
|
|
@then("the message should have a timestamp")
|
|
def step_timestamp_present(context):
|
|
assert context.last_messages
|
|
msg = context.last_messages[-1]
|
|
assert getattr(msg, "timestamp", None) is not None
|