CleverThis Engineering
5bbda89386
- Fix bare excepts (replace with except Exception:) - Remove wildcard star imports from test step files - Add explicit imports for ConfigurationError, AgentCreationError - Add 'from err' to raise statements in except blocks - Fix loop variable binding with default arguments (B023) - Organize imports, fix trailing whitespace and blank line whitespace - Bump Python to 3.13 and update tool configurations
578 lines
21 KiB
Python
578 lines
21 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.api.async_step import async_run_until_complete
|
|
|
|
from cleveractors.agents.tool import ToolAgent
|
|
from cleveractors.core.application import ReactiveCleverAgentsApp
|
|
from cleveractors.core.exceptions import AgentCreationError, ExecutionError
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
|
|
@given('a ToolAgent is configured with name "{name}" and config')
|
|
def step_tool_agent_config(context, name):
|
|
config = json.loads(context.text)
|
|
context.agent_config = config
|
|
context.agent_name = name
|
|
if not hasattr(context, "template_renderer"):
|
|
context.template_renderer = TemplateRenderer()
|
|
|
|
|
|
async def _create_tool_agent_helper(context):
|
|
"""Helper function for creating ToolAgent."""
|
|
context.error = None
|
|
try:
|
|
context.agent = ToolAgent(
|
|
name=context.agent_name,
|
|
config=context.agent_config,
|
|
template_renderer=context.template_renderer,
|
|
)
|
|
# Mock HTTP requests for testing to avoid network calls
|
|
if "http_request" in context.agent_config.get("tools", []):
|
|
# We'll handle HTTP mocking with aiohttp patches during execution
|
|
pass
|
|
except (AgentCreationError, Exception) as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I create the ToolAgent")
|
|
@async_run_until_complete
|
|
async def step_create_tool_agent(context):
|
|
await _create_tool_agent_helper(context)
|
|
|
|
|
|
@when("I try to create the ToolAgent")
|
|
@async_run_until_complete
|
|
async def step_try_to_create_tool_agent(context):
|
|
# This is an alias for the create step to make scenarios more readable
|
|
await _create_tool_agent_helper(context)
|
|
|
|
|
|
@when('I process a message with the ToolAgent: "{message}"')
|
|
@async_run_until_complete
|
|
async def step_process_message_tool_agent(context, message):
|
|
context.error = None
|
|
context.result = None
|
|
try:
|
|
test_context = (
|
|
{"_unsafe_mode": context.unsafe} if hasattr(context, "unsafe") else {}
|
|
)
|
|
|
|
# Mock shell commands for consistent testing across different systems
|
|
if context.agent_config.get("allow_shell", False) and message.strip():
|
|
# Check if this looks like a shell command execution
|
|
parts = message.strip().split()
|
|
first_part = parts[0] if parts else ""
|
|
|
|
# Commands that should be mocked
|
|
should_mock = (
|
|
"/bin/" in first_part
|
|
or "/usr/bin/" in first_part
|
|
or first_part in ["sleep", "false", "rm"]
|
|
)
|
|
|
|
if should_mock:
|
|
# Mock subprocess execution
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Determine the mock behavior based on the command
|
|
if "echo" in first_part:
|
|
# Mock echo command - return the arguments
|
|
mock_output = " ".join(parts[1:])
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(
|
|
return_value=(mock_output.encode(), b"")
|
|
)
|
|
mock_subprocess.return_value = mock_process
|
|
elif "sleep" in first_part:
|
|
# Mock sleep that takes too long (will trigger timeout)
|
|
async def timeout_sleep(*args, **kwargs):
|
|
await asyncio.sleep(10) # Will be caught by timeout
|
|
return (b"", b"")
|
|
|
|
mock_process = MagicMock()
|
|
mock_process.communicate = timeout_sleep
|
|
mock_subprocess.return_value = mock_process
|
|
elif "false" in first_part:
|
|
# Mock command that returns non-zero exit code
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = 1
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
elif "rm" in first_part:
|
|
# rm commands should be blocked by safe mode before execution
|
|
# But if we get here, mock it to fail
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = 1
|
|
mock_process.communicate = AsyncMock(
|
|
return_value=(b"", b"Permission denied")
|
|
)
|
|
mock_subprocess.return_value = mock_process
|
|
else:
|
|
# Default mock for other commands
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = 0
|
|
mock_process.communicate = AsyncMock(return_value=(b"", b""))
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
context.result = await context.agent.process_message(
|
|
message, context=test_context
|
|
)
|
|
else:
|
|
# Not a shell command, process normally
|
|
context.result = await context.agent.process_message(
|
|
message, context=test_context
|
|
)
|
|
else:
|
|
# No shell support or empty message, process normally
|
|
context.result = await context.agent.process_message(
|
|
message, context=test_context
|
|
)
|
|
except ExecutionError as e:
|
|
context.error = e
|
|
|
|
|
|
async def _process_json_message_helper(context):
|
|
"""Helper function for processing JSON messages."""
|
|
context.error = None
|
|
context.result = None
|
|
try:
|
|
message = json.loads(context.text)
|
|
|
|
# for file paths
|
|
if "file" in message.get("args", {}):
|
|
message["args"]["file"] = os.path.join(
|
|
context.scenario_temp, message["args"]["file"]
|
|
)
|
|
|
|
message_str = json.dumps(message)
|
|
test_context = (
|
|
{"_unsafe_mode": context.unsafe} if hasattr(context, "unsafe") else {}
|
|
)
|
|
|
|
# Mock HTTP requests if this is an HTTP request tool call
|
|
if message.get("tool") == "http_request":
|
|
url = message.get("args", {}).get("url", "")
|
|
|
|
# Handle specific error scenarios
|
|
if "delay" in url or context.agent_config.get("timeout", 5) <= 0.001:
|
|
# Simulate timeout error
|
|
async def mock_timeout_session():
|
|
import asyncio
|
|
|
|
raise asyncio.TimeoutError("Simulated timeout")
|
|
|
|
with patch("aiohttp.ClientSession") as mock_client:
|
|
mock_client.return_value.__aenter__ = AsyncMock(
|
|
side_effect=mock_timeout_session
|
|
)
|
|
context.result = await context.agent.process_message(
|
|
message_str, context=test_context
|
|
)
|
|
elif "invalid://" in url:
|
|
# Simulate invalid URL error
|
|
async def mock_invalid_session():
|
|
from aiohttp import InvalidURL
|
|
|
|
raise InvalidURL("Invalid URL scheme")
|
|
|
|
with patch("aiohttp.ClientSession") as mock_client:
|
|
mock_client.return_value.__aenter__ = AsyncMock(
|
|
side_effect=mock_invalid_session
|
|
)
|
|
context.result = await context.agent.process_message(
|
|
message_str, context=test_context
|
|
)
|
|
else:
|
|
# Normal successful response
|
|
mock_response = MagicMock()
|
|
mock_response.status = 200
|
|
mock_response.text = AsyncMock(
|
|
return_value='{"success": true, "url": "' + url + '"}'
|
|
)
|
|
|
|
mock_session = MagicMock()
|
|
mock_request = AsyncMock()
|
|
mock_request.__aenter__ = AsyncMock(return_value=mock_response)
|
|
mock_request.__aexit__ = AsyncMock(return_value=None)
|
|
mock_session.request.return_value = mock_request
|
|
|
|
mock_client_session = MagicMock()
|
|
mock_client_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_client_session.__aexit__ = AsyncMock(return_value=None)
|
|
|
|
with patch("aiohttp.ClientSession", return_value=mock_client_session):
|
|
context.result = await context.agent.process_message(
|
|
message_str, context=test_context
|
|
)
|
|
else:
|
|
context.result = await context.agent.process_message(
|
|
message_str, context=test_context
|
|
)
|
|
except (ExecutionError, json.JSONDecodeError) as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I process a JSON message with the ToolAgent")
|
|
@async_run_until_complete
|
|
async def step_process_json_message_tool_agent_no_params(context):
|
|
"""Step definition for processing JSON message without colon."""
|
|
await _process_json_message_helper(context)
|
|
|
|
|
|
@when("I process a JSON message with the ToolAgent:")
|
|
@async_run_until_complete
|
|
async def step_process_json_message_tool_agent(context):
|
|
await _process_json_message_helper(context)
|
|
|
|
|
|
@then('the result should be "{expected_result}"')
|
|
def step_check_result(context, expected_result):
|
|
assert context.result is not None, "Expected a result, but got None"
|
|
assert str(context.result) == expected_result, (
|
|
f"Expected '{expected_result}', but got '{context.result}'"
|
|
)
|
|
assert context.error is None, f"Expected no error, but got {context.error}"
|
|
|
|
|
|
@then('the result should be ""')
|
|
def step_check_empty_result(context):
|
|
assert context.result is not None, "Expected a result, but got None"
|
|
assert str(context.result) == "", (
|
|
f"Expected empty string, but got '{context.result}'"
|
|
)
|
|
assert context.error is None, f"Expected no error, but got {context.error}"
|
|
|
|
|
|
@then('the tool result should contain "{text}"')
|
|
def step_check_result_contains(context, text):
|
|
assert context.result is not None, "Expected a result, but got None"
|
|
assert text in str(context.result), (
|
|
f"Expected result to contain '{text}', but it was '{context.result}'"
|
|
)
|
|
assert context.error is None, f"Expected no error, but got {context.error}"
|
|
|
|
|
|
@then('tool execution should fail with message containing "{message}"')
|
|
def step_tool_execution_fail(context, message):
|
|
assert context.error is not None, "Expected an error, but none was raised"
|
|
assert isinstance(context.error, ExecutionError), (
|
|
f"Expected ExecutionError, but got {type(context.error)}"
|
|
)
|
|
assert message in str(context.error), (
|
|
f"Expected error message to contain '{message}', but it was '{str(context.error)}'"
|
|
)
|
|
|
|
|
|
@then('agent creation should fail with message containing "{message}"')
|
|
def step_agent_creation_fail(context, message):
|
|
assert context.error is not None, "Expected an error, but none was raised"
|
|
assert isinstance(context.error, AgentCreationError), (
|
|
f"Expected AgentCreationError, but got {type(context.error)}"
|
|
)
|
|
assert message in str(context.error), (
|
|
f"Expected error message to contain '{message}', but it was '{str(context.error)}'"
|
|
)
|
|
|
|
|
|
@given("I am running in unsafe mode")
|
|
def step_run_unsafe(context):
|
|
context.unsafe = True
|
|
|
|
|
|
@then('the agent capabilities should contain "{capability}"')
|
|
def step_check_capabilities(context, capability):
|
|
capabilities = context.agent.get_capabilities()
|
|
assert capability in capabilities, f"Expected '{capability}' in {capabilities}"
|
|
|
|
|
|
@then('the agent metadata "{key}" should be {value}')
|
|
def step_check_metadata(context, key, value):
|
|
metadata = context.agent.get_metadata()
|
|
# convert value string to bool/int if possible
|
|
if value.lower() == "true":
|
|
typed_value = True
|
|
elif value.lower() == "false":
|
|
typed_value = False
|
|
else:
|
|
try:
|
|
typed_value = int(value)
|
|
except ValueError:
|
|
typed_value = value # keep as string
|
|
|
|
assert key in metadata, f"Metadata key '{key}' not found."
|
|
assert metadata[key] == typed_value, (
|
|
f"Expected metadata['{key}'] to be {typed_value}, but got {metadata[key]}"
|
|
)
|
|
|
|
|
|
@when("I create a test file with content {content}")
|
|
def step_create_test_file(context, content):
|
|
"""Create a test file for file operations."""
|
|
|
|
# Create file in current working directory for the tool to find
|
|
filepath = "test_file.txt"
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write(content.strip('"'))
|
|
# Clean up after test - use direct dict access to avoid behave context issues
|
|
if not hasattr(context, "__dict__") or "_cleanup_files" not in context.__dict__:
|
|
context.__dict__["_cleanup_files"] = []
|
|
context.__dict__["_cleanup_files"].append(filepath)
|
|
|
|
|
|
@when("I create an absolute test file with content {content}")
|
|
def step_create_absolute_test_file(context, content):
|
|
"""Create a test file at absolute path for file operations."""
|
|
filepath = "/tmp/test_abs_file.txt"
|
|
with open(filepath, "w", encoding="utf-8") as f:
|
|
f.write(content.strip('"'))
|
|
# Clean up after test - use direct dict access to avoid behave context issues
|
|
if not hasattr(context, "__dict__") or "_cleanup_files" not in context.__dict__:
|
|
context.__dict__["_cleanup_files"] = []
|
|
context.__dict__["_cleanup_files"].append(filepath)
|
|
|
|
|
|
@when("I process a JSON message with unsafe context with the ToolAgent")
|
|
@async_run_until_complete
|
|
async def step_process_json_message_unsafe_context(context):
|
|
"""Process JSON message with unsafe context."""
|
|
context.error = None
|
|
context.result = None
|
|
try:
|
|
message = json.loads(context.text)
|
|
test_context = {"_unsafe_mode": True}
|
|
|
|
# for file paths - keep absolute paths as is
|
|
message_str = json.dumps(message)
|
|
context.result = await context.agent.process_message(
|
|
message_str, context=test_context
|
|
)
|
|
except (ExecutionError, json.JSONDecodeError) as e:
|
|
context.error = e
|
|
|
|
|
|
@when('I process a message with the ToolAgent: ""')
|
|
@async_run_until_complete
|
|
async def step_process_empty_message_tool_agent(context):
|
|
"""Process an empty message with the ToolAgent."""
|
|
context.error = None
|
|
context.result = None
|
|
try:
|
|
test_context = (
|
|
{"_unsafe_mode": context.unsafe} if hasattr(context, "unsafe") else {}
|
|
)
|
|
context.result = await context.agent.process_message("", context=test_context)
|
|
except ExecutionError as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I process a malformed message causing general exception")
|
|
@async_run_until_complete
|
|
async def step_process_malformed_message(context):
|
|
"""Process a message that causes a general exception."""
|
|
context.error = None
|
|
context.result = None
|
|
try:
|
|
# This will cause an exception in the tool processing due to invalid tool name
|
|
context.result = await context.agent.process_message("nonexistent_tool")
|
|
except ExecutionError as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I process a JSON message with unsafe context with the ToolAgent:")
|
|
@async_run_until_complete
|
|
async def step_process_json_unsafe_context_with_colon(context):
|
|
"""Process JSON message with unsafe context using ToolAgent."""
|
|
# Call the function implementation directly to avoid nested async issues
|
|
import json
|
|
|
|
context.error = None
|
|
context.result = None
|
|
try:
|
|
message = json.loads(context.text)
|
|
test_context = {"_unsafe_mode": True}
|
|
|
|
# for file paths - keep absolute paths as is
|
|
message_str = json.dumps(message)
|
|
context.result = await context.agent.process_message(
|
|
message_str, context=test_context
|
|
)
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
# File Write Mode Steps
|
|
|
|
|
|
@then('the result should contain "hello from file"')
|
|
def step_result_contains_hello_from_file(context):
|
|
"""Verify the result contains hello from file."""
|
|
assert context.result is not None
|
|
assert "hello from file" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "test file content"')
|
|
def step_result_contains_test_file_content(context):
|
|
"""Verify the result contains test file content."""
|
|
assert context.result is not None
|
|
assert "test file content" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "absolute file content"')
|
|
def step_result_contains_absolute_file_content(context):
|
|
"""Verify the result contains absolute file content."""
|
|
assert context.result is not None
|
|
assert "absolute file content" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "initial content appended content"')
|
|
def step_result_contains_appended_content(context):
|
|
"""Verify the result contains initial content appended content."""
|
|
assert context.result is not None
|
|
assert "initial content appended content" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "inserted line"')
|
|
def step_result_contains_inserted_line(context):
|
|
"""Verify the result contains inserted line."""
|
|
assert context.result is not None
|
|
assert "inserted line" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "first line"')
|
|
def step_result_contains_first_line(context):
|
|
"""Verify the result contains first line."""
|
|
assert context.result is not None
|
|
assert "first line" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "middle line"')
|
|
def step_result_contains_middle_line(context):
|
|
"""Verify the result contains middle line."""
|
|
assert context.result is not None
|
|
assert "middle line" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
@then('the result should contain "absolute test content"')
|
|
def step_result_contains_absolute_test_content(context):
|
|
"""Verify the result contains absolute test content."""
|
|
assert context.result is not None
|
|
assert "absolute test content" in str(context.result)
|
|
assert context.error is None
|
|
|
|
|
|
# Steps for application-level tool execution testing
|
|
|
|
|
|
@given("an application with no tool agents")
|
|
def step_app_with_no_tool_agents(context):
|
|
"""Create application without tool agents."""
|
|
# Ensure temp_dir exists
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
|
|
config = """
|
|
agents:
|
|
test_agent:
|
|
type: llm
|
|
config:
|
|
provider: openai
|
|
model: gpt-3.5-turbo
|
|
|
|
routes:
|
|
main:
|
|
type: stream
|
|
stream_type: cold
|
|
operators:
|
|
- type: map
|
|
params:
|
|
agent: test_agent
|
|
publications:
|
|
- __output__
|
|
|
|
merges:
|
|
- sources: [__input__]
|
|
target: main
|
|
"""
|
|
config_file = context.temp_dir / "config.yaml"
|
|
config_file.write_text(config)
|
|
# Enable verbose mode to get detailed error messages for testing
|
|
context.app = ReactiveCleverAgentsApp([config_file], verbose=1)
|
|
|
|
|
|
@given("an application with tool agent")
|
|
def step_app_with_tool_agent(context):
|
|
"""Create application with tool agent."""
|
|
# Ensure temp_dir exists
|
|
if not hasattr(context, "temp_dir"):
|
|
context.temp_dir = Path(tempfile.mkdtemp())
|
|
|
|
config = """
|
|
agents:
|
|
tool_agent:
|
|
type: tool
|
|
config:
|
|
tools:
|
|
- name: test_tool
|
|
description: Test tool
|
|
function: print
|
|
|
|
routes:
|
|
main:
|
|
type: stream
|
|
stream_type: cold
|
|
operators:
|
|
- type: map
|
|
params:
|
|
agent: tool_agent
|
|
publications:
|
|
- __output__
|
|
|
|
merges:
|
|
- sources: [__input__]
|
|
target: main
|
|
"""
|
|
config_file = context.temp_dir / "config.yaml"
|
|
config_file.write_text(config)
|
|
context.app = ReactiveCleverAgentsApp([config_file])
|
|
|
|
|
|
@when('executing tool "{tool_name}" with empty params')
|
|
def step_execute_tool_empty_params(context, tool_name):
|
|
"""Execute tool with empty parameters."""
|
|
try:
|
|
context.result = context.app._execute_single_tool(tool_name, {})
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when("processing content with invalid tool JSON")
|
|
def step_process_invalid_tool_json(context):
|
|
"""Process content with invalid tool JSON."""
|
|
content = "[TOOL_EXECUTE:test_tool] invalid{json [/TOOL_EXECUTE]"
|
|
context.result = context.app._process_tool_commands(content)
|
|
|
|
|
|
@then('tool execution returns "{error_msg}" error')
|
|
def step_tool_execution_error(context, error_msg):
|
|
"""Verify tool execution error."""
|
|
assert error_msg in context.result
|
|
|
|
|
|
@then('result contains "{error_msg}" error')
|
|
def step_result_contains_error(context, error_msg):
|
|
"""Verify result contains error."""
|
|
assert error_msg in context.result
|