MCP Odoo Shell

""" Unit tests for OdooShellManager class. """ import queue import subprocess import threading import time from unittest.mock import MagicMock, Mock, patch import pytest from mcp_odoo_shell.shell_manager import OdooShellManager class TestOdooShellManagerInit: """Test OdooShellManager initialization.""" def test_init_with_all_params(self, shell_manager_args): """Test initialization with all parameters.""" manager = OdooShellManager(**shell_manager_args) assert manager.odoo_bin_path == "/fake/odoo-bin" assert manager.addons_path == "/fake/addons" assert manager.db_name == "test_db" assert manager.config_file == "/fake/config.conf" assert manager.process is None assert isinstance(manager.input_queue, queue.Queue) assert isinstance(manager.output_queue, queue.Queue) assert isinstance(manager.session_vars, dict) assert len(manager.session_vars) == 0 def test_init_without_config_file(self, shell_manager_args): """Test initialization without config file.""" del shell_manager_args["config_file"] manager = OdooShellManager(**shell_manager_args) assert manager.config_file is None def test_command_construction_with_config(self, shell_manager_args): """Test that command is constructed correctly with config file.""" manager = OdooShellManager(**shell_manager_args) with patch('subprocess.Popen') as mock_popen: mock_process = Mock() mock_popen.return_value = mock_process with patch('threading.Thread'): with patch.object(manager, '_wait_for_prompt'): manager.start_shell() # Verify subprocess.Popen was called with correct arguments expected_cmd = [ "/fake/odoo-bin", "shell", "--addons-path", "/fake/addons", "--database", "test_db", "--no-http", "--config", "/fake/config.conf" ] mock_popen.assert_called_once() call_args = mock_popen.call_args assert call_args[0][0] == expected_cmd def test_command_construction_without_config(self, shell_manager_args): """Test that command is constructed correctly without config file.""" del shell_manager_args["config_file"] manager = OdooShellManager(**shell_manager_args) with patch('subprocess.Popen') as mock_popen: mock_process = Mock() mock_popen.return_value = mock_process with patch('threading.Thread'): with patch.object(manager, '_wait_for_prompt'): manager.start_shell() expected_cmd = [ "/fake/odoo-bin", "shell", "--addons-path", "/fake/addons", "--database", "test_db", "--no-http" ] mock_popen.assert_called_once() call_args = mock_popen.call_args assert call_args[0][0] == expected_cmd class TestOdooShellManagerLifecycle: """Test shell lifecycle management.""" def test_start_shell_creates_process(self, shell_manager_args, mock_subprocess): """Test that start_shell creates a process and threads.""" manager = OdooShellManager(**shell_manager_args) with patch('threading.Thread') as mock_thread: with patch.object(manager, '_wait_for_prompt'): manager.start_shell() assert manager.process is mock_subprocess # Should create 2 threads (input and output) assert mock_thread.call_count == 2 def test_stop_terminates_process(self, shell_manager_args): """Test that stop terminates the process.""" manager = OdooShellManager(**shell_manager_args) mock_process = Mock() manager.process = mock_process manager.stop() mock_process.terminate.assert_called_once() mock_process.wait.assert_called_once() def test_stop_handles_no_process(self, shell_manager_args): """Test that stop handles case when no process exists.""" manager = OdooShellManager(**shell_manager_args) # No process set # Should not raise exception manager.stop() def test_wait_for_prompt_success(self, shell_manager_args): """Test successful prompt detection.""" manager = OdooShellManager(**shell_manager_args) # Simulate prompt appearing in output manager.output_queue.put("Starting Odoo shell...\n") manager.output_queue.put(">>>\n") # Should complete without timeout manager._wait_for_prompt() def test_wait_for_prompt_ipython_style(self, shell_manager_args): """Test prompt detection with IPython-style prompt.""" manager = OdooShellManager(**shell_manager_args) # Simulate IPython prompt manager.output_queue.put("In [1]: ") # Should complete without timeout manager._wait_for_prompt() def test_wait_for_prompt_timeout(self, shell_manager_args): """Test timeout when prompt doesn't appear.""" manager = OdooShellManager(**shell_manager_args) # No output in queue, should timeout with pytest.raises(TimeoutError, match="Timeout waiting for Odoo shell prompt"): manager._wait_for_prompt() class TestThreads: """Test threading behavior.""" def test_input_thread_sends_code(self, shell_manager_args, mock_subprocess): """Test that input thread sends code to process stdin.""" manager = OdooShellManager(**shell_manager_args) manager.process = mock_subprocess # Put code in input queue manager.input_queue.put("print('test')") manager.input_queue.put(None) # Signal to stop # Run input thread method directly with patch.object(manager.process, 'poll', return_value=None): # Simulate one iteration try: code = manager.input_queue.get(timeout=1) if code: manager.process.stdin.write(code + '\n') manager.process.stdin.flush() except queue.Empty: pass mock_subprocess.stdin.write.assert_called_with("print('test')\n") mock_subprocess.stdin.flush.assert_called_once() def test_output_thread_reads_data(self, shell_manager_args, mock_subprocess): """Test that output thread reads data from process stdout.""" manager = OdooShellManager(**shell_manager_args) manager.process = mock_subprocess # Mock stdout to return characters mock_subprocess.stdout.read.side_effect = ['h', 'e', 'l', 'l', 'o', '\n', ''] # Run a few iterations of output thread logic buffer = [] for _ in range(6): # Read 'hello\n' try: char = manager.process.stdout.read(1) if char: buffer.append(char) if char == '\n' or len(buffer) > 1000: output = ''.join(buffer) manager.output_queue.put(output) buffer = [] except Exception: break # Check that output was queued assert not manager.output_queue.empty() output = manager.output_queue.get() assert output == "hello\n"