pluto
8b3e2e5067
Add DockerOps class wrapping the Docker SDK with methods for listing, starting, stopping, restarting containers, fetching logs, and pulling images. Swarm containers are detected via labels and receive special handling (scale/force-update instead of direct container ops). Includes 16 passing tests using mocked Docker client. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
390 lines
12 KiB
Python
390 lines
12 KiB
Python
"""Tests for agent.docker_ops.DockerOps — Docker operations wrapper.
|
|
|
|
Uses unittest.mock to mock the Docker SDK client so no real Docker
|
|
daemon is required.
|
|
"""
|
|
|
|
import datetime
|
|
from unittest.mock import MagicMock, patch, PropertyMock
|
|
|
|
import pytest
|
|
from docker.errors import NotFound
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers to build mock container objects
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_container(
|
|
id="abc123def456",
|
|
name="my-app",
|
|
status="running",
|
|
image_tag="nginx:latest",
|
|
created=None,
|
|
labels=None,
|
|
):
|
|
"""Return a MagicMock that behaves like a docker Container object."""
|
|
c = MagicMock()
|
|
c.id = id
|
|
c.short_id = id[:12]
|
|
c.name = name
|
|
c.status = status
|
|
|
|
img = MagicMock()
|
|
img.tags = [image_tag]
|
|
c.image = img
|
|
c.attrs = {
|
|
"Created": created or "2026-03-04T12:00:00.000000000Z",
|
|
"State": {
|
|
"StartedAt": "2026-03-04T12:00:00.000000000Z",
|
|
},
|
|
}
|
|
c.labels = labels or {}
|
|
return c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListContainers:
|
|
"""DockerOps.list_containers returns structured container info."""
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_list_containers_returns_expected_fields(self, mock_from_env):
|
|
"""Each dict has id, name, status, image, created, uptime,
|
|
is_swarm, swarm_service."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container()
|
|
mock_client.containers.list.return_value = [container]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.list_containers()
|
|
|
|
assert len(result) == 1
|
|
c = result[0]
|
|
assert c["id"] == container.short_id
|
|
assert c["name"] == "my-app"
|
|
assert c["status"] == "running"
|
|
assert c["image"] == "nginx:latest"
|
|
assert "created" in c
|
|
assert "uptime" in c
|
|
assert c["is_swarm"] is False
|
|
assert c["swarm_service"] is None
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_list_containers_detects_swarm(self, mock_from_env):
|
|
"""Containers with com.docker.swarm.service.name label are
|
|
marked is_swarm=True with the service name populated."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(
|
|
name="mystack_web.1.xyz",
|
|
labels={"com.docker.swarm.service.name": "mystack_web"},
|
|
)
|
|
mock_client.containers.list.return_value = [container]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.list_containers()
|
|
|
|
assert result[0]["is_swarm"] is True
|
|
assert result[0]["swarm_service"] == "mystack_web"
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_list_containers_includes_stopped(self, mock_from_env):
|
|
"""Exited containers should still appear with status 'exited'."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(status="exited")
|
|
mock_client.containers.list.return_value = [container]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.list_containers()
|
|
|
|
assert result[0]["status"] == "exited"
|
|
# Verify list was called with all=True to include stopped containers
|
|
mock_client.containers.list.assert_called_once_with(all=True)
|
|
|
|
|
|
class TestStartContainer:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_start_container_success(self, mock_from_env):
|
|
"""start_container calls container.start() and returns success."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(status="exited")
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.start_container("abc123def456")
|
|
|
|
container.start.assert_called_once()
|
|
assert result["success"] is True
|
|
assert "message" in result
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_start_nonexistent_container(self, mock_from_env):
|
|
"""start_container on missing container returns success=False."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
mock_client.containers.get.side_effect = NotFound("not found")
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.start_container("nonexistent")
|
|
|
|
assert result["success"] is False
|
|
assert "message" in result
|
|
|
|
|
|
class TestStopContainer:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_stop_container_success(self, mock_from_env):
|
|
"""stop_container calls container.stop() and returns success."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container()
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.stop_container("abc123def456")
|
|
|
|
container.stop.assert_called_once()
|
|
assert result["success"] is True
|
|
assert "message" in result
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_stop_nonexistent_container(self, mock_from_env):
|
|
"""stop_container on missing container returns success=False."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
mock_client.containers.get.side_effect = NotFound("not found")
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.stop_container("nonexistent")
|
|
|
|
assert result["success"] is False
|
|
|
|
|
|
class TestRestartContainer:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_restart_container_success(self, mock_from_env):
|
|
"""restart_container calls container.restart() and returns success."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container()
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.restart_container("abc123def456")
|
|
|
|
container.restart.assert_called_once()
|
|
assert result["success"] is True
|
|
assert "message" in result
|
|
|
|
|
|
class TestGetLogs:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_get_logs_returns_container_and_logs(self, mock_from_env):
|
|
"""get_logs returns {container: name, logs: string}."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container()
|
|
container.logs.return_value = b"line1\nline2\nline3"
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.get_logs("abc123def456")
|
|
|
|
assert result["container"] == "my-app"
|
|
assert "line1" in result["logs"]
|
|
container.logs.assert_called_once_with(tail=100)
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_get_logs_custom_tail(self, mock_from_env):
|
|
"""get_logs respects the tail parameter."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container()
|
|
container.logs.return_value = b"log data"
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
ops.get_logs("abc123def456", tail=50)
|
|
|
|
container.logs.assert_called_once_with(tail=50)
|
|
|
|
|
|
class TestPullImage:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_pull_image_success(self, mock_from_env):
|
|
"""pull_image pulls the container's current image tag."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(image_tag="nginx:1.25")
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.pull_image("abc123def456")
|
|
|
|
mock_client.images.pull.assert_called_once_with("nginx:1.25")
|
|
assert result["success"] is True
|
|
assert "message" in result
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_pull_image_no_tags(self, mock_from_env):
|
|
"""pull_image handles containers with no image tags gracefully."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container()
|
|
container.image.tags = []
|
|
mock_client.containers.get.return_value = container
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.pull_image("abc123def456")
|
|
|
|
assert result["success"] is False
|
|
assert "message" in result
|
|
|
|
|
|
class TestSwarmServiceOperations:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_stop_swarm_service_scales_to_zero(self, mock_from_env):
|
|
"""Stopping a Swarm container scales the service to 0 replicas."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(
|
|
labels={"com.docker.swarm.service.name": "mystack_web"},
|
|
)
|
|
mock_client.containers.get.return_value = container
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.name = "mystack_web"
|
|
mock_client.services.list.return_value = [mock_service]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.stop_container("abc123def456")
|
|
|
|
# Should scale service to 0 instead of calling container.stop()
|
|
mock_service.scale.assert_called_once_with(0)
|
|
container.stop.assert_not_called()
|
|
assert result["success"] is True
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_start_swarm_service_scales_to_one(self, mock_from_env):
|
|
"""Starting a Swarm container scales the service to 1 replica."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(
|
|
status="exited",
|
|
labels={"com.docker.swarm.service.name": "mystack_web"},
|
|
)
|
|
mock_client.containers.get.return_value = container
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.name = "mystack_web"
|
|
mock_client.services.list.return_value = [mock_service]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.start_container("abc123def456")
|
|
|
|
# Should scale service to 1 instead of calling container.start()
|
|
mock_service.scale.assert_called_once_with(1)
|
|
container.start.assert_not_called()
|
|
assert result["success"] is True
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
def test_restart_swarm_service_force_updates(self, mock_from_env):
|
|
"""Restarting a Swarm container force-updates the service."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
|
|
container = _make_container(
|
|
labels={"com.docker.swarm.service.name": "mystack_web"},
|
|
)
|
|
mock_client.containers.get.return_value = container
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.name = "mystack_web"
|
|
mock_client.services.list.return_value = [mock_service]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.restart_container("abc123def456")
|
|
|
|
mock_service.force_update.assert_called_once()
|
|
container.restart.assert_not_called()
|
|
assert result["success"] is True
|
|
|
|
|
|
class TestGetHealth:
|
|
|
|
@patch("docker.DockerClient.from_env")
|
|
@patch("socket.gethostname", return_value="hf-pdocker-01")
|
|
def test_get_health(self, mock_hostname, mock_from_env):
|
|
"""get_health returns hostname and container count."""
|
|
mock_client = MagicMock()
|
|
mock_from_env.return_value = mock_client
|
|
mock_client.containers.list.return_value = [
|
|
_make_container(),
|
|
_make_container(id="def456", name="another"),
|
|
]
|
|
|
|
from agent.docker_ops import DockerOps
|
|
|
|
ops = DockerOps()
|
|
result = ops.get_health()
|
|
|
|
assert result["status"] == "healthy"
|
|
assert result["hostname"] == "hf-pdocker-01"
|
|
assert result["containers_total"] == 2
|