|
from unittest.mock import Mock, patch
|
|
|
|
from rp.commands.handlers import (
|
|
clear_caches,
|
|
collaborate_agents_command,
|
|
execute_agent_task,
|
|
execute_workflow_command,
|
|
handle_background_command,
|
|
handle_command,
|
|
kill_background_session,
|
|
list_background_sessions,
|
|
obfuscate_file,
|
|
refactor_file,
|
|
review_file,
|
|
search_knowledge,
|
|
send_session_input,
|
|
show_agents,
|
|
show_background_events,
|
|
show_cache_stats,
|
|
show_conversation_history,
|
|
show_session_output,
|
|
show_session_status,
|
|
show_system_stats,
|
|
show_workflows,
|
|
start_background_session,
|
|
store_knowledge,
|
|
)
|
|
|
|
|
|
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("rp.commands.handlers.run_autonomous_mode")
|
|
def test_handle_edit(self, mock_run):
|
|
with patch("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.commands.handlers.read_file")
|
|
@patch("rp.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("rp.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("rp.commands.handlers.read_file")
|
|
@patch("rp.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("rp.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("rp.commands.handlers.read_file")
|
|
@patch("rp.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("rp.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("rp.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("rp.commands.handlers.start_background_session")
|
|
def test_handle_bg_start(self, mock_start):
|
|
handle_background_command(self.assistant, "/bg start ls -la")
|
|
|
|
@patch("rp.commands.handlers.list_background_sessions")
|
|
def test_handle_bg_list(self, mock_list):
|
|
handle_background_command(self.assistant, "/bg list")
|
|
|
|
@patch("rp.commands.handlers.show_session_status")
|
|
def test_handle_bg_status(self, mock_status):
|
|
handle_background_command(self.assistant, "/bg status session1")
|
|
|
|
@patch("rp.commands.handlers.show_session_output")
|
|
def test_handle_bg_output(self, mock_output):
|
|
handle_background_command(self.assistant, "/bg output session1")
|
|
|
|
@patch("rp.commands.handlers.send_session_input")
|
|
def test_handle_bg_input(self, mock_input):
|
|
handle_background_command(self.assistant, "/bg input session1 test input")
|
|
|
|
@patch("rp.commands.handlers.kill_background_session")
|
|
def test_handle_bg_kill(self, mock_kill):
|
|
handle_background_command(self.assistant, "/bg kill session1")
|
|
|
|
@patch("rp.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("rp.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("rp.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("rp.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("rp.multiplexer.get_all_sessions")
|
|
@patch("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.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("rp.multiplexer.kill_session")
|
|
def test_kill_success(self, mock_kill):
|
|
mock_kill.return_value = {"status": "success"}
|
|
kill_background_session(self.assistant, "session1")
|
|
|
|
@patch("rp.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("rp.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("rp.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("rp.ui.display.display_background_event"):
|
|
show_background_events(self.assistant)
|
|
|
|
@patch("rp.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("rp.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)
|
|
|