MCP Odoo Shell

""" Pytest configuration and fixtures for testing the Odoo Shell MCP server. """ import queue import threading import time from typing import Dict, List, Optional from unittest.mock import MagicMock, Mock import pytest class FakeShellProcess: """ Fake shell process that simulates Odoo shell behavior for testing. """ def __init__(self): self.stdin = Mock() self.stdout = Mock() self.stderr = Mock() self.responses: Dict[str, str] = {} self.input_history: List[str] = [] self.is_running = True self.delay = 0.0 # Simulate processing delay def add_response(self, input_code: str, output: str): """Add a canned response for specific input.""" self.responses[input_code.strip()] = output def poll(self): """Return None if process is running, 0 if terminated.""" return None if self.is_running else 0 def terminate(self): """Terminate the fake process.""" self.is_running = False def wait(self): """Wait for process to terminate.""" while self.is_running: time.sleep(0.01) return 0 def simulate_input(self, code: str): """Simulate receiving input and generating output.""" self.input_history.append(code) if self.delay > 0: time.sleep(self.delay) # Return canned response or default response = self.responses.get(code.strip(), f"Result of: {code}\n>>>") return response @pytest.fixture def fake_process(): """ Fixture providing a fake shell process with common responses. """ process = FakeShellProcess() # Add common shell responses process.add_response("2+2", "4\n>>>") process.add_response("print('hello')", "hello\n>>>") process.add_response("print('test')", "test\n>>>") process.add_response("", ">>>") # Empty input return process @pytest.fixture def mock_subprocess(monkeypatch, fake_process): """ Fixture that mocks subprocess.Popen to return our fake process. """ def mock_popen(*args, **kwargs): return fake_process monkeypatch.setattr("subprocess.Popen", mock_popen) return fake_process @pytest.fixture def shell_manager_args(): """ Fixture providing default arguments for OdooShellManager. """ return { "odoo_bin_path": "/fake/odoo-bin", "addons_path": "/fake/addons", "db_name": "test_db", "config_file": "/fake/config.conf" } @pytest.fixture def queue_manager(): """ Fixture to help manage and inspect queue contents in tests. """ class QueueManager: def __init__(self): self.input_queue = queue.Queue() self.output_queue = queue.Queue() def put_output(self, data: str): """Put data into output queue.""" self.output_queue.put(data) def get_all_input(self) -> List[str]: """Get all items from input queue.""" items = [] while not self.input_queue.empty(): try: items.append(self.input_queue.get_nowait()) except queue.Empty: break return items def get_all_output(self) -> List[str]: """Get all items from output queue.""" items = [] while not self.output_queue.empty(): try: items.append(self.output_queue.get_nowait()) except queue.Empty: break return items return QueueManager() @pytest.fixture def reset_global_state(): """ Fixture to reset global shell manager state before and after tests. """ import mcp_odoo_shell # Save original state original_shell_manager = mcp_odoo_shell.server.shell_manager # Reset before test mcp_odoo_shell.server.shell_manager = None yield # Reset after test if mcp_odoo_shell.server.shell_manager: try: mcp_odoo_shell.server.shell_manager.stop() except Exception: pass mcp_odoo_shell.server.shell_manager = original_shell_manager