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)