"""Coverage step definitions for routing_adapter, dynamic_router, message_router, config_parser.""" import asyncio from unittest.mock import MagicMock, patch from behave import given, then, when @given("a fresh integration test context") def step_fresh_integration(context): context.results = {} # ═══ RoutingAdapter ═════════════════════════════════════════════════════ @when("I exercise all routing_adapter parse_routing_command code paths") def step_all_routing_adapter(context): from cleveractors.langgraph.routing_adapter import RoutingAdapter adapter = RoutingAdapter() # Line 40-41: non-string → (None, str) t, m = adapter.parse_routing_command(42) context.results["ra_nonstr"] = t is None and m == "42" # Line 44-48: GOTO_COMMAND_HANDLER t, m = adapter.parse_routing_command("GOTO_COMMAND_HANDLER:do something") context.results["ra_goto_cmd"] = t == "command_handler" # Lines 55-65: GOTO_ known targets for target in [ "intro", "discovery", "brainstorming", "vetting", "structure", "section_writing", "paper_review", "latex_generation", ]: t, m = adapter.parse_routing_command(f"GOTO_{target.upper()}:data") if t != target: context.results[f"ra_goto_{target}"] = t # Line 68: GOTO_ unknown target t, m = adapter.parse_routing_command("GOTO_CUSTOM:data") context.results["ra_goto_unknown"] = t == "custom" or t is not None # Lines 71-90: ROUTE_ patterns for rt in [ "ask_topic", "ask_length", "ask_audience", "ask_publication", "ask_format", "ask_other", ]: t, m = adapter.parse_routing_command(f"ROUTE_{rt.upper()}:query") if t != rt: context.results[f"ra_route_{rt}"] = t # Line 93: no routing t, m = adapter.parse_routing_command("just a normal message") context.results["ra_no_route"] = t is None and m == "just a normal message" # Lines 106-143: create_routing_node router_fn = adapter.create_routing_node() state1 = {"messages": ["GOTO_INTRO:hello there"]} state2 = {"messages": [{"content": "ROUTE_ASK_TOPIC:what topic?"}]} state3 = {"messages": ["normal message"]} state4 = {"messages": []} state5 = {"messages": [42]} async def _run(): r1 = await router_fn(state1) r2 = await router_fn(state2) r3 = await router_fn(state3) r4 = await router_fn(state4) r5 = await router_fn(state5) context.results["ra_routing_node_goto"] = r1.get("next_node") == "intro" context.results["ra_routing_node_route"] = r2.get("next_node") == "ask_topic" context.results["ra_routing_node_normal"] = ( r3.get("next_node", "NOT_SET") is None or r3.get("next_node") == "NOT_SET" ) context.results["ra_routing_node_empty"] = "messages" in r4 context.results["ra_routing_node_nonstr"] = "messages" in r5 asyncio.get_event_loop().run_until_complete(_run()) # Lines 147-166: create_conditional_router cond_fn = adapter.create_conditional_router( {"intro": "intro_handler", "discovery": "disc_handler"} ) context.results["ra_cond_matched"] = ( cond_fn({"next_node": "intro"}) == "intro_handler" ) context.results["ra_cond_default"] = cond_fn({"next_node": "unknown"}) == "end" context.results["ra_cond_no_next"] = cond_fn({}) == "end" # Lines 171-253: create_langgraph_config_from_dynamic from cleveractors.langgraph.routing_adapter import ( create_langgraph_config_from_dynamic, ) config1 = {"agents": {"agent1": {}, "agent2": {}}} result1 = create_langgraph_config_from_dynamic(config1) context.results["ra_dynamic_config"] = result1.get("name") == "converted_graph" config2 = {"agents": {"workflow_controller": {}, "agent1": {}}} result2 = create_langgraph_config_from_dynamic(config2) context.results["ra_dynamic_config_wf"] = "workflow_controller" in result2.get( "nodes", {} ) @then("all routing_adapter paths should execute correctly") def step_ra_all_ok(context): assert context.results.get("ra_nonstr"), "non-string parse" assert context.results.get("ra_goto_cmd"), "GOTO command_handler" assert context.results.get("ra_goto_unknown"), "GOTO unknown handled" assert context.results.get("ra_no_route"), "no-route fallthrough" assert context.results.get("ra_routing_node_goto"), "routing node goto" assert context.results.get("ra_routing_node_route"), "routing node route" assert context.results.get("ra_cond_matched"), "cond router matched" assert context.results.get("ra_cond_default"), "cond router default" assert context.results.get("ra_dynamic_config"), "dynamic config created" # ═══ DynamicRouterNode ═══════════════════════════════════════════════════ @when("I exercise all DynamicRouterNode execute code paths") def step_all_dynamic_router(context): from cleveractors.langgraph.dynamic_router import ( DynamicRouterNode, RoutePattern, create_dynamic_router_config, extend_graph_with_router, ) routes = [ RoutePattern(pattern="GOTO_DISCOVERY", target="discovery"), RoutePattern(pattern="SET_TOPIC", target="topic_setter", extract_message=True), ] node = DynamicRouterNode(routes) async def _run(): # Lines 57-60: empty messages r1 = await node.execute({"messages": []}) context.results["dr_empty"] = "messages" in r1 # Lines 62-66: string message r2 = await node.execute({"messages": ["GOTO_DISCOVERY:find the topic"]}) context.results["dr_str"] = r2.get("next_node") == "discovery" # Lines 62-66: dict message r3 = await node.execute( {"messages": [{"content": "GOTO_DISCOVERY:find topic"}]} ) context.results["dr_dict"] = r3.get("next_node") == "discovery" # Lines 79-89: extract_message + colon r4 = await node.execute({"messages": [{"content": "SET_TOPIC:my topic is AI"}]}) context.results["dr_extract"] = r4.get("next_node") == "topic_setter" context.results["dr_extract_content"] = r4["messages"][-1].get("content", "") # Lines 79-89: extract_message with string message r5 = await node.execute({"messages": ["SET_TOPIC:some topic"]}) context.results["dr_extract_str"] = r5.get("next_node") == "topic_setter" # Line 93-95: no pattern matched r6 = await node.execute({"messages": ["nothing matches here"]}) context.results["dr_no_match"] = r6.get("next_node", "NOT_SET") is not None asyncio.get_event_loop().run_until_complete(_run()) # Lines 98-113: create_dynamic_router_config cfg = create_dynamic_router_config({"PAT_A": "node_a", "PAT_B": "node_b"}) context.results["dr_cfg_type"] = cfg.get("type") == "dynamic_router" context.results["dr_cfg_routes"] = len(cfg.get("routes", [])) == 2 # Lines 116-259: extend_graph_with_router (exercise the long function) graph_cfg = { "nodes": {"workflow_controller": {"type": "agent"}}, "edges": [{"source": "workflow_controller", "target": "discovery"}], } extended = extend_graph_with_router(graph_cfg) context.results["dr_extended"] = "dynamic_router" in extended.get("nodes", {}) context.results["dr_extended_edges"] = len(extended.get("edges", [])) > 1 # Empty graph config empty_cfg = {} ext2 = extend_graph_with_router(empty_cfg) context.results["dr_extended_empty"] = "dynamic_router" in ext2.get("nodes", {}) @then("all DynamicRouterNode paths should execute correctly") def step_dr_all_ok(context): assert context.results.get("dr_empty"), "empty messages handled" assert context.results.get("dr_str"), "string message routed" assert context.results.get("dr_dict"), "dict message routed" assert context.results.get("dr_extract"), "message extracted" assert context.results.get("dr_no_match"), "no-match handled" assert context.results.get("dr_cfg_type"), "config type correct" assert context.results.get("dr_extended"), "graph extended" assert context.results.get("dr_extended_empty"), "empty graph extended" # ═══ MessageRouter ═══════════════════════════════════════════════════════ @when("I exercise all MessageRouter code paths") def step_all_message_router(context): from cleveractors.langgraph.message_router import ( MessageRouter, RouteRule, create_router_node, ) # Lines 60-64: invalid regex rules = [ RouteRule(match_type="regex", pattern="[invalid", target_node="handler"), ] router = MessageRouter(rules) context.results["mr_invalid_regex"] = True # All match types rules2 = [ RouteRule(match_type="regex", pattern=r"go\s+to", target_node="regex_ok"), RouteRule(match_type="exact", pattern="hello", target_node="exact_ok"), RouteRule(match_type="contains", pattern="help", target_node="contains_ok"), RouteRule( match_type="prefix", pattern="CMD:", target_node="prefix_ok", separator=":" ), RouteRule(match_type="suffix", pattern=".end", target_node="suffix_ok"), ] router2 = MessageRouter(rules2) # Line 78: non-string message r0 = router2.route(42, {}) context.results["mr_nonstr"] = isinstance(r0, dict) # Regex match r1 = router2.route("go to the store", {}) context.results["mr_regex"] = r1.get("next_node") == "regex_ok" # Exact match r2 = router2.route("hello", {}) context.results["mr_exact"] = r2.get("next_node") == "exact_ok" # Contains match r3 = router2.route("need help please", {}) context.results["mr_contains"] = r3.get("next_node") == "contains_ok" # Prefix match r4 = router2.route("CMD:format", {}) context.results["mr_prefix"] = r4.get("next_node") == "prefix_ok" # Suffix match r5 = router2.route("finish.end", {}) context.results["mr_suffix"] = r5.get("next_node") == "suffix_ok" # No match r6 = router2.route("nothing to see here", {}) context.results["mr_no_match"] = isinstance(r6, dict) # extract_message=True rules3 = [ RouteRule( match_type="exact", pattern="ping", target_node="pong", extract_message=True ) ] router3 = MessageRouter(rules3) r7 = router3.route("ping:the rest of message", {}) context.results["mr_extract"] = isinstance(r7, dict) # set_state rules4 = [ RouteRule( match_type="exact", pattern="setit", target_node="setter", set_state={"flag": True}, ) ] router4 = MessageRouter(rules4) r8 = router4.route("setit", {}) context.results["mr_set_state"] = r8.get("flag") is True # Lines 169-226: create_router_node from features.steps._coverage_utils import _run node_fn = create_router_node( {"rules": [{"match_type": "exact", "pattern": "hi", "target": "greeter"}]} ) async def _run_node(): nr = await node_fn({"messages": ["hi"]}) context.results["mr_router_node"] = True _run(_run_node()) # Lines 218-267: MessageRouterNodeConfig from cleveractors.langgraph.message_router import MessageRouterNodeConfig ncfg = MessageRouterNodeConfig.from_config( {"rules": [{"match_type": "exact", "pattern": "bye", "target": "ender"}]} ) context.results["mr_node_config"] = isinstance(ncfg, dict) @then("all MessageRouter paths should execute correctly") def step_mr_all_ok(context): assert context.results.get("mr_invalid_regex"), "invalid regex handled" assert context.results.get("mr_nonstr"), "non-string message" assert context.results.get("mr_regex"), "regex match" assert context.results.get("mr_exact"), "exact match" assert context.results.get("mr_contains"), "contains match" assert context.results.get("mr_prefix"), "prefix match" assert context.results.get("mr_suffix"), "suffix match" assert context.results.get("mr_no_match"), "no-match handled" assert context.results.get("mr_router_node"), "router node created" assert context.results.get("mr_node_config"), "node config created"