forked from HAL9000/cleveragents-core
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
import asyncio
|
|
import os
|
|
import tempfile
|
|
import textwrap
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock
|
|
from unittest.mock import MagicMock
|
|
from unittest.mock import patch
|
|
|
|
import yaml
|
|
from behave import given
|
|
from behave import then
|
|
from behave import when
|
|
|
|
from cleveragents.core.application import CleverAgentsApp
|
|
from tests.mocks.llm_providers import MockOpenAIResponse
|
|
|
|
|
|
@given("a configuration with an LLM agent having a specific system message")
|
|
def step_impl(context):
|
|
"""
|
|
Creates a temporary YAML configuration file with a specific system message.
|
|
"""
|
|
context.expected_role = "You are a test classifier. Your only job is to work."
|
|
|
|
# Define the configuration as a Python dictionary to ensure
|
|
# correct YAML structure and avoid indentation issues.
|
|
config_data = {
|
|
"version": "1.0",
|
|
"logging": {"level": "INFO"},
|
|
"cleveragents": {"default_router": "main_router"},
|
|
"agents": {
|
|
"classifier": {
|
|
"type": "llm",
|
|
"config": {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"api_key": "some_key",
|
|
"role": context.expected_role,
|
|
"temperature": 0.0,
|
|
},
|
|
}
|
|
},
|
|
"routes": {"main_router": [{"source": "input", "destination": "classifier"}]},
|
|
}
|
|
|
|
# Create a temporary file for the configuration
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", delete=False, suffix=".yaml", encoding="utf-8"
|
|
) as f:
|
|
yaml.dump(config_data, f)
|
|
context.config_file_path = f.name
|
|
|
|
|
|
@then("the LLM provider should have been called with the specific system message")
|
|
def step_impl(context):
|
|
"""
|
|
Asserts that the mocked aiohttp `post` call was made with the correct system message.
|
|
"""
|
|
try:
|
|
context.mock_post.assert_called_once()
|
|
|
|
# Extract the keyword arguments the mock was called with.
|
|
_, kwargs = context.mock_post.call_args
|
|
json_payload = kwargs.get("json", {})
|
|
messages = json_payload.get("messages", [])
|
|
|
|
assert len(messages) > 0, "No messages were sent to the LLM."
|
|
|
|
role = next((m for m in messages if m.get("role") == "system"), None)
|
|
|
|
assert role is not None, "No system message was sent in the call to the LLM."
|
|
|
|
actual_role = role.get("content")
|
|
|
|
assert actual_role == context.expected_role, (
|
|
f"Incorrect system message. Expected: '{context.expected_role}', "
|
|
f"Got: '{actual_role}'"
|
|
)
|
|
finally:
|
|
# Clean up the temporary file.
|
|
if hasattr(context, "config_file_path"):
|
|
os.remove(context.config_file_path)
|