marcus
24b97338ba
Live test on this DSM firmware: SYNO.Docker.Container has no pause/
unpause method ("Method does not exist"). The Container Manager GUI
action menu only exposes Start / Stop / Force-Stop / Restart / Reset —
pause/resume simply isn't a feature here.
The two tools were briefly shipped in 0.4.0 (implemented by symmetry
with the verified stop call) and have now been removed rather than
left as a broken surface. The remaining lifecycle tools
(start_container, stop_container, restart_container) are unaffected.
Tool count: 33 → 31. Closes #7 (won't fix — DSM limitation).
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
855 lines
28 KiB
Python
855 lines
28 KiB
Python
"""Tests for modules/containers.py."""
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
|
|
def make_mock_mcp():
|
|
tools: dict = {}
|
|
|
|
class MockMCP:
|
|
def tool(self):
|
|
def decorator(fn):
|
|
tools[fn.__name__] = fn
|
|
return fn
|
|
|
|
return decorator
|
|
|
|
return MockMCP(), tools
|
|
|
|
|
|
def make_config():
|
|
from mcp_synology_container.config import AppConfig, ConnectionConfig
|
|
|
|
return AppConfig(
|
|
schema_version=1,
|
|
connection=ConnectionConfig(host="nas.local", port=443, https=True, verify_ssl=True),
|
|
)
|
|
|
|
|
|
SAMPLE_CONTAINERS_DATA = {
|
|
"containers": [
|
|
{
|
|
"name": "myapp_web",
|
|
"status": "running",
|
|
"image": "nginx:alpine",
|
|
"project_name": "myapp",
|
|
},
|
|
{
|
|
"name": "myapp_db",
|
|
"status": "running",
|
|
"image": "postgres:15",
|
|
"project_name": "myapp",
|
|
},
|
|
{
|
|
"name": "other_svc",
|
|
"status": "stopped",
|
|
"image": "redis:7",
|
|
"project_name": "other",
|
|
},
|
|
]
|
|
}
|
|
|
|
# Container data where DSM returns hash-prefixed names
|
|
HASH_PREFIXED_CONTAINERS_DATA = {
|
|
"containers": [
|
|
{
|
|
"name": "f93cb8b504f7_jenkins",
|
|
"status": "running",
|
|
"image": "jenkins/jenkins:lts",
|
|
"project_name": "frostiq",
|
|
},
|
|
]
|
|
}
|
|
|
|
SAMPLE_LOGS_DATA = {
|
|
"logs": [
|
|
{
|
|
"created": "2025-01-01T10:00:00Z",
|
|
"stream": "stdout",
|
|
"text": "Server started",
|
|
"docid": "1",
|
|
},
|
|
{
|
|
"created": "2025-01-01T10:00:01Z",
|
|
"stream": "stderr",
|
|
"text": "Warning: deprecated option",
|
|
"docid": "2",
|
|
},
|
|
],
|
|
"total": 2,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_containers_all():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_CONTAINERS_DATA
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["list_containers"]()
|
|
assert "myapp_web" in result
|
|
assert "myapp_db" in result
|
|
assert "other_svc" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_containers_filtered_by_project():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_CONTAINERS_DATA
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["list_containers"](project_name="myapp")
|
|
assert "myapp_web" in result
|
|
assert "myapp_db" in result
|
|
assert "other_svc" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_containers_empty():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {"containers": []}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["list_containers"]()
|
|
assert "No containers found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_logs():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_LOGS_DATA
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["get_container_logs"]("myapp_web", tail=50)
|
|
assert "myapp_web" in result
|
|
assert "Server started" in result
|
|
assert "Warning: deprecated option" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_logs_with_keyword():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_LOGS_DATA
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
await tools["get_container_logs"]("myapp_web", tail=100, keyword="error")
|
|
call_params = client.request.call_args[1]["params"]
|
|
assert call_params["keyword"] == "error"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_in_container_requires_confirmation():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["exec_in_container"]("myapp_web", "ls /app", confirmed=False)
|
|
assert "confirmed=True" in result
|
|
client.request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_in_container_confirmed():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {"output": "file1.py\nfile2.py", "exit_code": 0}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["exec_in_container"]("myapp_web", "ls /app", confirmed=True)
|
|
assert "file1.py" in result
|
|
assert "Exit code: 0" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# container_stats
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
# Realistic stats snapshot for a container named "glance"
|
|
SAMPLE_STATS = {
|
|
"ee220111cff2": {
|
|
"id": "ee220111cff2",
|
|
"name": "/glance",
|
|
"cpu_stats": {
|
|
"cpu_usage": {
|
|
"total_usage": 1_022_653_189,
|
|
"percpu_usage": [286394951, 245078386, 304613157, 186566695],
|
|
},
|
|
"system_cpu_usage": 990_015_100_000_000,
|
|
"online_cpus": 4,
|
|
},
|
|
"precpu_stats": {
|
|
"cpu_usage": {"total_usage": 1_000_000_000},
|
|
"system_cpu_usage": 989_000_000_000_000,
|
|
},
|
|
"memory_stats": {
|
|
"usage": 21_127_168,
|
|
"limit": 4_079_349_760,
|
|
},
|
|
"networks": {
|
|
"eth0": {"rx_bytes": 876, "tx_bytes": 0},
|
|
"eth1": {"rx_bytes": 1024, "tx_bytes": 512},
|
|
},
|
|
"blkio_stats": {
|
|
"io_service_bytes_recursive": [
|
|
{"op": "Read", "value": 4096},
|
|
{"op": "Write", "value": 8192},
|
|
{"op": "Total", "value": 12288},
|
|
]
|
|
},
|
|
}
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_found():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_STATS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["container_stats"]("glance")
|
|
assert "glance" in result
|
|
assert "CPU" in result
|
|
assert "%" in result
|
|
assert "Memory" in result
|
|
assert "Net I/O" in result
|
|
assert "Block I/O" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# delete_container
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_container_preview():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_container"]("myapp_web", confirmed=False)
|
|
assert "Preview" in result
|
|
assert "myapp_web" in result
|
|
client.request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_container_not_found():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = None
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_container"]("nonexistent", confirmed=True)
|
|
assert "not found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_container_running_blocked():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {
|
|
"details": {"State": {"Running": True, "Status": "running"}},
|
|
"profile": {"image": "nginx:latest"},
|
|
}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_container"]("myapp_web", confirmed=True)
|
|
assert "running" in result.lower()
|
|
assert "Cannot delete" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_container_stopped_confirmed():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {
|
|
"details": {"State": {"Running": False, "Status": "exited"}},
|
|
"profile": {"image": "nginx:latest"},
|
|
}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_container"]("myapp_web", confirmed=True)
|
|
assert "Deleted" in result
|
|
assert "myapp_web" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_container_sends_required_params():
|
|
"""SYNO.Docker.Container/delete must include name (json-encoded), force=false,
|
|
and preserve_profile=false — without them DSM returns error 114."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {
|
|
"details": {"State": {"Running": False, "Status": "exited"}},
|
|
"profile": {"image": "nginx:latest"},
|
|
}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
await tools["delete_container"]("myapp_web", confirmed=True)
|
|
|
|
# Find the delete call (second call — first is Container/get)
|
|
delete_calls = [c for c in client.request.call_args_list if c.args[1] == "delete"]
|
|
assert len(delete_calls) == 1
|
|
params = delete_calls[0].kwargs.get("params") or {}
|
|
assert params["name"] == '"myapp_web"'
|
|
assert params["force"] == "false"
|
|
assert params["preserve_profile"] == "false"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_cpu_calculation():
|
|
"""CPU% is computed via the standard Docker formula."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_STATS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["container_stats"]("glance")
|
|
|
|
# cpu_delta = 1_022_653_189 - 1_000_000_000 = 22_653_189
|
|
# system_delta = 990_015_100_000_000 - 989_000_000_000_000 = 1_015_100_000_000
|
|
# cpu_pct = (22_653_189 / 1_015_100_000_000) * 4 * 100 ≈ 0.0089 %
|
|
assert "0.00" in result or "%" in result # value near zero but formatted
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_memory_human_readable():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_STATS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["container_stats"]("glance")
|
|
# 21_127_168 bytes ≈ 20 MiB; limit 4_079_349_760 ≈ 3.8 GiB
|
|
assert "MiB" in result or "GiB" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_name_with_slash():
|
|
"""Container name matching strips leading slash from DSM response."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_STATS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
# Should match even if called without the leading slash
|
|
result = await tools["container_stats"]("glance")
|
|
assert "not found" not in result.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_not_found():
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_STATS
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["container_stats"]("nonexistent")
|
|
assert "not found" in result.lower()
|
|
assert "glance" in result # shows available containers
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("API error", code=102)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["container_stats"]("glance")
|
|
assert "Error" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Bug 1: hash-prefix stripping
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_strip_hash_prefix_strips_prefix():
|
|
from mcp_synology_container.modules.containers import _strip_hash_prefix
|
|
|
|
assert _strip_hash_prefix("f93cb8b504f7_jenkins") == "jenkins"
|
|
|
|
|
|
def test_strip_hash_prefix_no_prefix():
|
|
from mcp_synology_container.modules.containers import _strip_hash_prefix
|
|
|
|
assert _strip_hash_prefix("jenkins") == "jenkins"
|
|
|
|
|
|
def test_strip_hash_prefix_leading_slash():
|
|
from mcp_synology_container.modules.containers import _strip_hash_prefix
|
|
|
|
assert _strip_hash_prefix("/jenkins") == "jenkins"
|
|
|
|
|
|
def test_strip_hash_prefix_slash_with_hash():
|
|
from mcp_synology_container.modules.containers import _strip_hash_prefix
|
|
|
|
assert _strip_hash_prefix("/f93cb8b504f7_jenkins") == "jenkins"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_containers_strips_hash_prefix():
|
|
"""list_containers must display the clean name without the hash prefix."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = HASH_PREFIXED_CONTAINERS_DATA
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["list_containers"]()
|
|
assert "jenkins" in result
|
|
assert "f93cb8b504f7_jenkins" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_strips_hash_prefix():
|
|
"""container_stats must match even when DSM returns a hash-prefixed name."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
stats_with_hash = {
|
|
"abc123": {
|
|
"name": "/f93cb8b504f7_jenkins",
|
|
"cpu_stats": {
|
|
"cpu_usage": {"total_usage": 500_000, "percpu_usage": [500_000]},
|
|
"system_cpu_usage": 100_000_000_000,
|
|
"online_cpus": 1,
|
|
},
|
|
"precpu_stats": {
|
|
"cpu_usage": {"total_usage": 0},
|
|
"system_cpu_usage": 0,
|
|
},
|
|
"memory_stats": {"usage": 1024, "limit": 2048},
|
|
"networks": {},
|
|
"blkio_stats": {"io_service_bytes_recursive": []},
|
|
}
|
|
}
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = stats_with_hash
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
# User passes clean name; must still match the hash-prefixed entry
|
|
result = await tools["container_stats"]("jenkins")
|
|
assert "not found" not in result.lower()
|
|
assert "CPU" in result
|
|
|
|
|
|
DSM_CONTAINER_RESPONSE = {
|
|
"details": {
|
|
"State": {
|
|
"Status": "running",
|
|
"Running": True,
|
|
"StartedAt": "2025-01-01T10:00:00Z",
|
|
"FinishedAt": "0001-01-01T00:00:00Z",
|
|
"ExitCode": 0,
|
|
},
|
|
"NetworkSettings": {
|
|
"Networks": {
|
|
"bridge": {"IPAddress": "172.17.0.2"},
|
|
}
|
|
},
|
|
"Mounts": [
|
|
{
|
|
"Type": "bind",
|
|
"Source": "/volume1/docker/jenkins",
|
|
"Destination": "/var/jenkins_home",
|
|
"RW": True,
|
|
}
|
|
],
|
|
},
|
|
"profile": {
|
|
"image": "jenkins/jenkins:2.558-jdk21",
|
|
"port_bindings": [
|
|
{"host_port": 8080, "container_port": 8080, "type": "tcp"},
|
|
{"host_port": 50000, "container_port": 50000, "type": "tcp"},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_status_uses_clean_name():
|
|
"""get_container_status strips hash prefix and calls get with clean name."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Container" and method == "get":
|
|
assert kwargs["params"]["name"] == "jenkins"
|
|
return DSM_CONTAINER_RESPONSE
|
|
return {}
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
# User passes hash-prefixed name → stripped before get call
|
|
result = await tools["get_container_status"]("f93cb8b504f7_jenkins")
|
|
assert "running" in result
|
|
assert "f93cb8b504f7" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_status_details_profile_structure():
|
|
"""get_container_status reads status from details.State, image from profile."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = DSM_CONTAINER_RESPONSE
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["get_container_status"]("jenkins")
|
|
assert "running" in result
|
|
assert "jenkins/jenkins:2.558-jdk21" in result
|
|
assert "True" in result # Running field
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_status_shows_ip():
|
|
"""get_container_status shows IP address from NetworkSettings."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = DSM_CONTAINER_RESPONSE
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["get_container_status"]("jenkins")
|
|
assert "172.17.0.2" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_status_shows_ports():
|
|
"""get_container_status shows port bindings from profile."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = DSM_CONTAINER_RESPONSE
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["get_container_status"]("jenkins")
|
|
assert "8080" in result
|
|
assert "50000" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_status_shows_mounts():
|
|
"""get_container_status shows mount paths from details.Mounts."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = DSM_CONTAINER_RESPONSE
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["get_container_status"]("jenkins")
|
|
assert "/volume1/docker/jenkins" in result
|
|
assert "/var/jenkins_home" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_container_logs_resolves_hash_prefix():
|
|
"""get_container_logs resolves 'jenkins' to 'f93cb8b504f7_jenkins' for DSM call."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Container" and method == "list":
|
|
return HASH_PREFIXED_CONTAINERS_DATA
|
|
if api == "SYNO.Docker.Container.Log" and method == "get":
|
|
assert kwargs["params"]["name"] == "f93cb8b504f7_jenkins"
|
|
return {
|
|
"logs": [{"created": "2025-01-01", "stream": "stdout", "text": "started"}],
|
|
"total": 1,
|
|
}
|
|
return {}
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["get_container_logs"]("jenkins")
|
|
assert "started" in result
|
|
assert "f93cb8b504f7" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_exec_in_container_resolves_hash_prefix():
|
|
"""exec_in_container resolves 'jenkins' to 'f93cb8b504f7_jenkins' for DSM call."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Container" and method == "list":
|
|
return HASH_PREFIXED_CONTAINERS_DATA
|
|
if api == "SYNO.Docker.Container" and method == "exec":
|
|
assert kwargs["params"]["name"] == "f93cb8b504f7_jenkins"
|
|
return {"output": "ok", "exit_code": 0}
|
|
return {}
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["exec_in_container"]("jenkins", "echo ok", confirmed=True)
|
|
assert "ok" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_container_stats_no_precpu_graceful():
|
|
"""When precpu_stats has no system_cpu_usage (first poll), CPU% = 0."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
stats_no_pre = {
|
|
"abc123": {
|
|
"name": "/myapp",
|
|
"cpu_stats": {
|
|
"cpu_usage": {"total_usage": 500_000, "percpu_usage": [500_000]},
|
|
"system_cpu_usage": 100_000_000_000,
|
|
"online_cpus": 1,
|
|
},
|
|
"precpu_stats": {
|
|
"cpu_usage": {"total_usage": 0},
|
|
# system_cpu_usage absent → system_delta = 0
|
|
},
|
|
"memory_stats": {"usage": 1024, "limit": 2048},
|
|
"networks": {},
|
|
"blkio_stats": {"io_service_bytes_recursive": []},
|
|
}
|
|
}
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = stats_no_pre
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools["container_stats"]("myapp")
|
|
assert "0.00%" in result # graceful fallback to 0
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# Lifecycle: start / stop / restart
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
# Tools that don't have a confirmation gate
|
|
_NO_CONFIRM_LIFECYCLE = [
|
|
("start_container", "start", "Started"),
|
|
]
|
|
|
|
# Tools that require confirmed=True
|
|
_CONFIRM_LIFECYCLE = [
|
|
("stop_container", "stop", "Stopped", "stop"),
|
|
("restart_container", "restart", "Restarted", "restart"),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("tool_name, dsm_method, success_word", _NO_CONFIRM_LIFECYCLE)
|
|
@pytest.mark.asyncio
|
|
async def test_lifecycle_no_confirm_calls_dsm(tool_name, dsm_method, success_word):
|
|
"""start_container calls DSM directly with json-encoded name."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools[tool_name]("myapp_web")
|
|
|
|
# Find the matching lifecycle call (a list() may also occur via _resolve_container_name)
|
|
calls = [c for c in client.request.call_args_list if c.args[1] == dsm_method]
|
|
assert len(calls) == 1
|
|
call = calls[0]
|
|
assert call.args[0] == "SYNO.Docker.Container"
|
|
assert call.kwargs["version"] == 1
|
|
assert call.kwargs["params"] == {"name": '"myapp_web"'}
|
|
assert success_word in result
|
|
assert "myapp_web" in result
|
|
|
|
|
|
@pytest.mark.parametrize("tool_name, dsm_method, success_word, _action", _CONFIRM_LIFECYCLE)
|
|
@pytest.mark.asyncio
|
|
async def test_lifecycle_confirmed_calls_dsm(tool_name, dsm_method, success_word, _action):
|
|
"""stop/restart call DSM with confirmed=True using json-encoded name."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools[tool_name]("myapp_web", confirmed=True)
|
|
|
|
calls = [c for c in client.request.call_args_list if c.args[1] == dsm_method]
|
|
assert len(calls) == 1
|
|
call = calls[0]
|
|
assert call.args[0] == "SYNO.Docker.Container"
|
|
assert call.kwargs["version"] == 1
|
|
assert call.kwargs["params"] == {"name": '"myapp_web"'}
|
|
assert success_word in result
|
|
assert "myapp_web" in result
|
|
|
|
|
|
@pytest.mark.parametrize("tool_name, _dsm_method, _success_word, action", _CONFIRM_LIFECYCLE)
|
|
@pytest.mark.asyncio
|
|
async def test_lifecycle_preview_without_confirmation(
|
|
tool_name, _dsm_method, _success_word, action
|
|
):
|
|
"""stop/restart without confirmed=True must NOT call DSM."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools[tool_name]("myapp_web")
|
|
assert "Preview" in result
|
|
assert action in result
|
|
assert "myapp_web" in result
|
|
assert "confirmed=True" in result
|
|
client.request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"tool_name, dsm_method, kwargs",
|
|
[
|
|
("start_container", "start", {}),
|
|
("stop_container", "stop", {"confirmed": True}),
|
|
("restart_container", "restart", {"confirmed": True}),
|
|
],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_lifecycle_resolves_hash_prefix(tool_name, dsm_method, kwargs):
|
|
"""All lifecycle tools resolve 'jenkins' to 'f93cb8b504f7_jenkins' for the DSM call."""
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
async def mock_request(api, method, **kw):
|
|
if api == "SYNO.Docker.Container" and method == "list":
|
|
return HASH_PREFIXED_CONTAINERS_DATA
|
|
if api == "SYNO.Docker.Container" and method == dsm_method:
|
|
assert kw["params"]["name"] == '"f93cb8b504f7_jenkins"'
|
|
assert kw["version"] == 1
|
|
return {}
|
|
return {}
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
# User passes the clean name; resolved to the hash-prefixed name for DSM,
|
|
# but the display name in the success message must be hash-stripped.
|
|
result = await tools[tool_name]("jenkins", **kwargs)
|
|
assert "jenkins" in result
|
|
assert "f93cb8b504f7" not in result
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"tool_name, kwargs, error_verb",
|
|
[
|
|
("start_container", {}, "starting"),
|
|
("stop_container", {"confirmed": True}, "stopping"),
|
|
("restart_container", {"confirmed": True}, "restarting"),
|
|
],
|
|
)
|
|
@pytest.mark.asyncio
|
|
async def test_lifecycle_api_error(tool_name, kwargs, error_verb):
|
|
"""API errors during lifecycle ops return a human-readable error message."""
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.containers import register_containers
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("API error", code=102)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_containers(mcp, make_config(), client)
|
|
|
|
result = await tools[tool_name]("myapp_web", **kwargs)
|
|
assert "Error" in result
|
|
assert error_verb in result
|
|
assert "myapp_web" in result
|