cleveractors-core/features/steps/validate_dict_helpers.py

"""
Shared helper functions for validate_dict BDD step definitions.

This module provides reusable config builders, the ``call_and_capture``
invocation helper, and the ``minimal_config`` factory.  It is imported
by the domain-specific step modules and is **not** itself a Behave step
file (it contains no @given/@when/@then decorators).
"""

from __future__ import annotations

import copy
from typing import Any

from behave.runner import Context

from cleveractors import validate_dict
from cleveractors.core.exceptions import ConfigurationError

# ── Minimal test configs ───────────────────────────────────────────────────

MINIMAL_CONFIG: dict[str, Any] = {
    "agents": {
        "echo": {
            "type": "tool",
            "config": {"tools": ["echo"]},
        }
    },
    "routes": {
        "main": {
            "type": "stream",
            "operators": [{"type": "map", "params": {"agent": "echo"}}],
            "publications": ["__output__"],
        }
    },
}


def minimal_config() -> dict[str, Any]:
    """Return a deep copy of the minimal valid config."""
    return copy.deepcopy(MINIMAL_CONFIG)


# ── Shared helper: call validate_dict and capture result/exception ──────


def call_and_capture(
    context: Context,
    config: Any,
    limits: Any,
) -> None:
    """Call validate_dict and store the result or exception on context.

    Accepts ``Any`` for *config* and *limits* so that test steps can
    intentionally pass invalid types (e.g. ``None``) to exercise the
    fail-fast argument guards in ``validate_dict``.
    """
    context.validate_dict = validate_dict
    context.ConfigurationError = ConfigurationError
    try:
        context.result = validate_dict(config, limits)
        context.raised_exception = None
    except Exception as exc:
        context.raised_exception = exc
        context.result = None


# ── Shared graph builders ────────────────────────────────────────────────


def build_linear_graph(depth: int) -> dict[str, Any]:
    """Build a config with a linear graph route of exactly 'depth' edges."""
    config = minimal_config()
    if depth <= 0:
        config["routes"]["depth_graph"] = {
            "type": "graph",
            "entry_point": "end_node",
            "nodes": {"end_node": {"type": "end"}},
            "edges": [],
        }
        return config
    nodes: dict[str, Any] = {}
    edges: list[dict[str, Any]] = []
    node_names = [f"node_{i}" for i in range(depth)]
    for name in node_names:
        nodes[name] = {"type": "agent", "agent": "echo"}
    nodes["end"] = {"type": "end"}
    for i in range(len(node_names) - 1):
        edges.append({"source": node_names[i], "target": node_names[i + 1]})
    if node_names:
        edges.append({"source": node_names[-1], "target": "end"})
    config["routes"]["depth_graph"] = {
        "type": "graph",
        "entry_point": node_names[0],
        "nodes": nodes,
        "edges": edges,
    }
    return config


def build_node_count_graph(count: int) -> dict[str, Any]:
    """Build a graph with exactly 'count' non-end nodes plus the end node."""
    config = minimal_config()
    nodes: dict[str, Any] = {}
    edges: list[dict[str, Any]] = []
    node_names = [f"node_{i}" for i in range(count)]
    for name in node_names:
        nodes[name] = {"type": "agent", "agent": "echo"}
    nodes["end"] = {"type": "end"}
    for i in range(len(node_names) - 1):
        edges.append({"source": node_names[i], "target": node_names[i + 1]})
    if node_names:
        edges.append({"source": node_names[-1], "target": "end"})
    config["routes"]["count_graph"] = {
        "type": "graph",
        "entry_point": node_names[0] if node_names else "end",
        "nodes": nodes,
        "edges": edges,
    }
    return config


def build_subgraph_chain(depth: int) -> dict[str, Any]:
    """Build a config with nested subgraph routes of 'depth' levels."""
    if depth <= 0:
        raise ValueError("depth must be >= 1")
    config = minimal_config()
    routes: dict[str, Any] = {}
    for level in range(depth):
        route_name = f"sub_level_{level}"
        nodes: dict[str, Any] = {
            "worker": {"type": "agent", "agent": "echo"},
            "end": {"type": "end"},
        }
        edges_list: list[dict[str, Any]] = [{"source": "worker", "target": "end"}]
        if level < depth - 1:
            next_route = f"sub_level_{level + 1}"
            nodes["sub_node"] = {"type": "subgraph", "subgraph": next_route}
            edges_list = [
                {"source": "worker", "target": "sub_node"},
                {"source": "sub_node", "target": "end"},
            ]
        routes[route_name] = {
            "type": "graph",
            "entry_point": "worker",
            "nodes": nodes,
            "edges": edges_list,
        }
    top_nodes: dict[str, Any] = {
        "top_worker": {"type": "agent", "agent": "echo"},
        "sub_entry": {"type": "subgraph", "subgraph": "sub_level_0"},
        "end": {"type": "end"},
    }
    top_edges: list[dict[str, Any]] = [
        {"source": "top_worker", "target": "sub_entry"},
        {"source": "sub_entry", "target": "end"},
    ]
    routes["top_graph"] = {
        "type": "graph",
        "entry_point": "top_worker",
        "nodes": top_nodes,
        "edges": top_edges,
    }
    config["routes"].update(routes)
    return config