forked from HAL9000/cleveragents-core
e801eb1ee8
- Rewrite .forgejo/workflows/ci.yml to route all jobs through nox sessions - Fix coverage_report nox session: serial behave mode replaces broken parallel mode (22% -> 97% accuracy), raise fail-under from 85% to 97% - Pass posargs through format nox session for CI --check support - Add 11 CI workflow validation scenarios (Behave) + Robot smoke test + ASV bench - Add 108 new Behave scenarios covering 6 largest coverage gaps to reach 97%: yaml_template_engine, actor/config, actor/registry, message_router, context_analysis, context_service - Update docs/development/ci-cd.md with nox-based CI docs and 97% threshold - Restore implementation_plan.md verbose style, check off completed CI tasks Verified: 1673 scenarios pass, 97% coverage, lint clean, typecheck clean
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""Step definitions for message_router_new_coverage.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from behave import then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
|
|
from cleveragents.langgraph.message_router import (
|
|
build_message_router_edges,
|
|
build_message_router_node,
|
|
)
|
|
from cleveragents.langgraph.nodes import NodeType
|
|
|
|
|
|
@when('I build a message router node named "{name}" with rules')
|
|
def step_build_node(context: Context, name: str) -> None:
|
|
rules: list[dict[str, Any]] = [
|
|
{"target": "agent_a", "condition": {"type": "is_question"}},
|
|
{"target": "agent_b", "condition": {"type": "is_command"}},
|
|
]
|
|
context.node = build_message_router_node(name, rules)
|
|
|
|
|
|
@then('the node name should be "{name}"')
|
|
def step_assert_node_name(context: Context, name: str) -> None:
|
|
assert context.node.name == name
|
|
|
|
|
|
@then("the node type should be MESSAGE_ROUTER")
|
|
def step_assert_node_type(context: Context) -> None:
|
|
assert context.node.type == NodeType.MESSAGE_ROUTER
|
|
|
|
|
|
@then("the node metadata should contain rules")
|
|
def step_assert_node_metadata(context: Context) -> None:
|
|
assert "rules" in context.node.metadata
|
|
|
|
|
|
@when('I build message router edges for "{name}" with two target rules')
|
|
def step_build_edges_two(context: Context, name: str) -> None:
|
|
rules: list[dict[str, Any]] = [
|
|
{"target": "agent_a", "condition": {"type": "cond_a"}},
|
|
{"target": "agent_b", "condition": {"type": "cond_b"}},
|
|
]
|
|
context.edges = build_message_router_edges(name, rules)
|
|
|
|
|
|
@then("the edge count should be {n}")
|
|
def step_assert_edge_count(context: Context, n: str) -> None:
|
|
expected = int(n)
|
|
assert len(context.edges) == expected, (
|
|
f"Expected {expected} edges, got {len(context.edges)}"
|
|
)
|
|
|
|
|
|
@then('the last edge target should be "{target}"')
|
|
def step_assert_last_edge(context: Context, target: str) -> None:
|
|
assert context.edges[-1].target == target
|
|
|
|
|
|
@when('I build message router edges for "{name}" with no rules')
|
|
def step_build_edges_empty(context: Context, name: str) -> None:
|
|
context.edges = build_message_router_edges(name, [])
|
|
|
|
|
|
@when('I build message router edges for "{name}" with a rule missing target')
|
|
def step_build_edges_no_target(context: Context, name: str) -> None:
|
|
rules: list[dict[str, Any]] = [{"condition": {"type": "something"}}]
|
|
context.edges = build_message_router_edges(name, rules)
|