diff --git a/pyproject.toml b/pyproject.toml index 6165865..40185ab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,3 +36,8 @@ warn_return_any = true warn_unused_configs = true disallow_untyped_defs = true disallow_incomplete_defs = true + +[dependency-groups] +dev = [ + "pytest>=8.3.5", +] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..af45445 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,160 @@ +""" +Common test fixtures for Monadical Container tests. +""" + +import uuid +import tempfile +import pytest +import docker +from pathlib import Path +from unittest.mock import patch + +from mcontainer.container import ContainerManager +from mcontainer.session import SessionManager +from mcontainer.config import ConfigManager +from mcontainer.models import Session, SessionStatus +from mcontainer.user_config import UserConfigManager + + +# Check if Docker is available +def is_docker_available(): + """Check if Docker is available and running.""" + try: + client = docker.from_env() + client.ping() + return True + except Exception: + return False + + +# Decorator to mark tests that require Docker +requires_docker = pytest.mark.skipif( + not is_docker_available(), + reason="Docker is not available or not running", +) + + +@pytest.fixture +def temp_dir(): + """Create a temporary directory for test files.""" + with tempfile.TemporaryDirectory() as tmp_dir: + yield Path(tmp_dir) + + +@pytest.fixture +def temp_config_dir(): + """Create a temporary directory for configuration files.""" + with tempfile.TemporaryDirectory() as temp_dir: + yield Path(temp_dir) + + +@pytest.fixture +def isolated_config(temp_config_dir): + """Provide an isolated UserConfigManager instance.""" + config_path = temp_config_dir / "config.yaml" + config_path.parent.mkdir(parents=True, exist_ok=True) + return UserConfigManager(str(config_path)) + + +@pytest.fixture +def isolated_session_manager(temp_config_dir): + """Create an isolated session manager for testing.""" + sessions_path = temp_config_dir / "sessions.yaml" + return SessionManager(sessions_path) + + +@pytest.fixture +def isolated_config_manager(): + """Create an isolated config manager for testing.""" + config_manager = ConfigManager() + # Ensure we're using the built-in drivers, not trying to load from user config + return config_manager + + +@pytest.fixture +def mock_session_manager(): + """Mock the SessionManager class.""" + with patch("mcontainer.cli.session_manager") as mock_manager: + yield mock_manager + + +@pytest.fixture +def mock_container_manager(): + """Mock the ContainerManager class with proper initialization.""" + timestamp = "2023-01-01T00:00:00Z" # Use fixed timestamp for reproducibility + mock_session = Session( + id="test-session-id", + name="test-session", + driver="goose", + status=SessionStatus.RUNNING, + ports={"8080": "8080"}, + project=None, + created_at=timestamp, + ) + + with patch("mcontainer.cli.container_manager") as mock_manager: + # Set behaviors to avoid TypeErrors + mock_manager.list_sessions.return_value = [] + mock_manager.create_session.return_value = mock_session + mock_manager.close_session.return_value = True + mock_manager.close_all_sessions.return_value = (3, True) + yield mock_manager + + +@pytest.fixture +def container_manager(isolated_session_manager, isolated_config_manager): + """Create a container manager with isolated components.""" + return ContainerManager( + config_manager=isolated_config_manager, session_manager=isolated_session_manager + ) + + +@pytest.fixture +def cli_runner(): + """Provide a CLI runner for testing commands.""" + from typer.testing import CliRunner + + return CliRunner() + + +@pytest.fixture +def test_file_content(temp_dir): + """Create a test file with content in the temporary directory.""" + test_content = "This is a test file for volume mounting" + test_file = temp_dir / "test_volume_file.txt" + with open(test_file, "w") as f: + f.write(test_content) + return test_file, test_content + + +@pytest.fixture +def test_network_name(): + """Generate a unique network name for testing.""" + return f"mc-test-network-{uuid.uuid4().hex[:8]}" + + +@pytest.fixture +def docker_test_network(test_network_name): + """Create a Docker network for testing and clean it up after.""" + if not is_docker_available(): + pytest.skip("Docker is not available") + return None + + client = docker.from_env() + network = client.networks.create(test_network_name, driver="bridge") + + yield test_network_name + + # Clean up + try: + network.remove() + except Exception: + # Network might be in use by other containers + pass + + +@pytest.fixture +def patched_config_manager(isolated_config): + """Patch the UserConfigManager in cli.py to use our isolated instance.""" + with patch("mcontainer.cli.user_config", isolated_config): + yield isolated_config diff --git a/tests/test_cli.py b/tests/test_cli.py index 4188568..78dc276 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,4 +1,5 @@ from typer.testing import CliRunner + from mcontainer.cli import app runner = CliRunner() @@ -15,8 +16,8 @@ def test_session_list() -> None: """Test session list command""" result = runner.invoke(app, ["session", "list"]) assert result.exit_code == 0 - # Could be either "No active sessions found" or a table of sessions - assert "sessions" in result.stdout.lower() or "no active" in result.stdout.lower() + # Could be either "No active sessions found" or a table with headers + assert "no active" in result.stdout.lower() or "id" in result.stdout.lower() def test_help() -> None: diff --git a/tests/test_config_commands.py b/tests/test_config_commands.py new file mode 100644 index 0000000..6088d9d --- /dev/null +++ b/tests/test_config_commands.py @@ -0,0 +1,190 @@ +""" +Tests for the configuration management commands. +""" + +from mcontainer.cli import app + + +def test_config_list(cli_runner, patched_config_manager): + """Test the 'mc config list' command.""" + result = cli_runner.invoke(app, ["config", "list"]) + + assert result.exit_code == 0 + assert "Configuration" in result.stdout + assert "Value" in result.stdout + + # Check for default configurations + assert "defaults.driver" in result.stdout + assert "defaults.connect" in result.stdout + assert "defaults.mount_local" in result.stdout + + +def test_config_get(cli_runner, patched_config_manager): + """Test the 'mc config get' command.""" + # Test getting an existing value + result = cli_runner.invoke(app, ["config", "get", "defaults.driver"]) + + assert result.exit_code == 0 + assert "defaults.driver" in result.stdout + assert "goose" in result.stdout + + # Test getting a non-existent value + result = cli_runner.invoke(app, ["config", "get", "nonexistent.key"]) + + assert result.exit_code == 0 + assert "not found" in result.stdout + + +def test_config_set(cli_runner, patched_config_manager): + """Test the 'mc config set' command.""" + # Test setting a string value + result = cli_runner.invoke(app, ["config", "set", "defaults.driver", "claude"]) + + assert result.exit_code == 0 + assert "Configuration updated" in result.stdout + assert patched_config_manager.get("defaults.driver") == "claude" + + # Test setting a boolean value + result = cli_runner.invoke(app, ["config", "set", "defaults.connect", "false"]) + + assert result.exit_code == 0 + assert "Configuration updated" in result.stdout + assert patched_config_manager.get("defaults.connect") is False + + # Test setting a new value + result = cli_runner.invoke(app, ["config", "set", "new.setting", "value"]) + + assert result.exit_code == 0 + assert "Configuration updated" in result.stdout + assert patched_config_manager.get("new.setting") == "value" + + +def test_volume_list_empty(cli_runner, patched_config_manager): + """Test the 'mc config volume list' command with no volumes.""" + result = cli_runner.invoke(app, ["config", "volume", "list"]) + + assert result.exit_code == 0 + assert "No default volumes configured" in result.stdout + + +def test_volume_add_and_list(cli_runner, patched_config_manager, temp_config_dir): + """Test adding a volume and then listing it.""" + # Create a test directory + test_dir = temp_config_dir / "test_dir" + test_dir.mkdir() + + # Add a volume + result = cli_runner.invoke( + app, ["config", "volume", "add", f"{test_dir}:/container/path"] + ) + + assert result.exit_code == 0 + assert "Added volume" in result.stdout + + # List volumes + result = cli_runner.invoke(app, ["config", "volume", "list"]) + + assert result.exit_code == 0 + assert str(test_dir) in result.stdout + assert "/container/path" in result.stdout + + +def test_volume_remove(cli_runner, patched_config_manager, temp_config_dir): + """Test removing a volume.""" + # Create a test directory + test_dir = temp_config_dir / "test_dir" + test_dir.mkdir() + + # Add a volume + patched_config_manager.set("defaults.volumes", [f"{test_dir}:/container/path"]) + + # Remove the volume + result = cli_runner.invoke(app, ["config", "volume", "remove", f"{test_dir}"]) + + assert result.exit_code == 0 + assert "Removed volume" in result.stdout + + # Verify it's gone + volumes = patched_config_manager.get("defaults.volumes") + assert len(volumes) == 0 + + +def test_volume_add_nonexistent_path(cli_runner, patched_config_manager, monkeypatch): + """Test adding a volume with a nonexistent path.""" + nonexistent_path = "/path/that/does/not/exist" + + # Mock typer.confirm to return True + monkeypatch.setattr("typer.confirm", lambda message: True) + + # Add a volume with nonexistent path + result = cli_runner.invoke( + app, ["config", "volume", "add", f"{nonexistent_path}:/container/path"] + ) + + assert result.exit_code == 0 + assert "Warning: Local path" in result.stdout + assert "Added volume" in result.stdout + + # Verify it was added + volumes = patched_config_manager.get("defaults.volumes") + assert f"{nonexistent_path}:/container/path" in volumes + + +def test_network_list_empty(cli_runner, patched_config_manager): + """Test the 'mc config network list' command with no networks.""" + result = cli_runner.invoke(app, ["config", "network", "list"]) + + assert result.exit_code == 0 + assert "No default networks configured" in result.stdout + + +def test_network_add_and_list(cli_runner, patched_config_manager): + """Test adding a network and then listing it.""" + # Add a network + result = cli_runner.invoke(app, ["config", "network", "add", "test-network"]) + + assert result.exit_code == 0 + assert "Added network" in result.stdout + + # List networks + result = cli_runner.invoke(app, ["config", "network", "list"]) + + assert result.exit_code == 0 + assert "test-network" in result.stdout + + +def test_network_remove(cli_runner, patched_config_manager): + """Test removing a network.""" + # Add a network + patched_config_manager.set("defaults.networks", ["test-network"]) + + # Remove the network + result = cli_runner.invoke(app, ["config", "network", "remove", "test-network"]) + + assert result.exit_code == 0 + assert "Removed network" in result.stdout + + # Verify it's gone + networks = patched_config_manager.get("defaults.networks") + assert len(networks) == 0 + + +def test_config_reset(cli_runner, patched_config_manager, monkeypatch): + """Test resetting the configuration.""" + # Set a custom value first + patched_config_manager.set("defaults.driver", "custom-driver") + + # Mock typer.confirm to return True + monkeypatch.setattr("typer.confirm", lambda message: True) + + # Reset config + result = cli_runner.invoke(app, ["config", "reset"]) + + assert result.exit_code == 0 + assert "Configuration reset to defaults" in result.stdout + + # Verify it was reset + assert patched_config_manager.get("defaults.driver") == "goose" + + +# patched_config_manager fixture is now in conftest.py diff --git a/tests/test_integration_docker.py b/tests/test_integration_docker.py new file mode 100644 index 0000000..c5a1f71 --- /dev/null +++ b/tests/test_integration_docker.py @@ -0,0 +1,102 @@ +""" +Integration tests for Docker interactions in Monadical Container. +These tests require Docker to be running. +""" + +import subprocess +import time +import uuid + +# Import the requires_docker decorator from conftest +from conftest import requires_docker + + +def execute_command_in_container(container_id, command): + """Execute a command in a Docker container and return the output.""" + result = subprocess.run( + ["docker", "exec", container_id, "bash", "-c", command], + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +@requires_docker +def test_integration_session_create_with_volumes(container_manager, test_file_content): + """Test creating a session with a volume mount.""" + test_file, test_content = test_file_content + session = None + + try: + # Create a session with a volume mount + session = container_manager.create_session( + driver_name="goose", + session_name=f"mc-test-volume-{uuid.uuid4().hex[:8]}", + mount_local=False, # Don't mount current directory + volumes={str(test_file): {"bind": "/test/volume_test.txt", "mode": "ro"}}, + ) + + assert session is not None + assert session.status == "running" + + # Give container time to fully start + time.sleep(2) + + # Verify the file exists in the container and has correct content + container_content = execute_command_in_container( + session.container_id, "cat /test/volume_test.txt" + ) + + assert container_content == test_content + + finally: + # Clean up the container + if session and session.container_id: + container_manager.close_session(session.id) + + +@requires_docker +def test_integration_session_create_with_networks( + container_manager, docker_test_network +): + """Test creating a session connected to a custom network.""" + session = None + + try: + # Create a session with the test network + session = container_manager.create_session( + driver_name="goose", + session_name=f"mc-test-network-{uuid.uuid4().hex[:8]}", + mount_local=False, # Don't mount current directory + networks=[docker_test_network], + ) + + assert session is not None + assert session.status == "running" + + # Give container time to fully start + time.sleep(2) + + # Verify the container is connected to the test network + # Use inspect to check network connections + import docker + + client = docker.from_env() + container = client.containers.get(session.container_id) + container_networks = container.attrs["NetworkSettings"]["Networks"] + + # Container should be connected to both the default mc-network and our test network + assert docker_test_network in container_networks + + # Verify network interface exists in container + network_interfaces = execute_command_in_container( + session.container_id, "ip link show | grep -v 'lo' | wc -l" + ) + + # Should have at least 2 interfaces (eth0 for mc-network, eth1 for test network) + assert int(network_interfaces) >= 2 + + finally: + # Clean up the container + if session and session.container_id: + container_manager.close_session(session.id) diff --git a/tests/test_session_commands.py b/tests/test_session_commands.py new file mode 100644 index 0000000..b93b0e1 --- /dev/null +++ b/tests/test_session_commands.py @@ -0,0 +1,123 @@ +""" +Tests for the session management commands. +""" + +from unittest.mock import patch + + +from mcontainer.cli import app + + +def test_session_list_empty(cli_runner, mock_container_manager): + """Test 'mc session list' with no active sessions.""" + mock_container_manager.list_sessions.return_value = [] + + result = cli_runner.invoke(app, ["session", "list"]) + + assert result.exit_code == 0 + assert "No active sessions found" in result.stdout + + +def test_session_list_with_sessions(cli_runner, mock_container_manager): + """Test 'mc session list' with active sessions.""" + # Create a mock session and set list_sessions to return it + from mcontainer.models import Session, SessionStatus + + mock_session = Session( + id="test-session-id", + name="test-session", + driver="goose", + status=SessionStatus.RUNNING, + ports={"8080": "8080"}, + project=None, + created_at="2023-01-01T00:00:00Z", + ) + mock_container_manager.list_sessions.return_value = [mock_session] + + result = cli_runner.invoke(app, ["session", "list"]) + + assert result.exit_code == 0 + assert "test-session-id" in result.stdout + assert "test-session" in result.stdout + assert "goose" in result.stdout + + +def test_session_create_basic(cli_runner, mock_container_manager): + """Test 'mc session create' with basic options.""" + # We need to patch user_config.get with a side_effect to handle different keys + with patch("mcontainer.cli.user_config") as mock_user_config: + # Handle different key requests appropriately + def mock_get_side_effect(key, default=None): + if key == "defaults.driver": + return "goose" + elif key == "defaults.volumes": + return [] # Return empty list for volumes + elif key == "defaults.connect": + return True + elif key == "defaults.mount_local": + return True + elif key == "defaults.networks": + return [] + return default + + mock_user_config.get.side_effect = mock_get_side_effect + mock_user_config.get_environment_variables.return_value = {} + + result = cli_runner.invoke(app, ["session", "create"]) + + if result.exit_code != 0: + print(f"Error: {result.exception}") + + assert result.exit_code == 0 + assert "Session created successfully" in result.stdout + + # Verify container_manager was called with the expected driver + mock_container_manager.create_session.assert_called_once() + assert ( + mock_container_manager.create_session.call_args[1]["driver_name"] == "goose" + ) + + +def test_session_close(cli_runner, mock_container_manager): + """Test 'mc session close' command.""" + mock_container_manager.close_session.return_value = True + + result = cli_runner.invoke(app, ["session", "close", "test-session-id"]) + + assert result.exit_code == 0 + assert "closed successfully" in result.stdout + mock_container_manager.close_session.assert_called_once_with("test-session-id") + + +def test_session_close_all(cli_runner, mock_container_manager): + """Test 'mc session close --all' command.""" + # Set up mock sessions + from mcontainer.models import Session, SessionStatus + + timestamp = "2023-01-01T00:00:00Z" + mock_sessions = [ + Session( + id=f"session-{i}", + name=f"Session {i}", + driver="goose", + status=SessionStatus.RUNNING, + ports={}, + project=None, + created_at=timestamp, + ) + for i in range(3) + ] + + mock_container_manager.list_sessions.return_value = mock_sessions + mock_container_manager.close_all_sessions.return_value = (3, True) + + result = cli_runner.invoke(app, ["session", "close", "--all"]) + + assert result.exit_code == 0 + assert "3 sessions closed successfully" in result.stdout + mock_container_manager.close_all_sessions.assert_called_once() + + +# For more complex tests that need actual Docker, +# we've implemented them in test_integration_docker.py +# They will run automatically if Docker is available diff --git a/uv.lock b/uv.lock index b7b3127..a6c4428 100644 --- a/uv.lock +++ b/uv.lock @@ -138,6 +138,11 @@ dev = [ { name = "ruff" }, ] +[package.dev-dependencies] +dev = [ + { name = "pytest" }, +] + [package.metadata] requires-dist = [ { name = "docker", specifier = ">=7.0.0" }, @@ -150,6 +155,9 @@ requires-dist = [ { name = "typer", specifier = ">=0.9.0" }, ] +[package.metadata.requires-dev] +dev = [{ name = "pytest", specifier = ">=8.3.5" }] + [[package]] name = "mdurl" version = "0.1.2"