chore: scaffold project with editorconfig, ci workflows, gitignore, pre-commit, changelog, contributing guide, license, and makefile
This commit is contained in:
@@ -0,0 +1,53 @@
|
||||
import pytest
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
yield tmpdir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_api_response():
|
||||
return {
|
||||
'choices': [
|
||||
{
|
||||
'message': {
|
||||
'role': 'assistant',
|
||||
'content': 'Test response'
|
||||
}
|
||||
}
|
||||
],
|
||||
'usage': {
|
||||
'prompt_tokens': 10,
|
||||
'completion_tokens': 5,
|
||||
'total_tokens': 15
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_args():
|
||||
args = MagicMock()
|
||||
args.message = None
|
||||
args.model = None
|
||||
args.api_url = None
|
||||
args.model_list_url = None
|
||||
args.interactive = False
|
||||
args.verbose = False
|
||||
args.no_syntax = False
|
||||
args.include_env = False
|
||||
args.context = None
|
||||
args.api_mode = False
|
||||
return args
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_context_file(temp_dir):
|
||||
context_path = os.path.join(temp_dir, '.rcontext.txt')
|
||||
with open(context_path, 'w') as f:
|
||||
f.write('Sample context content\n')
|
||||
return context_path
|
||||
@@ -0,0 +1,127 @@
|
||||
import pytest
|
||||
import time
|
||||
from pr.agents.agent_roles import AgentRole, get_agent_role, list_agent_roles
|
||||
from pr.agents.agent_manager import AgentManager, AgentInstance
|
||||
from pr.agents.agent_communication import AgentCommunicationBus, AgentMessage, MessageType
|
||||
|
||||
def test_get_agent_role():
|
||||
role = get_agent_role('coding')
|
||||
assert isinstance(role, AgentRole)
|
||||
assert role.name == 'coding'
|
||||
|
||||
def test_list_agent_roles():
|
||||
roles = list_agent_roles()
|
||||
assert isinstance(roles, dict)
|
||||
assert len(roles) > 0
|
||||
assert 'coding' in roles
|
||||
|
||||
def test_agent_role():
|
||||
role = AgentRole(name='test', description='test', system_prompt='test', allowed_tools=set(), specialization_areas=[])
|
||||
assert role.name == 'test'
|
||||
|
||||
def test_agent_instance():
|
||||
role = get_agent_role('coding')
|
||||
instance = AgentInstance(agent_id='test', role=role)
|
||||
assert instance.agent_id == 'test'
|
||||
assert instance.role == role
|
||||
|
||||
def test_agent_manager_init():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
assert mgr is not None
|
||||
|
||||
def test_agent_manager_create_agent():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
agent = mgr.create_agent('coding', 'test_agent')
|
||||
assert agent is not None
|
||||
|
||||
def test_agent_manager_get_agent():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
mgr.create_agent('coding', 'test_agent')
|
||||
agent = mgr.get_agent('test_agent')
|
||||
assert isinstance(agent, AgentInstance)
|
||||
|
||||
def test_agent_manager_remove_agent():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
mgr.create_agent('coding', 'test_agent')
|
||||
mgr.remove_agent('test_agent')
|
||||
agent = mgr.get_agent('test_agent')
|
||||
assert agent is None
|
||||
|
||||
def test_agent_manager_send_agent_message():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
mgr.create_agent('coding', 'a')
|
||||
mgr.create_agent('coding', 'b')
|
||||
mgr.send_agent_message('a', 'b', 'test')
|
||||
assert True
|
||||
|
||||
def test_agent_manager_get_agent_messages():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
mgr.create_agent('coding', 'test')
|
||||
messages = mgr.get_agent_messages('test')
|
||||
assert isinstance(messages, list)
|
||||
|
||||
def test_agent_manager_get_session_summary():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
summary = mgr.get_session_summary()
|
||||
assert isinstance(summary, str)
|
||||
|
||||
def test_agent_manager_collaborate_agents():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
result = mgr.collaborate_agents('orchestrator', 'task', ['coding', 'research'])
|
||||
assert result is not None
|
||||
|
||||
def test_agent_manager_execute_agent_task():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
mgr.create_agent('coding', 'test')
|
||||
result = mgr.execute_agent_task('test', 'task')
|
||||
assert result is not None
|
||||
|
||||
def test_agent_manager_clear_session():
|
||||
mgr = AgentManager(':memory:', None)
|
||||
mgr.clear_session()
|
||||
assert True
|
||||
|
||||
def test_agent_message():
|
||||
msg = AgentMessage(from_agent='a', to_agent='b', message_type=MessageType.REQUEST, content='test', metadata={}, timestamp=1.0, message_id='id')
|
||||
assert msg.from_agent == 'a'
|
||||
|
||||
def test_agent_message_to_dict():
|
||||
msg = AgentMessage(from_agent='a', to_agent='b', message_type=MessageType.REQUEST, content='test', metadata={}, timestamp=1.0, message_id='id')
|
||||
d = msg.to_dict()
|
||||
assert isinstance(d, dict)
|
||||
|
||||
def test_agent_message_from_dict():
|
||||
d = {'from_agent': 'a', 'to_agent': 'b', 'message_type': 'request', 'content': 'test', 'metadata': {}, 'timestamp': 1.0, 'message_id': 'id'}
|
||||
msg = AgentMessage.from_dict(d)
|
||||
assert isinstance(msg, AgentMessage)
|
||||
|
||||
def test_agent_communication_bus_init():
|
||||
bus = AgentCommunicationBus(':memory:')
|
||||
assert bus is not None
|
||||
|
||||
def test_agent_communication_bus_send_message():
|
||||
bus = AgentCommunicationBus(':memory:')
|
||||
msg = AgentMessage(from_agent='a', to_agent='b', message_type=MessageType.REQUEST, content='test', metadata={}, timestamp=1.0, message_id='id')
|
||||
bus.send_message(msg)
|
||||
assert True
|
||||
|
||||
def test_agent_communication_bus_receive_messages():
|
||||
bus = AgentCommunicationBus(':memory:')
|
||||
msg = AgentMessage(from_agent='a', to_agent='b', message_type=MessageType.REQUEST, content='test', metadata={}, timestamp=1.0, message_id='id')
|
||||
bus.send_message(msg)
|
||||
messages = bus.receive_messages('b')
|
||||
assert len(messages) == 1
|
||||
|
||||
def test_agent_communication_bus_get_conversation_history():
|
||||
bus = AgentCommunicationBus(':memory:')
|
||||
msg = AgentMessage(from_agent='a', to_agent='b', message_type=MessageType.REQUEST, content='test', metadata={}, timestamp=1.0, message_id='id')
|
||||
bus.send_message(msg)
|
||||
history = bus.get_conversation_history('a', 'b')
|
||||
assert len(history) == 1
|
||||
|
||||
def test_agent_communication_bus_mark_as_read():
|
||||
bus = AgentCommunicationBus(':memory:')
|
||||
msg = AgentMessage(from_agent='a', to_agent='b', message_type=MessageType.REQUEST, content='test', metadata={}, timestamp=1.0, message_id='id')
|
||||
bus.send_message(msg)
|
||||
bus.mark_as_read(msg.message_id)
|
||||
assert True
|
||||
@@ -0,0 +1,31 @@
|
||||
import pytest
|
||||
from pr import config
|
||||
|
||||
|
||||
class TestConfig:
|
||||
|
||||
def test_default_model_exists(self):
|
||||
assert hasattr(config, 'DEFAULT_MODEL')
|
||||
assert isinstance(config.DEFAULT_MODEL, str)
|
||||
assert len(config.DEFAULT_MODEL) > 0
|
||||
|
||||
def test_api_url_exists(self):
|
||||
assert hasattr(config, 'DEFAULT_API_URL')
|
||||
assert config.DEFAULT_API_URL.startswith('http')
|
||||
|
||||
def test_file_paths_exist(self):
|
||||
assert hasattr(config, 'DB_PATH')
|
||||
assert hasattr(config, 'LOG_FILE')
|
||||
assert hasattr(config, 'HISTORY_FILE')
|
||||
|
||||
def test_autonomous_config(self):
|
||||
assert hasattr(config, 'MAX_AUTONOMOUS_ITERATIONS')
|
||||
assert config.MAX_AUTONOMOUS_ITERATIONS > 0
|
||||
|
||||
assert hasattr(config, 'CONTEXT_COMPRESSION_THRESHOLD')
|
||||
assert config.CONTEXT_COMPRESSION_THRESHOLD > 0
|
||||
|
||||
def test_language_keywords(self):
|
||||
assert hasattr(config, 'LANGUAGE_KEYWORDS')
|
||||
assert 'python' in config.LANGUAGE_KEYWORDS
|
||||
assert isinstance(config.LANGUAGE_KEYWORDS['python'], list)
|
||||
@@ -0,0 +1,35 @@
|
||||
import pytest
|
||||
from pr.core.context import should_compress_context, compress_context
|
||||
from pr.config import RECENT_MESSAGES_TO_KEEP
|
||||
|
||||
|
||||
class TestContextManagement:
|
||||
|
||||
def test_should_compress_context_below_threshold(self):
|
||||
messages = [{'role': 'user', 'content': 'test'}] * 10
|
||||
assert should_compress_context(messages) is False
|
||||
|
||||
def test_should_compress_context_above_threshold(self):
|
||||
messages = [{'role': 'user', 'content': 'test'}] * 35
|
||||
assert should_compress_context(messages) is True
|
||||
|
||||
def test_compress_context_preserves_system_message(self):
|
||||
messages = [
|
||||
{'role': 'system', 'content': 'System prompt'},
|
||||
{'role': 'user', 'content': 'Hello'},
|
||||
{'role': 'assistant', 'content': 'Hi'},
|
||||
] * 40 # Ensure compression
|
||||
compressed = compress_context(messages)
|
||||
assert compressed[0]['role'] == 'system'
|
||||
assert 'System prompt' in compressed[0]['content']
|
||||
|
||||
def test_compress_context_keeps_recent_messages(self):
|
||||
messages = [{'role': 'user', 'content': f'msg{i}'} for i in range(40)]
|
||||
compressed = compress_context(messages)
|
||||
# Should keep recent messages
|
||||
recent = compressed[-RECENT_MESSAGES_TO_KEEP:]
|
||||
assert len(recent) == RECENT_MESSAGES_TO_KEEP
|
||||
# Check that the messages are the most recent ones
|
||||
for i, msg in enumerate(recent):
|
||||
expected_index = 40 - RECENT_MESSAGES_TO_KEEP + i
|
||||
assert msg['content'] == f'msg{expected_index}'
|
||||
@@ -0,0 +1,118 @@
|
||||
import pytest
|
||||
import os
|
||||
import tempfile
|
||||
from pr.tools.filesystem import read_file, write_file, list_directory, search_replace
|
||||
from pr.tools.patch import apply_patch, create_diff
|
||||
from pr.tools.base import get_tools_definition
|
||||
|
||||
|
||||
class TestFilesystemTools:
|
||||
|
||||
def test_write_and_read_file(self, temp_dir):
|
||||
filepath = os.path.join(temp_dir, 'test.txt')
|
||||
content = 'Hello, World!'
|
||||
|
||||
write_result = write_file(filepath, content)
|
||||
assert write_result['status'] == 'success'
|
||||
|
||||
read_result = read_file(filepath)
|
||||
assert read_result['status'] == 'success'
|
||||
assert content in read_result['content']
|
||||
|
||||
def test_read_nonexistent_file(self):
|
||||
result = read_file('/nonexistent/path/file.txt')
|
||||
assert result['status'] == 'error'
|
||||
|
||||
def test_list_directory(self, temp_dir):
|
||||
test_file = os.path.join(temp_dir, 'testfile.txt')
|
||||
with open(test_file, 'w') as f:
|
||||
f.write('test')
|
||||
|
||||
result = list_directory(temp_dir)
|
||||
assert result['status'] == 'success'
|
||||
assert any(item['name'] == 'testfile.txt' for item in result['items'])
|
||||
|
||||
def test_search_replace(self, temp_dir):
|
||||
filepath = os.path.join(temp_dir, 'test.txt')
|
||||
content = 'Hello, World!'
|
||||
with open(filepath, 'w') as f:
|
||||
f.write(content)
|
||||
|
||||
result = search_replace(filepath, 'World', 'Universe')
|
||||
assert result['status'] == 'success'
|
||||
|
||||
read_result = read_file(filepath)
|
||||
assert 'Hello, Universe!' in read_result['content']
|
||||
|
||||
|
||||
class TestPatchTools:
|
||||
|
||||
def test_create_diff(self, temp_dir):
|
||||
file1 = os.path.join(temp_dir, 'file1.txt')
|
||||
file2 = os.path.join(temp_dir, 'file2.txt')
|
||||
with open(file1, 'w') as f:
|
||||
f.write('line1\nline2\nline3\n')
|
||||
with open(file2, 'w') as f:
|
||||
f.write('line1\nline2 modified\nline3\n')
|
||||
|
||||
result = create_diff(file1, file2)
|
||||
assert result['status'] == 'success'
|
||||
assert 'line2' in result['diff']
|
||||
assert 'line2 modified' in result['diff']
|
||||
|
||||
def test_apply_patch(self, temp_dir):
|
||||
filepath = os.path.join(temp_dir, 'file.txt')
|
||||
with open(filepath, 'w') as f:
|
||||
f.write('line1\nline2\nline3\n')
|
||||
|
||||
# Create a simple patch
|
||||
patch_content = """--- a/file.txt
|
||||
+++ b/file.txt
|
||||
@@ -1,3 +1,3 @@
|
||||
line1
|
||||
-line2
|
||||
+line2 modified
|
||||
line3
|
||||
"""
|
||||
result = apply_patch(filepath, patch_content)
|
||||
assert result['status'] == 'success'
|
||||
|
||||
read_result = read_file(filepath)
|
||||
assert 'line2 modified' in read_result['content']
|
||||
|
||||
|
||||
class TestToolDefinitions:
|
||||
|
||||
def test_get_tools_definition_returns_list(self):
|
||||
tools = get_tools_definition()
|
||||
assert isinstance(tools, list)
|
||||
assert len(tools) > 0
|
||||
|
||||
def test_all_tools_have_required_fields(self):
|
||||
tools = get_tools_definition()
|
||||
|
||||
for tool in tools:
|
||||
assert 'type' in tool
|
||||
assert tool['type'] == 'function'
|
||||
assert 'function' in tool
|
||||
|
||||
func = tool['function']
|
||||
assert 'name' in func
|
||||
assert 'description' in func
|
||||
assert 'parameters' in func
|
||||
|
||||
def test_filesystem_tools_present(self):
|
||||
tools = get_tools_definition()
|
||||
tool_names = [t['function']['name'] for t in tools]
|
||||
|
||||
assert 'read_file' in tool_names
|
||||
assert 'write_file' in tool_names
|
||||
assert 'list_directory' in tool_names
|
||||
assert 'search_replace' in tool_names
|
||||
|
||||
def test_patch_tools_present(self):
|
||||
tools = get_tools_definition()
|
||||
tool_names = [t['function']['name'] for t in tools]
|
||||
|
||||
assert 'apply_patch' in tool_names
|
||||
assert 'create_diff' in tool_names
|
||||
Reference in New Issue
Block a user