pluto
6dd1a61687
Add agent/main.py with REST endpoints wrapping DockerOps: health, list containers, start/stop/restart, logs, and pull. DockerOps instantiation handles missing Docker socket gracefully for test environments. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ops():
|
|
with patch("agent.main.ops") as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def client(mock_ops):
|
|
from agent.main import app
|
|
return TestClient(app)
|
|
|
|
|
|
class TestHealthEndpoint:
|
|
def test_health_returns_ok(self, client, mock_ops):
|
|
mock_ops.get_health.return_value = {
|
|
"status": "ok", "hostname": "test-node", "containers_total": 5
|
|
}
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
|
|
|
|
class TestListContainers:
|
|
def test_list_containers(self, client, mock_ops):
|
|
mock_ops.list_containers.return_value = [{
|
|
"id": "abc123", "name": "test-app", "status": "running",
|
|
"image": "nginx:latest", "created": "2026-03-05T00:00:00Z",
|
|
"uptime": "2h 30m", "is_swarm": False, "swarm_service": None,
|
|
}]
|
|
resp = client.get("/containers")
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()) == 1
|
|
assert resp.json()[0]["name"] == "test-app"
|
|
|
|
|
|
class TestContainerActions:
|
|
def test_start(self, client, mock_ops):
|
|
mock_ops.start_container.return_value = {"success": True, "message": "Started"}
|
|
assert client.post("/containers/abc123/start").json()["success"] is True
|
|
|
|
def test_stop(self, client, mock_ops):
|
|
mock_ops.stop_container.return_value = {"success": True, "message": "Stopped"}
|
|
assert client.post("/containers/abc123/stop").json()["success"] is True
|
|
|
|
def test_restart(self, client, mock_ops):
|
|
mock_ops.restart_container.return_value = {"success": True, "message": "Restarted"}
|
|
assert client.post("/containers/abc123/restart").json()["success"] is True
|
|
|
|
|
|
class TestLogs:
|
|
def test_get_logs(self, client, mock_ops):
|
|
mock_ops.get_logs.return_value = {"container": "test-app", "logs": "line1\nline2"}
|
|
resp = client.get("/containers/abc123/logs?tail=50")
|
|
assert resp.status_code == 200
|
|
assert "line1" in resp.json()["logs"]
|
|
|
|
def test_get_logs_default_tail(self, client, mock_ops):
|
|
mock_ops.get_logs.return_value = {"container": "test-app", "logs": "logs"}
|
|
client.get("/containers/abc123/logs")
|
|
mock_ops.get_logs.assert_called_once_with("abc123", tail=200)
|
|
|
|
|
|
class TestPull:
|
|
def test_pull(self, client, mock_ops):
|
|
mock_ops.pull_image.return_value = {"success": True, "message": "Pulled nginx:latest"}
|
|
assert client.post("/containers/abc123/pull").json()["success"] is True
|