"""Step definitions for stream_router remaining coverage tests. Targets the uncovered lines and partial branches identified in build/coverage.xml for src/cleveragents/reactive/stream_router.py. """ from __future__ import annotations from typing import Any from unittest.mock import MagicMock, patch from behave import given, then, when from cleveragents.core.exceptions import StreamRoutingError from cleveragents.reactive.stream_router import ( ReactiveStreamRouter, SimpleLLMAgent, SimpleToolAgent, StreamConfig, StreamMessage, ) # --------------------------------------------------------------------------- # SimpleToolAgent.register_operation - invalid name (lines 121-122) # --------------------------------------------------------------------------- @given("a fresh SimpleToolAgent registry") def step_given_fresh_tool_agent_registry(context: Any) -> None: context.error = None @when('I register an operation named "bad-name!" with a lambda') def step_when_register_invalid_operation(context: Any) -> None: try: SimpleToolAgent.register_operation("bad-name!", lambda c, m, x: c) except ValueError as exc: context.error = exc @then("a ValueError should be raised about operation name format") def step_then_value_error_operation_name(context: Any) -> None: assert context.error is not None, "Expected ValueError but none was raised" assert isinstance(context.error, ValueError) assert "alphanumeric" in str(context.error).lower() # --------------------------------------------------------------------------- # SimpleToolAgent.process - tool is not a dict (lines 138-139) # --------------------------------------------------------------------------- @given("a SimpleToolAgent whose tools list contains a non-dict entry") def step_given_tool_agent_non_dict_tool(context: Any) -> None: # tools list where the first element is a string, not a dict context.tool_agent = SimpleToolAgent(tools=["not_a_dict"]) @when('I process content "{content}" through that SimpleToolAgent') def step_when_process_through_tool_agent(context: Any, content: str) -> None: context.tool_result = context.tool_agent.process(content) @then('the SimpleToolAgent should return "{expected}" unchanged') def step_then_tool_agent_returns_unchanged(context: Any, expected: str) -> None: assert context.tool_result == expected, ( f"Expected '{expected}', got '{context.tool_result}'" ) # --------------------------------------------------------------------------- # SimpleToolAgent.process - unknown operation name (lines 146-151) # --------------------------------------------------------------------------- @given('a SimpleToolAgent with an unknown operation "nonexistent_op"') def step_given_tool_agent_unknown_op(context: Any) -> None: context.tool_agent = SimpleToolAgent(tools=[{"operation": "nonexistent_op"}]) # --------------------------------------------------------------------------- # SimpleToolAgent.process - operation raises exception (lines 154-155) # --------------------------------------------------------------------------- @given("a SimpleToolAgent with an operation that raises an exception") def step_given_tool_agent_raising_op(context: Any) -> None: def _raise(content: Any, meta: Any, ctx: Any) -> None: raise RuntimeError("deliberate test explosion") SimpleToolAgent.register_operation("exploding_op", _raise) context.tool_agent = SimpleToolAgent(tools=[{"operation": "exploding_op"}]) @then("the SimpleToolAgent should return an empty string") def step_then_tool_agent_returns_empty(context: Any) -> None: assert context.tool_result == "", ( f"Expected empty string, got '{context.tool_result}'" ) # --------------------------------------------------------------------------- # ReactiveStreamRouter.register_transform - invalid name (lines 264-265) # --------------------------------------------------------------------------- @given("a fresh ReactiveStreamRouter registry") def step_given_fresh_router_registry(context: Any) -> None: context.error = None @when('I register a transform named "bad@name" with a lambda') def step_when_register_invalid_transform(context: Any) -> None: try: ReactiveStreamRouter.register_transform("bad@name", lambda x: x) except ValueError as exc: context.error = exc @then("a ValueError should be raised about transform name format") def step_then_value_error_transform_name(context: Any) -> None: assert context.error is not None, "Expected ValueError but none was raised" assert isinstance(context.error, ValueError) assert "alphanumeric" in str(context.error).lower() # --------------------------------------------------------------------------- # _create_operator transform - no fn or type (lines 430, 433) # --------------------------------------------------------------------------- @given("a reactive stream router for coverage") def step_given_router_for_coverage(context: Any) -> None: context.router = ReactiveStreamRouter() context.error = None context.result = None context.results = [] @given('a stream named "{name}" exists on the router') def step_given_stream_exists(context: Any, name: str) -> None: context.router.create_stream(StreamConfig(name=name)) @when("I create a transform operator with neither fn nor type") def step_when_create_transform_no_fn_no_type(context: Any) -> None: try: context.router._create_operator( {"type": "transform", "params": {"something_else": True}} ) except StreamRoutingError as exc: context.error = exc @then("a StreamRoutingError should be raised about missing transform params") def step_then_routing_error_transform_params(context: Any) -> None: assert context.error is not None, "Expected StreamRoutingError but none raised" assert isinstance(context.error, StreamRoutingError) assert "fn" in str(context.error) or "type" in str(context.error) # --------------------------------------------------------------------------- # SimpleLLMAgent._resolve_llm - no optional kwargs (branches 203, 205, 207) # --------------------------------------------------------------------------- @given("a SimpleLLMAgent with no temperature or max_tokens or max_retries") def step_given_llm_agent_no_optional_kwargs(context: Any) -> None: # Only provider and model set; temperature/max_tokens/max_retries absent context.llm_agent = SimpleLLMAgent( name="test-llm", config={"provider": "mock", "model": "mock-model"}, ) @given("a stub provider registry is active") def step_given_stub_provider_registry_active(context: Any) -> None: mock_llm = MagicMock() mock_llm.invoke.return_value = MagicMock(content="stub response") mock_registry = MagicMock() mock_registry.create_llm.return_value = mock_llm context._registry_patch = patch( "cleveragents.reactive.stream_router.get_provider_registry", return_value=mock_registry, ) context._registry_patch.start() context._mock_registry = mock_registry context._mock_llm = mock_llm if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(context._registry_patch.stop) @when("I resolve the LLM on that agent") def step_when_resolve_llm(context: Any) -> None: context.llm_agent._resolve_llm() @then("the LLM should be resolved without optional kwargs") def step_then_llm_resolved_no_optional(context: Any) -> None: call_kwargs = context._mock_registry.create_llm.call_args # The kwargs passed to create_llm should NOT include temperature, max_tokens, max_retries kw = call_kwargs.kwargs if hasattr(call_kwargs, "kwargs") else call_kwargs[1] assert "temperature" not in kw, f"temperature should not be in kwargs: {kw}" assert "max_tokens" not in kw, f"max_tokens should not be in kwargs: {kw}" assert "max_retries" not in kw, f"max_retries should not be in kwargs: {kw}" # --------------------------------------------------------------------------- # SimpleLLMAgent.process - empty system prompt (branch 230) # --------------------------------------------------------------------------- @given("a SimpleLLMAgent with an empty system prompt") def step_given_llm_agent_empty_system_prompt(context: Any) -> None: context.llm_agent = SimpleLLMAgent( name="test-llm", config={"provider": "mock", "model": "mock-model", "system_prompt": ""}, ) @when('I process content "test input" through that SimpleLLMAgent') def step_when_process_through_llm_agent(context: Any) -> None: context.llm_result = context.llm_agent.process("test input") @then("the LLM should receive only a HumanMessage") def step_then_llm_received_only_human(context: Any) -> None: from langchain_core.messages import HumanMessage call_args = context._mock_llm.invoke.call_args messages = call_args[0][0] # Should have exactly 1 message (HumanMessage), no SystemMessage assert len(messages) == 1, f"Expected 1 message, got {len(messages)}: {messages}" assert isinstance(messages[0], HumanMessage), ( f"Expected HumanMessage, got {type(messages[0])}" ) # --------------------------------------------------------------------------- # LangGraph bridge - bridge has factory method (branch 378) # --------------------------------------------------------------------------- @given("a LangGraph bridge stub with a graph_execute factory is registered") def step_given_langgraph_bridge_with_factory(context: Any) -> None: bridge = MagicMock() # The router looks for _operator_graph_execute on the bridge bridge._operator_graph_execute.return_value = lambda x: x context.router._langgraph_bridge = bridge context._bridge = bridge @when("I create a graph_execute operator via the bridge") def step_when_create_graph_execute_via_bridge(context: Any) -> None: try: context.result = context.router._create_operator( {"type": "graph_execute", "params": {"graph": "test"}} ) except StreamRoutingError as exc: context.error = exc @then("the bridge factory method should produce a valid operator") def step_then_bridge_factory_produces_operator(context: Any) -> None: assert context.error is None, f"Unexpected error: {context.error}" assert context.result is not None, "Expected an operator but got None" context._bridge._operator_graph_execute.assert_called_once() # --------------------------------------------------------------------------- # Switch operator - case_operators with None operator (branch 453) # --------------------------------------------------------------------------- @when("I run a switch operator whose case pipeline includes a None operator") def step_when_switch_case_null_operator(context: Any) -> None: """Build a switch operator where _create_operator returns None for a sub-op.""" results: list[Any] = [] # Build a switch config with case operators switch_config = { "type": "switch", "params": { "cases": [ { "condition": {}, # Always matches (empty condition = True) "operators": [ # This is a valid operator that returns identity {"type": "map", "params": {"transform": {"type": "identity"}}}, ], }, ], }, } # Patch _create_operator to return None for one of the case operators original_create = context.router._create_operator def _patched_create(config: dict[str, Any]) -> Any: if ( config.get("type") == "map" and config.get("params", {}).get("transform", {}).get("type") == "identity" ): return None return original_create(config) context.router._create_operator = _patched_create try: op = original_create(switch_config) # Create a message and run through the switch msg = StreamMessage(content="test", metadata={}) import rx rx.just(msg).pipe(op).subscribe(lambda x: results.append(x)) finally: context.router._create_operator = original_create context.results = results @then("the switch should still emit the message through the pipeline") def step_then_switch_emits_through_pipeline(context: Any) -> None: assert len(context.results) > 0, "Expected at least one message from switch" # --------------------------------------------------------------------------- # Switch operator - target_stream missing (branch 456) # --------------------------------------------------------------------------- @when("I run a switch with a matching case whose target stream does not exist") def step_when_switch_missing_target_stream(context: Any) -> None: results: list[Any] = [] switch_config = { "type": "switch", "params": { "cases": [ { "condition": {}, # Always matches "target": "nonexistent_stream", }, ], "default": None, }, } op = context.router._create_operator(switch_config) msg = StreamMessage(content="test", metadata={}) import rx rx.just(msg).pipe(op).subscribe(lambda x: results.append(x)) context.results = results @then("the switch should fall through to the default") def step_then_switch_falls_through(context: Any) -> None: # With no default either, the message should still come through via rx.just(msg) assert len(context.results) > 0, "Expected message to fall through" # --------------------------------------------------------------------------- # Switch operator - default_operators with None operator (branch 466) # --------------------------------------------------------------------------- @when("I run a switch operator whose default pipeline includes a None operator") def step_when_switch_default_null_operator(context: Any) -> None: results: list[Any] = [] switch_config = { "type": "switch", "params": { "cases": [ { "condition": {"equals": "will_never_match_this_value"}, "target": "nowhere", }, ], "default_operators": [ {"type": "map", "params": {"transform": {"type": "identity"}}}, ], }, } original_create = context.router._create_operator def _patched_create(config: dict[str, Any]) -> Any: if ( config.get("type") == "map" and config.get("params", {}).get("transform", {}).get("type") == "identity" ): return None return original_create(config) context.router._create_operator = _patched_create try: op = original_create(switch_config) msg = StreamMessage(content="test", metadata={}) import rx rx.just(msg).pipe(op).subscribe(lambda x: results.append(x)) finally: context.router._create_operator = original_create context.results = results @then("the switch should still emit the message through the default pipeline") def step_then_switch_emits_through_default(context: Any) -> None: assert len(context.results) > 0, ( "Expected at least one message from default pipeline" ) # --------------------------------------------------------------------------- # _apply_transform - replace with non-dict message (branch 526) # --------------------------------------------------------------------------- @when("I apply a replace transform to a non-dict message") def step_when_apply_replace_to_non_dict(context: Any) -> None: context.result = context.router._apply_transform( "a plain string", {"type": "replace", "target": "key", "value": "val"} ) @then("the message should be returned unchanged") def step_then_message_returned_unchanged(context: Any) -> None: assert context.result == "a plain string", ( f"Expected original, got {context.result}" ) # --------------------------------------------------------------------------- # _apply_transform - extract_field (branch 529) # --------------------------------------------------------------------------- @when("I apply an extract_field transform to a dict message with the field") def step_when_apply_extract_field_dict(context: Any) -> None: context.result = context.router._apply_transform( {"name": "Alice", "age": 30}, {"type": "extract_field", "field": "name"}, ) @then("the extracted field value should be returned") def step_then_extracted_field_value(context: Any) -> None: assert context.result == "Alice", f"Expected 'Alice', got {context.result}" @when("I apply an extract_field transform to a non-dict message") def step_when_apply_extract_field_non_dict(context: Any) -> None: context.result = context.router._apply_transform( "just a string", {"type": "extract_field", "field": "name"}, ) @then("the non-dict message should be returned from extract_field") def step_then_non_dict_returned_extract_field(context: Any) -> None: assert context.result == "just a string", ( f"Expected 'just a string', got '{context.result}'" ) # --------------------------------------------------------------------------- # _evaluate_condition - field condition with dict (branch 542) # --------------------------------------------------------------------------- @when("I evaluate a field condition against a dict message containing the field") def step_when_evaluate_field_condition_dict_has(context: Any) -> None: context.condition_result = context.router._evaluate_condition( {"status": "active"}, {"field": "status"} ) @then("the condition should return true") def step_then_condition_true(context: Any) -> None: assert context.condition_result is True, ( f"Expected True, got {context.condition_result}" ) @when("I evaluate a field condition against a dict message missing the field") def step_when_evaluate_field_condition_dict_missing(context: Any) -> None: context.condition_result = context.router._evaluate_condition( {"other": "value"}, {"field": "status"} ) @then("the condition should return false") def step_then_condition_false(context: Any) -> None: assert context.condition_result is False, ( f"Expected False, got {context.condition_result}" ) # --------------------------------------------------------------------------- # split_stream - target missing from streams (branch 609) # --------------------------------------------------------------------------- @when("I split the stream with a condition targeting a nonexistent stream") def step_when_split_missing_target(context: Any) -> None: results: list[Any] = [] context.router.split_stream( "split_source", [{"target": "does_not_exist", "condition": {}}], ) # Send a message - it shouldn't route anywhere msg = StreamMessage(content="test", metadata={}) context.router.streams["split_source"].on_next(msg) context.results = results @then("no error should occur and the message should not route") def step_then_no_error_no_route(context: Any) -> None: # If we got here without an exception, the test passes assert True # --------------------------------------------------------------------------- # merge_streams - target doesn't exist (branch 622) # --------------------------------------------------------------------------- @when("I merge streams into a target that does not exist yet") def step_when_merge_target_not_exist(context: Any) -> None: results: list[Any] = [] context.router.merge_streams(["merge_src_a", "merge_src_b"], "auto_created_target") # Subscribe to the auto-created target context.router.streams["auto_created_target"].subscribe(lambda x: results.append(x)) # Send messages through sources msg = StreamMessage(content="from_a", metadata={}) context.router.streams["merge_src_a"].on_next(msg) context.results = results @then("the target stream should be created and receive merged messages") def step_then_target_created_receives(context: Any) -> None: assert "auto_created_target" in context.router.streams, ( "Target stream was not auto-created" ) assert len(context.results) > 0, "Expected messages in merged target" # --------------------------------------------------------------------------- # merge_streams - source doesn't exist (branch 628) # --------------------------------------------------------------------------- @when("I merge streams including a nonexistent source into an existing target") def step_when_merge_nonexistent_source(context: Any) -> None: # Create a target first context.router.create_stream(StreamConfig(name="merge_target_c")) results: list[Any] = [] context.router.streams["merge_target_c"].subscribe(lambda x: results.append(x)) # Merge with one real source and one nonexistent context.router.merge_streams( ["merge_src_c", "nonexistent_source"], "merge_target_c" ) # Send a message through the real source msg = StreamMessage(content="from_c", metadata={}) context.router.streams["merge_src_c"].on_next(msg) context.results = results @then("only the existing source should be subscribed") def step_then_only_existing_subscribed(context: Any) -> None: # The real source should have routed successfully assert len(context.results) > 0, "Expected messages from existing source" # --------------------------------------------------------------------------- # dispose - stream without dispose attribute (branch 657) # --------------------------------------------------------------------------- @given("a stream that lacks a dispose method is injected into the router") def step_given_stream_without_dispose(context: Any) -> None: # Inject a plain object without a dispose method into the streams dict class NoDispose: pass context.router.streams["no_dispose_stream"] = NoDispose() @when("I dispose the router") def step_when_dispose_router(context: Any) -> None: try: context.router.dispose() context.error = None except Exception as exc: context.error = exc @then("the router should be fully cleared without errors") def step_then_router_cleared(context: Any) -> None: assert context.error is None, f"Unexpected error during dispose: {context.error}" assert len(context.router.streams) == 0, "Streams not cleared" assert len(context.router.observables) == 0, "Observables not cleared" assert len(context.router.agents) == 0, "Agents not cleared" # --------------------------------------------------------------------------- # _apply_transform - unknown transform type (branch 529->533) # --------------------------------------------------------------------------- @when("I apply a transform with unknown type to a message") def step_when_apply_unknown_transform_type(context: Any) -> None: context.result = context.router._apply_transform( {"key": "value"}, {"type": "unknown_type", "field": "key"} ) @then("the original message should be returned") def step_then_original_message_returned(context: Any) -> None: assert context.result == {"key": "value"}, ( f"Expected original dict, got {context.result}" ) # --------------------------------------------------------------------------- # _evaluate_condition - field condition with non-dict (branch 542->544) # --------------------------------------------------------------------------- @when("I evaluate a field condition against a non-dict message") def step_when_evaluate_field_non_dict(context: Any) -> None: # message is a StreamMessage object (not a dict), so isinstance check is False msg = StreamMessage(content="text", metadata={}) context.condition_result = context.router._evaluate_condition( msg, {"field": "status"} ) # --------------------------------------------------------------------------- # LangGraph bridge - bridge without factory method (branch 378->381) # --------------------------------------------------------------------------- @given("a LangGraph bridge stub without the requested factory method is registered") def step_given_bridge_without_factory(context: Any) -> None: # Create a bridge object that does NOT have _operator_graph_execute bridge = MagicMock(spec=[]) context.router._langgraph_bridge = bridge @when("I attempt to create a graph_execute operator via the bridge") def step_when_attempt_graph_execute_no_factory(context: Any) -> None: try: context.result = context.router._create_operator( {"type": "graph_execute", "params": {"graph": "test"}} ) except StreamRoutingError as exc: context.error = exc @then("a StreamRoutingError should be raised about LangGraph bridge") def step_then_routing_error_langgraph(context: Any) -> None: assert context.error is not None, "Expected StreamRoutingError but none raised" assert isinstance(context.error, StreamRoutingError) assert "LangGraph" in str(context.error)