test: add unit tests

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",
+]

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

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:

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

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)

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

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"