mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-04-26 01:06:27 +02:00
fix mock ut and fix err ut
This commit is contained in:
张建生
parent
991e49e2d1
commit
3b57fdd466
5 changed files with 262 additions and 544 deletions
|
|
@ -1,43 +1,14 @@
|
|||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
from metagpt.actions.di.execute_nb_code import ExecuteNbCode
|
||||
from metagpt.actions.di.write_analysis_code import WriteAnalysisCode
|
||||
from metagpt.logs import logger
|
||||
from metagpt.roles.di.data_analyst import DataAnalyst
|
||||
from metagpt.roles.di.role_zero import RoleZero
|
||||
from metagpt.strategy.task_type import TaskType
|
||||
from metagpt.tools.tool_recommend import BM25ToolRecommender
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def data_analyst():
|
||||
analyst = DataAnalyst()
|
||||
analyst.planner = MagicMock()
|
||||
analyst.planner.plan = MagicMock()
|
||||
analyst.rc = MagicMock()
|
||||
analyst.rc.working_memory = MagicMock()
|
||||
analyst.rc.memory = MagicMock()
|
||||
return analyst
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_execute_code():
|
||||
with patch('metagpt.actions.di.execute_nb_code.ExecuteNbCode') as mock:
|
||||
instance = mock.return_value
|
||||
instance.init_code = AsyncMock()
|
||||
instance.run = AsyncMock()
|
||||
yield instance
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_write_code():
|
||||
with patch('metagpt.actions.di.write_analysis_code.WriteAnalysisCode') as mock:
|
||||
instance = mock.return_value
|
||||
instance.run = AsyncMock()
|
||||
yield instance
|
||||
|
||||
|
||||
class TestDataAnalyst:
|
||||
def test_init(self):
|
||||
analyst = DataAnalyst()
|
||||
|
|
@ -48,96 +19,60 @@ class TestDataAnalyst:
|
|||
assert isinstance(analyst.execute_code, ExecuteNbCode)
|
||||
|
||||
def test_set_custom_tool(self):
|
||||
# 测试有自定义工具的情况
|
||||
analyst = DataAnalyst()
|
||||
analyst.custom_tools = ["web scraping", "Terminal"]
|
||||
analyst.custom_tool_recommender = None # 确保初始值为None
|
||||
analyst.set_custom_tool()
|
||||
assert isinstance(analyst.custom_tool_recommender, BM25ToolRecommender)
|
||||
|
||||
# 测试没有自定义工具的情况
|
||||
analyst = DataAnalyst()
|
||||
analyst.custom_tools = []
|
||||
analyst.custom_tool_recommender = BM25ToolRecommender(tools=["some_tool"], force=True) # 设置一个初始值
|
||||
analyst.set_custom_tool()
|
||||
assert isinstance(analyst.custom_tool_recommender, BM25ToolRecommender) # 验证即使没有自定义工具，现有的推荐器也保持不变
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_and_exec_code_no_task(self, data_analyst):
|
||||
data_analyst.planner.current_task = None
|
||||
result = await data_analyst.write_and_exec_code()
|
||||
async def test_write_and_exec_code_no_task(self):
|
||||
analyst = DataAnalyst()
|
||||
result = await analyst.write_and_exec_code()
|
||||
logger.info(result)
|
||||
assert "No current_task found" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_and_exec_code_success(self, data_analyst, mock_execute_code, mock_write_code):
|
||||
# Setup mocks
|
||||
data_analyst.planner.current_task = MagicMock()
|
||||
data_analyst.planner.get_plan_status.return_value = "Plan status"
|
||||
data_analyst.custom_tool_recommender = MagicMock()
|
||||
data_analyst.custom_tool_recommender.get_recommended_tool_info = AsyncMock(return_value="Tool info")
|
||||
|
||||
mock_write_code.run.return_value = "test code"
|
||||
mock_execute_code.run.return_value = ("Success result", True)
|
||||
|
||||
result = await data_analyst.write_and_exec_code("test instruction")
|
||||
async def test_write_and_exec_code_success(self):
|
||||
analyst = DataAnalyst()
|
||||
await analyst.execute_code.init_code()
|
||||
analyst.planner.plan.goal = "construct a two-dimensional array"
|
||||
analyst.planner.plan.append_task(
|
||||
task_id="1",
|
||||
dependent_task_ids=[],
|
||||
instruction="construct a two-dimensional array",
|
||||
assignee="David",
|
||||
task_type="DATA_ANALYSIS",
|
||||
)
|
||||
|
||||
result = await analyst.write_and_exec_code("construct a two-dimensional array")
|
||||
logger.info(result)
|
||||
assert "Success" in result
|
||||
assert mock_execute_code.init_code.called
|
||||
assert mock_write_code.run.called
|
||||
data_analyst.rc.working_memory.add.assert_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_write_and_exec_code_failure(self, data_analyst, mock_execute_code, mock_write_code):
|
||||
# Setup mocks
|
||||
data_analyst.planner.current_task = MagicMock()
|
||||
data_analyst.planner.get_plan_status.return_value = "Plan status"
|
||||
data_analyst.custom_tool_recommender = None
|
||||
async def test_write_and_exec_code_failure(self):
|
||||
analyst = DataAnalyst()
|
||||
await analyst.execute_code.init_code()
|
||||
analyst.planner.plan.goal = "Execute a code that fails"
|
||||
|
||||
mock_write_code.run.return_value = "test code"
|
||||
mock_execute_code.run.return_value = ("Failed result", False)
|
||||
analyst.planner.plan.append_task(
|
||||
task_id="1", dependent_task_ids=[], instruction="Execute a code that fails", assignee="David"
|
||||
)
|
||||
|
||||
result = await data_analyst.write_and_exec_code()
|
||||
analyst.execute_code.run = AsyncMock(return_value=("Error: Division by zero", False))
|
||||
|
||||
result = await analyst.write_and_exec_code("divide by zero")
|
||||
|
||||
logger.info(result)
|
||||
assert "Failed" in result
|
||||
assert mock_execute_code.run.call_count == 3 # Should retry 3 times
|
||||
assert "Error: Division by zero" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_data_no_tasks(self, data_analyst):
|
||||
data_analyst.planner.plan.get_finished_tasks.return_value = []
|
||||
await data_analyst._check_data()
|
||||
assert not data_analyst.rc.working_memory.add.called
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_data_with_data_task(self, data_analyst, mock_execute_code):
|
||||
# Setup task with DATA_PREPROCESS type
|
||||
task = MagicMock()
|
||||
task.task_type = TaskType.DATA_PREPROCESS.type_name
|
||||
data_analyst.planner.plan.get_finished_tasks.return_value = [task]
|
||||
data_analyst.planner.plan.current_task = task
|
||||
|
||||
with patch('metagpt.actions.di.write_analysis_code.CheckData') as mock_check:
|
||||
mock_check.return_value.run = AsyncMock(return_value="check code")
|
||||
mock_execute_code.run.return_value = ("check result", True)
|
||||
|
||||
await data_analyst._check_data()
|
||||
|
||||
assert mock_check.return_value.run.called
|
||||
assert mock_execute_code.run.called
|
||||
data_analyst.rc.working_memory.add.assert_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_special_command(self, data_analyst):
|
||||
data_analyst.planner.plan.is_plan_finished.return_value = False
|
||||
async def test_run_special_command(self):
|
||||
analyst = DataAnalyst()
|
||||
|
||||
analyst.planner.plan.goal = "test goal"
|
||||
analyst.planner.plan.append_task(task_id="1", dependent_task_ids=[], instruction="test task", assignee="David")
|
||||
assert not analyst.planner.plan.is_plan_finished()
|
||||
cmd = {"command_name": "end"}
|
||||
with patch.object(RoleZero, '_run_special_command', return_value="base result"):
|
||||
result = await data_analyst._run_special_command(cmd)
|
||||
|
||||
result = await analyst._run_special_command(cmd)
|
||||
assert "All tasks are finished" in result
|
||||
assert data_analyst.planner.plan.finish_all_tasks.called
|
||||
|
||||
# Test non-end command
|
||||
cmd = {"command_name": "other"}
|
||||
with patch.object(RoleZero, '_run_special_command', return_value="base result"):
|
||||
result = await data_analyst._run_special_command(cmd)
|
||||
assert result == "base result"
|
||||
assert analyst.planner.plan.is_plan_finished()
|
||||
|
|
|
|||
|
|
@ -1,265 +1,41 @@
|
|||
from typing import List
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from metagpt.actions import UserRequirement
|
||||
from metagpt.logs import logger
|
||||
from metagpt.roles.di.role_zero import RoleZero
|
||||
from metagpt.schema import Message, UserMessage, AIMessage
|
||||
from metagpt.tools.libs.browser import Browser
|
||||
|
||||
|
||||
class MockConfig:
|
||||
"""Mock configuration for RoleZero testing"""
|
||||
|
||||
class RoleZeroConfig:
|
||||
enable_longterm_memory = True
|
||||
longterm_memory_persist_path = "/tmp/test_memory"
|
||||
memory_k = 5
|
||||
similarity_top_k = 3
|
||||
use_llm_ranker = False
|
||||
|
||||
role_zero = RoleZeroConfig()
|
||||
|
||||
|
||||
class MockLLM:
|
||||
"""Mock LLM for testing"""
|
||||
|
||||
def __init__(self, responses: List[str] = None):
|
||||
self.responses = responses or ["Mock LLM Response"]
|
||||
self.response_index = 0
|
||||
|
||||
async def aask(self, *args, **kwargs):
|
||||
response = self.responses[self.response_index]
|
||||
self.response_index = (self.response_index + 1) % len(self.responses)
|
||||
return response
|
||||
|
||||
def support_image_input(self):
|
||||
return True
|
||||
|
||||
def format_msg(self, msgs):
|
||||
return msgs
|
||||
|
||||
|
||||
class MockToolRecommender:
|
||||
"""Mock tool recommender for testing"""
|
||||
|
||||
async def recommend_tools(self):
|
||||
return []
|
||||
|
||||
|
||||
class MockMemory:
|
||||
"""Mock memory for testing"""
|
||||
|
||||
def add(self, msg):
|
||||
pass
|
||||
|
||||
def get(self, k=None):
|
||||
return []
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_role_zero():
|
||||
"""Fixture providing a configured RoleZero instance for testing"""
|
||||
role = RoleZero()
|
||||
role.llm = MockLLM()
|
||||
role.config = MockConfig()
|
||||
role.tool_recommender = MockToolRecommender()
|
||||
role.rc.working_memory = MockMemory()
|
||||
role.rc.memory = MockMemory()
|
||||
return role
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_message():
|
||||
"""Fixture providing a test message"""
|
||||
return Message(content="Test message", role="user")
|
||||
from metagpt.schema import Message
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_model_validators(mock_role_zero):
|
||||
async def test_model_validators():
|
||||
"""Test all model validators"""
|
||||
role = RoleZero()
|
||||
# Test set_plan_and_tool
|
||||
assert mock_role_zero.react_mode == "react"
|
||||
mock_role_zero = await mock_role_zero.set_plan_and_tool()
|
||||
assert mock_role_zero.planner is not None
|
||||
assert role.react_mode == "react"
|
||||
assert role.planner is not None
|
||||
|
||||
# Test set_tool_execution
|
||||
mock_role_zero = await mock_role_zero.set_tool_execution()
|
||||
assert "Plan.append_task" in mock_role_zero.tool_execution_map
|
||||
assert "RoleZero.ask_human" in mock_role_zero.tool_execution_map
|
||||
assert "Plan.append_task" in role.tool_execution_map
|
||||
assert "RoleZero.ask_human" in role.tool_execution_map
|
||||
|
||||
# Test set_longterm_memory
|
||||
mock_role_zero = await mock_role_zero.set_longterm_memory()
|
||||
assert mock_role_zero.rc.memory is not None
|
||||
assert role.rc.memory is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_think_react_cycle(mock_role_zero):
|
||||
async def test_think_react_cycle():
|
||||
"""Test the think-react cycle"""
|
||||
# Setup test conditions
|
||||
mock_role_zero.rc.todo = True
|
||||
mock_role_zero.planner.plan.goal = "Test goal"
|
||||
mock_role_zero.respond_language = "English"
|
||||
role = RoleZero(tools=["Plan"])
|
||||
role.rc.todo = True
|
||||
role.planner.plan.goal = "Test goal"
|
||||
role.respond_language = "English"
|
||||
|
||||
# Test _think
|
||||
with patch('metagpt.roles.di.role_zero.ThoughtReporter'):
|
||||
result = await mock_role_zero._think()
|
||||
assert result is True
|
||||
result = await role._think()
|
||||
assert result is True
|
||||
|
||||
# Test _react
|
||||
mock_role_zero.rc.news = [Message(content="Test", cause_by=UserRequirement())]
|
||||
with patch.object(mock_role_zero, '_quick_think', return_value=(None, "TASK")):
|
||||
result = await mock_role_zero._react()
|
||||
assert isinstance(result, Message)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_command_parsing(mock_role_zero):
|
||||
"""Test command parsing functionality"""
|
||||
# Test valid JSON parsing
|
||||
valid_commands = '''[
|
||||
{"command_name": "Editor.read", "args": {"filename": "test.txt"}},
|
||||
{"command_name": "Plan.finish_current_task", "args": {}}
|
||||
]'''
|
||||
commands, ok, rsp = await mock_role_zero._parse_commands(valid_commands)
|
||||
assert ok is True
|
||||
assert len(commands) == 2
|
||||
|
||||
# Test invalid JSON
|
||||
invalid_commands = "Invalid JSON"
|
||||
with patch.object(mock_role_zero.llm, 'aask') as mock_aask:
|
||||
mock_aask.return_value = valid_commands
|
||||
commands, ok, rsp = await mock_role_zero._parse_commands(invalid_commands)
|
||||
assert ok is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_command_execution(mock_role_zero):
|
||||
"""Test command execution"""
|
||||
# Test special commands
|
||||
special_commands = [
|
||||
{"command_name": "Plan.finish_current_task", "args": {}},
|
||||
{"command_name": "end", "args": {}}
|
||||
]
|
||||
|
||||
with patch.object(mock_role_zero, '_run_special_command') as mock_special:
|
||||
mock_special.return_value = "Special command executed"
|
||||
result = await mock_role_zero._run_commands(special_commands)
|
||||
assert "Command Plan.finish_current_task executed" in result
|
||||
|
||||
# Test normal commands
|
||||
normal_commands = [
|
||||
{"command_name": "Editor.read", "args": {"filename": "test.txt"}}
|
||||
]
|
||||
with patch.object(mock_role_zero.editor, 'read', return_value="File content"):
|
||||
result = await mock_role_zero._run_commands(normal_commands)
|
||||
assert "Command Editor.read executed" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_message_handling(mock_role_zero):
|
||||
"""Test message parsing and handling"""
|
||||
# Test browser action parsing
|
||||
mock_browser = AsyncMock(spec=Browser)
|
||||
mock_browser.is_empty_page = False
|
||||
mock_browser.view.return_value = "Browser content"
|
||||
mock_role_zero.browser = mock_browser
|
||||
|
||||
browser_memory = [
|
||||
UserMessage(content="Command Browser.goto executed"),
|
||||
UserMessage(content="Other message")
|
||||
]
|
||||
result = await mock_role_zero.parse_browser_actions(browser_memory)
|
||||
assert len(result) == 3
|
||||
|
||||
# Test editor result parsing
|
||||
editor_memory = [
|
||||
UserMessage(content="Command Editor.read executed: content"),
|
||||
UserMessage(content="Normal message")
|
||||
]
|
||||
result = await mock_role_zero.parse_editor_result(editor_memory)
|
||||
assert len(result) == 2
|
||||
|
||||
# Test image parsing
|
||||
image_memory = [
|
||||
UserMessage(content="Message with "),
|
||||
UserMessage(content="Normal message")
|
||||
]
|
||||
result = mock_role_zero.parse_images(image_memory)
|
||||
assert len(result) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_error_cases(mock_role_zero):
|
||||
"""Test error handling in various scenarios"""
|
||||
# Test invalid command execution
|
||||
invalid_commands = [
|
||||
{"command_name": "NonExistentCommand", "args": {}}
|
||||
]
|
||||
result = await mock_role_zero._run_commands(invalid_commands)
|
||||
assert "Command NonExistentCommand not found" in result
|
||||
|
||||
# Test command parsing with malformed JSON
|
||||
malformed_json = '[{"command_name": "test", "args": {}]' # Missing closing brace
|
||||
with patch.object(mock_role_zero.llm, 'aask') as mock_aask:
|
||||
mock_aask.return_value = '[{"command_name": "fixed", "args": {}}]' # Valid JSON response
|
||||
commands, ok, rsp = await mock_role_zero._parse_commands(malformed_json)
|
||||
assert ok is True
|
||||
|
||||
# Test command parsing with improper command structure
|
||||
invalid_format = '[{"not_a_command": true}]' # Valid JSON but wrong format
|
||||
with patch.object(mock_role_zero.llm, 'aask') as mock_aask:
|
||||
mock_aask.return_value = invalid_format
|
||||
commands, ok, rsp = await mock_role_zero._parse_commands(invalid_format)
|
||||
assert ok is False
|
||||
|
||||
# Test think with no todo
|
||||
mock_role_zero.rc.todo = False
|
||||
result = await mock_role_zero._think()
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_special_commands(mock_role_zero):
|
||||
"""Test special command handling"""
|
||||
# Test Plan.finish_current_task
|
||||
finish_command = {"command_name": "Plan.finish_current_task", "args": {}}
|
||||
result = await mock_role_zero._run_special_command(finish_command)
|
||||
assert "Current task is finished" in result
|
||||
|
||||
# Test end command
|
||||
end_command = {"command_name": "end", "args": {}}
|
||||
with patch.object(mock_role_zero.llm, 'aask', return_value="Summary"):
|
||||
result = await mock_role_zero._run_special_command(end_command)
|
||||
assert result
|
||||
|
||||
# Test ask_human command
|
||||
ask_command = {"command_name": "RoleZero.ask_human", "args": {"question": "Test?"}}
|
||||
result = await mock_role_zero._run_special_command(ask_command)
|
||||
assert "Not in MGXEnv" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_quick_think(mock_role_zero):
|
||||
"""Test quick think functionality"""
|
||||
mock_role_zero.rc.news = [Message(content="Test", cause_by=UserRequirement())]
|
||||
|
||||
with patch.object(mock_role_zero.llm, 'aask') as mock_aask:
|
||||
mock_aask.side_effect = ["QUICK", "Quick response"]
|
||||
result, intent = await mock_role_zero._quick_think()
|
||||
assert isinstance(result, AIMessage)
|
||||
assert intent == "QUICK"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_experience_retrieval(mock_role_zero):
|
||||
"""Test experience retrieval functionality"""
|
||||
# Test with empty memory
|
||||
result = mock_role_zero._retrieve_experience()
|
||||
assert isinstance(result, str)
|
||||
|
||||
# Test with mock experience retriever
|
||||
mock_role_zero.experience_retriever.retrieve = MagicMock(return_value="Test experience")
|
||||
result = mock_role_zero._retrieve_experience()
|
||||
assert result == "Test experience"
|
||||
role.rc.news = [Message(content="Test", cause_by=UserRequirement())]
|
||||
result = await role._react()
|
||||
logger.info(result)
|
||||
assert isinstance(result, Message)
|
||||
|
|
|
|||
|
|
@ -1,143 +1,47 @@
|
|||
import json
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from metagpt.roles.di.swe_agent import SWEAgent
|
||||
from metagpt.schema import Message
|
||||
from metagpt.tools.libs.terminal import Bash
|
||||
from metagpt.environment.mgx.mgx_env import MGXEnv
|
||||
from metagpt.roles.di.team_leader import TeamLeader
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_terminal():
|
||||
terminal = AsyncMock(spec=Bash)
|
||||
terminal.run = AsyncMock()
|
||||
return terminal
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_extract_patch():
|
||||
with patch('metagpt.tools.swe_agent_commands.swe_agent_utils.extract_patch') as mock:
|
||||
mock.return_value = 'test_patch'
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def swe_agent(mock_terminal):
|
||||
agent = SWEAgent()
|
||||
agent.terminal = mock_terminal
|
||||
# Mock super()._think and super()._act
|
||||
agent._think = AsyncMock(return_value=True)
|
||||
agent._act = AsyncMock(return_value=Message(content='test'))
|
||||
return agent
|
||||
def env():
|
||||
test_env = MGXEnv()
|
||||
tl = TeamLeader()
|
||||
test_env.add_roles(
|
||||
[
|
||||
tl,
|
||||
SWEAgent()
|
||||
]
|
||||
)
|
||||
return test_env
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initialization():
|
||||
"""Test SWEAgent initialization and attributes"""
|
||||
agent = SWEAgent()
|
||||
assert agent.name == 'Swen'
|
||||
assert agent.profile == 'Issue Solver'
|
||||
assert isinstance(agent.terminal, Bash)
|
||||
assert agent.output_diff == ''
|
||||
assert agent.max_react_loop == 40
|
||||
assert agent.run_eval is False
|
||||
async def test_swe_agent(env):
|
||||
requirement = "Fix bug in the calculator app"
|
||||
swe = env.get_role("Swen")
|
||||
|
||||
message = Message(content=requirement, send_to={swe.name})
|
||||
env.publish_message(message)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_think(swe_agent):
|
||||
"""Test _think method with mocked dependencies"""
|
||||
# Mock _format_instruction
|
||||
swe_agent._format_instruction = AsyncMock()
|
||||
await swe.run()
|
||||
|
||||
result = await swe_agent._think()
|
||||
assert result is True
|
||||
swe_agent._format_instruction.assert_called_once()
|
||||
history = env.history.get()
|
||||
agent_messages = [msg for msg in history if msg.sent_from == swe.name]
|
||||
|
||||
assert swe.name == "Swen"
|
||||
assert swe.profile == "Issue Solver"
|
||||
assert isinstance(swe.terminal, Bash)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_instruction(swe_agent):
|
||||
"""Test _format_instruction with mocked terminal response"""
|
||||
mock_state = {"key": "value"}
|
||||
swe_agent.terminal.run.return_value = json.dumps(mock_state)
|
||||
assert "Bash" in swe.tools
|
||||
assert "git_create_pull" in swe.tool_execution_map
|
||||
|
||||
await swe_agent._format_instruction()
|
||||
swe_agent.terminal.run.assert_called_with('state')
|
||||
assert isinstance(swe_agent.cmd_prompt_current_state, str)
|
||||
def is_valid_instruction_message(msg: Message) -> bool:
|
||||
content = msg.content.lower()
|
||||
return any(word in content for word in ["git", "bash", "check", "fix"])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_format_instruction_error(swe_agent):
|
||||
"""Test _format_instruction with invalid JSON response"""
|
||||
swe_agent.terminal.run.return_value = 'invalid json'
|
||||
|
||||
with pytest.raises(json.JSONDecodeError):
|
||||
await swe_agent._format_instruction()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_act_with_eval(swe_agent):
|
||||
"""Test _act method with run_eval=True"""
|
||||
swe_agent.run_eval = True
|
||||
swe_agent._parse_commands_for_eval = AsyncMock()
|
||||
|
||||
result = await swe_agent._act()
|
||||
assert isinstance(result, Message)
|
||||
swe_agent._parse_commands_for_eval.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_act_without_eval(swe_agent):
|
||||
"""Test _act method with run_eval=False"""
|
||||
swe_agent.run_eval = False
|
||||
swe_agent._parse_commands_for_eval = AsyncMock()
|
||||
|
||||
result = await swe_agent._act()
|
||||
assert isinstance(result, Message)
|
||||
swe_agent._parse_commands_for_eval.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parse_commands_for_eval_with_diff(swe_agent, mock_extract_patch):
|
||||
"""Test _parse_commands_for_eval with git diff output"""
|
||||
swe_agent.rc.todo = False
|
||||
swe_agent.terminal.run.return_value = 'test diff output'
|
||||
|
||||
await swe_agent._parse_commands_for_eval()
|
||||
assert swe_agent.output_diff == 'test_patch'
|
||||
mock_extract_patch.assert_called_with('test diff output')
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parse_commands_for_eval_with_error(swe_agent):
|
||||
"""Test _parse_commands_for_eval error handling"""
|
||||
swe_agent.rc.todo = False
|
||||
swe_agent.terminal.run.side_effect = Exception('test error')
|
||||
|
||||
await swe_agent._parse_commands_for_eval()
|
||||
assert swe_agent.output_diff == ''
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parse_commands_for_eval_with_todo(swe_agent):
|
||||
"""Test _parse_commands_for_eval when todo is True"""
|
||||
swe_agent.rc.todo = True
|
||||
|
||||
await swe_agent._parse_commands_for_eval()
|
||||
swe_agent.terminal.run.assert_not_called()
|
||||
|
||||
|
||||
def test_retrieve_experience(swe_agent):
|
||||
"""Test _retrieve_experience returns MINIMAL_EXAMPLE"""
|
||||
from metagpt.prompts.di.swe_agent import MINIMAL_EXAMPLE
|
||||
|
||||
result = swe_agent._retrieve_experience()
|
||||
assert result == MINIMAL_EXAMPLE
|
||||
|
||||
|
||||
def test_update_tool_execution(swe_agent):
|
||||
"""Test _update_tool_execution adds required tools"""
|
||||
swe_agent._update_tool_execution()
|
||||
|
||||
assert 'Bash.run' in swe_agent.tool_execution_map
|
||||
assert 'git_create_pull' in swe_agent.tool_execution_map
|
||||
assert swe_agent.tool_execution_map['Bash.run'] == swe_agent.terminal.run
|
||||
assert any(is_valid_instruction_message(msg) for msg in agent_messages), "Should have valid instruction messages"
|
||||
|
|
|
|||
|
|
@ -40,78 +40,70 @@ def env():
|
|||
@pytest.mark.asyncio
|
||||
async def test_plan_for_software_requirement(env):
|
||||
requirement = "create a 2048 game"
|
||||
|
||||
tl = env.get_role("Team Leader")
|
||||
tl = env.get_role("Mike")
|
||||
env.publish_message(Message(content=requirement, send_to=tl.name))
|
||||
await tl.run()
|
||||
|
||||
# TL should assign tasks to 5 members first, then send message to the first assignee, 6 commands in total
|
||||
assert len(tl.commands) == 6
|
||||
plan_cmd = tl.commands[:5]
|
||||
route_cmd = tl.commands[5]
|
||||
history = env.history.get()
|
||||
|
||||
task_assignment = [task["args"]["assignee"] for task in plan_cmd]
|
||||
assert task_assignment == [
|
||||
ProductManager().name,
|
||||
Architect().name,
|
||||
ProjectManager().name,
|
||||
Engineer().name,
|
||||
QaEngineer().name,
|
||||
]
|
||||
|
||||
assert route_cmd["command_name"] == "publish_message"
|
||||
assert route_cmd["args"]["send_to"] == ProductManager().name
|
||||
messages_to_team = [msg for msg in history if msg.sent_from == tl.name]
|
||||
pm_messages = [msg for msg in messages_to_team if "Alice" in msg.send_to]
|
||||
assert len(pm_messages) > 0, "Should have message sent to Product Manager"
|
||||
found_task_msg = False
|
||||
for msg in messages_to_team:
|
||||
if "prd" in msg.content.lower() and any(role in msg.content for role in ["Alice", "Bob", "Alex", "David"]):
|
||||
found_task_msg = True
|
||||
break
|
||||
assert found_task_msg, "Should have task assignment message"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plan_for_data_related_requirement(env):
|
||||
requirement = "I want to use yolov5 for target detection, yolov5 all the information from the following link, please help me according to the content of the link (https://github.com/ultralytics/yolov5), set up the environment and download the model parameters, and finally provide a few pictures for inference, the inference results will be saved!"
|
||||
|
||||
tl = env.get_role("Team Leader")
|
||||
tl = env.get_role("Mike")
|
||||
env.publish_message(Message(content=requirement, send_to=tl.name))
|
||||
await tl.run()
|
||||
|
||||
# TL should assign 1 task to Data Analyst and send message to it
|
||||
assert len(tl.commands) == 2
|
||||
plan_cmd = tl.commands[0]
|
||||
route_cmd = tl.commands[-1]
|
||||
history = env.history.get()
|
||||
messages_from_tl = [msg for msg in history if msg.sent_from == tl.name]
|
||||
da_messages = [msg for msg in messages_from_tl if "David" in msg.send_to]
|
||||
assert len(da_messages) > 0
|
||||
|
||||
da = env.get_role("Data Analyst")
|
||||
assert plan_cmd["command_name"] == "append_task"
|
||||
assert plan_cmd["args"]["assignee"] == da.name
|
||||
da_message = da_messages[0]
|
||||
assert "https://github.com/ultralytics/yolov5" in da_message.content
|
||||
|
||||
assert route_cmd["command_name"] == "publish_message"
|
||||
assert "https://github.com" in route_cmd["args"]["content"] # necessary info must be in the message
|
||||
assert route_cmd["args"]["send_to"] == da.name
|
||||
def is_valid_task_message(msg: Message) -> bool:
|
||||
content = msg.content.lower()
|
||||
has_model_info = "yolov5" in content
|
||||
has_task_info = any(word in content for word in ["detection", "inference", "environment", "parameters"])
|
||||
has_link = "github.com" in content
|
||||
return has_model_info and has_task_info and has_link
|
||||
|
||||
assert is_valid_task_message(da_message)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_plan_for_mixed_requirement(env):
|
||||
requirement = "Search the web for the new game 2048X, then replicate it"
|
||||
|
||||
tl = env.get_role("Team Leader")
|
||||
tl = env.get_role("Mike")
|
||||
env.publish_message(Message(content=requirement, send_to=tl.name))
|
||||
await tl.run()
|
||||
|
||||
# TL should assign 6 tasks, first to Data Analyst to search the web, following by the software team sequence
|
||||
# TL should send message to Data Analyst after task assignment
|
||||
assert len(tl.commands) == 7
|
||||
plan_cmd = tl.commands[:6]
|
||||
route_cmd = tl.commands[-1]
|
||||
history = env.history.get()
|
||||
messages_from_tl = [msg for msg in history if msg.sent_from == tl.name]
|
||||
|
||||
task_assignment = [task["args"]["assignee"] for task in plan_cmd]
|
||||
da = env.get_role("Data Analyst")
|
||||
assert task_assignment == [
|
||||
da.name,
|
||||
ProductManager().name,
|
||||
Architect().name,
|
||||
ProjectManager().name,
|
||||
Engineer().name,
|
||||
QaEngineer().name,
|
||||
]
|
||||
da_messages = [msg for msg in messages_from_tl if "David" in msg.send_to]
|
||||
assert len(da_messages) > 0
|
||||
|
||||
assert route_cmd["command_name"] == "publish_message"
|
||||
assert route_cmd["args"]["send_to"] == da.name
|
||||
da_message = da_messages[0]
|
||||
|
||||
def is_valid_search_task(msg: Message) -> bool:
|
||||
content = msg.content.lower()
|
||||
return "2048x" in content and "search" in content
|
||||
|
||||
assert is_valid_search_task(da_message)
|
||||
|
||||
|
||||
PRD_MSG_CONTENT = """{'docs': {'20240424153821.json': {'root_path': 'docs/prd', 'filename': '20240424153821.json', 'content': '{"Language":"en_us","Programming Language":"Python","Original Requirements":"create a 2048 game","Project Name":"game_2048","Product Goals":["Develop an intuitive and addictive 2048 game variant","Ensure the game is accessible and performs well on various devices","Design a visually appealing and modern user interface"],"User Stories":["As a player, I want to be able to undo my last move so I can correct mistakes","As a player, I want to see my high scores to track my progress over time","As a player, I want to be able to play the game without any internet connection"],"Competitive Analysis":["2048 Original: Classic gameplay, minimalistic design, lacks social sharing features","2048 Hex: Unique hexagon board, but not mobile-friendly","2048 Multiplayer: Offers real-time competition, but overwhelming ads","2048 Bricks: Innovative gameplay with bricks, but poor performance on older devices","2048.io: Multiplayer battle royale mode, but complicated UI for new players","2048 Animated: Animated tiles add fun, but the game consumes a lot of battery","2048 3D: 3D version of the game, but has a steep learning curve"],"Competitive Quadrant Chart":"quadrantChart\\n title \\"User Experience and Feature Set of 2048 Games\\"\\n x-axis \\"Basic Features\\" --> \\"Rich Features\\"\\n y-axis \\"Poor Experience\\" --> \\"Great Experience\\"\\n quadrant-1 \\"Need Improvement\\"\\n quadrant-2 \\"Feature-Rich but Complex\\"\\n quadrant-3 \\"Simplicity with Poor UX\\"\\n quadrant-4 \\"Balanced\\"\\n \\"2048 Original\\": [0.2, 0.7]\\n \\"2048 Hex\\": [0.3, 0.4]\\n \\"2048 Multiplayer\\": [0.6, 0.5]\\n \\"2048 Bricks\\": [0.4, 0.3]\\n \\"2048.io\\": [0.7, 0.4]\\n \\"2048 Animated\\": [0.5, 0.6]\\n \\"2048 3D\\": [0.6, 0.3]\\n \\"Our Target Product\\": [0.8, 0.9]","Requirement Analysis":"The game must be engaging and retain players, which requires a balance of simplicity and challenge. Accessibility on various devices is crucial for a wider reach. A modern UI is needed to attract and retain the modern user. The ability to play offline is important for users on the go. High score tracking and the ability to undo moves are features that will enhance user experience.","Requirement Pool":[["P0","Implement core 2048 gameplay mechanics"],["P0","Design responsive UI for multiple devices"],["P1","Develop undo move feature"],["P1","Integrate high score tracking system"],["P2","Enable offline gameplay capability"]],"UI Design draft":"The UI will feature a clean and modern design with a minimalist color scheme. The game board will be center-aligned with smooth tile animations. Score and high score will be displayed at the top. Undo and restart buttons will be easily accessible. The design will be responsive to fit various screen sizes.","Anything UNCLEAR":"The monetization strategy for the game is not specified. Further clarification is needed on whether the game should include advertisements, in-app purchases, or be completely free."}'}}}"""
|
||||
|
|
@ -122,48 +114,60 @@ DESIGN_CONTENT = """{"docs":{"20240424214432.json":{"root_path":"docs/system_des
|
|||
async def test_plan_update_and_routing(env):
|
||||
requirement = "create a 2048 game"
|
||||
|
||||
tl = env.get_role("Team Leader")
|
||||
tl = env.get_role("Mike")
|
||||
env.publish_message(Message(content=requirement))
|
||||
await tl.run()
|
||||
|
||||
# Assuming Product Manager finishes its task
|
||||
env.publish_message(Message(content=PRD_MSG_CONTENT, role="Alice(Product Manager)", sent_from="Alice"))
|
||||
# Verify message routing after PM completes task
|
||||
env.publish_message(Message(content=PRD_MSG_CONTENT, sent_from="Alice", send_to={"<all>"}))
|
||||
await tl.run()
|
||||
|
||||
# TL should mark current task as finished, and forward Product Manager's message to Architect
|
||||
# Current task should be updated to the second task
|
||||
plan_cmd = tl.commands[0]
|
||||
route_cmd = tl.commands[-1]
|
||||
assert plan_cmd["command_name"] == "finish_current_task"
|
||||
assert route_cmd["command_name"] == "publish_message"
|
||||
assert route_cmd["args"]["send_to"] == Architect().name
|
||||
assert tl.planner.plan.current_task_id == "2"
|
||||
# Get message history
|
||||
history = env.history.get()
|
||||
messages_from_tl = [msg for msg in history if msg.sent_from == tl.name]
|
||||
|
||||
# Next step, assuming Architect finishes its task
|
||||
env.publish_message(Message(content=DESIGN_CONTENT, role="Bob(Architect)", sent_from="Bob"))
|
||||
# Verify messages sent to architect
|
||||
architect_messages = [msg for msg in messages_from_tl if "Bob" in msg.send_to]
|
||||
assert len(architect_messages) > 0, "Should have message forwarded to architect"
|
||||
|
||||
# Verify message content contains PRD info
|
||||
architect_message = architect_messages[-1]
|
||||
assert "2048 game based on the PRD" in architect_message.content, "Message to architect should contain PRD info"
|
||||
|
||||
# Verify message routing after architect completes task
|
||||
env.publish_message(Message(content=DESIGN_CONTENT, sent_from="Bob", send_to={"<all>"}))
|
||||
await tl.run()
|
||||
plan_cmd = tl.commands[0]
|
||||
route_cmd = tl.commands[-1]
|
||||
assert plan_cmd["command_name"] == "finish_current_task"
|
||||
assert route_cmd["command_name"] == "publish_message"
|
||||
assert route_cmd["args"]["send_to"] == ProjectManager().name
|
||||
assert tl.planner.plan.current_task_id == "3"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reply_to_human(env):
|
||||
requirement = "create a 2048 game"
|
||||
|
||||
tl = env.get_role("Team Leader")
|
||||
tl = env.get_role("Mike")
|
||||
env.publish_message(Message(content=requirement))
|
||||
await tl.run()
|
||||
|
||||
# Assuming Product Manager finishes its task
|
||||
env.publish_message(Message(content=PRD_MSG_CONTENT, role="Alice(Product Manager)", sent_from="Alice"))
|
||||
# PM finishes task
|
||||
env.publish_message(Message(content=PRD_MSG_CONTENT, sent_from="Alice", send_to={"<all>"}))
|
||||
await tl.run()
|
||||
|
||||
# Human inquires about the progress
|
||||
env.publish_message(Message(content="Who is working? How does the project go?"))
|
||||
# Get history before human inquiry
|
||||
history_before = env.history.get()
|
||||
|
||||
# Human inquires about progress
|
||||
env.publish_message(Message(content="Who is working? How does the project go?", send_to={tl.name}))
|
||||
await tl.run()
|
||||
|
||||
assert tl.commands[0]["command_name"] == "reply_to_human"
|
||||
# Get new messages after human inquiry
|
||||
history_after = env.history.get()
|
||||
new_messages = [msg for msg in history_after if msg not in history_before]
|
||||
|
||||
# Verify team leader's response
|
||||
tl_responses = [msg for msg in new_messages if msg.sent_from == tl.name]
|
||||
assert len(tl_responses) > 0, "Should have response from team leader"
|
||||
|
||||
# Verify response contains project status
|
||||
response = tl_responses[0].content
|
||||
assert any(
|
||||
keyword in response.lower() for keyword in ["progress", "status", "working"]
|
||||
), "Response should contain project status information"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue