forked from cleveragents/cleveragents-core
159 lines
4.5 KiB
Python
159 lines
4.5 KiB
Python
import asyncio
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
from unittest.mock import patch
|
|
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.agents.base import Agent
|
|
from cleveragents.interactive.session import InteractiveSession
|
|
from cleveragents.routing.router import Router
|
|
from cleveragents.templates.renderer import TemplateRenderer
|
|
|
|
|
|
class MockAgent(Agent):
|
|
async def process(self, message, context=None):
|
|
return f"Response to: {message}"
|
|
|
|
def get_capabilities(self):
|
|
return ["mock"]
|
|
|
|
|
|
@given("the CleverAgents application is configured")
|
|
def step_impl(context):
|
|
context.template_renderer = TemplateRenderer()
|
|
|
|
# Create mock agent
|
|
agent = MockAgent("agent", {}, context.template_renderer)
|
|
|
|
# Create router
|
|
context.router = Router("test_router", {"agent": agent}, context.template_renderer)
|
|
|
|
# Add routes
|
|
context.router.add_routes(
|
|
[
|
|
{"from": "input", "to": "agent"},
|
|
{"from": "agent", "to": "output"},
|
|
]
|
|
)
|
|
|
|
|
|
@when("I start an interactive session")
|
|
def step_impl(context):
|
|
context.session = InteractiveSession(
|
|
routers={"test_router": context.router}, initial_route_name="test_router"
|
|
)
|
|
|
|
# Mock the run method to avoid actually running the interactive loop
|
|
context.session.run = MagicMock()
|
|
|
|
|
|
@then("I should be able to send messages and receive responses")
|
|
def step_impl(context):
|
|
response = asyncio.run(context.session.process_message("Hello"))
|
|
assert response is not None
|
|
assert "Response to: Hello" in response
|
|
|
|
|
|
@then("I should be able to end the session")
|
|
def step_impl(context):
|
|
# This is a placeholder - in a real test, you would verify that the session can be ended
|
|
pass
|
|
|
|
|
|
@given("an interactive session with history")
|
|
def step_impl(context):
|
|
context.template_renderer = TemplateRenderer()
|
|
|
|
# Create mock agent
|
|
agent = MockAgent("agent", {}, context.template_renderer)
|
|
|
|
# Create router
|
|
context.router = Router("test_router", {"agent": agent}, context.template_renderer)
|
|
|
|
# Create a temporary file for history
|
|
with tempfile.NamedTemporaryFile(suffix=".json", delete=False) as f:
|
|
context.history_file = Path(f.name)
|
|
|
|
# Create session
|
|
context.session = InteractiveSession(
|
|
routers={"test_router": context.router},
|
|
initial_route_name="test_router",
|
|
history_file=context.history_file,
|
|
)
|
|
|
|
# Add some history
|
|
context.session.add_to_history("user", "Hello")
|
|
context.session.add_to_history("assistant", "Hi there!")
|
|
|
|
|
|
@when("I save the conversation history")
|
|
def step_impl(context):
|
|
context.session.save_history()
|
|
|
|
|
|
@when("I load the conversation history in a new session")
|
|
def step_impl(context):
|
|
context.new_session = InteractiveSession(
|
|
routers={"test_router": context.router},
|
|
initial_route_name="test_router",
|
|
history_file=context.history_file,
|
|
)
|
|
context.new_session.load_history()
|
|
|
|
|
|
@then("the new session should contain the previous conversation")
|
|
def step_impl(context):
|
|
assert len(context.new_session.history) == 2
|
|
assert context.new_session.history[0]["role"] == "user"
|
|
assert context.new_session.history[0]["content"] == "Hello"
|
|
assert context.new_session.history[1]["role"] == "assistant"
|
|
assert context.new_session.history[1]["content"] == "Hi there!"
|
|
|
|
|
|
@given("an active interactive session")
|
|
def step_impl(context):
|
|
context.template_renderer = TemplateRenderer()
|
|
|
|
# Create mock agent
|
|
agent = MockAgent("agent", {}, context.template_renderer)
|
|
|
|
# Create router
|
|
context.router = Router("test_router", {"agent": agent}, context.template_renderer)
|
|
|
|
# Create session
|
|
context.session = InteractiveSession(
|
|
routers={"test_router": context.router}, initial_route_name="test_router"
|
|
)
|
|
|
|
|
|
@when('I enter a command prefixed with "/"')
|
|
def step_impl(context):
|
|
# Mock the process_command method
|
|
original_process_command = context.session.process_command
|
|
|
|
def mock_process_command(command_str):
|
|
context.command_processed = True
|
|
context.command = command_str
|
|
return True
|
|
|
|
context.session.process_command = mock_process_command
|
|
|
|
# Process a command
|
|
context.session.process_command("/help")
|
|
|
|
|
|
@then("the command should be executed")
|
|
def step_impl(context):
|
|
assert context.command_processed
|
|
assert context.command == "/help"
|
|
|
|
|
|
@then("the appropriate response should be displayed")
|
|
def step_impl(context):
|
|
# This is a placeholder - in a real test, you would verify the response
|
|
pass
|