"""Step definitions targeting uncovered lines in reactive/route.py. Covers: - lines 90-99: to_stream_config() success path - line 113: MESSAGE_ROUTER rules propagation in to_graph_config() - line 177: from_graph_config() fallback when node type has no .value """ from __future__ import annotations from behave import given, then, when from cleveragents.langgraph.graph import GraphConfig from cleveragents.langgraph.nodes import NodeConfig, NodeType from cleveragents.reactive.route import RouteConfig, RouteType from cleveragents.reactive.stream_router import StreamType @given("the reactive route coverage module is loaded") def step_module_loaded(context): """Ensure the route module is importable and set up context.""" context.route_config = None context.stream_config_result = None context.graph_config_result = None context.result_route_config = None # ── Scenario: Convert a STREAM RouteConfig to StreamConfig successfully ── @given("I have a stream RouteConfig with full configuration") def step_full_stream_route_config(context): context.route_config = RouteConfig( name="full_stream", type=RouteType.STREAM, stream_type=StreamType.HOT, operators=[ {"type": "map", "params": {"function": "transform"}}, {"type": "filter", "params": {"predicate": "is_valid"}}, ], subscriptions=["input_topic"], publications=["output_topic"], agents=["agent_a", "agent_b"], initial_value="init_val", buffer_size=16, template_config={"template": "my_template"}, ) @when("I convert the stream RouteConfig to a StreamConfig") def step_convert_route_to_stream_config(context): context.stream_config_result = context.route_config.to_stream_config() @then("the StreamConfig should have the correct name") def step_stream_config_name(context): assert context.stream_config_result.name == context.route_config.name @then("the StreamConfig should have the correct stream type") def step_stream_config_type(context): assert context.stream_config_result.type == context.route_config.stream_type @then("the StreamConfig should preserve all operator definitions") def step_stream_config_operators(context): assert context.stream_config_result.operators == context.route_config.operators @then("the StreamConfig should preserve subscriptions and publications") def step_stream_config_subs_pubs(context): assert ( context.stream_config_result.subscriptions == context.route_config.subscriptions ) assert ( context.stream_config_result.publications == context.route_config.publications ) @then("the StreamConfig should preserve agents and buffer size") def step_stream_config_agents_buffer(context): assert context.stream_config_result.agents == context.route_config.agents assert context.stream_config_result.buffer_size == context.route_config.buffer_size @then("the StreamConfig should preserve initial value and template config") def step_stream_config_initial_template(context): assert ( context.stream_config_result.initial_value == context.route_config.initial_value ) assert ( context.stream_config_result.template_config == context.route_config.template_config ) # ── Scenario: Convert a STREAM RouteConfig with defaults to StreamConfig ── @given("I have a minimal stream RouteConfig with no explicit stream type") def step_minimal_stream_route_config(context): context.route_config = RouteConfig( name="minimal_stream", type=RouteType.STREAM, ) @then("the StreamConfig stream type should be COLD") def step_stream_config_default_cold(context): assert context.stream_config_result.type == StreamType.COLD # ── Scenario: MESSAGE_ROUTER rules propagation ── @given( "I have a graph RouteConfig with a MESSAGE_ROUTER node containing top-level rules" ) def step_graph_route_with_message_router_rules(context): context.route_config = RouteConfig( name="router_graph", type=RouteType.GRAPH, nodes={ "start_node": {"type": "agent", "agent": "agentA"}, "router_node": { "type": "message_router", "rules": [ {"target": "end", "condition": "done"}, {"target": "start_node", "condition": "retry"}, ], "metadata": {"existing_key": "existing_value"}, }, }, edges=[ {"source": "start_node", "target": "router_node"}, {"source": "router_node", "target": "start_node"}, ], ) @when("I convert the graph RouteConfig to a GraphConfig") def step_convert_route_to_graph_config(context): context.graph_config_result = context.route_config.to_graph_config() @then("the resulting MESSAGE_ROUTER node metadata should contain the rules") def step_router_node_metadata_has_rules(context): router_node = context.graph_config_result.nodes["router_node"] assert "rules" in router_node.metadata assert len(router_node.metadata["rules"]) == 2 assert router_node.metadata["rules"][0]["target"] == "end" # Verify existing metadata is also preserved assert router_node.metadata.get("existing_key") == "existing_value" # ── Scenario: from_graph_config fallback when node type is None ── @given("I have a GraphConfig with a node whose type attribute is None") def step_graph_config_node_type_none(context): # Build a stub node config object where .type is None class _StubNodeConfig: def __init__(self): self.type = None self.agent = None self.function = "my_func" self.tools = [] self.retry_policy = None self.timeout = None self.parallel = False self.metadata = {} context.source_graph_config = GraphConfig( name="none_type_graph", nodes={"stub_node": NodeConfig(name="stub_node", type=NodeType.FUNCTION)}, edges=[], entry_point="start", ) # Replace the node value with our stub that has type=None # GraphConfig.nodes is dict[str, NodeConfig] but from_graph_config uses # getattr() so it works with any object that has the right attributes. context.source_graph_config.nodes["stub_node"] = _StubNodeConfig() # type: ignore[assignment] @when("I create a RouteConfig from that GraphConfig") def step_route_from_graph_config(context): context.result_route_config = RouteConfig.from_graph_config( context.source_graph_config ) @then('the node type in the resulting RouteConfig should be "{expected_type}"') def step_node_type_in_route(context, expected_type): node_dict = context.result_route_config.nodes.get("stub_node") assert node_dict is not None, "stub_node not found in resulting RouteConfig" assert node_dict["type"] == expected_type # ── Scenario: from_graph_config fallback when node type has no .value attribute ── @given("I have a GraphConfig with a node whose type is a plain string") def step_graph_config_node_type_plain_string(context): class _StubNodeConfig: def __init__(self): self.type = "custom_type" # plain string, no .value attribute self.agent = None self.function = None self.tools = None self.retry_policy = None self.timeout = None self.parallel = False self.metadata = None context.source_graph_config = GraphConfig( name="string_type_graph", nodes={"stub_node": NodeConfig(name="stub_node", type=NodeType.FUNCTION)}, edges=[], entry_point="start", ) context.source_graph_config.nodes["stub_node"] = _StubNodeConfig() # type: ignore[assignment]