test_micro_agents.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. import json
  2. import os
  3. from unittest.mock import MagicMock
  4. import pytest
  5. import yaml
  6. from pytest import TempPathFactory
  7. from agenthub.micro.registry import all_microagents
  8. from opendevin.controller.agent import Agent
  9. from opendevin.controller.state.state import State
  10. from opendevin.events import EventSource
  11. from opendevin.events.action import MessageAction
  12. from opendevin.events.stream import EventStream
  13. from opendevin.memory.history import ShortTermHistory
  14. from opendevin.storage import get_file_store
  15. @pytest.fixture
  16. def temp_dir(tmp_path_factory: TempPathFactory) -> str:
  17. return str(tmp_path_factory.mktemp('test_micro_agents'))
  18. @pytest.fixture
  19. def event_stream(temp_dir):
  20. file_store = get_file_store('local', temp_dir)
  21. event_stream = EventStream('asdf', file_store)
  22. yield event_stream
  23. # clear after each test
  24. event_stream.clear()
  25. def test_all_agents_are_loaded():
  26. assert all_microagents is not None
  27. assert len(all_microagents) > 1
  28. base = os.path.join('agenthub', 'micro')
  29. full_path = os.path.dirname(__file__) + '/../../' + base
  30. agent_names = set()
  31. for root, _, files in os.walk(full_path):
  32. for file in files:
  33. if file == 'agent.yaml':
  34. file_path = os.path.join(root, file)
  35. with open(file_path, 'r') as yaml_file:
  36. data = yaml.safe_load(yaml_file)
  37. agent_names.add(data['name'])
  38. assert agent_names == set(all_microagents.keys())
  39. def test_coder_agent_with_summary(event_stream: EventStream):
  40. """Coder agent should render code summary as part of prompt"""
  41. mock_llm = MagicMock()
  42. content = json.dumps({'action': 'finish', 'args': {}})
  43. mock_llm.completion.return_value = {'choices': [{'message': {'content': content}}]}
  44. coder_agent = Agent.get_cls('CoderAgent')(llm=mock_llm)
  45. assert coder_agent is not None
  46. task = 'This is a dummy task'
  47. history = ShortTermHistory()
  48. history.set_event_stream(event_stream)
  49. event_stream.add_event(MessageAction(content=task), EventSource.USER)
  50. summary = 'This is a dummy summary about this repo'
  51. state = State(history=history, inputs={'summary': summary})
  52. coder_agent.step(state)
  53. mock_llm.completion.assert_called_once()
  54. _, kwargs = mock_llm.completion.call_args
  55. prompt = kwargs['messages'][0]['content'][0]['text']
  56. assert task in prompt
  57. assert "Here's a summary of the codebase, as it relates to this task" in prompt
  58. assert summary in prompt
  59. def test_coder_agent_without_summary(event_stream: EventStream):
  60. """When there's no codebase_summary available, there shouldn't be any prompt
  61. about 'code summary'
  62. """
  63. mock_llm = MagicMock()
  64. content = json.dumps({'action': 'finish', 'args': {}})
  65. mock_llm.completion.return_value = {'choices': [{'message': {'content': content}}]}
  66. coder_agent = Agent.get_cls('CoderAgent')(llm=mock_llm)
  67. assert coder_agent is not None
  68. task = 'This is a dummy task'
  69. history = ShortTermHistory()
  70. history.set_event_stream(event_stream)
  71. event_stream.add_event(MessageAction(content=task), EventSource.USER)
  72. # set state without codebase summary
  73. state = State(history=history)
  74. coder_agent.step(state)
  75. mock_llm.completion.assert_called_once()
  76. _, kwargs = mock_llm.completion.call_args
  77. prompt = kwargs['messages'][0]['content'][0]['text']
  78. assert task in prompt
  79. assert "Here's a summary of the codebase, as it relates to this task" not in prompt