linting and jinja2

Author: prassanna-ravishankar

examples/tutorials/20_behavior_testing/000_basic_sync_testing/test_sync_agent.py

@@ -17,10 +17,10 @@
 """
 
 from agentex.lib.testing import (
+    test_sync_agent,
+    assert_valid_agent_response,
     assert_agent_response_contains,
     assert_conversation_maintains_context,
-    assert_valid_agent_response,
-    test_sync_agent,
 )
 
 # TODO: Replace with your actual sync agent name from 'agentex agents list'

examples/tutorials/20_behavior_testing/010_agentic_testing/test_agentic_agent.py

@@ -18,7 +18,7 @@
 
 import pytest
 
-from agentex.lib.testing import assert_valid_agent_response, test_agentic_agent
+from agentex.lib.testing import test_agentic_agent, assert_valid_agent_response
 
 # TODO: Replace with your actual agent name from 'agentex agents list'
 AGENT_NAME = "ab000-hello-acp"

src/agentex/lib/cli/commands/init.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from enum import Enum
 from typing import Any, Dict
 from pathlib import Path
@@ -23,18 +25,14 @@ class TemplateType(str, Enum):
     SYNC = "sync"
 
 
-def render_template(
-    template_path: str, context: Dict[str, Any], template_type: TemplateType
-) -> str:
+def render_template(template_path: str, context: Dict[str, Any], template_type: TemplateType) -> str:
     """Render a template with the given context"""
     env = Environment(loader=FileSystemLoader(TEMPLATES_DIR / template_type.value))
     template = env.get_template(template_path)
     return template.render(**context)
 
 
-def create_project_structure(
-    path: Path, context: Dict[str, Any], template_type: TemplateType, use_uv: bool
-):
+def create_project_structure(path: Path, context: Dict[str, Any], template_type: TemplateType, use_uv: bool):
     """Create the project structure from templates"""
     # Create project directory
     project_dir: Path = path / context["project_name"]
@@ -79,6 +77,9 @@ def create_project_structure(
     # Add development notebook for agents
     root_templates["dev.ipynb.j2"] = "dev.ipynb"
 
+    # Add test file
+    root_templates["test_agent.py.j2"] = "test_agent.py"
+
     for template, output in root_templates.items():
         output_path = project_dir / output
         output_path.write_text(render_template(template, context, template_type))
@@ -97,10 +98,7 @@ def get_project_context(answers: Dict[str, Any], project_path: Path, manifest_ro
     return {
         **answers,
         "project_name": project_name,
-        "workflow_class": "".join(
-            word.capitalize() for word in answers["agent_name"].split("-")
-        )
-        + "Workflow",
+        "workflow_class": "".join(word.capitalize() for word in answers["agent_name"].split("-")) + "Workflow",
         "workflow_name": answers["agent_name"],
         "queue_name": project_name + "_queue",
         "project_path_from_build_root": project_path_from_build_root,
@@ -155,9 +153,7 @@ def validate_agent_name(text: str) -> bool | str:
     if not template_type:
         return
 
-    project_path = questionary.path(
-        "Where would you like to create your project?", default="."
-    ).ask()
+    project_path = questionary.path("Where would you like to create your project?", default=".").ask()
     if not project_path:
         return
 
@@ -175,9 +171,7 @@ def validate_agent_name(text: str) -> bool | str:
     if not agent_directory_name:
         return
 
-    description = questionary.text(
-        "Provide a brief description of your agent:", default="An AgentEx agent"
-    ).ask()
+    description = questionary.text("Provide a brief description of your agent:", default="An AgentEx agent").ask()
     if not description:
         return
 
@@ -208,9 +202,7 @@ def validate_agent_name(text: str) -> bool | str:
     context["use_uv"] = answers["use_uv"]
 
     # Create project structure
-    create_project_structure(
-        project_path, context, answers["template_type"], answers["use_uv"]
-    )
+    create_project_structure(project_path, context, answers["template_type"], answers["use_uv"])
 
     # Show next steps
     console.print("\n[bold green]✨ Project created successfully![/bold green]")
@@ -226,7 +218,8 @@ def validate_agent_name(text: str) -> bool | str:
         console.print("4. Run locally:")
         console.print("   agentex agents run --manifest manifest.yaml")
 
-    console.print("5. Deploy your agent:")
-    console.print(
-        "   agentex agents deploy --cluster your-cluster --namespace your-namespace"
-    )
+    console.print("5. Test your agent:")
+    console.print("   pytest test_agent.py -v")
+
+    console.print("6. Deploy your agent:")
+    console.print("   agentex agents deploy --cluster your-cluster --namespace your-namespace")

src/agentex/lib/cli/templates/default/test_agent.py.j2

@@ -0,0 +1,79 @@
+"""
+Tests for {{ agent_name }} agent
+
+This file provides basic behavior tests for your agent using the AgentEx testing framework.
+
+Prerequisites:
+    - AgentEx services running (make dev)
+    - Agent running: agentex agents run --manifest manifest.yaml
+
+Run tests:
+    pytest test_agent.py -v
+"""
+
+import pytest
+
+from agentex.lib.testing import (
+    test_agentic_agent,
+    assert_valid_agent_response,
+    assert_agent_response_contains,
+)
+
+AGENT_NAME = "{{ agent_name }}"
+
+
+@pytest.mark.asyncio
+async def test_agent_responds():
+    """Test that agent responds to basic messages."""
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        response = await test.send_event(
+            "Hello! Please respond briefly.",
+            timeout_seconds=30.0
+        )
+
+        assert_valid_agent_response(response)
+        assert len(response.content) > 0
+        print(f"✓ Agent responded: {response.content[:80]}...")
+
+
+@pytest.mark.asyncio
+async def test_agent_multi_turn():
+    """Test that agent handles multi-turn conversation."""
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        # Turn 1
+        response1 = await test.send_event("Hello!", timeout_seconds=30.0)
+        assert_valid_agent_response(response1)
+
+        # Turn 2
+        response2 = await test.send_event("Thank you!", timeout_seconds=30.0)
+        assert_valid_agent_response(response2)
+
+        # Verify history
+        history = await test.get_conversation_history()
+        assert len(history) >= 4  # 2 user + 2 agent messages
+        print(f"✓ Conversation: {len(history)} messages")
+
+
+@pytest.mark.asyncio
+async def test_agent_custom_scenario():
+    """
+    Add your custom test scenarios here.
+
+    Example: Test specific functionality of your agent
+    """
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        # Customize this test for your agent's specific behavior
+        response = await test.send_event(
+            "Your custom test message here",
+            timeout_seconds=30.0
+        )
+
+        assert_valid_agent_response(response)
+
+        # Add assertions specific to your agent's expected behavior
+        # assert_agent_response_contains(response, "expected text")
+
+
+if __name__ == "__main__":
+    print(f"Run with: pytest test_agent.py -v")
+    print(f"Testing agent: {AGENT_NAME}")

src/agentex/lib/cli/templates/sync/test_agent.py.j2

@@ -0,0 +1,72 @@
+"""
+Tests for {{ agent_name }} sync agent
+
+This file provides basic behavior tests for your sync agent using the AgentEx testing framework.
+
+Prerequisites:
+    - AgentEx services running (make dev)
+    - Agent running: agentex agents run --manifest manifest.yaml
+
+Run tests:
+    pytest test_agent.py -v
+"""
+
+from agentex.lib.testing import (
+    test_sync_agent,
+    assert_valid_agent_response,
+    assert_agent_response_contains,
+)
+
+AGENT_NAME = "{{ agent_name }}"
+
+
+def test_agent_responds():
+    """Test that sync agent responds to basic messages."""
+    with test_sync_agent(agent_name=AGENT_NAME) as test:
+        response = test.send_message("Hello! Please respond briefly.")
+
+        assert_valid_agent_response(response)
+        assert len(response.content) > 0
+        print(f"✓ Agent responded: {response.content[:80]}...")
+
+
+def test_agent_multi_turn():
+    """Test that sync agent handles multi-turn conversation."""
+    with test_sync_agent(agent_name=AGENT_NAME) as test:
+        # Turn 1
+        response1 = test.send_message("Hello!")
+        assert_valid_agent_response(response1)
+
+        # Turn 2
+        response2 = test.send_message("How are you?")
+        assert_valid_agent_response(response2)
+
+        # Turn 3
+        response3 = test.send_message("Thank you!")
+        assert_valid_agent_response(response3)
+
+        # Verify history
+        history = test.get_conversation_history()
+        assert len(history) >= 6  # 3 user + 3 agent messages
+        print(f"✓ Conversation: {len(history)} messages")
+
+
+def test_agent_custom_scenario():
+    """
+    Add your custom test scenarios here.
+
+    Example: Test specific functionality of your sync agent
+    """
+    with test_sync_agent(agent_name=AGENT_NAME) as test:
+        # Customize this test for your agent's specific behavior
+        response = test.send_message("Your custom test message here")
+
+        assert_valid_agent_response(response)
+
+        # Add assertions specific to your agent's expected behavior
+        # assert_agent_response_contains(response, "expected text")
+
+
+if __name__ == "__main__":
+    print(f"Run with: pytest test_agent.py -v")
+    print(f"Testing agent: {AGENT_NAME}")

src/agentex/lib/cli/templates/temporal/test_agent.py.j2

@@ -0,0 +1,107 @@
+"""
+Tests for {{ agent_name }} temporal agent
+
+This file provides basic behavior tests for your temporal agent using the AgentEx testing framework.
+
+Prerequisites:
+    - AgentEx services running (make dev)
+    - Temporal server running
+    - Agent running: agentex agents run --manifest manifest.yaml
+
+Run tests:
+    pytest test_agent.py -v
+
+Note: Temporal agents may need longer timeouts due to workflow orchestration overhead.
+"""
+
+import pytest
+
+from agentex.lib.testing import (
+    test_agentic_agent,
+    assert_valid_agent_response,
+    assert_agent_response_contains,
+)
+
+AGENT_NAME = "{{ agent_name }}"
+
+
+@pytest.mark.asyncio
+async def test_agent_responds():
+    """Test that agent responds to basic messages."""
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        response = await test.send_event(
+            "Hello! Please respond briefly.",
+            timeout_seconds=60.0  # Temporal agents may need more time
+        )
+
+        assert_valid_agent_response(response)
+        assert len(response.content) > 0
+        print(f"✓ Agent responded: {response.content[:80]}...")
+
+
+@pytest.mark.asyncio
+async def test_agent_multi_turn():
+    """Test that agent handles multi-turn conversation."""
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        # Turn 1
+        response1 = await test.send_event("Hello!", timeout_seconds=60.0)
+        assert_valid_agent_response(response1)
+
+        # Turn 2
+        response2 = await test.send_event("How are you?", timeout_seconds=60.0)
+        assert_valid_agent_response(response2)
+
+        # Turn 3
+        response3 = await test.send_event("Thank you!", timeout_seconds=60.0)
+        assert_valid_agent_response(response3)
+
+        # Verify history
+        history = await test.get_conversation_history()
+        assert len(history) >= 6  # 3 user + 3 agent messages
+        print(f"✓ Conversation: {len(history)} messages")
+
+
+@pytest.mark.asyncio
+async def test_agent_workflow_execution():
+    """
+    Test your temporal workflow execution.
+
+    Temporal agents can handle long-running tasks with retries and state management.
+    Adjust timeout based on your workflow's expected duration.
+    """
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        response = await test.send_event(
+            "Execute your workflow task here",
+            timeout_seconds=120.0  # Longer timeout for complex workflows
+        )
+
+        assert_valid_agent_response(response)
+
+        # Add assertions specific to your workflow's expected behavior
+        # assert_agent_response_contains(response, "workflow completed")
+
+
+@pytest.mark.asyncio
+async def test_agent_custom_scenario():
+    """
+    Add your custom test scenarios here.
+
+    Example: Test specific functionality of your temporal agent
+    """
+    async with test_agentic_agent(agent_name=AGENT_NAME) as test:
+        # Customize this test for your agent's specific behavior
+        response = await test.send_event(
+            "Your custom test message here",
+            timeout_seconds=60.0
+        )
+
+        assert_valid_agent_response(response)
+
+        # Add assertions specific to your agent's expected behavior
+        # assert_agent_response_contains(response, "expected text")
+
+
+if __name__ == "__main__":
+    print(f"Run with: pytest test_agent.py -v")
+    print(f"Testing agent: {AGENT_NAME}")
+    print("\nNote: Temporal agents may require longer timeouts")

src/agentex/lib/testing/__init__.py

@@ -46,9 +46,9 @@ async def test_my_agentic_agent():
     assert_conversation_maintains_context,
 )
 from agentex.lib.testing.exceptions import (
+    AgentTimeoutError,
     AgentNotFoundError,
     AgentSelectionError,
-    AgentTimeoutError,
 )
 
 __all__ = [

src/agentex/lib/testing/config.py

@@ -4,8 +4,8 @@
 Centralized configuration management with environment variable support.
 """
 
-import logging
 import os
+import logging
 from dataclasses import dataclass
 
 logger = logging.getLogger(__name__)