MCP Python REPL Server

"""Unit tests for MCP tools using the test_project.""" import pytest import asyncio from pathlib import Path import sys import os # Add src to path for testing sys.path.insert(0, str(Path(__file__).parent.parent / "src")) from mcp_py3repl.server import ( python_execute, python_inspect, session_create, session_list, session_switch, package_install, package_remove, package_list, file_read_project, file_write_project, load_env_file, set_env_var, list_env_vars, create_env_template, format_file, lint_file, run_tests ) from mcp_py3repl.session_manager import SessionManager @pytest.fixture def setup_session_manager(): """Setup all managers for testing.""" from mcp_py3repl import server from mcp_py3repl.package_manager import PackageManager from mcp_py3repl.file_handler import FileHandler from mcp_py3repl.env_handler import EnvHandler from mcp_py3repl.dev_tools import DevTools server.session_manager = SessionManager() server.package_manager = PackageManager() server.file_handler = FileHandler() server.env_handler = EnvHandler() server.dev_tools = DevTools() # Yield the session manager for tests to use yield server.session_manager # Cleanup server.session_manager.cleanup_all_sessions() if hasattr(server, 'env_handler') and server.env_handler: server.env_handler.cleanup_session_vars() @pytest.mark.asyncio async def test_session_create(setup_session_manager): """Test creating a session.""" session_manager = setup_session_manager # First create a session in the test_project directory manually test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) # Test creating another session through the MCP tool response = await session_create( session_name="test2", working_directory=str(Path(__file__).parent / "test_project") ) print("SESSION CREATE OUTPUT:", response) assert "Created session: test2" in response assert str(Path(__file__).parent / "test_project") in response # Should show venv activation status assert "Virtual environment:" in response # Should show helpful import examples assert "This session can now import" in response @pytest.mark.asyncio async def test_session_list(setup_session_manager): """Test listing sessions.""" session_manager = setup_session_manager # Create a test session first test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) response = await session_list() assert "Active sessions:" in response or "No active sessions" in response @pytest.mark.asyncio async def test_python_execute_simple(setup_session_manager): """Test simple Python execution.""" session_manager = setup_session_manager # Create a test session first test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) # Test simple expression response = await python_execute(code="2 + 2", session_id=session_id) assert "4" in response assert "Success: True" in response @pytest.mark.asyncio async def test_python_execute_import_test_project(setup_session_manager): """Test importing from test_project with new venv activation.""" session_manager = setup_session_manager # Create a test session first test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) # Test that test_project paths are in sys.path response1 = await python_execute( code="import sys; test_paths = [p for p in sys.path if 'test_project' in p]; print(f'Test project paths: {test_paths}')", session_id=session_id ) print("PATH TEST:", response1) assert "test_project" in response1 assert "Success: True" in response1 # Now test importing from test_project response2 = await python_execute( code="from app.config import settings; print(f'Database URL: {settings.database_url}')", session_id=session_id ) print("IMPORT TEST:", response2) # This should now work! assert "Database URL:" in response2 assert "Success: True" in response2 @pytest.mark.asyncio async def test_python_execute_with_error(setup_session_manager): """Test Python execution with error.""" session_manager = setup_session_manager # Create a test session first test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) # Test code that raises an error response = await python_execute(code="1 / 0", session_id=session_id) assert "ZeroDivisionError" in response assert "Success: False" in response @pytest.mark.asyncio async def test_python_inspect(setup_session_manager): """Test object inspection.""" session_manager = setup_session_manager # Create a test session first test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) # First create a variable to inspect await python_execute(code="test_var = 'hello world'", session_id=session_id) # Then inspect it response = await python_inspect(object_name="test_var", session_id=session_id) assert "Object: test_var" in response assert "Type: str" in response assert "hello world" in response @pytest.mark.asyncio async def test_variable_persistence(setup_session_manager): """Test that variables persist across executions.""" session_manager = setup_session_manager # Create a test session first test_project_dir = Path(__file__).parent / "test_project" session_id = await session_manager.create_session("test", str(test_project_dir)) # Set a variable response1 = await python_execute(code="x = 42", session_id=session_id) assert "Success: True" in response1 # Use the variable in another execution response2 = await python_execute(code="print(f'x = {x}')", session_id=session_id) assert "x = 42" in response2 @pytest.mark.asyncio async def test_session_switch(setup_session_manager): """Test switching between sessions.""" session_id = setup_session_manager # Create another session response1 = await session_create(session_name="switch_test") assert "Created session: switch_test" in response1 # Extract session ID from response (simple approach for testing) lines = response1.split('\n') new_session_id = None for line in lines: if "Created session: switch_test" in line: # Extract session ID from line like "Created session: switch_test (uuid-here)" start = line.find('(') + 1 end = line.find(')') if start > 0 and end > start: new_session_id = line[start:end] break assert new_session_id is not None # Switch to the new session response2 = await session_switch(session_id=new_session_id) assert "Switched to session: switch_test" in response2 @pytest.mark.asyncio async def test_package_list(setup_session_manager): """Test listing packages.""" # Note: This test depends on the current environment having packages installed response = await package_list() # Should either show packages or indicate no packages assert "packages" in response.lower() or "no packages" in response.lower() @pytest.mark.asyncio async def test_package_install_invalid_package(setup_session_manager): """Test installing a non-existent package.""" # Try to install a clearly non-existent package response = await package_install(packages=["nonexistent-package-12345"]) # Should show error message assert "❌" in response or "error" in response.lower() @pytest.mark.asyncio async def test_package_install_and_remove(setup_session_manager): """Test installing and then removing a package.""" # This test will only work if uv is available and we're in a proper environment # Let's test with a simple, harmless package # First, get the current package list initial_packages = await package_list() # Try to install a small, harmless package install_response = await package_install(packages=["typing-extensions"]) # If installation succeeded, try to remove it if "✅" in install_response: # Package was installed successfully remove_response = await package_remove(packages=["typing-extensions"]) assert "✅" in remove_response or "❌" in remove_response # Should show some result else: # Installation failed - this is also acceptable for testing assert "❌" in install_response or "error" in install_response.lower() @pytest.mark.asyncio async def test_package_operations_with_dev_flag(setup_session_manager): """Test installing packages with dev flag.""" # Test dev dependency installation response = await package_install(packages=["pytest"], dev=True) # Should show some result (success or failure) assert "✅" in response or "❌" in response @pytest.mark.asyncio async def test_file_read_project_existing_file(setup_session_manager): """Test reading an existing file.""" # Read the pyproject.toml file which should exist response = await file_read_project("pyproject.toml") # Should show file content assert "📄 File: pyproject.toml" in response assert "✅ File exists" in response assert "📊 Size:" in response assert "🔤 Encoding:" in response assert "📝 Content:" in response @pytest.mark.asyncio async def test_file_read_project_nonexistent_file(setup_session_manager): """Test reading a non-existent file.""" response = await file_read_project("nonexistent_file.txt") # Should show error assert "📄 File: nonexistent_file.txt" in response assert "❌ Error:" in response assert "does not exist" in response @pytest.mark.asyncio async def test_file_write_and_read(setup_session_manager): """Test writing a file and then reading it back.""" test_content = "# Test file\n\nprint('Hello from test file!')\n" test_file = "test_temp_file.py" # Write the file write_response = await file_write_project(test_file, test_content) # Check write was successful assert "📄 File: test_temp_file.py" in write_response assert "✅ File written successfully" in write_response assert "📊 Size:" in write_response # Read the file back read_response = await file_read_project(test_file) # Check read was successful and content matches assert "✅ File exists" in read_response assert "# Test file" in read_response assert "print('Hello from test file!')" in read_response # Clean up - remove the test file try: from pathlib import Path Path(test_file).unlink() except FileNotFoundError: pass # File might have been cleaned up already @pytest.mark.asyncio async def test_file_write_with_encoding(setup_session_manager): """Test writing a file with specific encoding.""" test_content = "# Test with UTF-8\nprint('Hello with unicode: ñáéíóú')\n" test_file = "test_encoding.py" # Write with explicit UTF-8 encoding response = await file_write_project(test_file, test_content, encoding="utf-8") # Check write was successful assert "✅ File written successfully" in response assert "🔤 Encoding: utf-8" in response # Clean up try: from pathlib import Path Path(test_file).unlink() except FileNotFoundError: pass @pytest.mark.asyncio async def test_file_path_security(setup_session_manager): """Test that file operations respect project directory boundaries.""" # Try to read a file outside the project directory response = await file_read_project("../../../etc/passwd") # Should show error about path being outside project assert "❌ Error:" in response assert "outside project" in response @pytest.mark.asyncio async def test_set_and_list_env_vars(setup_session_manager): """Test setting and listing environment variables.""" # Set a test environment variable set_response = await set_env_var("TEST_VAR", "test_value", session_specific=True) # Check set was successful assert "🔧 Environment Variable: TEST_VAR" in set_response assert "✅" in set_response assert "💾 Value: test_value" in set_response assert "🎯 Scope: Session-specific" in set_response # List environment variables list_response = await list_env_vars(include_system=False) # Should include our test variable assert "🌍 Environment Variables:" in list_response assert "TEST_VAR" in list_response assert "test_value" in list_response @pytest.mark.asyncio async def test_load_env_file_nonexistent(setup_session_manager): """Test loading a non-existent .env file.""" response = await load_env_file(".env.nonexistent") # Should show error assert "🌍 Environment File: .env.nonexistent" in response assert "❌" in response assert "not found" in response @pytest.mark.asyncio async def test_create_env_template_from_test_project(setup_session_manager): """Test creating env template from test_project settings.""" # Try to create template from test_project's settings response = await create_env_template("app.config.Settings", "test_template.env") # This might succeed or fail depending on whether test_project is accessible assert "📋 Environment Template: test_template.env" in response # Clean up if file was created try: from pathlib import Path Path("test_template.env").unlink() except FileNotFoundError: pass @pytest.mark.asyncio async def test_create_env_template_invalid_class(setup_session_manager): """Test creating env template with invalid class.""" response = await create_env_template("nonexistent.module.Settings") # Should show error assert "❌" in response assert "Failed to import" in response or "ImportError" in response @pytest.mark.asyncio async def test_env_var_sensitive_masking(setup_session_manager): """Test that sensitive environment variables are masked.""" # Set a sensitive variable await set_env_var("SECRET_KEY", "super_secret_value") await set_env_var("PASSWORD", "my_password") await set_env_var("NORMAL_VAR", "normal_value") # List variables response = await list_env_vars() # Sensitive values should be masked assert "SECRET_KEY = ***" in response assert "PASSWORD = ***" in response # Normal values should not be masked assert "NORMAL_VAR = normal_value" in response # Should not contain actual sensitive values assert "super_secret_value" not in response assert "my_password" not in response @pytest.mark.asyncio async def test_format_file_existing(setup_session_manager): """Test formatting an existing file.""" # Create a test file with messy formatting test_content = "def hello_world( ):return'Hello, World!'" test_file = "test_format.py" # Write the messy file await file_write_project(test_file, test_content) # Format the file response = await format_file(test_file) # Should show formatting results assert "🎨 File Formatting" in response assert f"📄 File: {test_file}" in response assert "✅" in response or "❌" in response # Clean up try: from pathlib import Path Path(test_file).unlink() except FileNotFoundError: pass @pytest.mark.asyncio async def test_format_file_nonexistent(setup_session_manager): """Test formatting a non-existent file.""" response = await format_file("nonexistent_file.py") # Should show error assert "🎨 File Formatting" in response assert "📄 File: nonexistent_file.py" in response assert "❌ Formatting failed" in response assert "Cannot read file" in response @pytest.mark.asyncio async def test_lint_file_clean(setup_session_manager): """Test linting clean code file.""" clean_code = "def hello_world():\n return 'Hello, World!'\n" # Create a test file await file_write_project("test_clean.py", clean_code) response = await lint_file("test_clean.py") # Should show linting results assert "🔍 File Linting" in response assert "test_clean.py" in response assert "✅" in response or "❌" in response @pytest.mark.asyncio async def test_lint_file_with_issues(setup_session_manager): """Test linting code file with potential issues.""" problematic_code = "import os\nimport sys\ndef hello():pass" # Create a test file await file_write_project("test_problematic.py", problematic_code) response = await lint_file("test_problematic.py") # Should show linting results assert "🔍 File Linting" in response assert "test_problematic.py" in response # Might find unused imports or other issues assert "✅" in response or "❌" in response @pytest.mark.asyncio async def test_run_tests_basic(setup_session_manager): """Test running tests.""" response = await run_tests() # Should show test execution results assert "🧪 Test Execution" in response # Will either find tests or show no tests found assert "✅" in response or "❓" in response or "❌" in response @pytest.mark.asyncio async def test_run_tests_specific_path(setup_session_manager): """Test running tests with specific path.""" response = await run_tests(test_path="tests/", verbose=True) # Should show test execution with path assert "🧪 Test Execution" in response assert "📁 Path: tests/" in response @pytest.mark.asyncio async def test_run_tests_with_pattern(setup_session_manager): """Test running tests with pattern.""" response = await run_tests(pattern="test_env", verbose=True) # Should show test execution with pattern assert "🧪 Test Execution" in response assert "🔍 Pattern: test_env" in response