"""Robot Framework keyword library for email categorization actor graph tests. Provides keywords for loading YAML configs, creating executors, executing graph actors, and validating results against the email categorization use case. """ from __future__ import annotations import asyncio import json from pathlib import Path from typing import Any import yaml from cleveractors.config_utils import merge_configs from cleveractors.core.exceptions import ( AgentCreationError, ConfigurationError, ExecutionError, ) from cleveractors.result import ActorResult from cleveractors.runtime import create_executor _FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures" class EmailGraphLib: """Keyword library for email categorization graph integration tests. Wraps the cleveractors runtime API (create_executor, execute) with keywords specific to the email actor graph scenario. """ ROBOT_LIBRARY_SCOPE = "TEST SUITE" def __init__(self) -> None: self._executor: Any = None self._last_result: ActorResult | None = None # -- config loading ------------------------------------------------------- def load_email_graph_config(self, graph_name: str) -> Any: """Load an email graph YAML config fixture by name. Valid names: ``local_graph``, ``remote_graph``, ``mixed_graph``. Returns the parsed YAML as a dict. """ path = _FIXTURES / "email_graph" / f"{graph_name}.yaml" with open(path, encoding="utf-8") as fh: return yaml.safe_load(fh) def load_config_file(self, config_path: str) -> Any: """Load a YAML config file from ``tests/fixtures/``.""" path = _FIXTURES / config_path with open(path, encoding="utf-8") as fh: return yaml.safe_load(fh) # -- executor creation ---------------------------------------------------- def create_executor_for_email_graph(self, graph_name: str) -> None: """Create an Executor from the named email graph config.""" config = self.load_email_graph_config(graph_name) self._executor = create_executor( config_dict=config, credentials=None, limits={}, pricing={}, ) def create_executor_from_config_file(self, config_path: str) -> None: """Create an Executor from a fixture config file.""" config = self.load_config_file(config_path) self._executor = create_executor( config_dict=config, credentials=None, limits={}, pricing={}, ) def create_executor_from_merged_configs( self, graph_name: str, overlay_path: str ) -> None: """Create an Executor from a merged graph + overlay config.""" graph = self.load_email_graph_config(graph_name) overlay = self.load_config_file(overlay_path) merged = merge_configs(graph, overlay) self._executor = create_executor( config_dict=merged, credentials=None, limits={}, pricing={}, ) def create_executor_with_limits( self, graph_name: str, max_depth: int | None = None ) -> None: """Create an Executor with execution limits.""" config = self.load_email_graph_config(graph_name) limits: dict[str, Any] = {} if max_depth is not None: limits["max_depth"] = max_depth self._executor = create_executor( config_dict=config, credentials=None, limits=limits, pricing={}, ) def create_executor_from_dict(self, config_dict_str: str) -> None: """Create an Executor from a JSON string dict.""" config = json.loads(config_dict_str) self._executor = create_executor( config_dict=config, credentials=None, limits={}, pricing={}, ) def create_executor_from_config_dict(self, config_dict: dict[str, Any]) -> None: """Create an Executor directly from a Python dict.""" self._executor = create_executor( config_dict=config_dict, credentials=None, limits={}, pricing={}, ) # -- execution ------------------------------------------------------------ async def _execute_async(self, message: str) -> ActorResult: result = await self._executor.execute(message) self._last_result = result return result def execute_email_graph(self, email_json: str) -> None: """Execute the email graph with an email JSON input.""" self._last_result = asyncio.run(self._execute_async(email_json)) async def _execute_stream_async(self, message: str) -> str: chunks: list[str] = [] async for token in self._executor.execute_stream(message): chunks.append(token) return "".join(chunks) def execute_email_graph_stream(self, email_json: str) -> None: """Execute via streaming path and collect tokens.""" response = asyncio.run(self._execute_stream_async(email_json)) self._last_result = self._executor.last_result # -- result validation ---------------------------------------------------- def result_is_valid_actor_result(self) -> None: """Assert last_result is an ActorResult instance.""" assert isinstance(self._last_result, ActorResult), ( f"Expected ActorResult, got {type(self._last_result)}" ) def result_response_contains(self, text: str) -> None: """Assert the response string contains the given text.""" assert self._last_result is not None assert text in self._last_result.response, ( f"Expected '{text}' in response, got: {self._last_result.response[:200]}" ) def result_response_contains_json_key(self, key: str) -> None: """Assert the response contains JSON with the given top-level key.""" assert self._last_result is not None try: parsed = json.loads(self._last_result.response) except json.JSONDecodeError: raise AssertionError( f"Response is not valid JSON: {self._last_result.response[:200]}" ) assert key in parsed, ( f"Key '{key}' not found in response JSON. Keys: {list(parsed.keys())}" ) def result_json_key_equals(self, key: str, expected: str) -> None: """Assert a JSON key in the response has an expected value.""" assert self._last_result is not None parsed = json.loads(self._last_result.response) actual = parsed.get(key) assert str(actual) == expected, ( f"Key '{key}': expected '{expected}', got '{actual}'" ) def result_has_nodes(self) -> None: """Assert the result has at least one NodeUsage entry.""" assert self._last_result is not None assert len(self._last_result.nodes) > 0, "Expected at least one node in result" def result_node_count_equals(self, count: int) -> None: """Assert result has exactly count NodeUsage entries.""" assert self._last_result is not None actual = len(self._last_result.nodes) assert actual == count, f"Expected {count} nodes, got {actual}" def result_has_state(self) -> None: """Assert the result has a non-None state field.""" assert self._last_result is not None assert self._last_result.state is not None, "Expected non-None state" def result_state_contains_key(self, key: str) -> None: """Assert the result state dict contains the given key.""" assert self._last_result is not None assert self._last_result.state is not None assert key in self._last_result.state, ( f"Key '{key}' not in state: {list(self._last_result.state.keys())}" ) # -- error validation (negative tests) ------------------------------------ def executor_creation_should_fail(self, graph_name: str) -> None: """Assert creating an executor fails with ConfigurationError.""" try: self.create_executor_for_email_graph(graph_name) raise AssertionError("Expected ConfigurationError but no exception raised") except (ConfigurationError, AgentCreationError): pass def execution_should_raise(self, email_json: str, error_type: str) -> None: """Assert executing with the given input raises a specific error.""" try: asyncio.run(self._execute_async(email_json)) raise AssertionError(f"Expected {error_type} but no exception raised") except (ConfigurationError, AgentCreationError, ExecutionError) as e: actual_type = type(e).__name__ if error_type not in actual_type: raise AssertionError( f"Expected error type containing '{error_type}', got '{actual_type}: {e}'" ) def executor_from_dict_should_fail(self, config_dict_str: str) -> None: """Assert creating executor from dict raises ConfigurationError.""" try: self.create_executor_from_dict(config_dict_str) raise AssertionError("Expected ConfigurationError but no exception raised") except (ConfigurationError, AgentCreationError): pass def executor_from_python_dict_should_fail( self, config_dict: dict[str, Any] ) -> None: """Assert creating executor from a Python dict raises ConfigurationError.""" try: self.create_executor_from_config_dict(config_dict) raise AssertionError("Expected ConfigurationError but no exception raised") except (ConfigurationError, AgentCreationError): pass def executor_non_dict_should_fail(self, non_dict_value: str) -> None: """Assert non-dict value raises ConfigurationError.""" try: create_executor( config_dict=non_dict_value, credentials=None, limits={}, pricing={}, ) raise AssertionError("Expected ConfigurationError but no exception raised") except (ConfigurationError, AgentCreationError): pass # -- negative test configs ------------------------------------------------ def load_invalid_graph_no_routes(self) -> dict[str, Any]: """Return a config dict missing the routes key.""" return {"actors": {}} def load_invalid_graph_no_actors(self) -> dict[str, Any]: """Return a config dict missing the actors key.""" return {"routes": {"main": {"nodes": {}, "edges": []}}} def load_invalid_graph_duplicate_nodes(self) -> dict[str, Any]: """Return a graph config with duplicate node IDs.""" return { "routes": { "main": { "entry_point": "start", "nodes": { "n1": {"id": "start", "type": "agent", "agent": "echo"}, "n2": {"id": "start", "type": "agent", "agent": "echo"}, }, "edges": [], } }, "actors": {}, } def load_invalid_graph_missing_agent_ref(self) -> dict[str, Any]: """Return a graph config referencing a nonexistent agent.""" return { "routes": { "main": { "entry_point": "start", "nodes": { "n1": { "id": "start", "type": "agent", "agent": "nonexistent_agent", } }, "edges": [], } }, "actors": {"echo": {"type": "tool", "config": {}}}, } def load_invalid_config_nondict(self) -> str: """Return a non-dict config for type validation.""" return "not-a-dict" # -- merge_configs keywords ----------------------------------------------- def merge_email_graph_configs(self, graph_name: str, overlay_name: str) -> Any: """Merge two email graph configs and return the result.""" base = self.load_email_graph_config(graph_name) overlay = self.load_email_graph_config(overlay_name) return merge_configs(base, overlay) def merge_email_graph_configs_invalid(self, graph_name: str) -> Any: """Attempt merge_configs with one valid and one non-existent name. This tests that passing an invalid (non-existant) file to the file loader raises FileNotFoundError before merge_configs is called.""" base = self.load_email_graph_config(graph_name) return merge_configs(base, "not_a_valid_dict") def merged_config_has_nested_key(self, merged: Any, *keys: str) -> None: """Assert nested keys exist in a merged config dict.""" current = merged for key in keys: assert isinstance(current, dict), ( f"Expected dict at key path, got {type(current)}" ) assert key in current, ( f"Key '{key}' not found. Available: {list(current.keys())[:10]}" ) current = current[key] def merged_config_key_count(self, merged: Any, section: str, minimum: int) -> None: """Assert a config section has at least minimum keys.""" assert section in merged, f"Section '{section}' not in merged config" actual = len(merged[section]) assert actual >= minimum, ( f"Expected >= {minimum} keys in '{section}', got {actual}" ) # -- component-based graph assembly --------------------------------------- def load_all_email_components(self) -> Any: """Load all 7 component YAML files and merge them into one config. Returns the fully assembled graph config ready for Executor. """ base = _FIXTURES / "email_graph" / "components" component_files = [ "email_categorizer.yaml", "component_orders.yaml", "personal_summarizer.yaml", "business_email.yaml", "supplier.yaml", "client.yaml", "graph_routes.yaml", ] configs = [] for fname in component_files: with open(base / fname, encoding="utf-8") as fh: configs.append(yaml.safe_load(fh)) result: dict[str, Any] = {} for cfg in configs: result = merge_configs(result, cfg) return result def create_executor_from_components(self) -> None: """Create an Executor from the assembled component graph.""" config = self.load_all_email_components() self._executor = create_executor( config_dict=config, credentials=None, limits={}, pricing={}, ) def component_graph_actor_count_equals(self, expected: int) -> None: """Assert the assembled component graph has exactly expected actors.""" config = self.load_all_email_components() actual = len(config.get("actors", {})) assert actual == expected, f"Expected {expected} actors, got {actual}" # -- namespaced component graph with reference resolver ------------------- def assemble_namespaced_graph(self) -> None: """Assemble the namespaced email graph by resolving all local: refs. Uses LocalPackageStore + ReferenceResolver to resolve each ``agent_ref: local:...`` in the graph routes into fully inlined actor configs, then creates an Executor from the assembled config. """ from cleveractors.registry.local_store import LocalPackageStore from cleveractors.registry.resolver import ReferenceResolver base_dir = _FIXTURES / "email_graph" / "namespaced" store = LocalPackageStore(base_dir) # Resolve each local: reference via the store graph_path = base_dir / "email_graph.yaml" with open(graph_path, encoding="utf-8") as fh: graph_config = yaml.safe_load(fh) resolver = ReferenceResolver(local_store=store) actors: dict[str, Any] = {} nodes = graph_config.get("routes", {}).get("main", {}).get("nodes", {}) for _node_id, node_def in nodes.items(): agent_ref = node_def.get("agent_ref", "") if agent_ref.startswith("local:"): ref_name = agent_ref[len("local:") :] pkg = store.resolve_package(ref_name) content = pkg.content actor_name = content.get("name", ref_name) actors[actor_name] = { "type": content.get("type", "tool"), "config": content.get("config", {}), } graph_config["actors"] = actors for node_def in ( graph_config.get("routes", {}).get("main", {}).get("nodes", {}).values() ): if "agent_ref" in node_def and node_def["agent_ref"].startswith("local:"): ref_name = node_def["agent_ref"][len("local:") :] pkg = store.resolve_package(ref_name) node_def["agent"] = pkg.content.get("name", ref_name) del node_def["agent_ref"] self._executor = create_executor( config_dict=graph_config, credentials=None, limits={}, pricing={}, ) def namespaced_graph_actor_count_equals(self, expected: int) -> None: """Assert the namespaced graph resolves to exactly expected actors.""" base = _FIXTURES / "email_graph" / "namespaced" with open(base / "email_graph.yaml", encoding="utf-8") as fh: graph_config = yaml.safe_load(fh) ref_count = 0 nodes = graph_config.get("routes", {}).get("main", {}).get("nodes", {}) for _nid, ndef in nodes.items(): if ndef.get("agent_ref", "").startswith("local:"): ref_count += 1 assert ref_count == expected, ( f"Expected {expected} agent_ref references in graph, got {ref_count}" ) # -- agent factory keywords ----------------------------------------------- def create_agents_from_config(self, graph_name: str) -> int: """Create all agents from the named graph config and return count.""" from cleveractors.agents.factory import AgentFactory from cleveractors.templates.renderer import TemplateRenderer config = self.load_email_graph_config(graph_name) config.setdefault("agents", {}) config["agents"].update(config.get("actors", {})) renderer = TemplateRenderer() factory = AgentFactory(config=config, template_renderer=renderer) agents = factory.create_agents_from_config() return len(agents) def factory_validates_configuration(self, graph_name: str) -> None: """Assert AgentFactory.validate_configuration succeeds.""" from cleveractors.agents.factory import AgentFactory from cleveractors.templates.renderer import TemplateRenderer config = self.load_email_graph_config(graph_name) config.setdefault("agents", {}) config["agents"].update(config.get("actors", {})) renderer = TemplateRenderer() factory = AgentFactory(config=config, template_renderer=renderer) factory.validate_configuration() def factory_agent_has_metadata_key( self, graph_name: str, agent_name: str, key: str ) -> None: """Assert an agent's metadata contains a key.""" from cleveractors.agents.factory import AgentFactory from cleveractors.templates.renderer import TemplateRenderer config = self.load_email_graph_config(graph_name) config.setdefault("agents", {}) config["agents"].update(config.get("actors", {})) renderer = TemplateRenderer() factory = AgentFactory(config=config, template_renderer=renderer) meta = factory.get_agent_metadata(agent_name) assert key in meta, f"Key '{key}' not in metadata. Keys: {list(meta.keys())}"