forked from HAL9000/cleveragents-core
123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
import asyncio
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock
|
|
from unittest.mock import Mock
|
|
from unittest.mock import patch
|
|
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.core.config import ConfigurationManager
|
|
from cleveragents.network import AgentNetwork
|
|
from cleveragents.routing.router import Router
|
|
|
|
|
|
@given("I create an AgentNetwork without configuration files")
|
|
def step_impl(context):
|
|
with patch.object(ConfigurationManager, "load_files") as mock_load, patch.object(
|
|
ConfigurationManager, "validate"
|
|
) as mock_validate:
|
|
context.network = AgentNetwork()
|
|
mock_load.assert_not_called()
|
|
mock_validate.assert_not_called()
|
|
|
|
|
|
@given("an AgentNetwork")
|
|
def step_impl(context):
|
|
context.network = AgentNetwork()
|
|
|
|
|
|
@given("the AgentNetwork has a mock router")
|
|
def step_impl(context):
|
|
context.mock_router = Mock(spec=Router)
|
|
context.network.router = context.mock_router
|
|
|
|
|
|
@given("the AgentNetwork has a mock router that responds based on input")
|
|
def step_impl(context):
|
|
context.mock_router = Mock(spec=Router)
|
|
|
|
async def mock_process_message(message, context_data=None):
|
|
return f"mock response to {message}"
|
|
|
|
context.mock_router.process_message = AsyncMock(side_effect=mock_process_message)
|
|
context.network.router = context.mock_router
|
|
|
|
|
|
@when('I create an AgentNetwork with the configuration file "{filename}"')
|
|
def step_impl(context, filename):
|
|
config_path = context.temp_dir / filename
|
|
with patch.object(ConfigurationManager, "load_files") as mock_load, patch.object(
|
|
ConfigurationManager, "validate"
|
|
) as mock_validate:
|
|
context.network = AgentNetwork(config_files=[config_path])
|
|
context.mock_load_files = mock_load
|
|
context.mock_validate = mock_validate
|
|
context.used_config_path = config_path
|
|
|
|
|
|
@when('I process the message "{message}" with the AgentNetwork')
|
|
def step_impl(context, message):
|
|
async def run():
|
|
context.response = await context.network.process(message)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
@when("I try to process a message with the AgentNetwork")
|
|
def step_impl(context):
|
|
context.error = None
|
|
try:
|
|
|
|
async def run():
|
|
await context.network.process("some message")
|
|
|
|
asyncio.run(run())
|
|
except Exception as e:
|
|
context.error = e
|
|
|
|
|
|
@when("I get the router from the AgentNetwork")
|
|
def step_impl(context):
|
|
context.returned_router = context.network.get_router()
|
|
|
|
|
|
@then("the AgentNetwork should have a ConfigurationManager")
|
|
def step_impl(context):
|
|
assert hasattr(context.network, "config_manager")
|
|
assert isinstance(context.network.config_manager, ConfigurationManager)
|
|
|
|
|
|
@then("the AgentNetwork's agent factory should be None")
|
|
def step_impl(context):
|
|
assert context.network.agent_factory is None
|
|
|
|
|
|
@then("the AgentNetwork's router should be None")
|
|
def step_impl(context):
|
|
assert context.network.router is None
|
|
|
|
|
|
@then("the configuration should be loaded and validated")
|
|
def step_impl(context):
|
|
context.mock_load_files.assert_called_once_with([context.used_config_path])
|
|
context.mock_validate.assert_called_once()
|
|
|
|
|
|
@then('the router\'s process_message method should be called with "{message}"')
|
|
def step_impl(context, message):
|
|
context.mock_router.process_message.assert_called_once_with(message, None)
|
|
|
|
|
|
@then('an AssertionError should be raised with message "{message}"')
|
|
def step_impl(context, message):
|
|
assert context.error is not None, "Expected an exception to be raised"
|
|
assert isinstance(context.error, AssertionError)
|
|
assert str(context.error) == message
|
|
|
|
|
|
@then("the returned router should be the mock router")
|
|
def step_impl(context):
|
|
assert context.returned_router is context.mock_router
|