feat: enhance error handling and add comprehensive session tests
This commit is contained in:
@@ -103,7 +103,7 @@ class InteractiveSession:
|
||||
try:
|
||||
return await self.router.process_message(message, self.context)
|
||||
except Exception as e:
|
||||
raise InteractiveSessionError(f"Failed to process message: {str(e)}")
|
||||
raise InteractiveSessionError(f"Failed to process message: {str(e)}") from e
|
||||
|
||||
def add_to_history(self, role: str, content: str) -> None:
|
||||
"""
|
||||
@@ -186,9 +186,15 @@ class InteractiveSession:
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nInterrupted by user.")
|
||||
self.running = False
|
||||
break
|
||||
except InteractiveSessionError as e:
|
||||
if e.__cause__:
|
||||
print(f"Error: {str(e.__cause__)}")
|
||||
else:
|
||||
print(f"Error: {str(e)}")
|
||||
except Exception as e:
|
||||
print(f"Error: {str(e)}")
|
||||
print(f"An unexpected error occurred: {e}")
|
||||
|
||||
self.save_history()
|
||||
print("Session ended.")
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
Feature: Comprehensive Interactive Session Functionality
|
||||
As a developer, I want to test all aspects of the InteractiveSession class,
|
||||
including its main run loop and error handling, to ensure its robustness.
|
||||
|
||||
Scenario: Session handles failure during history loading
|
||||
Given an interactive session with a corrupted history file
|
||||
When I try to load the history
|
||||
Then an InteractiveSessionError should be raised with a message about loading history
|
||||
|
||||
Scenario: Session handles failure during history saving
|
||||
Given an interactive session with a read-only history file
|
||||
When I try to save the history
|
||||
Then an InteractiveSessionError should be raised with a message about saving history
|
||||
|
||||
Scenario: Session handles failure during message processing
|
||||
Given an interactive session where the router will fail
|
||||
When I try to process a message through the session
|
||||
Then an InteractiveSessionError should be raised with a message about processing a message
|
||||
|
||||
Scenario: Session run loop processes user input and commands
|
||||
Given an interactive session with a mock router
|
||||
And a sequence of user inputs: "hello", "/exit"
|
||||
When I run the session's main loop
|
||||
Then the router should have processed the message "hello"
|
||||
And the session should have stopped running
|
||||
|
||||
Scenario: Session run loop handles verbose mode
|
||||
Given an interactive session in verbose mode
|
||||
And a sequence of user inputs: "test verbose", "/exit"
|
||||
When I run the session's main loop
|
||||
Then the output should contain "Processing message..."
|
||||
|
||||
Scenario: Session run loop handles KeyboardInterrupt gracefully
|
||||
Given an interactive session
|
||||
And the user input will raise a KeyboardInterrupt
|
||||
When I run the session's main loop
|
||||
Then the output should contain "Interrupted by user."
|
||||
And the session should have stopped running
|
||||
|
||||
Scenario: Session run loop handles generic exceptions gracefully
|
||||
Given an interactive session where message processing will raise a generic error
|
||||
And a sequence of user inputs: "cause error", "/exit"
|
||||
When I run the session's main loop
|
||||
Then the output should contain "Error: Test processing error"
|
||||
|
||||
Scenario: Session run loop handles startup failure
|
||||
Given an interactive session that will fail to load history on startup
|
||||
When I try to run the session's main loop
|
||||
Then an InteractiveSessionError should be raised with a message about running the session
|
||||
|
||||
Scenario: Session displays history correctly
|
||||
Given an interactive session with some history entries
|
||||
When I display the history
|
||||
Then the output must contain "You: Hello"
|
||||
And the output must contain "Agent: Hi there!"
|
||||
|
||||
Scenario: Session handles history operations when no file is provided
|
||||
Given an interactive session without a history file
|
||||
When I call load_history and save_history
|
||||
Then no errors should be raised
|
||||
@@ -0,0 +1,98 @@
|
||||
Feature: Extended coverage for interactive commands
|
||||
In order to ensure the robustness of the interactive CLI, all command parsing
|
||||
and execution paths must be thoroughly tested.
|
||||
|
||||
Scenario Outline: Parsing various command aliases and formats
|
||||
Given I have a command parser
|
||||
When I parse the command "<command_str>"
|
||||
Then the command should be identified as "<expected_command>"
|
||||
And the arguments should be <expected_args>
|
||||
|
||||
Examples:
|
||||
| command_str | expected_command | expected_args |
|
||||
| /h | help | [] |
|
||||
| /? | help | [] |
|
||||
| /quit | exit | [] |
|
||||
| /q | exit | [] |
|
||||
| /hist 15 | history | ['15'] |
|
||||
| /cls | clear | [] |
|
||||
| /save | save | [] |
|
||||
| /v on | verbose | ['on'] |
|
||||
| /conf show | config_show | [] |
|
||||
| /cfg set k v | config_set | ['k', 'v'] |
|
||||
| /config | config | [] |
|
||||
| /agent ls | agents_list | [] |
|
||||
| /agents info foo | agent_info | ['foo'] |
|
||||
| /agents | agents | [] |
|
||||
| /rt | route | [] |
|
||||
| /sw bar | switch | ['bar'] |
|
||||
| /unknown | unknown | [] |
|
||||
| an_unknown_command | an_unknown_command | [] |
|
||||
|
||||
Scenario: Executing verbose command variants
|
||||
Given I have a command executor
|
||||
When I execute the command "verbose" with arguments "on"
|
||||
Then the result should be successful
|
||||
And the message should be "Verbose output enabled."
|
||||
When I execute the command "verbose" with arguments "off"
|
||||
Then the result should be successful
|
||||
And the message should be "Verbose output disabled."
|
||||
When I execute the command "verbose" with no arguments
|
||||
Then the result should be successful
|
||||
And the message should be "Toggled verbose output."
|
||||
|
||||
Scenario: Executing config command variants
|
||||
Given I have a command executor
|
||||
When I execute the command "config_show" with no arguments
|
||||
Then the result should be successful
|
||||
And the message should be "Showing configuration for path: "
|
||||
When I execute the command "config_show" with arguments "some.path"
|
||||
Then the result should be successful
|
||||
And the message should be "Showing configuration for path: some.path"
|
||||
When I execute the command "config_set" with arguments "path"
|
||||
Then the result should be unsuccessful
|
||||
And the message should be "Usage: /config set path value"
|
||||
When I execute the command "config" with no arguments
|
||||
Then the result should be successful
|
||||
And the message should be "Available config commands: show, set"
|
||||
|
||||
Scenario: Executing agents command variants
|
||||
Given I have a command executor
|
||||
When I execute the command "agent_info" with no arguments
|
||||
Then the result should be successful
|
||||
And the message should be "Showing information for agent: "
|
||||
When I execute the command "agents" with no arguments
|
||||
Then the result should be successful
|
||||
And the message should be "Available agent commands: list, info"
|
||||
|
||||
Scenario: Executing commands that depend on a session when no session is available
|
||||
Given I have a command executor without a session
|
||||
When I execute the command "save" with no arguments
|
||||
Then the result message should be "History not available."
|
||||
When I execute the command "route" with no arguments
|
||||
Then the result message should be "Route command is not available."
|
||||
|
||||
Scenario: Executing route command with a session
|
||||
Given I have a command executor with a complete mock session
|
||||
And the session has routes "default", "backup" and active route "default"
|
||||
When I execute the command "route" with no arguments
|
||||
Then the result should be successful
|
||||
And the message should be "Current route: 'default'. Available: default, backup."
|
||||
When I execute the command "route" with arguments "nonexistent"
|
||||
Then the result should be unsuccessful
|
||||
And the message should be "Error: Route 'nonexistent' not found."
|
||||
|
||||
Scenario: Executing switch command variants
|
||||
Given I have a command executor
|
||||
When I execute the command "switch" with arguments "new_agent"
|
||||
Then the result should be successful
|
||||
And the message should be "Switching to agent: new_agent"
|
||||
When I execute the command "switch" with no arguments
|
||||
Then the result should be unsuccessful
|
||||
And the message should be "Usage: /switch agent_name"
|
||||
|
||||
Scenario: Executing an unknown command
|
||||
Given I have a command executor
|
||||
When I execute the command "phantom_command" with no arguments
|
||||
Then the result should be unsuccessful
|
||||
And the message should be "Unknown command: phantom_command"
|
||||
@@ -1,15 +1,19 @@
|
||||
import ast
|
||||
import shlex
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import Mock
|
||||
|
||||
from behave import given
|
||||
from behave import then
|
||||
from behave import when
|
||||
|
||||
from cleveragents.interactive.commands import execute_command
|
||||
from cleveragents.interactive.commands import format_help_text
|
||||
from cleveragents.interactive import commands
|
||||
from cleveragents.interactive.commands import parse_command
|
||||
|
||||
|
||||
# --- Parsing Steps ---
|
||||
@given("I have a command parser")
|
||||
def step_impl(context):
|
||||
# Nothing to do, we'll use the parse_command function directly
|
||||
pass
|
||||
|
||||
|
||||
@@ -39,111 +43,25 @@ def step_impl(context, arg1, arg2):
|
||||
assert arg2 in args, f"Expected '{arg2}' in args, got {args}"
|
||||
|
||||
|
||||
@then("the arguments should be {expected_args}")
|
||||
def step_impl(context, expected_args):
|
||||
expected_list = ast.literal_eval(expected_args)
|
||||
assert context.parsed_command["args"] == expected_list, (
|
||||
f"Expected {expected_list}, " f"got {context.parsed_command['args']}"
|
||||
)
|
||||
|
||||
|
||||
# --- Executor Setup Steps ---
|
||||
@given("I have a command executor")
|
||||
def step_impl(context):
|
||||
# Nothing to do, we'll use the execute_command function directly
|
||||
pass
|
||||
context.executor = commands.execute_command
|
||||
context.context = {"session": None}
|
||||
|
||||
|
||||
@when('I execute the "{command}" command')
|
||||
def step_impl(context, command):
|
||||
context.command_result = execute_command({"command": command, "args": []})
|
||||
|
||||
|
||||
@when('I execute the "{command}" command with args "{args_str}"')
|
||||
def step_impl(context, command, args_str):
|
||||
args = args_str.split()
|
||||
context.command_result = execute_command({"command": command, "args": args})
|
||||
|
||||
|
||||
@when('I execute the "{command}" command without args')
|
||||
def step_impl(context, command):
|
||||
context.command_result = execute_command({"command": command, "args": []})
|
||||
|
||||
|
||||
@then("I should receive help text containing available commands")
|
||||
@given("I have a command executor without a session")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Available Commands:" in context.command_result["message"]
|
||||
), "Help text doesn't contain 'Available Commands:'"
|
||||
assert (
|
||||
"/help" in context.command_result["message"]
|
||||
), "Help text doesn't mention the help command"
|
||||
assert (
|
||||
"/exit" in context.command_result["message"]
|
||||
), "Help text doesn't mention the exit command"
|
||||
|
||||
|
||||
@then("I should receive a message about exiting")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Exiting" in context.command_result["message"]
|
||||
), f"Expected message about exiting, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("the continue flag should be false")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
context.command_result["continue"] is False
|
||||
), "Expected continue flag to be False"
|
||||
|
||||
|
||||
@then("I should receive a message about showing configuration")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Showing configuration" in context.command_result["message"]
|
||||
), f"Expected message about showing configuration, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about setting configuration")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Setting configuration" in context.command_result["message"]
|
||||
), f"Expected message about setting configuration, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about listing agents")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Listing available agents" in context.command_result["message"]
|
||||
), f"Expected message about listing agents, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about showing agent information")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Showing information for agent" in context.command_result["message"]
|
||||
), f"Expected message about showing agent information, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about switching agents")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Switching to agent" in context.command_result["message"]
|
||||
), f"Expected message about switching agents, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive an error message about usage")
|
||||
def step_impl(context):
|
||||
assert not context.command_result[
|
||||
"success"
|
||||
], "Expected command to fail but it succeeded"
|
||||
assert (
|
||||
"Usage:" in context.command_result["message"]
|
||||
), f"Expected usage information, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive an error message about unknown command")
|
||||
def step_impl(context):
|
||||
assert not context.command_result[
|
||||
"success"
|
||||
], "Expected command to fail but it succeeded"
|
||||
assert (
|
||||
"Unknown command" in context.command_result["message"]
|
||||
), f"Expected unknown command message, got '{context.command_result['message']}'"
|
||||
|
||||
|
||||
from unittest.mock import Mock
|
||||
context.executor = commands.execute_command
|
||||
context.context = {"session": None}
|
||||
|
||||
|
||||
@given("I have a command executor with a mock session")
|
||||
@@ -151,20 +69,153 @@ def step_impl(context):
|
||||
context.mock_session = Mock()
|
||||
context.mock_session.routers = {"main": Mock(), "secondary": Mock()}
|
||||
context.mock_session.active_route_name = "main"
|
||||
context.context = {"session": context.mock_session}
|
||||
context.executor = commands.execute_command
|
||||
|
||||
def execute(command_str):
|
||||
# The command parser expects a leading slash
|
||||
if not command_str.startswith("/"):
|
||||
command_str = f"/{command_str}"
|
||||
command_dict = parse_command(command_str)
|
||||
return execute_command(command_dict, {"session": context.mock_session})
|
||||
|
||||
context.executor = execute
|
||||
@given("I have a command executor with a complete mock session")
|
||||
def step_impl(context):
|
||||
context.executor = commands.execute_command
|
||||
context.mock_session = MagicMock()
|
||||
context.context = {"session": context.mock_session}
|
||||
|
||||
|
||||
@given(
|
||||
'the session has routes "{route1}", "{route2}" and active route "{active_route}"'
|
||||
)
|
||||
def step_impl(context, route1, route2, active_route):
|
||||
context.mock_session.routers = {route1: Mock(), route2: Mock()}
|
||||
context.mock_session.active_route_name = active_route
|
||||
|
||||
|
||||
# --- Execution Steps ---
|
||||
@when('I execute the command "{command}" with arguments "{arguments}"')
|
||||
def step_impl(context, command, arguments):
|
||||
args = shlex.split(arguments)
|
||||
command_dict = {"command": command, "args": args}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
@when('I execute the command "{command}" with no arguments')
|
||||
def step_impl(context, command):
|
||||
command_dict = {"command": command, "args": []}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
@when('I execute the "{command}" command')
|
||||
def step_impl(context, command):
|
||||
command_dict = {"command": command, "args": []}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
@when('I execute the "{command}" command with args "{args_str}"')
|
||||
def step_impl(context, command, args_str):
|
||||
args = shlex.split(args_str)
|
||||
command_dict = {"command": command, "args": args}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
@when('I execute the "{command}" command without args')
|
||||
def step_impl(context, command):
|
||||
command_dict = {"command": command, "args": []}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
@when('I execute the route command with arg "{route_name}"')
|
||||
def step_impl(context, route_name):
|
||||
context.result = context.executor(f"route {route_name}")
|
||||
command_dict = {"command": "route", "args": [route_name]}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
@when("I execute the route command without args")
|
||||
def step_impl(context):
|
||||
command_dict = {"command": "route", "args": []}
|
||||
context.result = context.executor(command_dict, context.context)
|
||||
|
||||
|
||||
# --- Assertion Steps ---
|
||||
@then("the result should be successful")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
context.result["success"] is True
|
||||
), f"Command was not successful: {context.result['message']}"
|
||||
|
||||
|
||||
@then("the result should be unsuccessful")
|
||||
def step_impl(context):
|
||||
assert context.result["success"] is False, "Command was unexpectedly successful"
|
||||
|
||||
|
||||
@then('the message should be "{expected_message}"')
|
||||
def step_impl(context, expected_message):
|
||||
assert context.result["message"] == expected_message, (
|
||||
f"Expected '{expected_message}', " f"got '{context.result['message']}'"
|
||||
)
|
||||
|
||||
|
||||
@then('the result message should be "{expected_message}"')
|
||||
def step_impl(context, expected_message):
|
||||
assert context.result["message"] == expected_message
|
||||
|
||||
|
||||
@then("I should receive help text containing available commands")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Available Commands:" in context.result["message"]
|
||||
), "Help text doesn't contain 'Available Commands:'"
|
||||
assert (
|
||||
"/help" in context.result["message"]
|
||||
), "Help text doesn't mention the help command"
|
||||
assert (
|
||||
"/exit" in context.result["message"]
|
||||
), "Help text doesn't mention the exit command"
|
||||
|
||||
|
||||
@then("I should receive a message about exiting")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Exiting" in context.result["message"]
|
||||
), f"Expected message about exiting, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then("the continue flag should be false")
|
||||
def step_impl(context):
|
||||
assert context.result["continue"] is False, "Expected continue flag to be False"
|
||||
|
||||
|
||||
@then("I should receive a message about showing configuration")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Showing configuration" in context.result["message"]
|
||||
), f"Expected message about showing configuration, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about setting configuration")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Setting configuration" in context.result["message"]
|
||||
), f"Expected message about setting configuration, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about listing agents")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Listing available agents" in context.result["message"]
|
||||
), f"Expected message about listing agents, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about showing agent information")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Showing information for agent" in context.result["message"]
|
||||
), f"Expected message about showing agent information, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive a message about switching agents")
|
||||
def step_impl(context):
|
||||
assert (
|
||||
"Switching to agent" in context.result["message"]
|
||||
), f"Expected message about switching agents, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then('the active route should be switched to "{route_name}"')
|
||||
@@ -173,12 +224,23 @@ def step_impl(context, route_name):
|
||||
assert context.mock_session.active_route_name == route_name
|
||||
|
||||
|
||||
@when("I execute the route command without args")
|
||||
def step_impl(context):
|
||||
context.result = context.executor("route")
|
||||
|
||||
|
||||
@then("I should receive info about the current and available routes")
|
||||
def step_impl(context):
|
||||
assert "Current route: 'main'" in context.result["message"]
|
||||
assert "Available: main, secondary" in context.result["message"]
|
||||
|
||||
|
||||
@then("I should receive an error message about usage")
|
||||
def step_impl(context):
|
||||
assert not context.result["success"], "Expected command to fail but it succeeded"
|
||||
assert (
|
||||
"Usage:" in context.result["message"]
|
||||
), f"Expected usage information, got '{context.result['message']}'"
|
||||
|
||||
|
||||
@then("I should receive an error message about unknown command")
|
||||
def step_impl(context):
|
||||
assert not context.result["success"], "Expected command to fail but it succeeded"
|
||||
assert (
|
||||
"Unknown command" in context.result["message"]
|
||||
), f"Expected unknown command message, got '{context.result['message']}'"
|
||||
|
||||
@@ -0,0 +1,265 @@
|
||||
import asyncio
|
||||
import io
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import patch
|
||||
|
||||
from behave import given
|
||||
from behave import then
|
||||
from behave import when
|
||||
|
||||
from cleveragents.core.exceptions import InteractiveSessionError
|
||||
from cleveragents.interactive.session import InteractiveSession
|
||||
from cleveragents.routing.router import Router
|
||||
|
||||
|
||||
# --- Givens ---
|
||||
@given("an interactive session with a corrupted history file")
|
||||
def step_impl(context):
|
||||
context.temp_dir = Path("build/tests/temp_behave_test_files")
|
||||
context.temp_dir.mkdir(exist_ok=True, parents=True)
|
||||
if not hasattr(context, "cleanup_dirs"):
|
||||
context.cleanup_dirs = []
|
||||
context.cleanup_dirs.append(context.temp_dir)
|
||||
|
||||
history_file = context.temp_dir / "corrupted_history.json"
|
||||
history_file.write_text("this is not valid json")
|
||||
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": MagicMock(spec=Router)},
|
||||
initial_route_name="main",
|
||||
history_file=history_file,
|
||||
)
|
||||
|
||||
|
||||
@given("an interactive session with a read-only history file")
|
||||
def step_impl(context):
|
||||
context.temp_dir = Path("build/tests/temp_behave_test_files")
|
||||
context.temp_dir.mkdir(exist_ok=True, parents=True)
|
||||
if not hasattr(context, "cleanup_dirs"):
|
||||
context.cleanup_dirs = []
|
||||
context.cleanup_dirs.append(context.temp_dir)
|
||||
|
||||
history_file = context.temp_dir / "readonly_history.json"
|
||||
history_file.touch()
|
||||
history_file.chmod(0o444) # Read-only permissions
|
||||
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": MagicMock(spec=Router)},
|
||||
initial_route_name="main",
|
||||
history_file=history_file,
|
||||
)
|
||||
|
||||
|
||||
@given("an interactive session where the router will fail")
|
||||
def step_impl(context):
|
||||
mock_router = MagicMock(spec=Router)
|
||||
mock_router.process_message = AsyncMock(
|
||||
side_effect=Exception("Router processing failed")
|
||||
)
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": mock_router}, initial_route_name="main"
|
||||
)
|
||||
|
||||
|
||||
@given("an interactive session with a mock router")
|
||||
def step_impl(context):
|
||||
context.mock_router = MagicMock(spec=Router)
|
||||
context.mock_router.process_message = AsyncMock(return_value="Mock response")
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": context.mock_router}, initial_route_name="main"
|
||||
)
|
||||
|
||||
|
||||
@given('a sequence of user inputs: "{input1}", "{input2}"')
|
||||
def step_impl(context, input1, input2):
|
||||
context.user_inputs = [input1, input2]
|
||||
|
||||
|
||||
@given("an interactive session in verbose mode")
|
||||
def step_impl(context):
|
||||
context.mock_router = MagicMock(spec=Router)
|
||||
context.mock_router.process_message = AsyncMock(return_value="Verbose response")
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": context.mock_router},
|
||||
initial_route_name="main",
|
||||
verbose=True,
|
||||
)
|
||||
|
||||
|
||||
@given("an interactive session")
|
||||
def step_impl(context):
|
||||
context.mock_router = MagicMock(spec=Router)
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": context.mock_router}, initial_route_name="main"
|
||||
)
|
||||
|
||||
|
||||
@given("the user input will raise a KeyboardInterrupt")
|
||||
def step_impl(context):
|
||||
context.user_inputs = [KeyboardInterrupt()]
|
||||
|
||||
|
||||
@given("an interactive session where message processing will raise a generic error")
|
||||
def step_impl(context):
|
||||
context.mock_router = MagicMock(spec=Router)
|
||||
context.mock_router.process_message = AsyncMock(
|
||||
side_effect=Exception("Test processing error")
|
||||
)
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": context.mock_router}, initial_route_name="main"
|
||||
)
|
||||
|
||||
|
||||
@given("an interactive session that will fail to load history on startup")
|
||||
def step_impl(context):
|
||||
# Patch load_history to raise an error during the run method
|
||||
context.load_history_patch = patch.object(
|
||||
InteractiveSession, "load_history", side_effect=Exception("Failed to load")
|
||||
)
|
||||
context.load_history_patch.start()
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": MagicMock(spec=Router)}, initial_route_name="main"
|
||||
)
|
||||
|
||||
|
||||
@given("an interactive session with some history entries")
|
||||
def step_impl(context):
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": MagicMock(spec=Router)}, initial_route_name="main"
|
||||
)
|
||||
context.session.add_to_history("user", "Hello")
|
||||
context.session.add_to_history("assistant", "Hi there!")
|
||||
|
||||
|
||||
@given("an interactive session without a history file")
|
||||
def step_impl(context):
|
||||
context.session = InteractiveSession(
|
||||
routers={"main": MagicMock(spec=Router)},
|
||||
initial_route_name="main",
|
||||
history_file=None, # Explicitly None
|
||||
)
|
||||
|
||||
|
||||
# --- Whens ---
|
||||
@when("I try to load the history")
|
||||
def step_impl(context):
|
||||
context.error = None
|
||||
try:
|
||||
context.session.load_history()
|
||||
except InteractiveSessionError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I try to save the history")
|
||||
def step_impl(context):
|
||||
context.error = None
|
||||
try:
|
||||
context.session.save_history()
|
||||
except InteractiveSessionError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I try to process a message through the session")
|
||||
def step_impl(context):
|
||||
context.error = None
|
||||
try:
|
||||
asyncio.run(context.session.process_message("test"))
|
||||
except InteractiveSessionError as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
@when("I run the session's main loop")
|
||||
def step_impl(context):
|
||||
# Mock input() and print() to control the loop and capture output
|
||||
with patch("builtins.input", side_effect=context.user_inputs), patch(
|
||||
"builtins.print"
|
||||
) as mock_print:
|
||||
context.mock_print = mock_print
|
||||
# We don't need to catch errors here, the run method should handle them
|
||||
asyncio.run(context.session.run())
|
||||
|
||||
|
||||
@when("I try to run the session's main loop")
|
||||
def step_impl(context):
|
||||
context.error = None
|
||||
try:
|
||||
asyncio.run(context.session.run())
|
||||
except InteractiveSessionError as e:
|
||||
context.error = e
|
||||
finally:
|
||||
if hasattr(context, "load_history_patch"):
|
||||
context.load_history_patch.stop()
|
||||
|
||||
|
||||
@when("I display the history")
|
||||
def step_impl(context):
|
||||
with patch("sys.stdout", new_callable=io.StringIO) as mock_stdout:
|
||||
context.session.display_history()
|
||||
context.output = mock_stdout.getvalue()
|
||||
|
||||
|
||||
@when("I call load_history and save_history")
|
||||
def step_impl(context):
|
||||
context.error = None
|
||||
try:
|
||||
# Patch open to ensure it's not called
|
||||
with patch("builtins.open") as mock_open:
|
||||
context.session.load_history()
|
||||
context.session.save_history()
|
||||
mock_open.assert_not_called()
|
||||
except Exception as e:
|
||||
context.error = e
|
||||
|
||||
|
||||
# --- Thens ---
|
||||
@then("an InteractiveSessionError should be raised with a message about {action}")
|
||||
def step_impl(context, action):
|
||||
assert context.error is not None, "Expected an error but none was raised"
|
||||
assert isinstance(context.error, InteractiveSessionError)
|
||||
action_map = {
|
||||
"loading history": "Failed to load history file",
|
||||
"saving history": "Failed to save history file",
|
||||
"processing a message": "Failed to process message",
|
||||
"running the session": "Failed to run interactive session",
|
||||
}
|
||||
assert action_map[action] in str(context.error)
|
||||
|
||||
|
||||
@then('the router should have processed the message "{message}"')
|
||||
def step_impl(context, message):
|
||||
context.mock_router.process_message.assert_called_once_with(
|
||||
message, context.session.context
|
||||
)
|
||||
|
||||
|
||||
@then("the session should have stopped running")
|
||||
def step_impl(context):
|
||||
assert context.session.running is False
|
||||
|
||||
|
||||
@then('the output should contain "{text}"')
|
||||
def step_impl(context, text):
|
||||
if hasattr(context, "output"):
|
||||
# From display_history
|
||||
assert text in context.output
|
||||
else:
|
||||
# From run loop
|
||||
mock_print_calls = [call.args[0] for call in context.mock_print.call_args_list]
|
||||
assert any(text in call for call in mock_print_calls)
|
||||
|
||||
|
||||
@then("no errors should be raised")
|
||||
def step_impl(context):
|
||||
assert context.error is None, f"An unexpected error was raised: {context.error}"
|
||||
|
||||
|
||||
@then('the output must contain "You: {user_message}"')
|
||||
def step_impl(context, user_message):
|
||||
assert f"You: {user_message}" in context.output
|
||||
|
||||
|
||||
@then('the output must contain "Agent: {agent_message}"')
|
||||
def step_impl(context, agent_message):
|
||||
assert f"Agent: {agent_message}" in context.output
|
||||
Reference in New Issue
Block a user