scripts.generate_pytest
Reference › scripts › generate_pytest
scripts/generate_pytest.py is the AST-based unit-test generator for orchestrata-ultimate: it analyses a source file or directory and renders pytest modules from Jinja2 templates. A human runs it, usually per package; it has no internal imports and is exercised by four test modules (test_generate_pytest_coverage_wave, test_generate_pytest_package_flag, test_generate_pytest_signature_compat and test_generate_pytest_determinism).
main takes an optional positional source (a file or directory), --package (a top-level package name such as agents, core or gateway; it resolves the source directory to src/orchestrata_ultimate/<package> and sets the default output to tests/generated_<package>), --output or -o (default tests/generated), --template-dir or -t (a directory of templates) and --force-templates (overwrite customised template files with the built-in defaults; by default a customised template is preserved). The module is large, about 1900 lines: it parses the source with ast, builds a description of each class, function and signature, and fills the templates. The determinism test pins that the same input gives the same output. It reads the source tree and writes only the generated test directory.
At a glance
| Source | scripts/generate_pytest.py |
| Lines of code | 1924 |
| Classes | 3 |
| Functions | 1 |
| Imports (internal) | none |
| Imported by | none |
| Tests that import it | tests/scripts/test_generate_pytest_coverage_wave.py, tests/scripts/test_generate_pytest_package_flag.py, tests/scripts/test_generate_pytest_signature_compat.py, tests/test_generate_pytest_determinism.py |
Class diagram
Dependency diagram
No internal imports and no importers.
API
Classes
MethodInfo
Line 25
Information about a method extracted from AST.
| Attribute | Type | Default |
|---|---|---|
_KNOWN_CLASS_CTORS |
dict[str, tuple[str, set[str]]] |
{'AgentHierarchy': ('AgentHierarchy()', {'from orchestrata_… |
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__() -> None |
|
__repr__ |
__repr__() -> str |
|
get_parameter_types |
get_parameter_types() -> dict[str, str] |
Get mapping of parameter names to their types. |
has_type_annotations |
has_type_annotations() -> bool |
Check if any parameters have type annotations. |
has_parameter_type_annotations |
has_parameter_type_annotations() -> bool |
Check if any parameters have type annotations (excluding return type). |
get_import_path |
get_import_path() -> str |
Get import path for this method. |
get_test_prefix |
get_test_prefix() -> str |
Get prefix for test function names. |
get_snake_case_prefix |
get_snake_case_prefix() -> str |
Get snake_case prefix for test function names. |
get_required_imports_from_defaults |
get_required_imports_from_defaults() -> set[str] |
Get imports required for default values in parameters. |
inherits_from_pydantic |
inherits_from_pydantic() -> bool |
Check if the containing class inherits from pydantic.BaseModel. |
is_dataclass |
is_dataclass() -> bool |
Check if the containing class is decorated with @dataclass. |
should_validate_constructor |
should_validate_constructor() -> bool |
Check if constructor validation tests should be generated. |
has_validation_block |
has_validation_block() -> bool |
Check if any parameter warrants an invalid-value validation test. |
get_validation_error_imports |
get_validation_error_imports() -> set[str] |
Get imports needed for validation error testing. |
get_dummy_value |
get_dummy_value(param: dict[str, Any]) -> tuple[str, set[str]] |
Return a Python literal and required imports for a parameter's dummy value. |
ASTAnalyzer
Line 587
Analyzes Python files using AST to extract method information.
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__() -> None |
|
analyze_file |
analyze_file(file_path: Path) -> list[MethodInfo] |
Analyze a single Python file and extract method information. |
_file_to_module_path |
_file_to_module_path(file_path: Path) -> str |
Convert file path to Python module import path. |
analyze_directory |
analyze_directory(directory: Path) -> list[MethodInfo] |
Analyze all Python files in a directory. |
_visit_node |
_visit_node(node: ast.AST) -> None |
Visit AST node and extract method information. |
_visit_class |
_visit_class(class_node: ast.ClassDef, prefix: str | None = None) -> None |
Visit a class definition and extract its methods. |
_extract_base_names |
_extract_base_names(bases: list[ast.expr]) -> list[str] |
Extract base class names from AST base expressions. |
_extract_class_fields |
_extract_class_fields(class_node: ast.ClassDef) -> list[dict[str, Any]] |
Extract annotated fields from a class body (for dataclasses/Pydantic). |
_extract_function_info |
_extract_function_info(func_node: ast.FunctionDef, class_name: str) -> MethodInfo | None |
Extract method information from a FunctionDef node. |
_extract_async_function_info |
_extract_async_function_info(func_node: ast.AsyncFunctionDef, class_name: str) -> MethodInfo | None |
Extract method information from an AsyncFunctionDef node. |
_extract_method_info_base |
_extract_method_info_base(func_node: ast.FunctionDef | ast.AsyncFunctionDef, class_name: str, is_async: bool) -> MethodInfo | None |
Base method for extracting method information. |
_extract_parameters |
_extract_parameters(args_node: ast.arguments) -> list[dict[str, Any]] |
Extract parameter information from arguments node. |
_create_parameter_info |
_create_parameter_info(arg: ast.arg, args_node: ast.arguments, index: int, param_type: str, is_kwonly: bool = False) -> dict[str, Any] |
Create parameter information dictionary. |
_type_to_string |
_type_to_string(node: ast.AST) -> str |
Convert AST type node to string representation. |
_binop_to_string |
_binop_to_string(op: ast.operator) -> str |
Convert binary operator to string. |
_node_to_string |
_node_to_string(node: ast.AST) -> str |
Convert AST node to string representation. |
TestGenerator
Line 1137
Generates unit tests from method information using Jinja templates.
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__(template_dir: Path | None = None, force_templates: bool = False) |
|
_create_default_templates |
_create_default_templates(force: bool = False) -> None |
Create default Jinja templates for test generation. |
generate_tests |
generate_tests(methods: list[MethodInfo], output_dir: Path, prune_duplicates: bool = False, source_root: Path | None = None) -> dict[str, str] |
Generate test files for each method. |
_lint_and_format_generated_files |
_lint_and_format_generated_files(output_dir: Path, file_paths: list[Path]) -> None |
Run ruff and black on generated test files. |
_generate_test_file_content |
_generate_test_file_content(methods: list[MethodInfo], source_path: Path) -> str |
Generate content for a test file. |
Functions
| Function | Signature | Summary |
|---|---|---|
main |
main() -> None |
Main entry point. |
Source
#!/usr/bin/env python3
"""
AST-based test generator for orchestrata-ultimate - CORRECTED VERSION.
For each method in the code, generates 2 unit tests from Jinja templates:
1. Interface stability test - breaks on interface changes
2. Parameter validation test - validates parameter types
Fixes:
1. Correct import path generation
2. Avoid duplicate method extraction
3. Better template handling
"""
import ast
import re
import subprocess
import sys
from pathlib import Path
from typing import Any
import jinja2
class MethodInfo:
"""Information about a method extracted from AST."""
def __init__(self) -> None:
self.class_name: str = ""
self.method_name: str = ""
self.is_async: bool = False
self.parameters: list[dict[str, Any]] = []
self.return_type: str | None = None
self.file_path: str = ""
self.line_number: int = 0
self.is_classmethod: bool = False
self.is_staticmethod: bool = False
self.is_property: bool = False
self.is_protocol: bool = False # True if containing class is a Protocol
self.is_synthetic_init: bool = False # synthesized dataclass/Pydantic __init__
# Class names defined in the same module (stamped at generation time)
self.local_class_names: set[str] = set()
self.docstring: str | None = None
self.module_path: str = "" # Python import path
self.class_bases: list[str] = [] # Base classes of the containing class
self._is_dataclass: bool = False
self.ctor_params: list[dict[str, Any]] = []
def __repr__(self) -> str:
props = []
if self.is_async:
props.append("async")
if self.is_classmethod:
props.append("classmethod")
if self.is_staticmethod:
props.append("staticmethod")
if self.is_property:
props.append("property")
prop_str = f" ({', '.join(props)})" if props else ""
return f"MethodInfo({self.class_name}.{self.method_name}, params={len(self.parameters)}{prop_str})"
def get_parameter_types(self) -> dict[str, str]:
"""Get mapping of parameter names to their types."""
types = {}
for param in self.parameters:
if param["type"]:
types[param["name"]] = param["type"]
else:
types[param["name"]] = "Any"
return types
def has_type_annotations(self) -> bool:
"""Check if any parameters have type annotations."""
return (
any(param["type"] for param in self.parameters)
or self.return_type is not None
)
def has_parameter_type_annotations(self) -> bool:
"""Check if any parameters have type annotations (excluding return type)."""
return any(param["type"] for param in self.parameters)
def get_import_path(self) -> str:
"""Get import path for this method.
For nested classes (e.g. LRUCache.CacheEntry) only the top-level class
is imported; templates reference the method as Outer.Inner.method.
"""
if self.class_name:
top_level = self.class_name.split(".")[0]
return f"from {self.module_path} import {top_level}"
else:
return f"from {self.module_path} import {self.method_name}"
def get_test_prefix(self) -> str:
"""Get prefix for test function names."""
if self.class_name:
return f"{self.class_name}_{self.method_name}_line_{self.line_number}"
else:
return f"{self.method_name}_line_{self.line_number}"
def get_snake_case_prefix(self) -> str:
"""Get snake_case prefix for test function names.
Converts CamelCase class names so generated function names satisfy
snake_case style checks (e.g. ``SystemEvent`` -> ``system_event``).
"""
def _to_snake(name: str) -> str:
# Handle leading underscores and dunder names.
prefix = ""
while name.startswith("_"):
prefix += "_"
name = name[1:]
# Insert underscore before capital letters (except the first).
s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name)
s2 = re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1)
return prefix + s2.lower()
if self.class_name:
class_part = _to_snake(self.class_name)
method_part = _to_snake(self.method_name)
prefix = f"{class_part}_{method_part}_line_{self.line_number}"
else:
prefix = f"{_to_snake(self.method_name)}_line_{self.line_number}"
# Nested classes are referenced as "Outer.Inner"; dots are not valid
# in Python identifiers, so flatten them into underscores.
return prefix.replace(".", "_")
def get_required_imports_from_defaults(self) -> set[str]:
"""Get imports required for default values in parameters."""
imports = set()
for param in self.parameters:
default = param.get("default")
if default:
# Check for common imports in default values
if "timedelta" in default:
imports.add("from datetime import timedelta")
if "datetime" in default and "datetime.datetime" not in default:
imports.add("from datetime import datetime")
if "Path" in default and "pathlib.Path" not in default:
imports.add("from pathlib import Path")
if "Decimal" in default:
imports.add("from decimal import Decimal")
return imports
def inherits_from_pydantic(self) -> bool:
"""Check if the containing class inherits from pydantic.BaseModel."""
# Direct inheritance from BaseModel
if any("BaseModel" in base for base in self.class_bases):
return True
# Indirect inheritance through BaseAgent (our custom base class)
if any("BaseAgent" in base for base in self.class_bases):
return True
return False
def is_dataclass(self) -> bool:
"""Check if the containing class is decorated with @dataclass."""
return getattr(self, "_is_dataclass", False)
def should_validate_constructor(self) -> bool:
"""Check if constructor validation tests should be generated.
Always False (2026-07-14 regen-merge ruling): statically inferred
raise-expectations are unsound in every class of target —
plain dataclasses store wrong-typed fields silently, Pydantic lax
mode coerces some wrong types per-field, and a `raise` inside
__init__ proves nothing about TYPE validation (PrologEngine raises
on connection failure, not on bad argument types). Each of those
produced DID-NOT-RAISE failures in the 2026-07-14 regeneration.
Constructability is already covered by the generated smoke tests.
"""
return False
def has_validation_block(self) -> bool:
"""Check if any parameter warrants an invalid-value validation test."""
if not self.should_validate_constructor():
return False
for param in self.parameters:
if param["name"] in ("self", "cls") or not param.get("type"):
continue
ptype = param["type"]
if ptype == "Any" or "Optional" in ptype or "None" in ptype:
continue
if (
("int" in ptype and "float" not in ptype)
or "str" in ptype
or "bool" in ptype
or "list" in ptype.lower()
or "List[" in ptype
or "dict" in ptype.lower()
or "Dict[" in ptype
):
return True
return False
def get_validation_error_imports(self) -> set[str]:
"""Get imports needed for validation error testing."""
imports = set()
if self.inherits_from_pydantic():
imports.add("from pydantic import ValidationError")
return imports
# Known class constructors for generated smoke tests.
# Maps type name -> (expression, imports_set)
_KNOWN_CLASS_CTORS: dict[str, tuple[str, set[str]]] = {
"AgentHierarchy": (
"AgentHierarchy()",
{"from orchestrata_ultimate.agents.hierarchy import AgentHierarchy"},
),
"AgentCommunication": (
"AgentCommunication(hierarchy=AgentHierarchy())",
{
"from orchestrata_ultimate.agents.hierarchy import AgentHierarchy",
"from orchestrata_ultimate.agents.communication import AgentCommunication",
},
),
"AgentEscalation": (
"AgentEscalation(hierarchy=AgentHierarchy())",
{
"from orchestrata_ultimate.agents.hierarchy import AgentHierarchy",
"from orchestrata_ultimate.agents.escalation import AgentEscalation",
},
),
"AgentLifecycle": (
"AgentLifecycle()",
{"from orchestrata_ultimate.agents.lifecycle import AgentLifecycle"},
),
"ConstitutionalConstraints": (
"ConstitutionalConstraints()",
{
"from orchestrata_ultimate.agents.constraints import ConstitutionalConstraints"
},
),
"Constraint": (
'Constraint(id="c1", name="test", article=1)',
{"from orchestrata_ultimate.agents.constraints import Constraint"},
),
"CXPVolkAdapter": (
"CXPVolkAdapter()",
{"from orchestrata_ultimate.agents.cxp_adapter import CXPVolkAdapter"},
),
"PostmanAdapter": (
"PostmanAdapter()",
{"from orchestrata_ultimate.cxp.postman_adapter import PostmanAdapter"},
),
"VolkRegistry": (
"VolkRegistry()",
{"from orchestrata_ultimate.agents.volk_registry import VolkRegistry"},
),
"PrologBridge": (
'PrologBridge(rules_dir=Path("/tmp"))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
},
),
"MultiagentPrologBridge": (
'MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
"from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge",
},
),
"VolkRuntime": (
'VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp"))))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
"from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge",
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.agents.runtime import VolkRuntime",
},
),
"VolkPool": (
'VolkPool(runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp"))))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
"from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge",
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.agents.runtime import VolkRuntime",
"from orchestrata_ultimate.agents.pool import VolkPool",
},
),
"TaskExecutor": (
'TaskExecutor(pool=VolkPool(runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp"))))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
"from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge",
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.agents.runtime import VolkRuntime",
"from orchestrata_ultimate.agents.pool import VolkPool",
"from orchestrata_ultimate.agents.task_executor import TaskExecutor",
},
),
"HierarchyNode": (
'HierarchyNode(agent_id="test")',
{"from orchestrata_ultimate.agents.hierarchy import HierarchyNode"},
),
"LifecycleState": (
'LifecycleState(agent_id="test")',
{"from orchestrata_ultimate.agents.lifecycle import LifecycleState"},
),
"TaskDefinition": (
"TaskDefinition()",
{"from orchestrata_ultimate.agents.types import TaskDefinition"},
),
"TaskResult": (
'TaskResult(task_id="t1", success=True)',
{"from orchestrata_ultimate.agents.types import TaskResult"},
),
"VolkInstance": (
"VolkInstance()",
{"from orchestrata_ultimate.agents.types import VolkInstance"},
),
"ClassDesign": (
'ClassDesign(name="Test")',
{"from orchestrata_ultimate.agents.uml_generator import ClassDesign"},
),
"ModuleDesign": (
'ModuleDesign(name="test")',
{"from orchestrata_ultimate.agents.uml_generator import ModuleDesign"},
),
"RedisStateManager": (
"RedisStateManager()",
{"from orchestrata_ultimate.infra.redis_state import RedisStateManager"},
),
"VolkPostmanAdapter": (
'VolkPostmanAdapter(postman=PostmanAdapter(), executor=TaskExecutor(pool=VolkPool(runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
"from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge",
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.agents.runtime import VolkRuntime",
"from orchestrata_ultimate.agents.pool import VolkPool",
"from orchestrata_ultimate.agents.task_executor import TaskExecutor",
"from orchestrata_ultimate.cxp.postman_adapter import PostmanAdapter",
"from orchestrata_ultimate.agents.postman_adapter import VolkPostmanAdapter",
},
),
"IntentTaskAdapter": (
'IntentTaskAdapter(executor=TaskExecutor(pool=VolkPool(runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), runtime=VolkRuntime(registry=VolkRegistry(), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))), prolog=MultiagentPrologBridge(bridge=PrologBridge(rules_dir=Path("/tmp")))))',
{
"from pathlib import Path",
"from orchestrata_ultimate.infra.prolog_bridge import PrologBridge",
"from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge",
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.agents.runtime import VolkRuntime",
"from orchestrata_ultimate.agents.pool import VolkPool",
"from orchestrata_ultimate.agents.task_executor import TaskExecutor",
"from orchestrata_ultimate.agents.intent_adapter import IntentTaskAdapter",
},
),
"GatewayConfig": (
"GatewayConfig()",
{"from orchestrata_ultimate.gateway.config import GatewayConfig"},
),
"WebSocketGateway": (
"WebSocketGateway(config=GatewayConfig())",
{
"from orchestrata_ultimate.gateway.config import GatewayConfig",
"from orchestrata_ultimate.gateway.server import WebSocketGateway",
},
),
"RedisRingBuffer": (
"RedisRingBuffer(config=GatewayConfig())",
{
"from orchestrata_ultimate.gateway.config import GatewayConfig",
"from orchestrata_ultimate.gateway.ring_buffer import RedisRingBuffer",
},
),
"GraphClient": (
"GraphClient()",
{
"from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient"
},
),
"AnomalyDetector": (
"AnomalyDetector(graph_client=GraphClient())",
{
"from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient",
"from orchestrata_ultimate.knowledge_graph.anomaly_detector import AnomalyDetector",
},
),
"PolicyContext": (
'PolicyContext(session_id="s1", volk_id="v1", tool="t1", parameters={})',
{"from orchestrata_ultimate.orchestrator.engine import PolicyContext"},
),
"IntentValidator": (
"IntentValidator()",
{
"from orchestrata_ultimate.orchestrator.intent_parser import IntentValidator"
},
),
"IntentRouter": (
"IntentRouter(volk_registry=VolkRegistry())",
{
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.orchestrator.intent_router import IntentRouter",
},
),
"IntentDispatcher": (
"IntentDispatcher(volk_registry=VolkRegistry(), transport=None)",
{
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.orchestrator.intent_router import IntentDispatcher",
},
),
"IntentLifecycleManager": (
"IntentLifecycleManager(redis_host=REDIS_HOST, redis_port=REDIS_PORT)",
{
"from tests.infrastructure import REDIS_HOST, REDIS_PORT",
"from orchestrata_ultimate.orchestrator.intent_lifecycle import IntentLifecycleManager",
},
),
"IntentExtractor": (
"IntentExtractor()",
{
"from orchestrata_ultimate.orchestrator.intent_parser import IntentExtractor"
},
),
"IntentHistory": (
"IntentHistory(redis=RedisStateManager())",
{
"from orchestrata_ultimate.infra.redis_state import RedisStateManager",
"from orchestrata_ultimate.orchestrator.intent_history import IntentHistory",
},
),
"IntentMetricsCollector": (
"IntentMetricsCollector()",
{
"from orchestrata_ultimate.orchestrator.intent_metrics import IntentMetricsCollector"
},
),
"IntentSecurity": (
"IntentSecurity()",
{
"from orchestrata_ultimate.orchestrator.intent_security import IntentSecurity"
},
),
"IntentImpactAnalyzer": (
"IntentImpactAnalyzer()",
{
"from orchestrata_ultimate.orchestrator.impact.analyzer import IntentImpactAnalyzer"
},
),
"IntentBrokerConfig": (
"IntentBrokerConfig(validator=IntentValidator(), router=IntentRouter(volk_registry=VolkRegistry()), dispatcher=IntentDispatcher(volk_registry=VolkRegistry(), transport=None), lifecycle=IntentLifecycleManager(redis_host=REDIS_HOST, redis_port=REDIS_PORT))",
{
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.orchestrator.intent_parser import IntentValidator",
"from orchestrata_ultimate.orchestrator.intent_router import IntentRouter",
"from orchestrata_ultimate.orchestrator.intent_router import IntentDispatcher",
"from tests.infrastructure import REDIS_HOST, REDIS_PORT",
"from orchestrata_ultimate.orchestrator.intent_lifecycle import IntentLifecycleManager",
"from orchestrata_ultimate.orchestrator.intent_broker import IntentBrokerConfig",
},
),
"IntentBroker": (
"IntentBroker(config=IntentBrokerConfig(validator=IntentValidator(), router=IntentRouter(volk_registry=VolkRegistry()), dispatcher=IntentDispatcher(volk_registry=VolkRegistry(), transport=None), lifecycle=IntentLifecycleManager(redis_host=REDIS_HOST, redis_port=REDIS_PORT)))",
{
"from orchestrata_ultimate.agents.volk_registry import VolkRegistry",
"from orchestrata_ultimate.orchestrator.intent_parser import IntentValidator",
"from orchestrata_ultimate.orchestrator.intent_router import IntentRouter",
"from orchestrata_ultimate.orchestrator.intent_router import IntentDispatcher",
"from tests.infrastructure import REDIS_HOST, REDIS_PORT",
"from orchestrata_ultimate.orchestrator.intent_lifecycle import IntentLifecycleManager",
"from orchestrata_ultimate.orchestrator.intent_broker import IntentBrokerConfig",
"from orchestrata_ultimate.orchestrator.intent_broker import IntentBroker",
},
),
}
def get_dummy_value(self, param: dict[str, Any]) -> tuple[str, set[str]]:
"""Return a Python literal and required imports for a parameter's dummy value.
Uses type annotation to produce a more realistic default than bare None.
"""
ptype = param.get("type") or ""
default = param.get("default")
if default:
return default, set()
imports: set[str] = set()
# Optional / Union with None
if "Optional[" in ptype or "| None" in ptype or "None" in ptype:
return "None", set()
# Primitives
if ptype == "int":
return "0", set()
if ptype == "float":
return "0.0", set()
if ptype == "str":
name = param.get("name") or ""
if any(
k in name for k in ("path", "root", "dir", "file", "folder", "source")
):
return '"/nonexistent/generated-smoke"', set()
return '""', set()
if ptype == "bool":
return "False", set()
# Collections
if ptype.startswith("list[") or ptype.startswith("List["):
return "[]", set()
if ptype.startswith("dict[") or ptype.startswith("Dict["):
return "{}", set()
if ptype.startswith("set[") or ptype.startswith("Set["):
return "set()", set()
if ptype.startswith("tuple[") or ptype.startswith("Tuple["):
return "()", set()
# pathlib — a nonexistent path: Path(".") made generated smokes walk
# the whole repo (scan/RTM functions timed out at >120s)
if "Path" in ptype:
imports.add("from pathlib import Path")
return 'Path("/nonexistent/generated-smoke")', imports
# datetime
if ptype in ("datetime", "datetime.datetime"):
imports.add("from datetime import datetime")
return "datetime.now()", imports
if ptype == "date":
imports.add("from datetime import date")
return "date.today()", imports
if ptype == "timedelta":
imports.add("from datetime import timedelta")
return "timedelta(seconds=1)", imports
# UUID
if "UUID" in ptype:
imports.add("import uuid")
return "uuid.uuid4()", imports
# Enums — can't guess value, use None
if ptype.endswith("Enum") or "Enum" in ptype:
return "None", set()
# Callable / Coroutine
if "Callable" in ptype or "callable" in ptype:
return "lambda *a, **k: None", set()
if "Coroutine" in ptype:
return "None", set()
# Known orchestrata classes. A table entry keyed by bare class name
# must NOT hijack a same-named class defined in the module under
# test (PolicyContext exists in both orchestrator.engine and
# orchestrator.orchestrator — the table's engine import shadowed the
# local dataclass and broke its interface test).
local_names: set[str] = getattr(self, "local_class_names", set())
for cls_name, (expr, cls_imports) in self._KNOWN_CLASS_CTORS.items():
if cls_name in local_names:
continue
if (
ptype == cls_name
or ptype.endswith(f" | {cls_name}")
or cls_name in ptype.split(" | ")
):
return expr, set(cls_imports)
# Known service classes that need real infrastructure
if any(
svc in ptype
for svc in (
"GraphClient",
"Redis",
"Connection",
"Session",
"Engine",
"Pool",
"Client",
)
):
return "None", set()
# Fallback
return "None", set()
class ASTAnalyzer:
"""Analyzes Python files using AST to extract method information."""
def __init__(self) -> None:
self.methods: list[MethodInfo] = []
self._current_file: str = ""
self._current_module_path: str = ""
self.class_inheritance: dict[tuple[str, str], list[str]] = {}
self._current_class_bases: list[str] = []
def analyze_file(self, file_path: Path) -> list[MethodInfo]:
"""Analyze a single Python file and extract method information."""
with open(file_path, encoding="utf-8") as f:
content = f.read()
try:
tree = ast.parse(content)
except SyntaxError as e:
print(f"Syntax error in {file_path}: {e}")
return []
self._current_file = str(file_path)
# Calculate module import path
self._current_module_path = self._file_to_module_path(file_path)
self.methods = [] # Reset for each file
self._visit_node(tree)
return self.methods
def _file_to_module_path(self, file_path: Path) -> str:
"""Convert file path to Python module import path."""
# Convert to absolute path first
abs_path = file_path.absolute()
# Remove .py extension
path_str = str(abs_path).replace(".py", "")
# Find src/ directory in path
parts = path_str.split("/")
# Try to find 'src' in the path
try:
src_index = parts.index("src")
# Keep everything after src/ (including orchestrata_ultimate)
path_str = ".".join(parts[src_index + 1 :])
except ValueError:
# No src/ directory found, use relative path from current directory
# Remove leading ./ if present
if path_str.startswith("./"):
path_str = path_str[2:]
# Convert / to .
path_str = path_str.replace("/", ".")
# Ensure it doesn't start with .
if path_str.startswith("."):
# Remove leading dots
while path_str.startswith("."):
path_str = path_str[1:]
# If empty after removing dots, use a default
if not path_str:
path_str = "module"
# Final validation: ensure no leading dots
if path_str.startswith("."):
# Remove any remaining leading dots
path_str = path_str.lstrip(".")
# Remove .__init__ suffix if present (package __init__.py files)
if path_str.endswith(".__init__"):
path_str = path_str[:-9] # Remove last 9 characters
# Sanitize each dot-separated segment into a valid Python identifier.
# Real filesystem paths (pytest's tmp_path, or any kebab-case
# directory) can contain characters illegal in module paths, such
# as hyphens; left as-is, the "from X import Y" line emitted for
# the generated test file would be a syntax error.
def _sanitize_segment(segment: str) -> str:
sanitized = re.sub(r"\W", "_", segment)
if sanitized and sanitized[0].isdigit():
sanitized = f"_{sanitized}"
return sanitized or "_"
path_str = ".".join(_sanitize_segment(part) for part in path_str.split("."))
return path_str
def analyze_directory(self, directory: Path) -> list[MethodInfo]:
"""Analyze all Python files in a directory."""
all_methods = []
for py_file in directory.rglob("*.py"):
# Skip test files
if (
py_file.name.startswith("test_")
or py_file.parent.name.lower() == "tests"
):
continue
print(f"Analyzing {py_file}")
methods = self.analyze_file(py_file)
all_methods.extend(methods)
return all_methods
def _visit_node(self, node: ast.AST) -> None:
"""Visit AST node and extract method information."""
if isinstance(node, ast.ClassDef):
self._visit_class(node)
# Don't recursively visit class body - _visit_class already handles it
return
# Check for module-level functions (not inside classes)
elif isinstance(node, ast.FunctionDef):
method_info = self._extract_function_info(node, class_name="")
if method_info:
method_info.module_path = self._current_module_path
self.methods.append(method_info)
# Don't recursively visit function body (avoids nested functions)
return
elif isinstance(node, ast.AsyncFunctionDef):
method_info = self._extract_async_function_info(node, class_name="")
if method_info:
method_info.module_path = self._current_module_path
self.methods.append(method_info)
# Don't recursively visit function body (avoids nested functions)
return
# Recursively visit child nodes (but not inside classes - handled above)
for child in ast.iter_child_nodes(node):
self._visit_node(child)
def _visit_class(self, class_node: ast.ClassDef, prefix: str | None = None) -> None:
"""Visit a class definition and extract its methods."""
# Build the fully-qualified class name, including parent classes for
# nested class definitions (e.g. LRUCache.CacheEntry).
full_class_name = f"{prefix}.{class_node.name}" if prefix else class_node.name
# Check for @dataclass decorator
is_dataclass = any(
(isinstance(dec, ast.Name) and dec.id == "dataclass")
or (isinstance(dec, ast.Attribute) and dec.attr == "dataclass")
or (
isinstance(dec, ast.Call)
and isinstance(dec.func, ast.Name)
and dec.func.id == "dataclass"
)
for dec in class_node.decorator_list
)
# Check for Pydantic base classes
base_names = self._extract_base_names(class_node.bases)
is_pydantic = any(
"BaseModel" in base or "BaseAgent" in base for base in base_names
)
# Check for Protocol base class
is_protocol = any("Protocol" in base for base in base_names)
# Store class inheritance
key = (self._current_file, full_class_name)
self.class_inheritance[key] = base_names
# Extract class fields (for dataclass/Pydantic synthetic __init__)
class_fields = self._extract_class_fields(class_node)
# Save previous bases for nested classes
prev_bases = self._current_class_bases
self._current_class_bases = base_names
for item in class_node.body:
if isinstance(item, ast.FunctionDef):
method_info = self._extract_function_info(item, full_class_name)
if method_info:
method_info.module_path = self._current_module_path
# Attach class bases to method info
method_info.class_bases = base_names
self.methods.append(method_info)
elif isinstance(item, ast.AsyncFunctionDef):
method_info = self._extract_async_function_info(item, full_class_name)
if method_info:
method_info.module_path = self._current_module_path
method_info.class_bases = base_names
self.methods.append(method_info)
elif isinstance(item, ast.ClassDef):
# Handle nested class definitions
self._visit_class(item, prefix=full_class_name)
# Recursively visit other nodes (like if statements, assignments)
else:
self._visit_node(item)
# Restore previous bases
self._current_class_bases = prev_bases
# Tag all methods collected for this class with dataclass / protocol status
if is_dataclass:
for m in self.methods:
if (
m.class_name == full_class_name
and m.file_path == self._current_file
):
m._is_dataclass = True
if is_protocol:
for m in self.methods:
if (
m.class_name == full_class_name
and m.file_path == self._current_file
):
m.is_protocol = True
# Synthetic __init__ for dataclasses and Pydantic models without explicit __init__
if (is_dataclass or is_pydantic) and class_fields:
has_explicit_init = any(
m.class_name == full_class_name
and m.method_name == "__init__"
and m.file_path == self._current_file
for m in self.methods
)
if not has_explicit_init:
synthetic_init = MethodInfo()
synthetic_init.class_name = full_class_name
synthetic_init.method_name = "__init__"
synthetic_init.file_path = self._current_file
synthetic_init.line_number = class_node.lineno
synthetic_init.module_path = self._current_module_path
synthetic_init.class_bases = base_names
synthetic_init._is_dataclass = is_dataclass
synthetic_init.is_synthetic_init = True
# self parameter
synthetic_init.parameters = [
{
"name": "self",
"type": None,
"default": None,
"param_type": "positional",
"is_kwonly": False,
}
]
for field in class_fields:
synthetic_init.parameters.append(
{
"name": field["name"],
"type": field["type"],
"default": (
field["default"] if field["has_default"] else None
),
"param_type": "positional",
"is_kwonly": False,
}
)
self.methods.append(synthetic_init)
def _extract_base_names(self, bases: list[ast.expr]) -> list[str]:
"""Extract base class names from AST base expressions."""
base_names = []
for base in bases:
if isinstance(base, ast.Name):
base_names.append(base.id)
elif isinstance(base, ast.Attribute):
parts = []
node: ast.expr = base
while isinstance(node, ast.Attribute):
parts.append(node.attr)
node = node.value
if isinstance(node, ast.Name):
parts.append(node.id)
base_names.append(".".join(reversed(parts)))
else:
try:
base_names.append(ast.unparse(base))
except Exception:
pass
return base_names
def _extract_class_fields(self, class_node: ast.ClassDef) -> list[dict[str, Any]]:
"""Extract annotated fields from a class body (for dataclasses/Pydantic).
Skips names that never appear in the generated ``__init__``:
underscore-prefixed attributes (Pydantic PrivateAttr / private state),
``ClassVar`` annotations, and dataclass ``field(..., init=False)``.
"""
fields = []
for item in class_node.body:
if isinstance(item, ast.AnnAssign):
field_name = (
item.target.id if isinstance(item.target, ast.Name) else None
)
if not field_name or field_name.startswith("_"):
continue
field_type = (
ast.unparse(item.annotation)
if hasattr(ast, "unparse") and item.annotation
else None
)
if field_type and "ClassVar" in field_type:
continue
# dataclass field(..., init=False) never reaches __init__
if (
isinstance(item.value, ast.Call)
and isinstance(item.value.func, ast.Name)
and item.value.func.id == "field"
and any(
kw.arg == "init"
and isinstance(kw.value, ast.Constant)
and kw.value.value is False
for kw in item.value.keywords
)
):
continue
# Check if there's a default value
has_default = item.value is not None
default_value = (
ast.unparse(item.value)
if item.value is not None and hasattr(ast, "unparse")
else None
)
fields.append(
{
"name": field_name,
"type": field_type,
"has_default": has_default,
"default": default_value,
}
)
return fields
def _extract_function_info(
self, func_node: ast.FunctionDef, class_name: str
) -> MethodInfo | None:
"""Extract method information from a FunctionDef node."""
return self._extract_method_info_base(func_node, class_name, is_async=False)
def _extract_async_function_info(
self, func_node: ast.AsyncFunctionDef, class_name: str
) -> MethodInfo | None:
"""Extract method information from an AsyncFunctionDef node."""
return self._extract_method_info_base(func_node, class_name, is_async=True)
def _extract_method_info_base(
self,
func_node: ast.FunctionDef | ast.AsyncFunctionDef,
class_name: str,
is_async: bool,
) -> MethodInfo | None:
"""Base method for extracting method information."""
# Skip dunder methods except __init__
if func_node.name.startswith("__") and func_node.name != "__init__":
return None
# Skip private methods
if func_node.name.startswith("_") and not func_node.name.startswith("__"):
return None
method_info = MethodInfo()
method_info.class_name = class_name
method_info.method_name = func_node.name
method_info.is_async = is_async
method_info.file_path = self._current_file
method_info.line_number = func_node.lineno
# Check for decorators
method_info.is_classmethod = any(
isinstance(decorator, ast.Name) and decorator.id == "classmethod"
for decorator in func_node.decorator_list
)
method_info.is_staticmethod = any(
isinstance(decorator, ast.Name) and decorator.id == "staticmethod"
for decorator in func_node.decorator_list
)
method_info.is_property = any(
(isinstance(decorator, ast.Name) and decorator.id == "property")
or (
isinstance(decorator, ast.Attribute)
and decorator.attr in ("property", "setter", "deleter")
)
for decorator in func_node.decorator_list
)
# Properties: include them for interface stability but skip for
# parameter validation (they have no callable signature)
if method_info.is_property:
# Still return method_info so interface test can verify it exists
pass
# Extract docstring
if (
func_node.body
and isinstance(func_node.body[0], ast.Expr)
and isinstance(func_node.body[0].value, ast.Constant)
):
docstring_value = func_node.body[0].value.value
if isinstance(docstring_value, str):
method_info.docstring = docstring_value
else:
method_info.docstring = None
# Extract parameters
method_info.parameters = self._extract_parameters(func_node.args)
# Extract return type
if func_node.returns:
method_info.return_type = (
ast.unparse(func_node.returns)
if hasattr(ast, "unparse")
else self._type_to_string(func_node.returns)
)
return method_info
def _extract_parameters(self, args_node: ast.arguments) -> list[dict[str, Any]]:
"""Extract parameter information from arguments node."""
parameters = []
# Positional arguments (normal args before *args)
for i, arg in enumerate(args_node.args):
param = self._create_parameter_info(arg, args_node, i, "positional")
parameters.append(param)
# *args parameter (varargs) - appears after positional args, before keyword-only
if args_node.vararg:
param = self._create_parameter_info(
args_node.vararg, args_node, 0, "varargs"
)
parameters.append(param)
# Keyword-only arguments (after *args)
for i, arg in enumerate(args_node.kwonlyargs):
param = self._create_parameter_info(
arg, args_node, i, "keyword", is_kwonly=True
)
parameters.append(param)
# **kwargs parameter
if args_node.kwarg:
param = self._create_parameter_info(args_node.kwarg, args_node, 0, "kwargs")
parameters.append(param)
return parameters
def _create_parameter_info(
self,
arg: ast.arg,
args_node: ast.arguments,
index: int,
param_type: str,
is_kwonly: bool = False,
) -> dict[str, Any]:
"""Create parameter information dictionary."""
param = {
"name": arg.arg,
"type": None,
"default": None,
"param_type": param_type,
"is_kwonly": is_kwonly,
}
# Extract type annotation
if arg.annotation:
param["type"] = (
ast.unparse(arg.annotation)
if hasattr(ast, "unparse")
else self._type_to_string(arg.annotation)
)
# Extract default value
if param_type == "positional" and index >= len(args_node.args) - len(
args_node.defaults
):
default_index = index - (len(args_node.args) - len(args_node.defaults))
if default_index >= 0 and default_index < len(args_node.defaults):
default = args_node.defaults[default_index]
param["default"] = (
ast.unparse(default)
if hasattr(ast, "unparse")
else self._node_to_string(default)
)
elif (
param_type == "keyword"
and (kw_default := args_node.kw_defaults[index]) is not None
):
param["default"] = (
ast.unparse(kw_default)
if hasattr(ast, "unparse")
else self._node_to_string(kw_default)
)
return param
def _type_to_string(self, node: ast.AST) -> str:
"""Convert AST type node to string representation."""
try:
# First try ast.unparse (Python 3.9+)
if hasattr(ast, "unparse"):
return ast.unparse(node)
except AttributeError, NotImplementedError, RecursionError:
pass
# Fallback implementation for complex types or when ast.unparse fails
if isinstance(node, ast.Name):
return node.id
elif isinstance(node, ast.Attribute):
# Handle dotted names like typing.Optional
value = self._type_to_string(node.value)
return f"{value}.{node.attr}"
elif isinstance(node, ast.Subscript):
# Handle generic types like List[int], Dict[str, int]
value = self._type_to_string(node.value)
if isinstance(node.slice, ast.Tuple):
# Multiple type arguments like Dict[str, int]
slice_parts = [self._type_to_string(elt) for elt in node.slice.elts]
slice_str = ", ".join(slice_parts)
else:
# Single type argument like List[int]
slice_str = self._type_to_string(node.slice)
return f"{value}[{slice_str}]"
elif isinstance(node, ast.BinOp):
# Handle union types like str | None or int | float
left = self._type_to_string(node.left)
right = self._type_to_string(node.right)
op = self._binop_to_string(node.op)
return f"{left} {op} {right}"
elif isinstance(node, ast.Constant):
return str(node.value)
elif isinstance(node, ast.Tuple):
# Handle tuple types like (int, str)
elements = [self._type_to_string(elt) for elt in node.elts]
return f"({', '.join(elements)})"
else:
# Unknown node type
return "Any"
def _binop_to_string(self, op: ast.operator) -> str:
"""Convert binary operator to string."""
if isinstance(op, ast.BitOr):
return "|"
elif isinstance(op, ast.BitAnd):
return "&"
else:
return "|" # Default to union operator
def _node_to_string(self, node: ast.AST) -> str:
"""Convert AST node to string representation."""
try:
return ast.unparse(node)
except AttributeError, NotImplementedError:
if isinstance(node, ast.Constant):
return repr(node.value)
elif isinstance(node, ast.Name):
return node.id
else:
return str(node)
class TestGenerator:
"""Generates unit tests from method information using Jinja templates."""
def __init__(self, template_dir: Path | None = None, force_templates: bool = False):
if template_dir is None:
# Use default templates - templates are in project root
self.template_dir = Path(__file__).parent.parent / "templates"
else:
self.template_dir = template_dir
# Create template directory if it doesn't exist
self.template_dir.mkdir(exist_ok=True)
# Create default templates if they don't exist. Pass force_templates
# (--force-templates on the CLI) to overwrite existing template files
# with the defaults; otherwise any user customization is preserved.
self._create_default_templates(force=force_templates)
# Setup Jinja2 environment - disable autoescape for Python code generation
self.jinja_env = jinja2.Environment(
loader=jinja2.FileSystemLoader(self.template_dir),
trim_blocks=True,
lstrip_blocks=True,
autoescape=False,
)
def _create_default_templates(self, force: bool = False) -> None:
"""Create default Jinja templates for test generation.
By default, an existing template file (possibly hand-customized by a
user) is left untouched. Pass force=True to always overwrite it with
the built-in default, e.g. to pick up fixes to the generator itself.
"""
# Template 1: Interface stability test
interface_template = '''{# Interface stability test template #}
{% if method.is_property %}
def test_{{ method.get_snake_case_prefix() }}_interface() -> None:
"""Test that {{ method.class_name }}.{{ method.method_name }} property exists."""
# Verify property exists on class
attr = getattr({{ method.class_name }}, "{{ method.method_name }}", None)
assert attr is not None, (
"Property '{{ method.method_name }}' should exist on {{ method.class_name }}"
)
assert isinstance(attr, property), (
"'{{ method.method_name }}' should be a property, not a regular method"
)
{% else %}
{% if method.is_async %}async {% endif %}def test_{{ method.get_snake_case_prefix() }}_interface() -> None:
"""Test that {{ method.class_name if method.class_name else 'module' }}.{{ method.method_name }} interface hasn't changed."""
{% if method.is_synthetic_init and method.inherits_from_pydantic() %}
# Pydantic model: __init__ is synthesized (and may be **data for custom
# bases), so assert declared-field membership instead of the signature.
{% for param in method.parameters %}
{% if param.name != 'self' and param.name != 'cls' %}
assert "{{ param.name }}" in {{ method.class_name }}.model_fields, (
"Field '{{ param.name }}' missing from {{ method.class_name }}.model_fields"
)
{% endif %}
{% endfor %}
{% else %}
{% if method.class_name %}
{% if method.method_name == '__init__' %}
# Test constructor signature
sig = inspect.signature({{ method.class_name }}.__init__)
{% else %}
# Test method signature
sig = inspect.signature({{ method.class_name }}.{{ method.method_name }})
{% endif %}
{% else %}
# Test function signature
sig = inspect.signature({{ method.method_name }})
{% endif %}
{% if method.return_type is not none %}
# Check return type annotation if it exists in source
assert sig.return_annotation is not inspect.Parameter.empty, (
"Method should have return type annotation, expected '{{ method.return_type }}'"
)
{% endif %}
# Expected parameters (excluding 'self' for methods)
expected_params: list[str] = [
{% for param in method.parameters %}
{% if param.name != 'self' and param.name != 'cls' %}
{% set param_name = param.name %}
{% if method.class_name and param_name[:2] == '__' and param_name[-2:] != '__' %}
{% set param_name = '_' + method.class_name + param_name %}
{% endif %}
"{{ param_name }}",
{% endif %}
{% endfor %}
]
# Get actual parameters
actual_params = list(sig.parameters.keys())
{% if method.class_name %}
# Remove 'self' from actual params for instance methods
if actual_params and actual_params[0] in ("self", "cls"):
actual_params = actual_params[1:]
{% endif %}
# Backward-compatible signature check: every parameter recorded at
# generation time must still be present, in the same relative order.
# A parameter that wasn't recorded (added since generation) is only
# allowed when it can't break an existing caller -- i.e. it has a
# default, or it's a *args/**kwargs catch-all. This tolerates
# compatible evolution (added **kwargs, added an optional kwarg) while
# still failing on a removed parameter, a renamed positional, or a new
# *required* parameter that changes call arity.
removed_params = [p for p in expected_params if p not in actual_params]
assert not removed_params, (
f"Parameter(s) removed or renamed: {removed_params} "
f"(expected {expected_params}, got {actual_params})"
)
preserved_order = [p for p in actual_params if p in expected_params]
assert preserved_order == expected_params, (
f"Parameter order changed: expected {expected_params}, "
f"got {preserved_order}"
)
for extra_param_name in [p for p in actual_params if p not in expected_params]:
extra_param = sig.parameters[extra_param_name]
is_compatible_addition = (
extra_param.kind
in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
or extra_param.default is not inspect.Parameter.empty
)
assert is_compatible_addition, (
f"New required parameter '{extra_param_name}' breaks backward "
f"compatibility (no default, not *args/**kwargs): "
f"expected {expected_params}, got {actual_params}"
)
{% if method.has_parameter_type_annotations() %}
# Optional: Check that every source-annotated parameter is still annotated.
{% for src_param in method.parameters %}
{% if src_param.type and src_param.name not in ['self', 'cls'] %}
assert sig.parameters["{{ src_param.name }}"].annotation is not (
inspect.Parameter.empty
), "Parameter '{{ src_param.name }}' should have type annotation"
{% endif %}
{% endfor %}
{% endif %}
{% endif %}
{% endif %}
'''
# Template 2: Parameter validation test
validation_template = '''{% if method.class_name and method.method_name == '__init__' and method.should_validate_constructor() %}
{% if method.is_async %}async {% endif %}def test_{{ method.get_snake_case_prefix() }}_validation() -> None:
"""Test {{ method.class_name }} constructor parameter validation."""
{% if method.has_validation_block() %}
invalid_kwargs: dict[str, Any] = {}
{% endif %}
{% for param in method.parameters %}
{% if param.name not in ['self', 'cls'] and param.type and param.type != 'Any' %}
# Test {{ param.name }} parameter validation (type: {{ param.type }})
{% set is_pydantic = method.inherits_from_pydantic() %}
{% set exception_expr = 'ValidationError' if is_pydantic else '(TypeError, ValueError, AttributeError)' %}
{% if 'int' in param.type and 'float' not in param.type
and 'Optional' not in param.type and 'None' not in param.type %}
# Test that {{ param.name }} rejects non-integer values
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": 3.14} # Float instead of int
{{ method.class_name }}(**invalid_kwargs)
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": "string"} # String instead of int
{{ method.class_name }}(**invalid_kwargs)
{% elif 'str' in param.type and 'Optional' not in param.type
and 'None' not in param.type %}
# Test that {{ param.name }} rejects non-string values
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": 123} # Int instead of string
{{ method.class_name }}(**invalid_kwargs)
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": []} # List instead of string
{{ method.class_name }}(**invalid_kwargs)
{% elif 'bool' in param.type and 'Optional' not in param.type
and 'None' not in param.type %}
# Test that {{ param.name }} rejects non-boolean values
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": "string"} # String instead of bool
{{ method.class_name }}(**invalid_kwargs)
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": 123} # Int instead of bool
{{ method.class_name }}(**invalid_kwargs)
{% elif 'list' in param.type|lower or 'List[' in param.type %}
# Test that {{ param.name }} rejects non-list values
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": "string"} # String instead of list
{{ method.class_name }}(**invalid_kwargs)
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": 123} # Int instead of list
{{ method.class_name }}(**invalid_kwargs)
{% elif 'dict' in param.type|lower or 'Dict[' in param.type %}
# Test that {{ param.name }} rejects non-dict values
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": "string"} # String instead of dict
{{ method.class_name }}(**invalid_kwargs)
with pytest.raises({{ exception_expr }}):
invalid_kwargs = {"{{ param.name }}": []} # List instead of dict
{{ method.class_name }}(**invalid_kwargs)
{% else %}
# Generic type validation for {{ param.type }} not generated
{% endif %}
{% endif %}
{% endfor %}
# Test that valid parameters work
try:
# Use reasonable defaults for testing
kwargs: dict[str, Any] = {}
{% for param in method.parameters %}
{% if param.name not in ['self', 'cls'] %}
{% if param.type %}
{% if param.default and 'Field(' not in param.default and 'field(' not in param.default and '(' not in param.default and '.' not in param.default and not (param.default.upper() == param.default and param.default != param.default.lower()) %}
kwargs["{{ param.name }}"] = {{ param.default }}
{% elif 'int' in param.type and 'float' not in param.type %}
kwargs["{{ param.name }}"] = 0
{% elif 'str' in param.type %}
kwargs["{{ param.name }}"] = ""
{% elif 'bool' in param.type %}
kwargs["{{ param.name }}"] = False
{% elif 'list' in param.type|lower or 'List[' in param.type %}
kwargs["{{ param.name }}"] = []
{% elif 'dict' in param.type|lower or 'Dict[' in param.type %}
kwargs["{{ param.name }}"] = {}
{% elif 'Optional' in param.type or 'None' in param.type
or param.type.endswith('| None') %}
kwargs["{{ param.name }}"] = None
{% else %}
kwargs["{{ param.name }}"] = None
{% endif %}
{% else %}
kwargs["{{ param.name }}"] = None
{% endif %}
{% endif %}
{% endfor %}
instance: Any = {{ method.class_name }}(**kwargs)
assert instance is not None
except Exception as e:
pytest.fail(f"Constructor failed with valid parameters: {e}")
{% endif %}'''
# Template 3: Smoke test — actually calls methods to exercise code paths
smoke_template = '''{% if method.class_name and method.method_name != '__init__' and not method.is_property and not method.is_protocol %}
{% if method.is_async %}@pytest.mark.asyncio
async def test_{{ method.get_snake_case_prefix() }}_smoke() -> None:
{% else %}def test_{{ method.get_snake_case_prefix() }}_smoke() -> None:
{% endif %}
"""Smoke test: call {{ method.class_name }}.{{ method.method_name }} to exercise code."""
# Try to construct the class with typed dummy args
ctor_kwargs: dict[str, Any] = {}
{% for param in method.ctor_params | default([]) %}
{% if param.name not in ['self', 'cls'] %}
{% if not param.default %}
ctor_kwargs["{{ param.name }}"] = {{ param.dummy_value }}
{% endif %}
{% endif %}
{% endfor %}
try:
# Deliberately speculative: abstract classes (incl. inherited
# abstractness) and stale dummy args raise here at runtime and are
# tolerated below. Constructing through an Any-typed alias keeps
# mypy from flagging abstract instantiation statically without
# scattering per-class ignores (which go stale as unused-ignore).
cls_under_test: Any = {{ method.class_name }}
instance: Any = cls_under_test(**ctor_kwargs)
except Exception:
# Construction failed with dummy args; nothing further to exercise.
return
# Build kwargs and call — any error from dummy args (including stale
# dummy-constructor expressions) is expected and OK
try:
call_kwargs: dict[str, Any] = {}
{% for param in method.parameters %}
{% if param.name not in ['self', 'cls'] %}
{% if param.name == 'argv' %}
call_kwargs["{{ param.name }}"] = []
{% elif not param.default %}
call_kwargs["{{ param.name }}"] = {{ param.dummy_value }}
{% endif %}
{% endif %}
{% endfor %}
{% if method.is_async %}
# Time-bounded: a coroutine that blocks by design (serve loop,
# consumer) must never hang the suite. TimeoutError is an
# Exception subclass, so the handler below treats a timeout as
# "exercised without crashing" — the same contract as dummy-arg
# failures.
await asyncio.wait_for(
instance.{{ method.method_name }}(**call_kwargs), timeout=5.0
)
{% else %}
instance.{{ method.method_name }}(**call_kwargs)
{% endif %}
except Exception:
# Expected: dummy args don't match real requirements
return
{% endif %}
{% if not method.class_name and not method.is_property %}
{% if method.is_async %}@pytest.mark.asyncio
async def test_{{ method.get_snake_case_prefix() }}_smoke() -> None:
{% else %}def test_{{ method.get_snake_case_prefix() }}_smoke() -> None:
{% endif %}
"""Smoke test: call {{ method.method_name }} to exercise code."""
# Build kwargs and call — any error from dummy args (including stale
# dummy-constructor expressions) is expected and OK
try:
call_kwargs: dict[str, Any] = {}
{% for param in method.parameters %}
{% if param.name not in ['self', 'cls'] %}
{% if param.name == 'argv' %}
call_kwargs["{{ param.name }}"] = []
{% elif not param.default %}
call_kwargs["{{ param.name }}"] = {{ param.dummy_value }}
{% endif %}
{% endif %}
{% endfor %}
{% if method.is_async %}
# Time-bounded: a coroutine that blocks by design (serve loop,
# consumer) must never hang the suite. TimeoutError is an
# Exception subclass, so the handler below treats a timeout as
# "exercised without crashing" — the same contract as dummy-arg
# failures.
await asyncio.wait_for(
{{ method.method_name }}(**call_kwargs), timeout=5.0
)
{% else %}
{{ method.method_name }}(**call_kwargs)
{% endif %}
except Exception:
# Expected: dummy args don't match real requirements
return
{% endif %}'''
# Write templates to files. Preserve any existing (possibly
# user-customized) template by default — only overwrite when
# force=True (--force-templates), to pick up fixes on demand.
for filename, content in (
("interface_test.jinja2", interface_template),
("validation_test.jinja2", validation_template),
("smoke_test.jinja2", smoke_template),
):
template_path = self.template_dir / filename
if force or not template_path.exists():
template_path.write_text(content)
def generate_tests(
self,
methods: list[MethodInfo],
output_dir: Path,
prune_duplicates: bool = False,
source_root: Path | None = None,
) -> dict[str, str]:
"""Generate test files for each method.
By default, stale duplicate generated test files found elsewhere
under tests/ are only reported (printed), never deleted. Pass
prune_duplicates=True (--prune-duplicates on the CLI) to actually
remove them.
When source_root is provided, the relative path from source_root to
the source file is encoded into the generated test file name so that
files with the same stem in different subdirectories do not collide.
"""
output_dir.mkdir(parents=True, exist_ok=True)
generated_files = {}
# Group methods by file
methods_by_file: dict[str, list[MethodInfo]] = {}
for method in methods:
if method.file_path not in methods_by_file:
methods_by_file[method.file_path] = []
methods_by_file[method.file_path].append(method)
# Find the tests/ root so we can detect duplicate basenames
tests_root = output_dir
while tests_root.name != "tests" and tests_root.parent != tests_root:
tests_root = tests_root.parent
# Generate tests for each source file
for file_path, file_methods in methods_by_file.items():
# Determine test file name. Include the relative directory path
# so that nested modules with the same stem do not overwrite each
# other (e.g. agents/volks/pm/volk.py vs agents/volks/qa/volk.py).
source_path = Path(file_path)
if source_root is not None:
try:
rel_parts = (
source_path.relative_to(source_root).with_suffix("").parts
)
name_parts = "_".join(rel_parts)
except ValueError:
name_parts = source_path.stem
else:
name_parts = source_path.stem
test_file_name = f"test_{name_parts}_generated.py"
test_file_path = output_dir / test_file_name
# Check for duplicate basenames in sibling directories under tests/
if tests_root.name == "tests":
duplicates = [
p
for p in tests_root.rglob(test_file_name)
if p != test_file_path and p.is_file()
]
for dup in duplicates:
if prune_duplicates:
print(
f" WARN: removing stale duplicate {dup} "
f"(conflicts with {test_file_path})"
)
dup.unlink()
else:
print(
f" INFO: stale duplicate {dup} "
f"(conflicts with {test_file_path}); "
"pass --prune-duplicates to remove it"
)
# Create directory if it doesn't exist
test_file_path.parent.mkdir(parents=True, exist_ok=True)
# Generate test content
test_content = self._generate_test_file_content(file_methods, source_path)
# Write test file
test_file_path.write_text(test_content)
generated_files[str(test_file_path)] = test_content
print(f"Generated: {test_file_path}")
# Run linting/formatting on generated files
self._lint_and_format_generated_files(
output_dir, [Path(p) for p in generated_files]
)
return generated_files
def _lint_and_format_generated_files(
self, output_dir: Path, file_paths: list[Path]
) -> None:
"""Run ruff and black on generated test files."""
if not file_paths:
return
try:
# Find project root (where pyproject.toml is)
project_root = output_dir
while project_root.parent != project_root: # until root
if (project_root / "pyproject.toml").exists():
break
project_root = project_root.parent
else:
# Fallback to output_dir parent
project_root = output_dir.parent
abs_paths = [str(p.resolve()) for p in file_paths]
print(f"Running linting and formatting on {len(abs_paths)} file(s)...")
# Run ruff check --fix via uv so the project's ruff is used.
ruff_cmd = ["uv", "run", "ruff", "check", "--fix"] + abs_paths
result = subprocess.run(
ruff_cmd,
cwd=project_root,
capture_output=True,
text=True,
)
if result.returncode == 0:
print("✓ Ruff linting applied successfully")
if result.stdout:
print(f"Ruff output: {result.stdout}")
else:
print(f"⚠ Ruff encountered issues: {result.stderr}")
if result.stdout:
print(f"Ruff output: {result.stdout}")
# Run black formatting via uv so the project's black is used.
black_cmd = ["uv", "run", "black"] + abs_paths
result = subprocess.run(
black_cmd,
cwd=project_root,
capture_output=True,
text=True,
)
if result.returncode == 0:
print("✓ Black formatting applied successfully")
else:
print(f"⚠ Black encountered issues: {result.stderr}")
if result.stdout:
print(f"Black output: {result.stdout}")
except Exception as e:
print(f"⚠ Linting/formatting failed: {e}")
print("Generated files may need manual linting.")
def _generate_test_file_content(
self, methods: list[MethodInfo], source_path: Path
) -> str:
"""Generate content for a test file."""
if not methods:
return ""
# Load templates
interface_template = self.jinja_env.get_template("interface_test.jinja2")
validation_template = self.jinja_env.get_template("validation_test.jinja2")
smoke_template = self.jinja_env.get_template("smoke_test.jinja2")
# Classes defined in this module — table lookups must not shadow them
local_classes = {m.class_name.split(".")[0] for m in methods if m.class_name}
for m in methods:
m.local_class_names = local_classes
# Build ctor param map: class_name -> list of ctor params with dummy values
ctor_params_by_class: dict[str, list[dict[str, Any]]] = {}
for method in methods:
if method.class_name and method.method_name == "__init__":
enriched = []
for p in method.parameters:
val, _ = method.get_dummy_value(p)
enriched.append({**p, "dummy_value": val})
ctor_params_by_class[method.class_name] = enriched
# Generate test functions
test_functions = []
imports = set()
for method in methods:
# Collect import for this method
import_stmt = method.get_import_path()
imports.add(import_stmt)
# Collect any additional imports needed for default values
required_imports = method.get_required_imports_from_defaults()
imports.update(required_imports)
# Collect imports for validation errors (Pydantic)
validation_imports = method.get_validation_error_imports()
imports.update(validation_imports)
# Enrich parameters with dummy values for templates
for p in method.parameters:
val, extra_imports = method.get_dummy_value(p)
p["dummy_value"] = val
imports.update(extra_imports)
# Attach ctor params to method for smoke template
method.ctor_params = ctor_params_by_class.get(method.class_name, [])
# Generate interface test
interface_test = interface_template.render(method=method)
test_functions.append(interface_test)
# Generate validation test for methods with parameters
has_real_params = any(
p["name"] not in ["self", "cls"] for p in method.parameters
)
if has_real_params and method.parameters:
validation_test = validation_template.render(method=method)
if "def test_" in validation_test:
test_functions.append(validation_test)
# Generate smoke test to exercise code paths. Module-level
# blocking entry points are excluded: ``main`` parses sys.argv
# and exits (SystemExit under pytest); serve/``listen``/
# ``run_forever`` block until killed (timed out at >120s).
# ``serve`` matches by substring, not prefix — ``run_serve``
# hung the 2026-08-16 gate for 120s per run, and its historical
# 0.01s "passes" were EADDRINUSE collisions with the live
# gateway holding the port, not real coverage. Async smokes are
# additionally time-bounded in the template itself (the
# mechanism backstop for blockers this name list misses).
# Interface tests still cover their signatures.
blocking_entry = not method.class_name and (
method.method_name in ("main", "listen", "run_forever")
or "serve" in method.method_name
)
if not blocking_entry:
smoke_test = smoke_template.render(method=method)
if "def test_" in smoke_test:
test_functions.append(smoke_test)
if method.is_async:
imports.add("import asyncio")
# Decide whether pytest is actually used in the generated tests.
test_functions_text = "\n".join(test_functions)
needs_pytest = "pytest." in test_functions_text or any(
"@pytest.mark" in func for func in test_functions
)
# Smoke/validation templates use typing.Any in dict[str, Any] annotations.
needs_typing_any = re.search(r"\bAny\b", test_functions_text) is not None
# Deduplicate imports from same module
imports_by_module: dict[str, set[str]] = {}
for imp in imports:
if imp.startswith("from "):
parts = imp.split(" import ")
if len(parts) == 2:
mod = parts[0][5:]
names = {n.strip() for n in parts[1].split(",")}
imports_by_module.setdefault(mod, set()).update(names)
else:
imports_by_module.setdefault(imp, set())
else:
# Plain "import X" entries (e.g. from get_dummy_value's UUID
# branch) are stored bare here so the render step below
# doesn't double up the "import " keyword.
mod = imp[len("import ") :] if imp.startswith("import ") else imp
imports_by_module.setdefault(mod, set())
if needs_pytest:
imports_by_module.setdefault("pytest", set())
if needs_typing_any:
imports_by_module.setdefault("typing", set()).add("Any")
# Deduplicate imported names across modules to avoid ruff F811
# (redefinition of unused name). A given symbol only needs to be
# imported once per generated test file.
seen_names: set[str] = set()
deduped_imports: dict[str, set[str]] = {}
for mod, names in sorted(imports_by_module.items()):
unique_names = names - seen_names
if unique_names:
deduped_imports[mod] = unique_names
seen_names.update(unique_names)
elif not names:
# Preserve bare ``import mod`` entries (no from-names).
deduped_imports[mod] = set()
imports_by_module = deduped_imports
def _import_group(mod: str) -> int:
# Sort order: stdlib < third-party < local.
root = mod.split(".")[0]
if root in sys.stdlib_module_names:
return 0
if root == "orchestrata_ultimate":
return 2
return 1
import_lines: list[str] = []
current_group = -1
for mod, names in sorted(
imports_by_module.items(), key=lambda kv: (_import_group(kv[0]), kv[0])
):
group = _import_group(mod)
if group != current_group and import_lines:
import_lines.append("")
current_group = group
if names:
import_lines.append(f"from {mod} import {', '.join(sorted(names))}")
else:
import_lines.append(f"import {mod}")
imports_text = "\n".join(import_lines) if import_lines else ""
test_file_content = f'''"""
Generated unit tests for {source_path}.
DO NOT EDIT THIS FILE MANUALLY.
Generated by generate_pytest.py
"""
import inspect
{imports_text}
# Test functions
{"\n".join(test_functions)}
'''
return test_file_content
def main() -> None:
"""Main entry point."""
import argparse
parser = argparse.ArgumentParser(
description="Generate unit tests from AST analysis"
)
parser.add_argument(
"source",
nargs="?",
default=None,
help=(
"Source file or directory to analyze. Optional when --package "
"is given (the package resolves its own source directory)."
),
)
parser.add_argument(
"--package",
default=None,
help=(
"Top-level orchestrata_ultimate package to regenerate (e.g. "
"'agents', 'core', 'gateway'). Resolves the source directory to "
"src/orchestrata_ultimate/<package> (unless 'source' is also "
"given) and defaults --output to tests/generated_<package>."
),
)
parser.add_argument(
"--output",
"-o",
default=None,
help=(
"Output directory for generated tests (default: tests/generated, "
"or tests/generated_<package> when --package is given)"
),
)
parser.add_argument(
"--template-dir", "-t", help="Directory containing Jinja2 templates"
)
parser.add_argument(
"--force-templates",
action="store_true",
help=(
"Overwrite existing template files with the built-in defaults "
"(default: preserve any customized template)"
),
)
parser.add_argument(
"--prune-duplicates",
action="store_true",
help=(
"Delete stale duplicate generated test files found elsewhere "
"under tests/ (default: only report them, never delete)"
),
)
args = parser.parse_args()
if args.package:
package_root = Path("src/orchestrata_ultimate") / args.package
if not package_root.is_dir():
parser.error(
f"--package {args.package!r}: no such directory {package_root}"
)
source_path = Path(args.source) if args.source else package_root
output_dir = (
Path(args.output)
if args.output
else Path(f"tests/generated_{args.package}")
)
else:
if not args.source:
parser.error(
"the following arguments are required: source (or use --package)"
)
source_path = Path(args.source)
output_dir = Path(args.output) if args.output else Path("tests/generated")
if not source_path.exists():
parser.error(f"source path does not exist: {source_path}")
# Analyze source
analyzer = ASTAnalyzer()
if source_path.is_file():
methods = analyzer.analyze_file(source_path)
else:
methods = analyzer.analyze_directory(source_path)
print(f"Found {len(methods)} methods to test")
# Generate tests
generator = TestGenerator(
Path(args.template_dir) if args.template_dir else None,
force_templates=args.force_templates,
)
generated = generator.generate_tests(
methods,
output_dir,
prune_duplicates=args.prune_duplicates,
source_root=source_path,
)
print(f"Generated {len(generated)} test files")
if __name__ == "__main__":
main()