chore: migrate config paths to XDG base directory and add hit_count tracking to api_cache
This commit is contained in:
@@ -1,8 +1,11 @@
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from pr.ads import AsyncDataSet
|
||||
from pr.web.app import create_app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -41,3 +44,77 @@ def sample_context_file(temp_dir):
|
||||
with open(context_path, "w") as f:
|
||||
f.write("Sample context content\n")
|
||||
return context_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client(aiohttp_client, monkeypatch):
|
||||
"""Create a test client for the app."""
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as tmp:
|
||||
temp_db_file = tmp.name
|
||||
|
||||
# Monkeypatch the db
|
||||
monkeypatch.setattr("pr.web.views.base.db", AsyncDataSet(temp_db_file, f"{temp_db_file}.sock"))
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp_path = Path(tmpdir)
|
||||
static_dir = tmp_path / "static"
|
||||
templates_dir = tmp_path / "templates"
|
||||
repos_dir = tmp_path / "repos"
|
||||
|
||||
static_dir.mkdir()
|
||||
templates_dir.mkdir()
|
||||
repos_dir.mkdir()
|
||||
|
||||
# Monkeypatch the directories
|
||||
monkeypatch.setattr("pr.web.config.STATIC_DIR", static_dir)
|
||||
monkeypatch.setattr("pr.web.config.TEMPLATES_DIR", templates_dir)
|
||||
monkeypatch.setattr("pr.web.config.REPOS_DIR", repos_dir)
|
||||
monkeypatch.setattr("pr.web.config.REPOS_DIR", repos_dir)
|
||||
|
||||
# Create minimal templates
|
||||
(templates_dir / "index.html").write_text("<html>Index</html>")
|
||||
(templates_dir / "login.html").write_text("<html>Login</html>")
|
||||
(templates_dir / "register.html").write_text("<html>Register</html>")
|
||||
(templates_dir / "dashboard.html").write_text("<html>Dashboard</html>")
|
||||
(templates_dir / "repos.html").write_text("<html>Repos</html>")
|
||||
(templates_dir / "api_keys.html").write_text("<html>API Keys</html>")
|
||||
(templates_dir / "repo.html").write_text("<html>Repo</html>")
|
||||
(templates_dir / "file.html").write_text("<html>File</html>")
|
||||
(templates_dir / "edit_file.html").write_text("<html>Edit File</html>")
|
||||
(templates_dir / "deploy.html").write_text("<html>Deploy</html>")
|
||||
|
||||
app_instance = await create_app()
|
||||
client = await aiohttp_client(app_instance)
|
||||
|
||||
yield client
|
||||
|
||||
# Cleanup db
|
||||
os.unlink(temp_db_file)
|
||||
sock_file = f"{temp_db_file}.sock"
|
||||
if os.path.exists(sock_file):
|
||||
os.unlink(sock_file)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def authenticated_client(client):
|
||||
"""Create a client with an authenticated user."""
|
||||
# Register a user
|
||||
resp = await client.post(
|
||||
"/register",
|
||||
data={
|
||||
"username": "testuser",
|
||||
"email": "test@example.com",
|
||||
"password": "password123",
|
||||
"confirm_password": "password123",
|
||||
},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert resp.status == 302 # Redirect to login
|
||||
|
||||
# Login
|
||||
resp = await client.post(
|
||||
"/login", data={"username": "testuser", "password": "password123"}, allow_redirects=False
|
||||
)
|
||||
assert resp.status == 302 # Redirect to dashboard
|
||||
|
||||
return client
|
||||
|
||||
@@ -1,101 +1,97 @@
|
||||
from pr.core.advanced_context import AdvancedContextManager
|
||||
|
||||
|
||||
def test_adaptive_context_window_simple():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = [
|
||||
{"content": "short"},
|
||||
{"content": "this is a longer message with more words"},
|
||||
]
|
||||
window = mgr.adaptive_context_window(messages, "simple")
|
||||
assert isinstance(window, int)
|
||||
assert window >= 10
|
||||
class TestAdvancedContextManager:
|
||||
def setup_method(self):
|
||||
self.manager = AdvancedContextManager()
|
||||
|
||||
def test_init(self):
|
||||
manager = AdvancedContextManager(knowledge_store="test", conversation_memory="test")
|
||||
assert manager.knowledge_store == "test"
|
||||
assert manager.conversation_memory == "test"
|
||||
|
||||
def test_adaptive_context_window_medium():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = [
|
||||
{"content": "short"},
|
||||
{"content": "this is a longer message with more words"},
|
||||
]
|
||||
window = mgr.adaptive_context_window(messages, "medium")
|
||||
assert isinstance(window, int)
|
||||
assert window >= 20
|
||||
def test_adaptive_context_window_simple(self):
|
||||
messages = [{"content": "short message"}]
|
||||
result = self.manager.adaptive_context_window(messages, "simple")
|
||||
assert result >= 10
|
||||
|
||||
def test_adaptive_context_window_medium(self):
|
||||
messages = [{"content": "medium length message with some content"}]
|
||||
result = self.manager.adaptive_context_window(messages, "medium")
|
||||
assert result >= 20
|
||||
|
||||
def test_adaptive_context_window_complex():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = [
|
||||
{"content": "short"},
|
||||
{"content": "this is a longer message with more words"},
|
||||
]
|
||||
window = mgr.adaptive_context_window(messages, "complex")
|
||||
assert isinstance(window, int)
|
||||
assert window >= 35
|
||||
def test_adaptive_context_window_complex(self):
|
||||
messages = [
|
||||
{
|
||||
"content": "very long and complex message with many words and detailed information about various topics"
|
||||
}
|
||||
]
|
||||
result = self.manager.adaptive_context_window(messages, "complex")
|
||||
assert result >= 35
|
||||
|
||||
def test_adaptive_context_window_very_complex(self):
|
||||
messages = [
|
||||
{
|
||||
"content": "extremely long and very complex message with extensive vocabulary and detailed explanations"
|
||||
}
|
||||
]
|
||||
result = self.manager.adaptive_context_window(messages, "very_complex")
|
||||
assert result >= 50
|
||||
|
||||
def test_analyze_message_complexity():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = [{"content": "hello world"}, {"content": "hello again"}]
|
||||
score = mgr._analyze_message_complexity(messages)
|
||||
assert 0 <= score <= 1
|
||||
def test_adaptive_context_window_unknown_complexity(self):
|
||||
messages = [{"content": "test"}]
|
||||
result = self.manager.adaptive_context_window(messages, "unknown")
|
||||
assert result >= 20
|
||||
|
||||
def test_analyze_message_complexity(self):
|
||||
messages = [{"content": "This is a test message with some words."}]
|
||||
result = self.manager._analyze_message_complexity(messages)
|
||||
assert 0.0 <= result <= 1.0
|
||||
|
||||
def test_analyze_message_complexity_empty():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = []
|
||||
score = mgr._analyze_message_complexity(messages)
|
||||
assert score == 0
|
||||
def test_analyze_message_complexity_empty(self):
|
||||
messages = []
|
||||
result = self.manager._analyze_message_complexity(messages)
|
||||
assert result == 0.0
|
||||
|
||||
def test_extract_key_sentences(self):
|
||||
text = "First sentence. Second sentence is longer and more detailed. Third sentence."
|
||||
result = self.manager.extract_key_sentences(text, top_k=2)
|
||||
assert len(result) <= 2
|
||||
assert all(isinstance(s, str) for s in result)
|
||||
|
||||
def test_extract_key_sentences():
|
||||
mgr = AdvancedContextManager()
|
||||
text = "This is the first sentence. This is the second sentence. This is a longer third sentence with more words."
|
||||
sentences = mgr.extract_key_sentences(text, 2)
|
||||
assert len(sentences) <= 2
|
||||
assert all(isinstance(s, str) for s in sentences)
|
||||
def test_extract_key_sentences_empty(self):
|
||||
text = ""
|
||||
result = self.manager.extract_key_sentences(text)
|
||||
assert result == []
|
||||
|
||||
def test_advanced_summarize_messages(self):
|
||||
messages = [
|
||||
{"content": "First message with important information."},
|
||||
{"content": "Second message with more details."},
|
||||
]
|
||||
result = self.manager.advanced_summarize_messages(messages)
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
|
||||
def test_extract_key_sentences_empty():
|
||||
mgr = AdvancedContextManager()
|
||||
text = ""
|
||||
sentences = mgr.extract_key_sentences(text, 5)
|
||||
assert sentences == []
|
||||
def test_advanced_summarize_messages_empty(self):
|
||||
messages = []
|
||||
result = self.manager.advanced_summarize_messages(messages)
|
||||
assert result == "No content to summarize."
|
||||
|
||||
def test_score_message_relevance(self):
|
||||
message = {"content": "test message"}
|
||||
context = "test context"
|
||||
result = self.manager.score_message_relevance(message, context)
|
||||
assert 0.0 <= result <= 1.0
|
||||
|
||||
def test_advanced_summarize_messages():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = [{"content": "Hello"}, {"content": "How are you?"}]
|
||||
summary = mgr.advanced_summarize_messages(messages)
|
||||
assert isinstance(summary, str)
|
||||
def test_score_message_relevance_no_overlap(self):
|
||||
message = {"content": "apple banana"}
|
||||
context = "orange grape"
|
||||
result = self.manager.score_message_relevance(message, context)
|
||||
assert result == 0.0
|
||||
|
||||
|
||||
def test_advanced_summarize_messages_empty():
|
||||
mgr = AdvancedContextManager()
|
||||
messages = []
|
||||
summary = mgr.advanced_summarize_messages(messages)
|
||||
assert summary == "No content to summarize."
|
||||
|
||||
|
||||
def test_score_message_relevance():
|
||||
mgr = AdvancedContextManager()
|
||||
message = {"content": "hello world"}
|
||||
context = "world hello"
|
||||
score = mgr.score_message_relevance(message, context)
|
||||
assert 0 <= score <= 1
|
||||
|
||||
|
||||
def test_score_message_relevance_no_overlap():
|
||||
mgr = AdvancedContextManager()
|
||||
message = {"content": "hello"}
|
||||
context = "world"
|
||||
score = mgr.score_message_relevance(message, context)
|
||||
assert score == 0
|
||||
|
||||
|
||||
def test_score_message_relevance_empty():
|
||||
mgr = AdvancedContextManager()
|
||||
message = {"content": ""}
|
||||
context = ""
|
||||
score = mgr.score_message_relevance(message, context)
|
||||
assert score == 0
|
||||
def test_score_message_relevance_empty(self):
|
||||
message = {"content": ""}
|
||||
context = ""
|
||||
result = self.manager.score_message_relevance(message, context)
|
||||
assert result == 0.0
|
||||
|
||||
@@ -82,7 +82,10 @@ def test_agent_manager_get_agent_messages():
|
||||
def test_agent_manager_get_session_summary():
|
||||
mgr = AgentManager(":memory:", None)
|
||||
summary = mgr.get_session_summary()
|
||||
assert isinstance(summary, str)
|
||||
assert isinstance(summary, dict)
|
||||
assert "session_id" in summary
|
||||
assert "active_agents" in summary
|
||||
assert "agents" in summary
|
||||
|
||||
|
||||
def test_agent_manager_collaborate_agents():
|
||||
|
||||
@@ -0,0 +1,697 @@
|
||||
from unittest.mock import Mock, patch
|
||||
from pr.commands.handlers import (
|
||||
handle_command,
|
||||
review_file,
|
||||
refactor_file,
|
||||
obfuscate_file,
|
||||
show_workflows,
|
||||
execute_workflow_command,
|
||||
execute_agent_task,
|
||||
show_agents,
|
||||
collaborate_agents_command,
|
||||
search_knowledge,
|
||||
store_knowledge,
|
||||
show_conversation_history,
|
||||
show_cache_stats,
|
||||
clear_caches,
|
||||
show_system_stats,
|
||||
handle_background_command,
|
||||
start_background_session,
|
||||
list_background_sessions,
|
||||
show_session_status,
|
||||
show_session_output,
|
||||
send_session_input,
|
||||
kill_background_session,
|
||||
show_background_events,
|
||||
)
|
||||
|
||||
|
||||
class TestHandleCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
self.assistant.messages = [{"role": "system", "content": "test"}]
|
||||
self.assistant.verbose = False
|
||||
self.assistant.model = "test-model"
|
||||
self.assistant.model_list_url = "http://test.com"
|
||||
self.assistant.api_key = "test-key"
|
||||
|
||||
@patch("pr.commands.handlers.run_autonomous_mode")
|
||||
def test_handle_edit(self, mock_run):
|
||||
with patch("pr.commands.handlers.RPEditor") as mock_editor:
|
||||
mock_editor_instance = Mock()
|
||||
mock_editor.return_value = mock_editor_instance
|
||||
mock_editor_instance.get_text.return_value = "test task"
|
||||
handle_command(self.assistant, "/edit test.py")
|
||||
mock_editor.assert_called_once_with("test.py")
|
||||
mock_editor_instance.start.assert_called_once()
|
||||
mock_editor_instance.thread.join.assert_called_once()
|
||||
mock_run.assert_called_once_with(self.assistant, "test task")
|
||||
mock_editor_instance.stop.assert_called_once()
|
||||
|
||||
@patch("pr.commands.handlers.run_autonomous_mode")
|
||||
def test_handle_auto(self, mock_run):
|
||||
result = handle_command(self.assistant, "/auto test task")
|
||||
assert result is True
|
||||
mock_run.assert_called_once_with(self.assistant, "test task")
|
||||
|
||||
def test_handle_auto_no_args(self):
|
||||
result = handle_command(self.assistant, "/auto")
|
||||
assert result is True
|
||||
|
||||
def test_handle_exit(self):
|
||||
result = handle_command(self.assistant, "exit")
|
||||
assert result is False
|
||||
|
||||
@patch("pr.commands.help_docs.get_full_help")
|
||||
def test_handle_help(self, mock_help):
|
||||
mock_help.return_value = "full help"
|
||||
result = handle_command(self.assistant, "/help")
|
||||
assert result is True
|
||||
mock_help.assert_called_once()
|
||||
|
||||
@patch("pr.commands.help_docs.get_workflow_help")
|
||||
def test_handle_help_workflows(self, mock_help):
|
||||
mock_help.return_value = "workflow help"
|
||||
result = handle_command(self.assistant, "/help workflows")
|
||||
assert result is True
|
||||
mock_help.assert_called_once()
|
||||
|
||||
def test_handle_reset(self):
|
||||
self.assistant.messages = [
|
||||
{"role": "system", "content": "test"},
|
||||
{"role": "user", "content": "hi"},
|
||||
]
|
||||
result = handle_command(self.assistant, "/reset")
|
||||
assert result is True
|
||||
assert self.assistant.messages == [{"role": "system", "content": "test"}]
|
||||
|
||||
def test_handle_dump(self):
|
||||
result = handle_command(self.assistant, "/dump")
|
||||
assert result is True
|
||||
|
||||
def test_handle_verbose(self):
|
||||
result = handle_command(self.assistant, "/verbose")
|
||||
assert result is True
|
||||
assert self.assistant.verbose is True
|
||||
|
||||
def test_handle_model_get(self):
|
||||
result = handle_command(self.assistant, "/model")
|
||||
assert result is True
|
||||
|
||||
def test_handle_model_set(self):
|
||||
result = handle_command(self.assistant, "/model new-model")
|
||||
assert result is True
|
||||
assert self.assistant.model == "new-model"
|
||||
|
||||
@patch("pr.commands.handlers.list_models")
|
||||
def test_handle_models(self, mock_list):
|
||||
mock_list.return_value = [{"id": "model1"}, {"id": "model2"}]
|
||||
result = handle_command(self.assistant, "/models")
|
||||
assert result is True
|
||||
mock_list.assert_called_once_with("http://test.com", "test-key")
|
||||
|
||||
@patch("pr.commands.handlers.list_models")
|
||||
def test_handle_models_error(self, mock_list):
|
||||
mock_list.return_value = {"error": "test error"}
|
||||
result = handle_command(self.assistant, "/models")
|
||||
assert result is True
|
||||
|
||||
@patch("pr.commands.handlers.get_tools_definition")
|
||||
def test_handle_tools(self, mock_tools):
|
||||
mock_tools.return_value = [{"function": {"name": "tool1", "description": "desc"}}]
|
||||
result = handle_command(self.assistant, "/tools")
|
||||
assert result is True
|
||||
mock_tools.assert_called_once()
|
||||
|
||||
@patch("pr.commands.handlers.review_file")
|
||||
def test_handle_review(self, mock_review):
|
||||
result = handle_command(self.assistant, "/review test.py")
|
||||
assert result is True
|
||||
mock_review.assert_called_once_with(self.assistant, "test.py")
|
||||
|
||||
@patch("pr.commands.handlers.refactor_file")
|
||||
def test_handle_refactor(self, mock_refactor):
|
||||
result = handle_command(self.assistant, "/refactor test.py")
|
||||
assert result is True
|
||||
mock_refactor.assert_called_once_with(self.assistant, "test.py")
|
||||
|
||||
@patch("pr.commands.handlers.obfuscate_file")
|
||||
def test_handle_obfuscate(self, mock_obfuscate):
|
||||
result = handle_command(self.assistant, "/obfuscate test.py")
|
||||
assert result is True
|
||||
mock_obfuscate.assert_called_once_with(self.assistant, "test.py")
|
||||
|
||||
@patch("pr.commands.handlers.show_workflows")
|
||||
def test_handle_workflows(self, mock_show):
|
||||
result = handle_command(self.assistant, "/workflows")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.execute_workflow_command")
|
||||
def test_handle_workflow(self, mock_exec):
|
||||
result = handle_command(self.assistant, "/workflow test")
|
||||
assert result is True
|
||||
mock_exec.assert_called_once_with(self.assistant, "test")
|
||||
|
||||
@patch("pr.commands.handlers.execute_agent_task")
|
||||
def test_handle_agent(self, mock_exec):
|
||||
result = handle_command(self.assistant, "/agent coding test task")
|
||||
assert result is True
|
||||
mock_exec.assert_called_once_with(self.assistant, "coding", "test task")
|
||||
|
||||
def test_handle_agent_no_args(self):
|
||||
result = handle_command(self.assistant, "/agent")
|
||||
assert result is True
|
||||
|
||||
@patch("pr.commands.handlers.show_agents")
|
||||
def test_handle_agents(self, mock_show):
|
||||
result = handle_command(self.assistant, "/agents")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.collaborate_agents_command")
|
||||
def test_handle_collaborate(self, mock_collab):
|
||||
result = handle_command(self.assistant, "/collaborate test task")
|
||||
assert result is True
|
||||
mock_collab.assert_called_once_with(self.assistant, "test task")
|
||||
|
||||
@patch("pr.commands.handlers.search_knowledge")
|
||||
def test_handle_knowledge(self, mock_search):
|
||||
result = handle_command(self.assistant, "/knowledge test query")
|
||||
assert result is True
|
||||
mock_search.assert_called_once_with(self.assistant, "test query")
|
||||
|
||||
@patch("pr.commands.handlers.store_knowledge")
|
||||
def test_handle_remember(self, mock_store):
|
||||
result = handle_command(self.assistant, "/remember test content")
|
||||
assert result is True
|
||||
mock_store.assert_called_once_with(self.assistant, "test content")
|
||||
|
||||
@patch("pr.commands.handlers.show_conversation_history")
|
||||
def test_handle_history(self, mock_show):
|
||||
result = handle_command(self.assistant, "/history")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.show_cache_stats")
|
||||
def test_handle_cache(self, mock_show):
|
||||
result = handle_command(self.assistant, "/cache")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.clear_caches")
|
||||
def test_handle_cache_clear(self, mock_clear):
|
||||
result = handle_command(self.assistant, "/cache clear")
|
||||
assert result is True
|
||||
mock_clear.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.show_system_stats")
|
||||
def test_handle_stats(self, mock_show):
|
||||
result = handle_command(self.assistant, "/stats")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.handle_background_command")
|
||||
def test_handle_bg(self, mock_bg):
|
||||
result = handle_command(self.assistant, "/bg list")
|
||||
assert result is True
|
||||
mock_bg.assert_called_once_with(self.assistant, "/bg list")
|
||||
|
||||
def test_handle_unknown(self):
|
||||
result = handle_command(self.assistant, "/unknown")
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestReviewFile:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
@patch("pr.core.assistant.process_message")
|
||||
def test_review_file_success(self, mock_process, mock_read):
|
||||
mock_read.return_value = {"status": "success", "content": "test content"}
|
||||
review_file(self.assistant, "test.py")
|
||||
mock_read.assert_called_once_with("test.py")
|
||||
mock_process.assert_called_once()
|
||||
args = mock_process.call_args[0]
|
||||
assert "Please review this file" in args[1]
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
def test_review_file_error(self, mock_read):
|
||||
mock_read.return_value = {"status": "error", "error": "file not found"}
|
||||
review_file(self.assistant, "test.py")
|
||||
mock_read.assert_called_once_with("test.py")
|
||||
|
||||
|
||||
class TestRefactorFile:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
@patch("pr.core.assistant.process_message")
|
||||
def test_refactor_file_success(self, mock_process, mock_read):
|
||||
mock_read.return_value = {"status": "success", "content": "test content"}
|
||||
refactor_file(self.assistant, "test.py")
|
||||
mock_process.assert_called_once()
|
||||
args = mock_process.call_args[0]
|
||||
assert "Please refactor this code" in args[1]
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
def test_refactor_file_error(self, mock_read):
|
||||
mock_read.return_value = {"status": "error", "error": "file not found"}
|
||||
refactor_file(self.assistant, "test.py")
|
||||
|
||||
|
||||
class TestObfuscateFile:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
@patch("pr.core.assistant.process_message")
|
||||
def test_obfuscate_file_success(self, mock_process, mock_read):
|
||||
mock_read.return_value = {"status": "success", "content": "test content"}
|
||||
obfuscate_file(self.assistant, "test.py")
|
||||
mock_process.assert_called_once()
|
||||
args = mock_process.call_args[0]
|
||||
assert "Please obfuscate this code" in args[1]
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
def test_obfuscate_file_error(self, mock_read):
|
||||
mock_read.return_value = {"status": "error", "error": "file not found"}
|
||||
obfuscate_file(self.assistant, "test.py")
|
||||
|
||||
|
||||
class TestShowWorkflows:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_workflows_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_workflows(self.assistant)
|
||||
|
||||
def test_show_workflows_no_workflows(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_workflow_list.return_value = []
|
||||
show_workflows(self.assistant)
|
||||
|
||||
def test_show_workflows_with_workflows(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_workflow_list.return_value = [
|
||||
{"name": "wf1", "description": "desc1", "execution_count": 5}
|
||||
]
|
||||
show_workflows(self.assistant)
|
||||
|
||||
|
||||
class TestExecuteWorkflowCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_execute_workflow_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
execute_workflow_command(self.assistant, "test")
|
||||
|
||||
def test_execute_workflow_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.execute_workflow.return_value = {
|
||||
"execution_id": "123",
|
||||
"results": {"key": "value"},
|
||||
}
|
||||
execute_workflow_command(self.assistant, "test")
|
||||
|
||||
def test_execute_workflow_error(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.execute_workflow.return_value = {"error": "test error"}
|
||||
execute_workflow_command(self.assistant, "test")
|
||||
|
||||
|
||||
class TestExecuteAgentTask:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_execute_agent_task_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
execute_agent_task(self.assistant, "coding", "task")
|
||||
|
||||
def test_execute_agent_task_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.create_agent.return_value = "agent123"
|
||||
self.assistant.enhanced.agent_task.return_value = {"response": "done"}
|
||||
execute_agent_task(self.assistant, "coding", "task")
|
||||
|
||||
def test_execute_agent_task_error(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.create_agent.return_value = "agent123"
|
||||
self.assistant.enhanced.agent_task.return_value = {"error": "test error"}
|
||||
execute_agent_task(self.assistant, "coding", "task")
|
||||
|
||||
|
||||
class TestShowAgents:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_agents_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_agents(self.assistant)
|
||||
|
||||
def test_show_agents_with_agents(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_agent_summary.return_value = {
|
||||
"active_agents": 2,
|
||||
"agents": [{"agent_id": "a1", "role": "coding", "task_count": 3, "message_count": 10}],
|
||||
}
|
||||
show_agents(self.assistant)
|
||||
|
||||
|
||||
class TestCollaborateAgentsCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_collaborate_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
collaborate_agents_command(self.assistant, "task")
|
||||
|
||||
def test_collaborate_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.collaborate_agents.return_value = {
|
||||
"orchestrator": {"response": "orchestrator response"},
|
||||
"agents": [{"role": "coding", "response": "coding response"}],
|
||||
}
|
||||
collaborate_agents_command(self.assistant, "task")
|
||||
|
||||
|
||||
class TestSearchKnowledge:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_search_knowledge_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
search_knowledge(self.assistant, "query")
|
||||
|
||||
def test_search_knowledge_no_results(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.search_knowledge.return_value = []
|
||||
search_knowledge(self.assistant, "query")
|
||||
|
||||
def test_search_knowledge_with_results(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
mock_entry = Mock()
|
||||
mock_entry.category = "general"
|
||||
mock_entry.content = "long content here"
|
||||
mock_entry.access_count = 5
|
||||
self.assistant.enhanced.search_knowledge.return_value = [mock_entry]
|
||||
search_knowledge(self.assistant, "query")
|
||||
|
||||
|
||||
class TestStoreKnowledge:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_store_knowledge_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
store_knowledge(self.assistant, "content")
|
||||
|
||||
@patch("pr.memory.KnowledgeEntry")
|
||||
def test_store_knowledge_success(self, mock_entry):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.fact_extractor.categorize_content.return_value = ["general"]
|
||||
self.assistant.enhanced.knowledge_store = Mock()
|
||||
store_knowledge(self.assistant, "content")
|
||||
mock_entry.assert_called_once()
|
||||
|
||||
|
||||
class TestShowConversationHistory:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_history_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_conversation_history(self.assistant)
|
||||
|
||||
def test_show_history_no_history(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_conversation_history.return_value = []
|
||||
show_conversation_history(self.assistant)
|
||||
|
||||
def test_show_history_with_history(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_conversation_history.return_value = [
|
||||
{
|
||||
"conversation_id": "conv1",
|
||||
"started_at": 1234567890,
|
||||
"message_count": 5,
|
||||
"summary": "test summary",
|
||||
"topics": ["topic1", "topic2"],
|
||||
}
|
||||
]
|
||||
show_conversation_history(self.assistant)
|
||||
|
||||
|
||||
class TestShowCacheStats:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_cache_stats_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_cache_stats(self.assistant)
|
||||
|
||||
def test_show_cache_stats_with_stats(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_cache_statistics.return_value = {
|
||||
"api_cache": {
|
||||
"total_entries": 10,
|
||||
"valid_entries": 8,
|
||||
"expired_entries": 2,
|
||||
"total_cached_tokens": 1000,
|
||||
"total_cache_hits": 50,
|
||||
},
|
||||
"tool_cache": {
|
||||
"total_entries": 5,
|
||||
"valid_entries": 5,
|
||||
"total_cache_hits": 20,
|
||||
"by_tool": {"tool1": {"cached_entries": 3, "total_hits": 10}},
|
||||
},
|
||||
}
|
||||
show_cache_stats(self.assistant)
|
||||
|
||||
|
||||
class TestClearCaches:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_clear_caches_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
clear_caches(self.assistant)
|
||||
|
||||
def test_clear_caches_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
clear_caches(self.assistant)
|
||||
self.assistant.enhanced.clear_caches.assert_called_once()
|
||||
|
||||
|
||||
class TestShowSystemStats:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_system_stats_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_system_stats(self.assistant)
|
||||
|
||||
def test_show_system_stats_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_cache_statistics.return_value = {
|
||||
"api_cache": {"valid_entries": 10},
|
||||
"tool_cache": {"valid_entries": 5},
|
||||
}
|
||||
self.assistant.enhanced.get_knowledge_statistics.return_value = {
|
||||
"total_entries": 100,
|
||||
"total_categories": 5,
|
||||
"total_accesses": 200,
|
||||
"vocabulary_size": 1000,
|
||||
}
|
||||
self.assistant.enhanced.get_agent_summary.return_value = {"active_agents": 3}
|
||||
show_system_stats(self.assistant)
|
||||
|
||||
|
||||
class TestHandleBackgroundCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_handle_bg_no_args(self):
|
||||
handle_background_command(self.assistant, "/bg")
|
||||
|
||||
@patch("pr.commands.handlers.start_background_session")
|
||||
def test_handle_bg_start(self, mock_start):
|
||||
handle_background_command(self.assistant, "/bg start ls -la")
|
||||
|
||||
@patch("pr.commands.handlers.list_background_sessions")
|
||||
def test_handle_bg_list(self, mock_list):
|
||||
handle_background_command(self.assistant, "/bg list")
|
||||
|
||||
@patch("pr.commands.handlers.show_session_status")
|
||||
def test_handle_bg_status(self, mock_status):
|
||||
handle_background_command(self.assistant, "/bg status session1")
|
||||
|
||||
@patch("pr.commands.handlers.show_session_output")
|
||||
def test_handle_bg_output(self, mock_output):
|
||||
handle_background_command(self.assistant, "/bg output session1")
|
||||
|
||||
@patch("pr.commands.handlers.send_session_input")
|
||||
def test_handle_bg_input(self, mock_input):
|
||||
handle_background_command(self.assistant, "/bg input session1 test input")
|
||||
|
||||
@patch("pr.commands.handlers.kill_background_session")
|
||||
def test_handle_bg_kill(self, mock_kill):
|
||||
handle_background_command(self.assistant, "/bg kill session1")
|
||||
|
||||
@patch("pr.commands.handlers.show_background_events")
|
||||
def test_handle_bg_events(self, mock_events):
|
||||
handle_background_command(self.assistant, "/bg events")
|
||||
|
||||
def test_handle_bg_unknown(self):
|
||||
handle_background_command(self.assistant, "/bg unknown")
|
||||
|
||||
|
||||
class TestStartBackgroundSession:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.multiplexer.start_background_process")
|
||||
def test_start_background_success(self, mock_start):
|
||||
mock_start.return_value = {"status": "success", "pid": 123}
|
||||
start_background_session(self.assistant, "session1", "ls -la")
|
||||
|
||||
@patch("pr.multiplexer.start_background_process")
|
||||
def test_start_background_error(self, mock_start):
|
||||
mock_start.return_value = {"status": "error", "error": "failed"}
|
||||
start_background_session(self.assistant, "session1", "ls -la")
|
||||
|
||||
@patch("pr.multiplexer.start_background_process")
|
||||
def test_start_background_exception(self, mock_start):
|
||||
mock_start.side_effect = Exception("test")
|
||||
start_background_session(self.assistant, "session1", "ls -la")
|
||||
|
||||
|
||||
class TestListBackgroundSessions:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.multiplexer.get_all_sessions")
|
||||
@patch("pr.ui.display.display_multiplexer_status")
|
||||
def test_list_sessions_success(self, mock_display, mock_get):
|
||||
mock_get.return_value = {}
|
||||
list_background_sessions(self.assistant)
|
||||
|
||||
@patch("pr.multiplexer.get_all_sessions")
|
||||
def test_list_sessions_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
list_background_sessions(self.assistant)
|
||||
|
||||
|
||||
class TestShowSessionStatus:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.multiplexer.get_session_info")
|
||||
def test_show_status_found(self, mock_get):
|
||||
mock_get.return_value = {
|
||||
"status": "running",
|
||||
"pid": 123,
|
||||
"command": "ls",
|
||||
"start_time": 1234567890.0,
|
||||
}
|
||||
show_session_status(self.assistant, "session1")
|
||||
|
||||
@patch("pr.multiplexer.get_session_info")
|
||||
def test_show_status_not_found(self, mock_get):
|
||||
mock_get.return_value = None
|
||||
show_session_status(self.assistant, "session1")
|
||||
|
||||
@patch("pr.multiplexer.get_session_info")
|
||||
def test_show_status_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
show_session_status(self.assistant, "session1")
|
||||
|
||||
|
||||
class TestShowSessionOutput:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.multiplexer.get_session_output")
|
||||
def test_show_output_success(self, mock_get):
|
||||
mock_get.return_value = ["line1", "line2"]
|
||||
show_session_output(self.assistant, "session1")
|
||||
|
||||
@patch("pr.multiplexer.get_session_output")
|
||||
def test_show_output_no_output(self, mock_get):
|
||||
mock_get.return_value = None
|
||||
show_session_output(self.assistant, "session1")
|
||||
|
||||
@patch("pr.multiplexer.get_session_output")
|
||||
def test_show_output_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
show_session_output(self.assistant, "session1")
|
||||
|
||||
|
||||
class TestSendSessionInput:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.multiplexer.send_input_to_session")
|
||||
def test_send_input_success(self, mock_send):
|
||||
mock_send.return_value = {"status": "success"}
|
||||
send_session_input(self.assistant, "session1", "input")
|
||||
|
||||
@patch("pr.multiplexer.send_input_to_session")
|
||||
def test_send_input_error(self, mock_send):
|
||||
mock_send.return_value = {"status": "error", "error": "failed"}
|
||||
send_session_input(self.assistant, "session1", "input")
|
||||
|
||||
@patch("pr.multiplexer.send_input_to_session")
|
||||
def test_send_input_exception(self, mock_send):
|
||||
mock_send.side_effect = Exception("test")
|
||||
send_session_input(self.assistant, "session1", "input")
|
||||
|
||||
|
||||
class TestKillBackgroundSession:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.multiplexer.kill_session")
|
||||
def test_kill_success(self, mock_kill):
|
||||
mock_kill.return_value = {"status": "success"}
|
||||
kill_background_session(self.assistant, "session1")
|
||||
|
||||
@patch("pr.multiplexer.kill_session")
|
||||
def test_kill_error(self, mock_kill):
|
||||
mock_kill.return_value = {"status": "error", "error": "failed"}
|
||||
kill_background_session(self.assistant, "session1")
|
||||
|
||||
@patch("pr.multiplexer.kill_session")
|
||||
def test_kill_exception(self, mock_kill):
|
||||
mock_kill.side_effect = Exception("test")
|
||||
kill_background_session(self.assistant, "session1")
|
||||
|
||||
|
||||
class TestShowBackgroundEvents:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.core.background_monitor.get_global_monitor")
|
||||
def test_show_events_success(self, mock_get):
|
||||
mock_monitor = Mock()
|
||||
mock_monitor.get_events.return_value = [{"event": "test"}]
|
||||
mock_get.return_value = mock_monitor
|
||||
with patch("pr.ui.display.display_background_event"):
|
||||
show_background_events(self.assistant)
|
||||
|
||||
@patch("pr.core.background_monitor.get_global_monitor")
|
||||
def test_show_events_no_events(self, mock_get):
|
||||
mock_monitor = Mock()
|
||||
mock_monitor.get_events.return_value = []
|
||||
mock_get.return_value = mock_monitor
|
||||
show_background_events(self.assistant)
|
||||
|
||||
@patch("pr.core.background_monitor.get_global_monitor")
|
||||
def test_show_events_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
show_background_events(self.assistant)
|
||||
@@ -0,0 +1,693 @@
|
||||
from unittest.mock import Mock, patch
|
||||
from pr.commands.handlers import (
|
||||
handle_command,
|
||||
review_file,
|
||||
refactor_file,
|
||||
obfuscate_file,
|
||||
show_workflows,
|
||||
execute_workflow_command,
|
||||
execute_agent_task,
|
||||
show_agents,
|
||||
collaborate_agents_command,
|
||||
search_knowledge,
|
||||
store_knowledge,
|
||||
show_conversation_history,
|
||||
show_cache_stats,
|
||||
clear_caches,
|
||||
show_system_stats,
|
||||
handle_background_command,
|
||||
start_background_session,
|
||||
list_background_sessions,
|
||||
show_session_status,
|
||||
show_session_output,
|
||||
send_session_input,
|
||||
kill_background_session,
|
||||
show_background_events,
|
||||
)
|
||||
|
||||
|
||||
class TestHandleCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
self.assistant.messages = [{"role": "system", "content": "test"}]
|
||||
self.assistant.verbose = False
|
||||
self.assistant.model = "test-model"
|
||||
self.assistant.model_list_url = "http://test.com"
|
||||
self.assistant.api_key = "test-key"
|
||||
|
||||
@patch("pr.commands.handlers.run_autonomous_mode")
|
||||
def test_handle_edit(self, mock_run):
|
||||
with patch("pr.commands.handlers.RPEditor") as mock_editor:
|
||||
mock_editor_instance = Mock()
|
||||
mock_editor.return_value = mock_editor_instance
|
||||
mock_editor_instance.get_text.return_value = "test task"
|
||||
handle_command(self.assistant, "/edit test.py")
|
||||
mock_editor.assert_called_once_with("test.py")
|
||||
mock_editor_instance.start.assert_called_once()
|
||||
mock_editor_instance.thread.join.assert_called_once()
|
||||
mock_run.assert_called_once_with(self.assistant, "test task")
|
||||
mock_editor_instance.stop.assert_called_once()
|
||||
|
||||
@patch("pr.commands.handlers.run_autonomous_mode")
|
||||
def test_handle_auto(self, mock_run):
|
||||
result = handle_command(self.assistant, "/auto test task")
|
||||
assert result is True
|
||||
mock_run.assert_called_once_with(self.assistant, "test task")
|
||||
|
||||
def test_handle_auto_no_args(self):
|
||||
result = handle_command(self.assistant, "/auto")
|
||||
assert result is True
|
||||
|
||||
def test_handle_exit(self):
|
||||
result = handle_command(self.assistant, "exit")
|
||||
assert result is False
|
||||
|
||||
@patch("pr.commands.help_docs.get_full_help")
|
||||
def test_handle_help(self, mock_help):
|
||||
mock_help.return_value = "full help"
|
||||
result = handle_command(self.assistant, "/help")
|
||||
assert result is True
|
||||
mock_help.assert_called_once()
|
||||
|
||||
@patch("pr.commands.help_docs.get_workflow_help")
|
||||
def test_handle_help_workflows(self, mock_help):
|
||||
mock_help.return_value = "workflow help"
|
||||
result = handle_command(self.assistant, "/help workflows")
|
||||
assert result is True
|
||||
mock_help.assert_called_once()
|
||||
|
||||
def test_handle_reset(self):
|
||||
self.assistant.messages = [
|
||||
{"role": "system", "content": "test"},
|
||||
{"role": "user", "content": "hi"},
|
||||
]
|
||||
result = handle_command(self.assistant, "/reset")
|
||||
assert result is True
|
||||
assert self.assistant.messages == [{"role": "system", "content": "test"}]
|
||||
|
||||
def test_handle_dump(self):
|
||||
result = handle_command(self.assistant, "/dump")
|
||||
assert result is True
|
||||
|
||||
def test_handle_verbose(self):
|
||||
result = handle_command(self.assistant, "/verbose")
|
||||
assert result is True
|
||||
assert self.assistant.verbose is True
|
||||
|
||||
def test_handle_model_get(self):
|
||||
result = handle_command(self.assistant, "/model")
|
||||
assert result is True
|
||||
|
||||
def test_handle_model_set(self):
|
||||
result = handle_command(self.assistant, "/model new-model")
|
||||
assert result is True
|
||||
assert self.assistant.model == "new-model"
|
||||
|
||||
@patch("pr.core.api.list_models")
|
||||
@patch("pr.core.api.list_models")
|
||||
def test_handle_models(self, mock_list):
|
||||
mock_list.return_value = [{"id": "model1"}, {"id": "model2"}]
|
||||
with patch('pr.commands.handlers.list_models', mock_list):
|
||||
result = handle_command(self.assistant, "/models")
|
||||
assert result is True
|
||||
mock_list.assert_called_once_with("http://test.com", "test-key")ef test_handle_models_error(self, mock_list):
|
||||
mock_list.return_value = {"error": "test error"}
|
||||
result = handle_command(self.assistant, "/models")
|
||||
assert result is True
|
||||
|
||||
@patch("pr.tools.base.get_tools_definition")
|
||||
@patch("pr.tools.base.get_tools_definition")
|
||||
def test_handle_tools(self, mock_tools):
|
||||
mock_tools.return_value = [{"function": {"name": "tool1", "description": "desc"}}]
|
||||
with patch('pr.commands.handlers.get_tools_definition', mock_tools):
|
||||
result = handle_command(self.assistant, "/tools")
|
||||
assert result is True
|
||||
mock_tools.assert_called_once()ef test_handle_review(self, mock_review):
|
||||
result = handle_command(self.assistant, "/review test.py")
|
||||
assert result is True
|
||||
mock_review.assert_called_once_with(self.assistant, "test.py")
|
||||
|
||||
@patch("pr.commands.handlers.refactor_file")
|
||||
def test_handle_refactor(self, mock_refactor):
|
||||
result = handle_command(self.assistant, "/refactor test.py")
|
||||
assert result is True
|
||||
mock_refactor.assert_called_once_with(self.assistant, "test.py")
|
||||
|
||||
@patch("pr.commands.handlers.obfuscate_file")
|
||||
def test_handle_obfuscate(self, mock_obfuscate):
|
||||
result = handle_command(self.assistant, "/obfuscate test.py")
|
||||
assert result is True
|
||||
mock_obfuscate.assert_called_once_with(self.assistant, "test.py")
|
||||
|
||||
@patch("pr.commands.handlers.show_workflows")
|
||||
def test_handle_workflows(self, mock_show):
|
||||
result = handle_command(self.assistant, "/workflows")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.execute_workflow_command")
|
||||
def test_handle_workflow(self, mock_exec):
|
||||
result = handle_command(self.assistant, "/workflow test")
|
||||
assert result is True
|
||||
mock_exec.assert_called_once_with(self.assistant, "test")
|
||||
|
||||
@patch("pr.commands.handlers.execute_agent_task")
|
||||
def test_handle_agent(self, mock_exec):
|
||||
result = handle_command(self.assistant, "/agent coding test task")
|
||||
assert result is True
|
||||
mock_exec.assert_called_once_with(self.assistant, "coding", "test task")
|
||||
|
||||
def test_handle_agent_no_args(self):
|
||||
result = handle_command(self.assistant, "/agent")
|
||||
assert result is None assert result is True
|
||||
|
||||
@patch("pr.commands.handlers.show_agents")
|
||||
def test_handle_agents(self, mock_show):
|
||||
result = handle_command(self.assistant, "/agents")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.collaborate_agents_command")
|
||||
def test_handle_collaborate(self, mock_collab):
|
||||
result = handle_command(self.assistant, "/collaborate test task")
|
||||
assert result is True
|
||||
mock_collab.assert_called_once_with(self.assistant, "test task")
|
||||
|
||||
@patch("pr.commands.handlers.search_knowledge")
|
||||
def test_handle_knowledge(self, mock_search):
|
||||
result = handle_command(self.assistant, "/knowledge test query")
|
||||
assert result is True
|
||||
mock_search.assert_called_once_with(self.assistant, "test query")
|
||||
|
||||
@patch("pr.commands.handlers.store_knowledge")
|
||||
def test_handle_remember(self, mock_store):
|
||||
result = handle_command(self.assistant, "/remember test content")
|
||||
assert result is True
|
||||
mock_store.assert_called_once_with(self.assistant, "test content")
|
||||
|
||||
@patch("pr.commands.handlers.show_conversation_history")
|
||||
def test_handle_history(self, mock_show):
|
||||
result = handle_command(self.assistant, "/history")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.show_cache_stats")
|
||||
def test_handle_cache(self, mock_show):
|
||||
result = handle_command(self.assistant, "/cache")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.clear_caches")
|
||||
def test_handle_cache_clear(self, mock_clear):
|
||||
result = handle_command(self.assistant, "/cache clear")
|
||||
assert result is True
|
||||
mock_clear.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.show_system_stats")
|
||||
def test_handle_stats(self, mock_show):
|
||||
result = handle_command(self.assistant, "/stats")
|
||||
assert result is True
|
||||
mock_show.assert_called_once_with(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.handle_background_command")
|
||||
def test_handle_bg(self, mock_bg):
|
||||
result = handle_command(self.assistant, "/bg list")
|
||||
assert result is True
|
||||
mock_bg.assert_called_once_with(self.assistant, "/bg list")
|
||||
|
||||
def test_handle_unknown(self):
|
||||
result = handle_command(self.assistant, "/unknown")
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestReviewFile:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.tools.read_file")
|
||||
@patch("pr.core.assistant.process_message")
|
||||
def test_review_file_success(self, mock_process, mock_read):
|
||||
mock_read.return_value = {"status": "success", "content": "test content"}
|
||||
review_file(self.assistant, "test.py")
|
||||
mock_read.assert_called_once_with("test.py")
|
||||
mock_process.assert_called_once()
|
||||
args = mock_process.call_args[0]
|
||||
assert "Please review this file" in args[1]
|
||||
|
||||
@patch("pr.tools.read_file")ef test_review_file_error(self, mock_read):
|
||||
mock_read.return_value = {"status": "error", "error": "file not found"}
|
||||
review_file(self.assistant, "test.py")
|
||||
mock_read.assert_called_once_with("test.py")
|
||||
|
||||
|
||||
class TestRefactorFile:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.tools.read_file")
|
||||
@patch("pr.core.assistant.process_message")
|
||||
def test_refactor_file_success(self, mock_process, mock_read):
|
||||
mock_read.return_value = {"status": "success", "content": "test content"}
|
||||
refactor_file(self.assistant, "test.py")
|
||||
mock_process.assert_called_once()
|
||||
args = mock_process.call_args[0]
|
||||
assert "Please refactor this code" in args[1]
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
@patch("pr.tools.read_file") mock_read.return_value = {"status": "error", "error": "file not found"}
|
||||
refactor_file(self.assistant, "test.py")
|
||||
|
||||
|
||||
class TestObfuscateFile:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.tools.read_file")
|
||||
@patch("pr.core.assistant.process_message")
|
||||
def test_obfuscate_file_success(self, mock_process, mock_read):
|
||||
mock_read.return_value = {"status": "success", "content": "test content"}
|
||||
obfuscate_file(self.assistant, "test.py")
|
||||
mock_process.assert_called_once()
|
||||
args = mock_process.call_args[0]
|
||||
assert "Please obfuscate this code" in args[1]
|
||||
|
||||
@patch("pr.commands.handlers.read_file")
|
||||
def test_obfuscate_file_error(self, mock_read):
|
||||
mock_read.return_value = {"status": "error", "error": "file not found"}
|
||||
obfuscate_file(self.assistant, "test.py")
|
||||
|
||||
|
||||
class TestShowWorkflows:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_workflows_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_workflows(self.assistant)
|
||||
|
||||
def test_show_workflows_no_workflows(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_workflow_list.return_value = []
|
||||
show_workflows(self.assistant)
|
||||
|
||||
def test_show_workflows_with_workflows(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_workflow_list.return_value = [
|
||||
{"name": "wf1", "description": "desc1", "execution_count": 5}
|
||||
]
|
||||
show_workflows(self.assistant)
|
||||
|
||||
|
||||
class TestExecuteWorkflowCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_execute_workflow_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
execute_workflow_command(self.assistant, "test")
|
||||
|
||||
def test_execute_workflow_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.execute_workflow.return_value = {
|
||||
"execution_id": "123",
|
||||
"results": {"key": "value"},
|
||||
}
|
||||
execute_workflow_command(self.assistant, "test")
|
||||
|
||||
def test_execute_workflow_error(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.execute_workflow.return_value = {"error": "test error"}
|
||||
execute_workflow_command(self.assistant, "test")
|
||||
|
||||
|
||||
class TestExecuteAgentTask:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_execute_agent_task_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
execute_agent_task(self.assistant, "coding", "task")
|
||||
|
||||
def test_execute_agent_task_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.create_agent.return_value = "agent123"
|
||||
self.assistant.enhanced.agent_task.return_value = {"response": "done"}
|
||||
execute_agent_task(self.assistant, "coding", "task")
|
||||
|
||||
def test_execute_agent_task_error(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.create_agent.return_value = "agent123"
|
||||
self.assistant.enhanced.agent_task.return_value = {"error": "test error"}
|
||||
execute_agent_task(self.assistant, "coding", "task")
|
||||
|
||||
|
||||
class TestShowAgents:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_agents_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_agents(self.assistant)
|
||||
|
||||
def test_show_agents_with_agents(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_agent_summary.return_value = {
|
||||
"active_agents": 2,
|
||||
"agents": [{"agent_id": "a1", "role": "coding", "task_count": 3, "message_count": 10}],
|
||||
}
|
||||
show_agents(self.assistant)
|
||||
|
||||
|
||||
class TestCollaborateAgentsCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_collaborate_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
collaborate_agents_command(self.assistant, "task")
|
||||
|
||||
def test_collaborate_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.collaborate_agents.return_value = {
|
||||
"orchestrator": {"response": "orchestrator response"},
|
||||
"agents": [{"role": "coding", "response": "coding response"}],
|
||||
}
|
||||
collaborate_agents_command(self.assistant, "task")
|
||||
|
||||
|
||||
class TestSearchKnowledge:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_search_knowledge_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
search_knowledge(self.assistant, "query")
|
||||
|
||||
def test_search_knowledge_no_results(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.search_knowledge.return_value = []
|
||||
search_knowledge(self.assistant, "query")
|
||||
|
||||
def test_search_knowledge_with_results(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
mock_entry = Mock()
|
||||
mock_entry.category = "general"
|
||||
mock_entry.content = "long content here"
|
||||
mock_entry.access_count = 5
|
||||
self.assistant.enhanced.search_knowledge.return_value = [mock_entry]
|
||||
search_knowledge(self.assistant, "query")
|
||||
|
||||
|
||||
class TestStoreKnowledge:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_store_knowledge_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
store_knowledge(self.assistant, "content")
|
||||
|
||||
@patch("pr.memory.KnowledgeEntry")
|
||||
def test_store_knowledge_success(self, mock_entry):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.fact_extractor.categorize_content.return_value = ["general"]
|
||||
self.assistant.enhanced.knowledge_store = Mock()
|
||||
store_knowledge(self.assistant, "content")
|
||||
mock_entry.assert_called_once()
|
||||
|
||||
|
||||
class TestShowConversationHistory:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_history_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_conversation_history(self.assistant)
|
||||
|
||||
def test_show_history_no_history(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_conversation_history.return_value = []
|
||||
show_conversation_history(self.assistant)
|
||||
|
||||
def test_show_history_with_history(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_conversation_history.return_value = [
|
||||
{
|
||||
"conversation_id": "conv1",
|
||||
"started_at": 1234567890,
|
||||
"message_count": 5,
|
||||
"summary": "test summary",
|
||||
"topics": ["topic1", "topic2"],
|
||||
}
|
||||
]
|
||||
show_conversation_history(self.assistant)
|
||||
|
||||
|
||||
class TestShowCacheStats:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_cache_stats_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_cache_stats(self.assistant)
|
||||
|
||||
def test_show_cache_stats_with_stats(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_cache_statistics.return_value = {
|
||||
"api_cache": {
|
||||
"total_entries": 10,
|
||||
"valid_entries": 8,
|
||||
"expired_entries": 2,
|
||||
"total_cached_tokens": 1000,
|
||||
"total_cache_hits": 50,
|
||||
},
|
||||
"tool_cache": {
|
||||
"total_entries": 5,
|
||||
"valid_entries": 5,
|
||||
"total_cache_hits": 20,
|
||||
"by_tool": {"tool1": {"cached_entries": 3, "total_hits": 10}},
|
||||
},
|
||||
}
|
||||
show_cache_stats(self.assistant)
|
||||
|
||||
|
||||
class TestClearCaches:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_clear_caches_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
clear_caches(self.assistant)
|
||||
|
||||
def test_clear_caches_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
clear_caches(self.assistant)
|
||||
self.assistant.enhanced.clear_caches.assert_called_once()
|
||||
|
||||
|
||||
class TestShowSystemStats:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_show_system_stats_no_enhanced(self):
|
||||
delattr(self.assistant, "enhanced")
|
||||
show_system_stats(self.assistant)
|
||||
|
||||
def test_show_system_stats_success(self):
|
||||
self.assistant.enhanced = Mock()
|
||||
self.assistant.enhanced.get_cache_statistics.return_value = {
|
||||
"api_cache": {"valid_entries": 10},
|
||||
"tool_cache": {"valid_entries": 5},
|
||||
}
|
||||
self.assistant.enhanced.get_knowledge_statistics.return_value = {
|
||||
"total_entries": 100,
|
||||
"total_categories": 5,
|
||||
"total_accesses": 200,
|
||||
"vocabulary_size": 1000,
|
||||
}
|
||||
self.assistant.enhanced.get_agent_summary.return_value = {"active_agents": 3}
|
||||
show_system_stats(self.assistant)
|
||||
|
||||
|
||||
class TestHandleBackgroundCommand:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
def test_handle_bg_no_args(self):
|
||||
handle_background_command(self.assistant, "/bg")
|
||||
|
||||
@patch("pr.commands.handlers.start_background_session")
|
||||
def test_handle_bg_start(self, mock_start):
|
||||
handle_background_command(self.assistant, "/bg start ls -la")
|
||||
|
||||
@patch("pr.commands.handlers.list_background_sessions")
|
||||
def test_handle_bg_list(self, mock_list):
|
||||
handle_background_command(self.assistant, "/bg list")
|
||||
|
||||
@patch("pr.commands.handlers.show_session_status")
|
||||
def test_handle_bg_status(self, mock_status):
|
||||
handle_background_command(self.assistant, "/bg status session1")
|
||||
|
||||
@patch("pr.commands.handlers.show_session_output")
|
||||
def test_handle_bg_output(self, mock_output):
|
||||
handle_background_command(self.assistant, "/bg output session1")
|
||||
|
||||
@patch("pr.commands.handlers.send_session_input")
|
||||
def test_handle_bg_input(self, mock_input):
|
||||
handle_background_command(self.assistant, "/bg input session1 test input")
|
||||
|
||||
@patch("pr.commands.handlers.kill_background_session")
|
||||
def test_handle_bg_kill(self, mock_kill):
|
||||
handle_background_command(self.assistant, "/bg kill session1")
|
||||
|
||||
@patch("pr.commands.handlers.show_background_events")
|
||||
def test_handle_bg_events(self, mock_events):
|
||||
handle_background_command(self.assistant, "/bg events")
|
||||
|
||||
def test_handle_bg_unknown(self):
|
||||
handle_background_command(self.assistant, "/bg unknown")
|
||||
|
||||
|
||||
class TestStartBackgroundSession:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.start_background_process")
|
||||
def test_start_background_success(self, mock_start):
|
||||
mock_start.return_value = {"status": "success", "pid": 123}
|
||||
start_background_session(self.assistant, "session1", "ls -la")
|
||||
|
||||
@patch("pr.commands.handlers.start_background_process")
|
||||
def test_start_background_error(self, mock_start):
|
||||
mock_start.return_value = {"status": "error", "error": "failed"}
|
||||
start_background_session(self.assistant, "session1", "ls -la")
|
||||
|
||||
@patch("pr.commands.handlers.start_background_process")
|
||||
def test_start_background_exception(self, mock_start):
|
||||
mock_start.side_effect = Exception("test")
|
||||
start_background_session(self.assistant, "session1", "ls -la")
|
||||
|
||||
|
||||
class TestListBackgroundSessions:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.get_all_sessions")
|
||||
@patch("pr.commands.handlers.display_multiplexer_status")
|
||||
def test_list_sessions_success(self, mock_display, mock_get):
|
||||
mock_get.return_value = {}
|
||||
list_background_sessions(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.get_all_sessions")
|
||||
def test_list_sessions_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
list_background_sessions(self.assistant)
|
||||
|
||||
|
||||
class TestShowSessionStatus:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.get_session_info")
|
||||
def test_show_status_found(self, mock_get):
|
||||
mock_get.return_value = {
|
||||
"status": "running",
|
||||
"pid": 123,
|
||||
"command": "ls",
|
||||
"start_time": 1234567890.0,
|
||||
}
|
||||
show_session_status(self.assistant, "session1")
|
||||
|
||||
@patch("pr.commands.handlers.get_session_info")
|
||||
def test_show_status_not_found(self, mock_get):
|
||||
mock_get.return_value = None
|
||||
show_session_status(self.assistant, "session1")
|
||||
|
||||
@patch("pr.commands.handlers.get_session_info")
|
||||
def test_show_status_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
show_session_status(self.assistant, "session1")
|
||||
|
||||
|
||||
class TestShowSessionOutput:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.get_session_output")
|
||||
def test_show_output_success(self, mock_get):
|
||||
mock_get.return_value = ["line1", "line2"]
|
||||
show_session_output(self.assistant, "session1")
|
||||
|
||||
@patch("pr.commands.handlers.get_session_output")
|
||||
def test_show_output_no_output(self, mock_get):
|
||||
mock_get.return_value = None
|
||||
show_session_output(self.assistant, "session1")
|
||||
|
||||
@patch("pr.commands.handlers.get_session_output")
|
||||
def test_show_output_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
show_session_output(self.assistant, "session1")
|
||||
|
||||
|
||||
class TestSendSessionInput:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.send_input_to_session")
|
||||
def test_send_input_success(self, mock_send):
|
||||
mock_send.return_value = {"status": "success"}
|
||||
send_session_input(self.assistant, "session1", "input")
|
||||
|
||||
@patch("pr.commands.handlers.send_input_to_session")
|
||||
def test_send_input_error(self, mock_send):
|
||||
mock_send.return_value = {"status": "error", "error": "failed"}
|
||||
send_session_input(self.assistant, "session1", "input")
|
||||
|
||||
@patch("pr.commands.handlers.send_input_to_session")
|
||||
def test_send_input_exception(self, mock_send):
|
||||
mock_send.side_effect = Exception("test")
|
||||
send_session_input(self.assistant, "session1", "input")
|
||||
|
||||
|
||||
class TestKillBackgroundSession:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.kill_session")
|
||||
def test_kill_success(self, mock_kill):
|
||||
mock_kill.return_value = {"status": "success"}
|
||||
kill_background_session(self.assistant, "session1")
|
||||
|
||||
@patch("pr.commands.handlers.kill_session")
|
||||
def test_kill_error(self, mock_kill):
|
||||
mock_kill.return_value = {"status": "error", "error": "failed"}
|
||||
kill_background_session(self.assistant, "session1")
|
||||
|
||||
@patch("pr.commands.handlers.kill_session")
|
||||
def test_kill_exception(self, mock_kill):
|
||||
mock_kill.side_effect = Exception("test")
|
||||
kill_background_session(self.assistant, "session1")
|
||||
|
||||
|
||||
class TestShowBackgroundEvents:
|
||||
def setup_method(self):
|
||||
self.assistant = Mock()
|
||||
|
||||
@patch("pr.commands.handlers.get_global_monitor")
|
||||
def test_show_events_success(self, mock_get):
|
||||
mock_monitor = Mock()
|
||||
mock_monitor.get_pending_events.return_value = [{"event": "test"}]
|
||||
mock_get.return_value = mock_monitor
|
||||
with patch("pr.commands.handlers.display_background_event"):
|
||||
show_background_events(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.get_global_monitor")
|
||||
def test_show_events_no_events(self, mock_get):
|
||||
mock_monitor = Mock()
|
||||
mock_monitor.get_pending_events.return_value = []
|
||||
mock_get.return_value = mock_monitor
|
||||
show_background_events(self.assistant)
|
||||
|
||||
@patch("pr.commands.handlers.get_global_monitor")
|
||||
def test_show_events_exception(self, mock_get):
|
||||
mock_get.side_effect = Exception("test")
|
||||
show_background_events(self.assistant)
|
||||
@@ -77,9 +77,9 @@ def test_get_cache_statistics():
|
||||
mock_base = MagicMock()
|
||||
assistant = EnhancedAssistant(mock_base)
|
||||
assistant.api_cache = MagicMock()
|
||||
assistant.api_cache.get_statistics.return_value = {"hits": 10}
|
||||
assistant.api_cache.get_statistics.return_value = {"total_cache_hits": 10}
|
||||
assistant.tool_cache = MagicMock()
|
||||
assistant.tool_cache.get_statistics.return_value = {"misses": 5}
|
||||
assistant.tool_cache.get_statistics.return_value = {"total_cache_hits": 5}
|
||||
|
||||
stats = assistant.get_cache_statistics()
|
||||
assert "api_cache" in stats
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
import pytest
|
||||
from pr.core.exceptions import (
|
||||
PRException,
|
||||
APIException,
|
||||
APIConnectionError,
|
||||
APITimeoutError,
|
||||
APIResponseError,
|
||||
ConfigurationError,
|
||||
ToolExecutionError,
|
||||
FileSystemError,
|
||||
SessionError,
|
||||
ContextError,
|
||||
ValidationError,
|
||||
)
|
||||
|
||||
|
||||
class TestExceptions:
|
||||
def test_pre_exception(self):
|
||||
with pytest.raises(PRException):
|
||||
raise PRException("test")
|
||||
|
||||
def test_api_exception(self):
|
||||
with pytest.raises(APIException):
|
||||
raise APIException("test")
|
||||
|
||||
def test_api_connection_error(self):
|
||||
with pytest.raises(APIConnectionError):
|
||||
raise APIConnectionError("test")
|
||||
|
||||
def test_api_timeout_error(self):
|
||||
with pytest.raises(APITimeoutError):
|
||||
raise APITimeoutError("test")
|
||||
|
||||
def test_api_response_error(self):
|
||||
with pytest.raises(APIResponseError):
|
||||
raise APIResponseError("test")
|
||||
|
||||
def test_configuration_error(self):
|
||||
with pytest.raises(ConfigurationError):
|
||||
raise ConfigurationError("test")
|
||||
|
||||
def test_tool_execution_error(self):
|
||||
error = ToolExecutionError("test_tool", "test message")
|
||||
assert error.tool_name == "test_tool"
|
||||
assert str(error) == "Error executing tool 'test_tool': test message"
|
||||
|
||||
def test_file_system_error(self):
|
||||
with pytest.raises(FileSystemError):
|
||||
raise FileSystemError("test")
|
||||
|
||||
def test_session_error(self):
|
||||
with pytest.raises(SessionError):
|
||||
raise SessionError("test")
|
||||
|
||||
def test_context_error(self):
|
||||
with pytest.raises(ContextError):
|
||||
raise ContextError("test")
|
||||
|
||||
def test_validation_error(self):
|
||||
with pytest.raises(ValidationError):
|
||||
raise ValidationError("test")
|
||||
@@ -0,0 +1,46 @@
|
||||
from pr.commands.help_docs import (
|
||||
get_workflow_help,
|
||||
get_agent_help,
|
||||
get_knowledge_help,
|
||||
get_cache_help,
|
||||
get_background_help,
|
||||
get_full_help,
|
||||
)
|
||||
|
||||
|
||||
class TestHelpDocs:
|
||||
def test_get_workflow_help(self):
|
||||
result = get_workflow_help()
|
||||
assert isinstance(result, str)
|
||||
assert "WORKFLOWS" in result
|
||||
assert "AUTOMATED TASK EXECUTION" in result
|
||||
|
||||
def test_get_agent_help(self):
|
||||
result = get_agent_help()
|
||||
assert isinstance(result, str)
|
||||
assert "AGENTS" in result
|
||||
assert "SPECIALIZED AI ASSISTANTS" in result
|
||||
|
||||
def test_get_knowledge_help(self):
|
||||
result = get_knowledge_help()
|
||||
assert isinstance(result, str)
|
||||
assert "KNOWLEDGE BASE" in result
|
||||
assert "PERSISTENT INFORMATION STORAGE" in result
|
||||
|
||||
def test_get_cache_help(self):
|
||||
result = get_cache_help()
|
||||
assert isinstance(result, str)
|
||||
assert "CACHING SYSTEM" in result
|
||||
assert "PERFORMANCE OPTIMIZATION" in result
|
||||
|
||||
def test_get_background_help(self):
|
||||
result = get_background_help()
|
||||
assert isinstance(result, str)
|
||||
assert "BACKGROUND SESSIONS" in result
|
||||
assert "CONCURRENT TASK EXECUTION" in result
|
||||
|
||||
def test_get_full_help(self):
|
||||
result = get_full_help()
|
||||
assert isinstance(result, str)
|
||||
assert "R - PROFESSIONAL AI ASSISTANT" in result
|
||||
assert "BASIC COMMANDS" in result
|
||||
+69
-17
@@ -1,25 +1,77 @@
|
||||
from pr.core.logging import get_logger, setup_logging
|
||||
from unittest.mock import patch, MagicMock
|
||||
from pr.core.logging import setup_logging, get_logger
|
||||
|
||||
|
||||
def test_setup_logging_basic():
|
||||
logger = setup_logging(verbose=False)
|
||||
assert logger.name == "pr"
|
||||
assert logger.level == 20 # INFO
|
||||
class TestLogging:
|
||||
@patch("pr.core.logging.os.makedirs")
|
||||
@patch("pr.core.logging.os.path.dirname")
|
||||
@patch("pr.core.logging.os.path.exists")
|
||||
@patch("pr.core.logging.RotatingFileHandler")
|
||||
@patch("pr.core.logging.logging.getLogger")
|
||||
def test_setup_logging_basic(
|
||||
self, mock_get_logger, mock_handler, mock_exists, mock_dirname, mock_makedirs
|
||||
):
|
||||
mock_exists.return_value = False
|
||||
mock_dirname.return_value = "/tmp/logs"
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
mock_logger.handlers = []
|
||||
|
||||
result = setup_logging(verbose=False)
|
||||
|
||||
def test_setup_logging_verbose():
|
||||
logger = setup_logging(verbose=True)
|
||||
assert logger.name == "pr"
|
||||
assert logger.level == 10 # DEBUG
|
||||
# Should have console handler
|
||||
assert len(logger.handlers) >= 2
|
||||
mock_makedirs.assert_called_once_with("/tmp/logs", exist_ok=True)
|
||||
mock_get_logger.assert_called_once_with("pr")
|
||||
mock_logger.setLevel.assert_called_once_with(20) # INFO level
|
||||
mock_handler.assert_called_once()
|
||||
assert result == mock_logger
|
||||
|
||||
@patch("pr.core.logging.os.makedirs")
|
||||
@patch("pr.core.logging.os.path.dirname")
|
||||
@patch("pr.core.logging.os.path.exists")
|
||||
@patch("pr.core.logging.RotatingFileHandler")
|
||||
@patch("pr.core.logging.logging.StreamHandler")
|
||||
@patch("pr.core.logging.logging.getLogger")
|
||||
def test_setup_logging_verbose(
|
||||
self,
|
||||
mock_get_logger,
|
||||
mock_stream_handler,
|
||||
mock_file_handler,
|
||||
mock_exists,
|
||||
mock_dirname,
|
||||
mock_makedirs,
|
||||
):
|
||||
mock_exists.return_value = True
|
||||
mock_dirname.return_value = "/tmp/logs"
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
mock_logger.handlers = MagicMock()
|
||||
|
||||
def test_get_logger_default():
|
||||
logger = get_logger()
|
||||
assert logger.name == "pr"
|
||||
result = setup_logging(verbose=True)
|
||||
|
||||
mock_makedirs.assert_not_called()
|
||||
mock_get_logger.assert_called_once_with("pr")
|
||||
mock_logger.setLevel.assert_called_once_with(10) # DEBUG level
|
||||
mock_logger.handlers.clear.assert_called_once()
|
||||
mock_file_handler.assert_called_once()
|
||||
mock_stream_handler.assert_called_once()
|
||||
assert result == mock_logger
|
||||
|
||||
def test_get_logger_named():
|
||||
logger = get_logger("test")
|
||||
assert logger.name == "pr.test"
|
||||
@patch("pr.core.logging.logging.getLogger")
|
||||
def test_get_logger_default(self, mock_get_logger):
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
result = get_logger()
|
||||
|
||||
mock_get_logger.assert_called_once_with("pr")
|
||||
assert result == mock_logger
|
||||
|
||||
@patch("pr.core.logging.logging.getLogger")
|
||||
def test_get_logger_named(self, mock_get_logger):
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
result = get_logger("test")
|
||||
|
||||
mock_get_logger.assert_called_once_with("pr.test")
|
||||
assert result == mock_logger
|
||||
|
||||
@@ -0,0 +1,228 @@
|
||||
from unittest.mock import Mock, patch
|
||||
from pr.commands.multiplexer_commands import (
|
||||
show_sessions,
|
||||
attach_session,
|
||||
detach_session,
|
||||
kill_session,
|
||||
send_command,
|
||||
show_session_log,
|
||||
show_session_status,
|
||||
list_waiting_sessions,
|
||||
)
|
||||
|
||||
|
||||
class TestShowSessions:
|
||||
@patch("pr.commands.multiplexer_commands.list_active_sessions")
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
def test_show_sessions_no_sessions(self, mock_status, mock_list):
|
||||
mock_list.return_value = {}
|
||||
show_sessions()
|
||||
mock_list.assert_called_once()
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.list_active_sessions")
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
def test_show_sessions_with_sessions(self, mock_status, mock_list):
|
||||
mock_list.return_value = {
|
||||
"session1": {
|
||||
"metadata": {
|
||||
"process_type": "test",
|
||||
"start_time": 123.0,
|
||||
"interaction_count": 5,
|
||||
"state": "running",
|
||||
},
|
||||
"output_summary": {"stdout_lines": 10, "stderr_lines": 2},
|
||||
}
|
||||
}
|
||||
mock_status.return_value = {"is_active": True, "pid": 123}
|
||||
show_sessions()
|
||||
mock_list.assert_called_once()
|
||||
mock_status.assert_called_once_with("session1")
|
||||
|
||||
|
||||
class TestAttachSession:
|
||||
def test_attach_session_no_args(self):
|
||||
attach_session([])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
def test_attach_session_not_found(self, mock_status):
|
||||
mock_status.return_value = None
|
||||
attach_session(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
@patch("pr.commands.multiplexer_commands.read_session_output")
|
||||
def test_attach_session_success(self, mock_read, mock_status):
|
||||
mock_status.return_value = {"is_active": True, "metadata": {"process_type": "test"}}
|
||||
mock_read.return_value = {"stdout": "line1\nline2", "stderr": ""}
|
||||
attach_session(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
@patch("pr.commands.multiplexer_commands.read_session_output")
|
||||
def test_attach_session_with_stderr(self, mock_read, mock_status):
|
||||
mock_status.return_value = {"is_active": False, "metadata": {"process_type": "test"}}
|
||||
mock_read.return_value = {"stdout": "", "stderr": "error1\nerror2"}
|
||||
attach_session(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
@patch("pr.commands.multiplexer_commands.read_session_output")
|
||||
def test_attach_session_read_error(self, mock_read, mock_status):
|
||||
mock_status.return_value = {"is_active": True, "metadata": {"process_type": "test"}}
|
||||
mock_read.side_effect = Exception("test error")
|
||||
attach_session(["session1"])
|
||||
|
||||
|
||||
class TestDetachSession:
|
||||
def test_detach_session_no_args(self):
|
||||
detach_session([])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_multiplexer")
|
||||
def test_detach_session_not_found(self, mock_get):
|
||||
mock_get.return_value = None
|
||||
detach_session(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_multiplexer")
|
||||
def test_detach_session_success(self, mock_get):
|
||||
mock_mux = Mock()
|
||||
mock_get.return_value = mock_mux
|
||||
detach_session(["session1"])
|
||||
assert mock_mux.show_output is False
|
||||
|
||||
|
||||
class TestKillSession:
|
||||
def test_kill_session_no_args(self):
|
||||
kill_session([])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.close_interactive_session")
|
||||
def test_kill_session_success(self, mock_close):
|
||||
kill_session(["session1"])
|
||||
mock_close.assert_called_once_with("session1")
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.close_interactive_session")
|
||||
def test_kill_session_error(self, mock_close):
|
||||
mock_close.side_effect = Exception("test error")
|
||||
kill_session(["session1"])
|
||||
|
||||
|
||||
class TestSendCommand:
|
||||
def test_send_command_no_args(self):
|
||||
send_command([])
|
||||
|
||||
def test_send_command_insufficient_args(self):
|
||||
send_command(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.send_input_to_session")
|
||||
def test_send_command_success(self, mock_send):
|
||||
send_command(["session1", "ls", "-la"])
|
||||
mock_send.assert_called_once_with("session1", "ls -la")
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.send_input_to_session")
|
||||
def test_send_command_error(self, mock_send):
|
||||
mock_send.side_effect = Exception("test error")
|
||||
send_command(["session1", "ls"])
|
||||
|
||||
|
||||
class TestShowSessionLog:
|
||||
def test_show_session_log_no_args(self):
|
||||
show_session_log([])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.read_session_output")
|
||||
def test_show_session_log_success(self, mock_read):
|
||||
mock_read.return_value = {"stdout": "stdout content", "stderr": "stderr content"}
|
||||
show_session_log(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.read_session_output")
|
||||
def test_show_session_log_no_stderr(self, mock_read):
|
||||
mock_read.return_value = {"stdout": "stdout content", "stderr": ""}
|
||||
show_session_log(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.read_session_output")
|
||||
def test_show_session_log_error(self, mock_read):
|
||||
mock_read.side_effect = Exception("test error")
|
||||
show_session_log(["session1"])
|
||||
|
||||
|
||||
class TestShowSessionStatus:
|
||||
def test_show_session_status_no_args(self):
|
||||
show_session_status([])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
def test_show_session_status_not_found(self, mock_status):
|
||||
mock_status.return_value = None
|
||||
show_session_status(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
@patch("pr.commands.multiplexer_commands.get_global_detector")
|
||||
def test_show_session_status_success(self, mock_detector, mock_status):
|
||||
mock_status.return_value = {
|
||||
"is_active": True,
|
||||
"pid": 123,
|
||||
"metadata": {
|
||||
"process_type": "test",
|
||||
"start_time": 123.0,
|
||||
"last_activity": 456.0,
|
||||
"interaction_count": 5,
|
||||
"state": "running",
|
||||
},
|
||||
"output_summary": {"stdout_lines": 10, "stderr_lines": 2},
|
||||
}
|
||||
mock_detector_instance = Mock()
|
||||
mock_detector_instance.get_session_info.return_value = {
|
||||
"current_state": "waiting",
|
||||
"is_waiting": True,
|
||||
}
|
||||
mock_detector.return_value = mock_detector_instance
|
||||
show_session_status(["session1"])
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
@patch("pr.commands.multiplexer_commands.get_global_detector")
|
||||
def test_show_session_status_no_detector_info(self, mock_detector, mock_status):
|
||||
mock_status.return_value = {
|
||||
"is_active": False,
|
||||
"metadata": {
|
||||
"process_type": "test",
|
||||
"start_time": 123.0,
|
||||
"last_activity": 456.0,
|
||||
"interaction_count": 5,
|
||||
"state": "running",
|
||||
},
|
||||
"output_summary": {"stdout_lines": 10, "stderr_lines": 2},
|
||||
}
|
||||
mock_detector_instance = Mock()
|
||||
mock_detector_instance.get_session_info.return_value = None
|
||||
mock_detector.return_value = mock_detector_instance
|
||||
show_session_status(["session1"])
|
||||
|
||||
|
||||
class TestListWaitingSessions:
|
||||
@patch("pr.commands.multiplexer_commands.list_active_sessions")
|
||||
@patch("pr.commands.multiplexer_commands.get_global_detector")
|
||||
def test_list_waiting_sessions_no_sessions(self, mock_detector, mock_list):
|
||||
mock_list.return_value = {}
|
||||
mock_detector_instance = Mock()
|
||||
mock_detector_instance.is_waiting_for_input.return_value = False
|
||||
mock_detector.return_value = mock_detector_instance
|
||||
list_waiting_sessions()
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.list_active_sessions")
|
||||
@patch("pr.commands.multiplexer_commands.get_session_status")
|
||||
@patch("pr.commands.multiplexer_commands.get_global_detector")
|
||||
def test_list_waiting_sessions_with_waiting(self, mock_detector, mock_status, mock_list):
|
||||
mock_list.return_value = ["session1"]
|
||||
mock_detector_instance = Mock()
|
||||
mock_detector_instance.is_waiting_for_input.return_value = True
|
||||
mock_detector_instance.get_session_info.return_value = {
|
||||
"current_state": "waiting",
|
||||
"is_waiting": True,
|
||||
}
|
||||
mock_detector_instance.get_response_suggestions.return_value = ["yes", "no", "quit"]
|
||||
mock_detector.return_value = mock_detector_instance
|
||||
mock_status.return_value = {"metadata": {"process_type": "test"}}
|
||||
list_waiting_sessions()
|
||||
|
||||
@patch("pr.commands.multiplexer_commands.list_active_sessions")
|
||||
@patch("pr.commands.multiplexer_commands.get_global_detector")
|
||||
def test_list_waiting_sessions_no_waiting(self, mock_detector, mock_list):
|
||||
mock_list.return_value = ["session1"]
|
||||
mock_detector_instance = Mock()
|
||||
mock_detector_instance.is_waiting_for_input.return_value = False
|
||||
mock_detector.return_value = mock_detector_instance
|
||||
list_waiting_sessions()
|
||||
+68
-1
@@ -1,8 +1,12 @@
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from pr.tools.base import get_tools_definition
|
||||
from pr.tools.filesystem import list_directory, read_file, search_replace, write_file
|
||||
from pr.tools.command import run_command
|
||||
from pr.tools.filesystem import chdir, getpwd, list_directory, read_file, search_replace, write_file
|
||||
from pr.tools.interactive_control import start_interactive_session
|
||||
from pr.tools.patch import apply_patch, create_diff
|
||||
from pr.tools.python_exec import python_exec
|
||||
|
||||
|
||||
class TestFilesystemTools:
|
||||
@@ -43,6 +47,54 @@ class TestFilesystemTools:
|
||||
read_result = read_file(filepath)
|
||||
assert "Hello, Universe!" in read_result["content"]
|
||||
|
||||
def test_chdir_and_getpwd(self):
|
||||
original_cwd = getpwd()["path"]
|
||||
try:
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
result = chdir(temp_dir)
|
||||
assert result["status"] == "success"
|
||||
assert getpwd()["path"] == temp_dir
|
||||
finally:
|
||||
chdir(original_cwd)
|
||||
|
||||
|
||||
class TestCommandTools:
|
||||
|
||||
def test_run_command_with_cwd(self):
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
result = run_command("pwd", cwd=temp_dir)
|
||||
assert result["status"] == "success"
|
||||
assert temp_dir in result["stdout"].strip()
|
||||
|
||||
def test_run_command_basic(self):
|
||||
result = run_command("echo hello")
|
||||
assert result["status"] == "success"
|
||||
assert "hello" in result["stdout"]
|
||||
|
||||
|
||||
class TestInteractiveSessionTools:
|
||||
|
||||
def test_start_interactive_session_with_cwd(self):
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
session_name = start_interactive_session("pwd", cwd=temp_dir)
|
||||
assert session_name is not None
|
||||
# Note: In a real test, we'd need to interact with the session, but for now just check it starts
|
||||
|
||||
|
||||
class TestPythonExecTools:
|
||||
|
||||
def test_python_exec_with_cwd(self):
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
code = "import os; print(os.getcwd())"
|
||||
result = python_exec(code, {}, cwd=temp_dir)
|
||||
assert result["status"] == "success"
|
||||
assert temp_dir in result["output"].strip()
|
||||
|
||||
def test_python_exec_basic(self):
|
||||
result = python_exec("print('hello')", {})
|
||||
assert result["status"] == "success"
|
||||
assert "hello" in result["output"]
|
||||
|
||||
|
||||
class TestPatchTools:
|
||||
|
||||
@@ -108,6 +160,21 @@ class TestToolDefinitions:
|
||||
assert "write_file" in tool_names
|
||||
assert "list_directory" in tool_names
|
||||
assert "search_replace" in tool_names
|
||||
assert "chdir" in tool_names
|
||||
assert "getpwd" in tool_names
|
||||
|
||||
def test_command_tools_present(self):
|
||||
tools = get_tools_definition()
|
||||
tool_names = [t["function"]["name"] for t in tools]
|
||||
|
||||
assert "run_command" in tool_names
|
||||
assert "start_interactive_session" in tool_names
|
||||
|
||||
def test_python_exec_present(self):
|
||||
tools = get_tools_definition()
|
||||
tool_names = [t["function"]["name"] for t in tools]
|
||||
|
||||
assert "python_exec" in tool_names
|
||||
|
||||
def test_patch_tools_present(self):
|
||||
tools = get_tools_definition()
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
from pr.ui.output import OutputFormatter
|
||||
|
||||
|
||||
class TestOutputFormatter:
|
||||
def test_init(self):
|
||||
formatter = OutputFormatter()
|
||||
assert formatter.format_type == "text"
|
||||
assert formatter.quiet is False
|
||||
|
||||
formatter = OutputFormatter("json", True)
|
||||
assert formatter.format_type == "json"
|
||||
assert formatter.quiet is True
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_text_quiet(self, mock_print):
|
||||
formatter = OutputFormatter(quiet=True)
|
||||
formatter.output("test", "response")
|
||||
mock_print.assert_not_called()
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_text_error_quiet(self, mock_print):
|
||||
formatter = OutputFormatter(quiet=True)
|
||||
formatter.output("test", "error")
|
||||
mock_print.assert_called()
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_text_result_quiet(self, mock_print):
|
||||
formatter = OutputFormatter(quiet=True)
|
||||
formatter.output("test", "result")
|
||||
mock_print.assert_called()
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_json(self, mock_print):
|
||||
formatter = OutputFormatter("json")
|
||||
formatter.output("test data", "response")
|
||||
args = mock_print.call_args[0][0]
|
||||
data = json.loads(args)
|
||||
assert data["type"] == "response"
|
||||
assert data["data"] == "test data"
|
||||
assert "timestamp" in data
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_structured_dict(self, mock_print):
|
||||
formatter = OutputFormatter("structured")
|
||||
formatter.output({"key": "value", "key2": "value2"}, "response")
|
||||
assert mock_print.call_count == 2
|
||||
calls = [call[0][0] for call in mock_print.call_args_list]
|
||||
assert "key: value" in calls
|
||||
assert "key2: value2" in calls
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_structured_list(self, mock_print):
|
||||
formatter = OutputFormatter("structured")
|
||||
formatter.output(["item1", "item2"], "response")
|
||||
assert mock_print.call_count == 2
|
||||
calls = [call[0][0] for call in mock_print.call_args_list]
|
||||
assert "- item1" in calls
|
||||
assert "- item2" in calls
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_structured_other(self, mock_print):
|
||||
formatter = OutputFormatter("structured")
|
||||
formatter.output("plain text", "response")
|
||||
mock_print.assert_called_once_with("plain text")
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_text_dict(self, mock_print):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.output({"key": "value"}, "response")
|
||||
args = mock_print.call_args[0][0]
|
||||
data = json.loads(args)
|
||||
assert data == {"key": "value"}
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_text_list(self, mock_print):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.output(["item1"], "response")
|
||||
args = mock_print.call_args[0][0]
|
||||
data = json.loads(args)
|
||||
assert data == ["item1"]
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_output_text_other(self, mock_print):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.output("plain text", "response")
|
||||
mock_print.assert_called_once_with("plain text")
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_error_json(self, mock_print):
|
||||
formatter = OutputFormatter("json")
|
||||
formatter.error("test error")
|
||||
args = mock_print.call_args[0][0]
|
||||
data = json.loads(args)
|
||||
assert data["type"] == "error"
|
||||
assert data["data"]["error"] == "test error"
|
||||
|
||||
@patch("sys.stderr")
|
||||
def test_error_text(self, mock_stderr):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.error("test error")
|
||||
mock_stderr.write.assert_called()
|
||||
calls = mock_stderr.write.call_args_list
|
||||
assert any("Error: test error" in call[0][0] for call in calls)
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_success_json(self, mock_print):
|
||||
formatter = OutputFormatter("json")
|
||||
formatter.success("test success")
|
||||
args = mock_print.call_args[0][0]
|
||||
data = json.loads(args)
|
||||
assert data["type"] == "success"
|
||||
assert data["data"]["success"] == "test success"
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_success_text(self, mock_print):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.success("test success")
|
||||
mock_print.assert_called_once_with("test success")
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_success_quiet(self, mock_print):
|
||||
formatter = OutputFormatter("text", quiet=True)
|
||||
formatter.success("test success")
|
||||
mock_print.assert_not_called()
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_info_json(self, mock_print):
|
||||
formatter = OutputFormatter("json")
|
||||
formatter.info("test info")
|
||||
args = mock_print.call_args[0][0]
|
||||
data = json.loads(args)
|
||||
assert data["type"] == "info"
|
||||
assert data["data"]["info"] == "test info"
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_info_text(self, mock_print):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.info("test info")
|
||||
mock_print.assert_called_once_with("test info")
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_info_quiet(self, mock_print):
|
||||
formatter = OutputFormatter("text", quiet=True)
|
||||
formatter.info("test info")
|
||||
mock_print.assert_not_called()
|
||||
|
||||
@patch("builtins.print")
|
||||
def test_result(self, mock_print):
|
||||
formatter = OutputFormatter("text")
|
||||
formatter.result("test result")
|
||||
mock_print.assert_called_once_with("test result")
|
||||
Reference in New Issue
Block a user