mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-29 10:52:26 +00:00
620 lines
22 KiB
Python
620 lines
22 KiB
Python
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
|
|
import sys
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.agent.agent_model import agent_model
|
|
from app.model.chat import AgentModelConfig, Chat
|
|
from app.service.task import Agents
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
class TestAgentFactoryFunctions:
|
|
"""Test cases for agent factory functions."""
|
|
|
|
def test_agent_model_creation(self, sample_chat_data):
|
|
"""Test agent_model creates agent properly."""
|
|
options = Chat(**sample_chat_data)
|
|
agent_name = "TestAgent"
|
|
system_prompt = "You are a helpful assistant"
|
|
|
|
# Setup task lock in the registry before calling agent_model
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
task_locks[options.task_id] = mock_task_lock
|
|
mock_task_lock.put_queue = AsyncMock()
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent") as mock_listen_agent,
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch("asyncio.create_task"),
|
|
):
|
|
mock_agent = MagicMock()
|
|
mock_listen_agent.return_value = mock_agent
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
result = agent_model(agent_name, system_prompt, options, [])
|
|
|
|
assert result is mock_agent
|
|
mock_listen_agent.assert_called_once()
|
|
|
|
def test_codex_subscription_model_uses_responses_api(
|
|
self, monkeypatch, sample_chat_data
|
|
):
|
|
"""Codex subscription must call the Responses API, not chat completions."""
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"api_key": "",
|
|
"api_url": "",
|
|
"model_platform": "openai",
|
|
"model_type": "gpt-5.5",
|
|
"auth_source": "codex_subscription",
|
|
"model_config_dict": {"stream": False, "store": True},
|
|
}
|
|
)
|
|
monkeypatch.setenv("CODEX_RESOLVER_URL", "http://127.0.0.1:12345")
|
|
monkeypatch.setenv("CODEX_RESOLVER_SECRET", "resolver-secret")
|
|
|
|
from app.model.subscription_runtime import codex
|
|
from app.service.task import task_locks
|
|
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def json(self):
|
|
return {
|
|
"access_token": "fresh-local-token",
|
|
"token_type": "Bearer",
|
|
"status": "connected",
|
|
}
|
|
|
|
monkeypatch.setattr(
|
|
codex.httpx, "post", lambda *args, **kwargs: Response()
|
|
)
|
|
|
|
mock_task_lock = MagicMock()
|
|
task_locks[options.task_id] = mock_task_lock
|
|
mock_task_lock.put_queue = AsyncMock()
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent") as mock_listen_agent,
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch("asyncio.create_task"),
|
|
):
|
|
mock_listen_agent.return_value = MagicMock()
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model("TestAgent", "You are helpful", options, [])
|
|
|
|
_, kwargs = mock_model_factory.create.call_args
|
|
assert kwargs["model_platform"] == "openai"
|
|
assert kwargs["model_type"] == "gpt-5.5"
|
|
assert kwargs["url"] == "https://chatgpt.com/backend-api/codex"
|
|
assert kwargs["api_mode"] == "responses"
|
|
assert kwargs["model_config_dict"]["stream"] is True
|
|
assert kwargs["model_config_dict"]["store"] is False
|
|
assert kwargs["default_headers"]["originator"] == "codex_cli_rs"
|
|
|
|
def test_non_codex_model_does_not_inherit_subscription_runtime_params(
|
|
self, sample_chat_data
|
|
):
|
|
"""Switching away from Codex must not keep Codex-only runtime params."""
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"api_key": "legacy-key",
|
|
"api_url": "https://api.openai.com/v1",
|
|
"model_platform": "openai",
|
|
"model_type": "gpt-4o",
|
|
"auth_source": None,
|
|
"extra_params": {},
|
|
}
|
|
)
|
|
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
task_locks[options.task_id] = mock_task_lock
|
|
mock_task_lock.put_queue = AsyncMock()
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch("asyncio.create_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model("TestAgent", "You are helpful", options, [])
|
|
|
|
_, kwargs = mock_model_factory.create.call_args
|
|
model_config = kwargs["model_config_dict"] or {}
|
|
assert "api_mode" not in kwargs
|
|
assert "stream" not in model_config
|
|
assert "store" not in model_config
|
|
assert kwargs["url"] == "https://api.openai.com/v1"
|
|
|
|
def test_explicit_model_config_wins_over_legacy_extra_params(
|
|
self, sample_chat_data
|
|
):
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"extra_params": {
|
|
"temperature": 0.8,
|
|
"top_p": 0.6,
|
|
"max_retries": 7,
|
|
"model_config_dict": {
|
|
"temperature": 0.5,
|
|
"presence_penalty": 0.1,
|
|
},
|
|
},
|
|
"model_config_dict": {
|
|
"temperature": 0.2,
|
|
"max_tokens": 2048,
|
|
},
|
|
}
|
|
)
|
|
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
task_locks[options.project_id] = mock_task_lock
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model("TestAgent", "You are helpful", options, [])
|
|
|
|
kwargs = mock_model_factory.create.call_args.kwargs
|
|
assert kwargs["max_retries"] == 7
|
|
assert kwargs["model_config_dict"]["temperature"] == 0.2
|
|
assert kwargs["model_config_dict"]["top_p"] == 0.6
|
|
assert kwargs["model_config_dict"]["presence_penalty"] == 0.1
|
|
assert kwargs["model_config_dict"]["max_tokens"] == 2048
|
|
assert "max_retries" not in kwargs["model_config_dict"]
|
|
assert "model_config_dict" not in kwargs["model_config_dict"]
|
|
|
|
def test_runtime_owned_agent_model_values_override_user_config(
|
|
self, sample_chat_data
|
|
):
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"model_config_dict": {
|
|
"stream": False,
|
|
"parallel_tool_calls": True,
|
|
},
|
|
}
|
|
)
|
|
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
task_locks[options.project_id] = mock_task_lock
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model(Agents.task_agent, "Task agent", options, [])
|
|
agent_model(Agents.browser_agent, "Browser agent", options, [])
|
|
|
|
task_config = mock_model_factory.create.call_args_list[0].kwargs[
|
|
"model_config_dict"
|
|
]
|
|
browser_config = mock_model_factory.create.call_args_list[1].kwargs[
|
|
"model_config_dict"
|
|
]
|
|
assert task_config["stream"] is True
|
|
assert browser_config["parallel_tool_calls"] is False
|
|
|
|
def test_environment_effort_overrides_untrusted_model_config(
|
|
self, sample_chat_data
|
|
):
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"model_config_dict": {"reasoning_effort": "low"},
|
|
}
|
|
)
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
mock_task_lock.provider_effort_parameter_name = "reasoning_effort"
|
|
mock_task_lock.provider_effort_parameter_value = "xhigh"
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
agent_model("TestAgent", "You are helpful", options, [])
|
|
|
|
assert (
|
|
mock_model_factory.create.call_args.kwargs["model_config_dict"][
|
|
"reasoning_effort"
|
|
]
|
|
== "xhigh"
|
|
)
|
|
|
|
@pytest.mark.parametrize(
|
|
"model_type",
|
|
[
|
|
"gpt-5.6-sol",
|
|
"gpt-5.6-terra",
|
|
"gpt-5.6-luna",
|
|
"gpt-5.7-future",
|
|
],
|
|
)
|
|
def test_cloud_azure_uses_server_declared_responses_transport(
|
|
self, sample_chat_data, model_type
|
|
):
|
|
"""Cloud transport remains server-controlled for future models."""
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"model_platform": "azure",
|
|
"model_type": model_type,
|
|
"api_url": "https://proxy.eigent.ai",
|
|
"extra_params": {
|
|
"api_mode": "responses",
|
|
"stream_options": {"include_usage": True},
|
|
},
|
|
}
|
|
)
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
mock_task_lock.provider_effort_parameter_name = "reasoning_effort"
|
|
mock_task_lock.provider_effort_parameter_value = "high"
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
agent_model(
|
|
"TestAgent",
|
|
"You are helpful",
|
|
options,
|
|
[MagicMock()],
|
|
)
|
|
|
|
kwargs = mock_model_factory.create.call_args.kwargs
|
|
assert kwargs["model_platform"] == "openai-compatible-model"
|
|
assert kwargs["api_mode"] == "responses"
|
|
assert "api_version" not in kwargs
|
|
assert "azure_deployment_name" not in kwargs
|
|
assert kwargs["model_config_dict"]["reasoning"] == {"effort": "high"}
|
|
assert "reasoning_effort" not in kwargs["model_config_dict"]
|
|
assert "stream_options" not in kwargs["model_config_dict"]
|
|
|
|
def test_direct_azure_gpt_5_6_reasoning_with_tools_uses_responses_api(
|
|
self, sample_chat_data
|
|
):
|
|
"""A user-managed Azure endpoint keeps its native Responses support."""
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"model_platform": "azure",
|
|
"model_type": "gpt-5.6-sol",
|
|
"api_url": "https://customer-resource.openai.azure.com",
|
|
"extra_params": {"api_mode": "chat_completions"},
|
|
}
|
|
)
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
mock_task_lock.provider_effort_parameter_name = "reasoning_effort"
|
|
mock_task_lock.provider_effort_parameter_value = "high"
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
agent_model(
|
|
"TestAgent",
|
|
"You are helpful",
|
|
options,
|
|
[MagicMock()],
|
|
)
|
|
|
|
kwargs = mock_model_factory.create.call_args.kwargs
|
|
assert kwargs["model_platform"] == "azure"
|
|
assert kwargs["api_mode"] == "responses"
|
|
assert kwargs["model_config_dict"]["reasoning"] == {"effort": "high"}
|
|
assert "reasoning_effort" not in kwargs["model_config_dict"]
|
|
assert "stream_options" not in kwargs["model_config_dict"]
|
|
|
|
@pytest.mark.parametrize(
|
|
("model_type", "effort", "has_tools"),
|
|
[
|
|
("gpt-5.5", "high", True),
|
|
("gpt-5.6-sol", "provider_default", True),
|
|
("gpt-5.6-sol", "high", False),
|
|
],
|
|
)
|
|
def test_azure_responses_transport_is_not_applied_broadly(
|
|
self,
|
|
sample_chat_data,
|
|
model_type,
|
|
effort,
|
|
has_tools,
|
|
):
|
|
"""Unrelated Azure request shapes retain their configured transport."""
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"model_platform": "azure",
|
|
"model_type": model_type,
|
|
"api_url": "https://proxy.eigent.ai",
|
|
}
|
|
)
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
mock_task_lock.provider_effort_parameter_name = "reasoning_effort"
|
|
mock_task_lock.provider_effort_parameter_value = effort
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
agent_model(
|
|
"TestAgent",
|
|
"You are helpful",
|
|
options,
|
|
[MagicMock()] if has_tools else [],
|
|
)
|
|
|
|
kwargs = mock_model_factory.create.call_args.kwargs
|
|
assert "api_mode" not in kwargs
|
|
|
|
def test_per_agent_explicit_model_config_overrides_task_config(
|
|
self, sample_chat_data
|
|
):
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"model_config_dict": {"temperature": 0.7, "top_p": 0.8},
|
|
}
|
|
)
|
|
custom_config = AgentModelConfig(
|
|
model_config_dict={"temperature": 0.1}
|
|
)
|
|
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
task_locks[options.project_id] = mock_task_lock
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model(
|
|
"CustomAgent",
|
|
"Custom agent",
|
|
options,
|
|
[],
|
|
custom_model_config=custom_config,
|
|
)
|
|
|
|
model_config = mock_model_factory.create.call_args.kwargs[
|
|
"model_config_dict"
|
|
]
|
|
assert model_config["temperature"] == 0.1
|
|
assert "top_p" not in model_config
|
|
|
|
def test_per_agent_empty_credentials_do_not_inherit_task_credentials(
|
|
self, sample_chat_data
|
|
):
|
|
options = Chat(**sample_chat_data)
|
|
custom_config = AgentModelConfig(
|
|
model_platform="aws-bedrock-converse",
|
|
model_type="anthropic.claude-3-5-sonnet-20241022-v2:0",
|
|
api_key="",
|
|
api_url="",
|
|
extra_params={
|
|
"region_name": "us-east-1",
|
|
"aws_access_key_id": "worker-access-key",
|
|
"aws_secret_access_key": "worker-secret-key",
|
|
},
|
|
)
|
|
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
task_locks[options.project_id] = mock_task_lock
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model(
|
|
"BedrockAgent",
|
|
"Bedrock agent",
|
|
options,
|
|
[],
|
|
custom_model_config=custom_config,
|
|
)
|
|
|
|
kwargs = mock_model_factory.create.call_args.kwargs
|
|
assert kwargs["api_key"] == ""
|
|
assert kwargs["url"] == ""
|
|
assert kwargs["aws_access_key_id"] == "worker-access-key"
|
|
assert kwargs["aws_secret_access_key"] == "worker-secret-key"
|
|
|
|
def test_direct_per_agent_provider_is_not_treated_as_task_cloud_model(
|
|
self, sample_chat_data
|
|
):
|
|
options = Chat(
|
|
**{
|
|
**sample_chat_data,
|
|
"api_url": "https://eigent-proxy.example.com",
|
|
}
|
|
)
|
|
custom_config = AgentModelConfig(
|
|
model_platform="aws-bedrock-converse",
|
|
model_type="anthropic.claude-3-5-sonnet-20241022-v2:0",
|
|
api_key="",
|
|
api_url="https://bedrock-runtime.us-east-1.amazonaws.com",
|
|
extra_params={
|
|
"region_name": "eu-west-1",
|
|
"aws_access_key_id": "worker-access-key",
|
|
"aws_secret_access_key": "worker-secret-key",
|
|
},
|
|
)
|
|
|
|
from app.service.task import task_locks
|
|
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
task_locks[options.project_id] = mock_task_lock
|
|
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ListenChatAgent"),
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
patch.object(_m, "_schedule_async_task"),
|
|
):
|
|
mock_model_factory.create.return_value = MagicMock()
|
|
|
|
agent_model(
|
|
"BedrockAgent",
|
|
"Bedrock agent",
|
|
options,
|
|
[],
|
|
custom_model_config=custom_config,
|
|
)
|
|
|
|
kwargs = mock_model_factory.create.call_args.kwargs
|
|
assert kwargs["url"] == (
|
|
"https://bedrock-runtime.us-east-1.amazonaws.com"
|
|
)
|
|
assert kwargs["region_name"] == "eu-west-1"
|
|
assert "user" not in (kwargs["model_config_dict"] or {})
|
|
|
|
def test_agent_model_with_missing_options(self):
|
|
"""Test agent_model with missing required options."""
|
|
agent_name = "ErrorAgent"
|
|
system_prompt = "Test prompt"
|
|
|
|
# Missing required Chat options
|
|
with pytest.raises((AttributeError, KeyError)):
|
|
agent_model(agent_name, system_prompt, None, [])
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestAgentIntegration:
|
|
"""Integration tests for agent utilities."""
|
|
|
|
def setup_method(self):
|
|
"""Clean up before each test."""
|
|
from app.service.task import task_locks
|
|
|
|
task_locks.clear()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_agent_workflow(self, sample_chat_data):
|
|
"""Test complete agent creation and usage workflow."""
|
|
from app.service.task import task_locks
|
|
|
|
options = Chat(**sample_chat_data)
|
|
api_task_id = options.task_id
|
|
|
|
# Create task lock
|
|
mock_task_lock = MagicMock()
|
|
mock_task_lock.put_queue = MagicMock(return_value=None)
|
|
task_locks[api_task_id] = mock_task_lock
|
|
|
|
# Create agent
|
|
_m = sys.modules["app.agent.agent_model"]
|
|
with (
|
|
patch.object(_m, "ModelFactory") as mock_model_factory,
|
|
patch.object(_m, "_schedule_async_task"),
|
|
patch.object(_m, "ListenChatAgent") as mock_listen_agent,
|
|
patch.object(_m, "get_task_lock", return_value=mock_task_lock),
|
|
):
|
|
mock_model = MagicMock()
|
|
mock_model_factory.return_value = mock_model
|
|
|
|
mock_agent_instance = MagicMock()
|
|
mock_agent_instance.api_task_id = api_task_id
|
|
mock_listen_agent.return_value = mock_agent_instance
|
|
|
|
agent = agent_model(
|
|
"IntegrationAgent", "Test system prompt", options, []
|
|
)
|
|
|
|
assert agent is mock_agent_instance
|
|
assert agent.api_task_id == api_task_id
|
|
|
|
# Test step operation
|
|
mock_response = MagicMock()
|
|
mock_response.msg = MagicMock()
|
|
mock_response.msg.content = "Test response"
|
|
mock_response.info = {"usage": {"total_tokens": 50}}
|
|
|
|
agent.step = MagicMock(return_value=mock_response)
|
|
result = agent.step("Test message")
|
|
assert result is mock_response
|