"""Integration tests for Docker deployment functionality.
These tests verify the Docker deployer can create and manage
Velociraptor server containers. They require Docker to be available.
Note: These tests create real Docker containers. Cleanup is performed
automatically but containers may be left behind if tests crash.
"""
import asyncio
import uuid
from pathlib import Path
import pytest
# These tests require Docker
pytestmark = [pytest.mark.integration]
@pytest.fixture(scope="module")
def event_loop():
"""Create an event loop for async tests."""
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture
def unique_deployment_id():
"""Generate a unique deployment ID for test isolation."""
return f"test-deploy-{uuid.uuid4().hex[:8]}"
@pytest.fixture
def docker_deployer(docker_available, temp_deployment_dir):
"""Create a DockerDeployer for testing."""
if not docker_available:
pytest.skip("Docker not available")
try:
from megaraptor_mcp.deployment.deployers.docker_deployer import DockerDeployer
deployer = DockerDeployer(
storage_path=temp_deployment_dir,
)
return deployer
except ImportError as e:
pytest.skip(f"Docker deployer not available: {e}")
class TestDockerDeployerAvailability:
"""Tests for Docker availability checks."""
def test_docker_client_available(self, docker_available):
"""Test that Docker client is available."""
assert docker_available is True
def test_can_import_docker_module(self):
"""Test that docker Python module can be imported."""
try:
import docker
assert docker is not None
except ImportError:
pytest.skip("docker module not installed")
class TestDockerDeployerConfiguration:
"""Tests for Docker deployer configuration."""
def test_deployer_initialization(self, docker_deployer):
"""Test deployer initializes correctly."""
assert docker_deployer is not None
def test_deployer_has_client_property(self, docker_deployer):
"""Test deployer has a Docker client property."""
# The deployer should have a client property
assert hasattr(docker_deployer, "client")
# Accessing it should work (lazy initialization)
client = docker_deployer.client
assert client is not None
class TestDeploymentLifecycle:
"""Tests for the deployment status checking.
Note: Full deployment tests are skipped as they require
specific configuration objects that match the deployer's expected interface.
"""
@pytest.mark.slow
async def test_get_status_nonexistent(self, docker_deployer):
"""Test getting status of non-existent deployment."""
status = await docker_deployer.get_status("nonexistent-deployment")
# Should return None for non-existent deployment
assert status is None
@pytest.mark.slow
async def test_health_check_nonexistent(self, docker_deployer):
"""Test health check on non-existent deployment."""
health = await docker_deployer.health_check("nonexistent-deployment")
assert isinstance(health, dict)
assert health.get("healthy") is False
assert health.get("container_running") is False
class TestResourceLimits:
"""Tests for Docker resource limit enforcement."""
def test_profile_resource_limits(self):
"""Test that profiles define resource limits."""
from megaraptor_mcp.deployment.profiles import get_profile
rapid = get_profile("rapid")
assert "memory" in rapid.resource_limits
assert "cpus" in rapid.resource_limits
assert rapid.resource_limits["memory"] == "4g"
assert rapid.resource_limits["cpus"] == "2"
class TestContainerNaming:
"""Tests for container naming conventions."""
def test_container_name_format(self, unique_deployment_id):
"""Test that deployment IDs are valid container names."""
# Container names can contain lowercase letters, digits, hyphens, and underscores
import re
pattern = r'^[a-z0-9][a-z0-9_.-]*$'
assert re.match(pattern, unique_deployment_id.lower())
def test_container_name_generation(self, docker_deployer, unique_deployment_id):
"""Test container name generation from deployment ID."""
container_name = docker_deployer._container_name(unique_deployment_id)
assert container_name.startswith("velociraptor-")
assert unique_deployment_id in container_name
class TestTargetType:
"""Tests for deployment target type."""
def test_target_type_is_docker(self, docker_deployer):
"""Test that the deployer reports DOCKER as target type."""
from megaraptor_mcp.deployment.profiles import DeploymentTarget
assert docker_deployer.target_type == DeploymentTarget.DOCKER
class TestDockerDeployerMethods:
"""Tests for DockerDeployer method availability."""
def test_has_deploy_method(self, docker_deployer):
"""Test deployer has deploy method."""
assert hasattr(docker_deployer, "deploy")
assert callable(docker_deployer.deploy)
def test_has_destroy_method(self, docker_deployer):
"""Test deployer has destroy method."""
assert hasattr(docker_deployer, "destroy")
assert callable(docker_deployer.destroy)
def test_has_get_status_method(self, docker_deployer):
"""Test deployer has get_status method."""
assert hasattr(docker_deployer, "get_status")
assert callable(docker_deployer.get_status)
def test_has_health_check_method(self, docker_deployer):
"""Test deployer has health_check method."""
assert hasattr(docker_deployer, "health_check")
assert callable(docker_deployer.health_check)
def test_has_get_logs_method(self, docker_deployer):
"""Test deployer has get_logs method."""
assert hasattr(docker_deployer, "get_logs")
assert callable(docker_deployer.get_logs)
def test_has_restart_method(self, docker_deployer):
"""Test deployer has restart method."""
assert hasattr(docker_deployer, "restart")
assert callable(docker_deployer.restart)
