marcus
82e8167f67
Three live-NAS observations that the 0.4.2 implementation got wrong:
1. Header showed "?:?" — DSM Image/get returns image="" and tag=""
for name:tag lookups; the response fields are unreliable. The
header now echoes the user-supplied image_id, falling back to the
sha256 id only when image_id itself is empty.
2. Ports printed as raw Python dicts ({'port':'22','protocol':'tcp'}).
The ports array is a list of {port, protocol} objects — each entry
now renders as "22/tcp".
3. Environment printed as raw Python dicts ({'key':'PATH','value':...}).
The env array is a list of {key, value} objects — each entry now
renders as "PATH=/usr/local/...".
cmd, entrypoint, and volumes (plain string arrays) were already
correct and are untouched. Tests updated to match the live shape;
two new tests guard the header fallback. References #4 (already
closed by 0.4.2 — no need to reopen for a rendering polish).
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
815 lines
27 KiB
Python
815 lines
27 KiB
Python
"""Tests for modules/images.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_IMAGES = {
|
|
"images": [
|
|
{
|
|
"id": "sha256:aaaa",
|
|
"repository": "nginx",
|
|
"tags": ["1.24"],
|
|
"size": 50 * 1024 * 1024,
|
|
"created": 1700000000,
|
|
"upgradable": True,
|
|
},
|
|
{
|
|
"id": "sha256:bbbb",
|
|
"repository": "postgres",
|
|
"tags": ["15"],
|
|
"size": 80 * 1024 * 1024,
|
|
"created": 1700000000,
|
|
"upgradable": False,
|
|
},
|
|
{
|
|
"id": "sha256:cccc",
|
|
"repository": "redis",
|
|
"tags": ["7"],
|
|
"size": 30 * 1024 * 1024,
|
|
"created": 1700000000,
|
|
"upgradable": False,
|
|
},
|
|
]
|
|
}
|
|
|
|
SAMPLE_CONTAINERS = {
|
|
"containers": [
|
|
{"name": "my-nginx", "image_id": "sha256:aaaa", "status": "running"},
|
|
]
|
|
}
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# list_images
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_images_sorted_by_size():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return SAMPLE_CONTAINERS
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["list_images"]()
|
|
# postgres (80 MiB) should appear before nginx (50 MiB) before redis (30 MiB)
|
|
pos_postgres = result.index("postgres")
|
|
pos_nginx = result.index("nginx")
|
|
pos_redis = result.index("redis")
|
|
assert pos_postgres < pos_nginx < pos_redis
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_images_shows_in_use():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return SAMPLE_CONTAINERS
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["list_images"]()
|
|
assert "[in use]" in result
|
|
assert "[update available]" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_images_no_images():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image":
|
|
return {"images": []}
|
|
return {"containers": []}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["list_images"]()
|
|
assert "No local images found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_images_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("API unavailable", code=102)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["list_images"]()
|
|
assert "Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_images_container_error_graceful():
|
|
"""Container list failure must not prevent image listing."""
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image":
|
|
return SAMPLE_IMAGES
|
|
raise SynologyError("containers unavailable", code=102)
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["list_images"]()
|
|
assert "postgres" in result # images still listed
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# delete_image
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_preview():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return {"containers": []}
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="redis:7")
|
|
assert "Preview" in result
|
|
assert "redis:7" in result
|
|
# Should not have called the delete method
|
|
calls = [str(c) for c in client.request.call_args_list]
|
|
assert not any("delete" in c for c in calls)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_confirmed():
|
|
import json
|
|
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return {"containers": []}
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="redis:7", confirmed=True)
|
|
assert "Deleted" in result
|
|
assert "redis:7" in result
|
|
assert "freed" in result
|
|
|
|
# post_request must be called with images JSON param, not name/tag/id
|
|
client.post_request.assert_called_once()
|
|
params = client.post_request.call_args.kwargs.get("params", {})
|
|
images = json.loads(params["images"])
|
|
assert images == [{"repository": "redis", "tags": ["7"]}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_not_found():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="nonexistent:latest", confirmed=True)
|
|
assert "not found" in result
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_in_use_by_running_blocked():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return SAMPLE_CONTAINERS # nginx is in use (running)
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="nginx:1.24", confirmed=True)
|
|
assert "Cannot delete" in result
|
|
assert "running" in result.lower()
|
|
assert "my-nginx" in result
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_in_use_by_stopped_blocked():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return {
|
|
"containers": [
|
|
{"name": "stopped-nginx", "image_id": "sha256:aaaa", "status": "exited"}
|
|
]
|
|
}
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="nginx:1.24", confirmed=True)
|
|
assert "Cannot delete" in result
|
|
assert "stopped" in result.lower()
|
|
assert "stopped-nginx" in result
|
|
assert "system_prune" in result
|
|
client.post_request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_by_hash():
|
|
import json
|
|
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return {"containers": []}
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="sha256:cccc", confirmed=True)
|
|
assert "Deleted" in result
|
|
assert "redis" in result
|
|
|
|
# Verify images param uses the resolved name+tag (not the hash)
|
|
call_kwargs = client.post_request.call_args
|
|
params = call_kwargs.kwargs.get("params") or {}
|
|
images = json.loads(params.get("images", "[]"))
|
|
assert images == [{"repository": "redis", "tags": ["7"]}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_registry_prefixed_name():
|
|
"""Registry-prefixed image names (e.g. ghcr.io/foo/bar:v1) must split at last ':'."""
|
|
import json
|
|
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
registry_images = {
|
|
"images": [
|
|
{
|
|
"id": "sha256:dddd",
|
|
"repository": "ghcr.io/open-webui/open-webui",
|
|
"tags": ["v0.8.10"],
|
|
"size": 100 * 1024 * 1024,
|
|
"created": 1700000000,
|
|
"upgradable": False,
|
|
}
|
|
]
|
|
}
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(return_value={})
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return registry_images
|
|
if api == "SYNO.Docker.Container":
|
|
return {"containers": []}
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](
|
|
image_id="ghcr.io/open-webui/open-webui:v0.8.10", confirmed=True
|
|
)
|
|
assert "Deleted" in result
|
|
assert "open-webui" in result
|
|
|
|
# images param must use full registry-prefixed repository name
|
|
call_kwargs = client.post_request.call_args
|
|
params = call_kwargs.kwargs.get("params") or {}
|
|
images = json.loads(params.get("images", "[]"))
|
|
assert images == [{"repository": "ghcr.io/open-webui/open-webui", "tags": ["v0.8.10"]}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.post_request = AsyncMock(side_effect=SynologyError("delete failed", code=114))
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Container":
|
|
return {"containers": []}
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["delete_image"](image_id="redis:7", confirmed=True)
|
|
assert "Error" in result
|
|
assert "114" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# inspect_image
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
# Authoritative DSM SYNO.Docker.Image/get response shape (v1).
|
|
# Live-captured peculiarities:
|
|
# - `image` and `tag` come back as empty strings when the lookup is
|
|
# by name:tag — the header therefore has to fall back to image_id.
|
|
# - `ports` is an array of {"port", "protocol"} dicts, NOT strings.
|
|
# - `env` is an array of {"key", "value"} dicts, NOT strings.
|
|
# - `volumes` and `cmd`/`entrypoint` are plain string arrays.
|
|
# - There is no `layers` field on this endpoint.
|
|
SAMPLE_INSPECT = {
|
|
"image": "",
|
|
"tag": "",
|
|
"id": "sha256:aaaa1234567890abcdef",
|
|
"digest": "sha256:digestabcdef",
|
|
"size": 50 * 1024 * 1024,
|
|
"virtual_size": 50 * 1024 * 1024,
|
|
"author": "NGINX Team",
|
|
"docker_version": "20.10.0",
|
|
"cmd": ["nginx", "-g", "daemon off;"],
|
|
"entrypoint": ["/docker-entrypoint.sh"],
|
|
"env": [
|
|
{"key": "NGINX_VERSION", "value": "1.24"},
|
|
{"key": "PATH", "value": "/usr/local/sbin"},
|
|
],
|
|
"ports": [
|
|
{"port": "80", "protocol": "tcp"},
|
|
{"port": "443", "protocol": "tcp"},
|
|
],
|
|
"volumes": ["/var/cache/nginx"],
|
|
}
|
|
|
|
|
|
def _make_inspect_client(inspect_payload=None):
|
|
"""Build a mock DsmClient that returns inspect_payload for SYNO.Docker.Image/get."""
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "get":
|
|
return inspect_payload if inspect_payload is not None else SAMPLE_INSPECT
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
return client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_by_name_tag():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "nginx" in result
|
|
assert "1.24" in result
|
|
assert "MiB" in result # size formatted via _human_size
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_by_hash():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
# Inspect data shaped for redis returned by hash lookup.
|
|
# image/tag come back empty (matches live DSM behavior) — display
|
|
# falls back to image_id, then to the id hash.
|
|
redis_inspect = {
|
|
"image": "",
|
|
"tag": "",
|
|
"id": "sha256:ccccredis",
|
|
"digest": "sha256:rdigest",
|
|
"size": 30 * 1024 * 1024,
|
|
"virtual_size": 30 * 1024 * 1024,
|
|
"cmd": ["redis-server"],
|
|
"entrypoint": [],
|
|
"env": [],
|
|
"ports": [{"port": "6379", "protocol": "tcp"}],
|
|
"volumes": [],
|
|
}
|
|
client = _make_inspect_client(inspect_payload=redis_inspect)
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="sha256:cccc")
|
|
# The header echoes the user-supplied image_id; "redis" surfaces via the cmd line.
|
|
assert "sha256:cccc" in result
|
|
assert "redis" in result
|
|
assert "6379/tcp" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_header_uses_image_id_when_fields_empty():
|
|
"""DSM Image/get returns image='' and tag='' for name:tag lookups — the header
|
|
must fall back to the user-supplied image_id, NOT render '?:?'."""
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client() # SAMPLE_INSPECT has image='' / tag=''
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="gitea/gitea:1.26.1")
|
|
assert "Image: gitea/gitea:1.26.1" in result
|
|
assert "?:?" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_header_falls_back_to_id_when_image_id_empty():
|
|
"""If image_id were empty (defensive — should not happen in practice), the
|
|
header falls back to the sha256 id field."""
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client() # SAMPLE_INSPECT.id == sha256:aaaa1234567890abcdef
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="")
|
|
assert "Image: sha256:aaaa1234567890abcdef" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_uses_identity_param():
|
|
"""DSM SYNO.Docker.Image/get expects parameter 'identity' (JSON-encoded), version=1.
|
|
|
|
Using name/tag/id (the old shape) returned DSM error 114 — this test guards
|
|
against regressing to that contract.
|
|
"""
|
|
import json
|
|
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
await tools["inspect_image"](image_id="nginx:1.24")
|
|
|
|
get_calls = [
|
|
c for c in client.request.call_args_list if c.args[:2] == ("SYNO.Docker.Image", "get")
|
|
]
|
|
assert len(get_calls) == 1
|
|
call = get_calls[0]
|
|
assert call.kwargs.get("version") == 1
|
|
params = call.kwargs.get("params") or {}
|
|
assert params == {"identity": json.dumps("nginx:1.24")}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_empty_response_treated_as_not_found():
|
|
"""An empty response dict surfaces as a clean 'not found' message."""
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client(inspect_payload={})
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="bogus:latest")
|
|
assert "not found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_shows_env_vars():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "NGINX_VERSION=1.24" in result
|
|
assert "PATH=/usr/local/sbin" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_shows_exposed_ports():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "80/tcp" in result
|
|
assert "443/tcp" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_shows_volumes():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "/var/cache/nginx" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_shows_entrypoint_cmd():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "/docker-entrypoint.sh" in result
|
|
assert "nginx" in result
|
|
assert "daemon off;" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_shows_digest_author_docker_version():
|
|
"""Identity-block fields specific to the DSM Image/get response are surfaced."""
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = _make_inspect_client()
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "sha256:digestabcdef" in result
|
|
assert "NGINX Team" in result
|
|
assert "20.10.0" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_registry_prefixed():
|
|
"""A registry-prefixed identifier like 'ghcr.io/foo/bar:v1' is passed through verbatim
|
|
in the JSON-encoded identity parameter."""
|
|
import json
|
|
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
registry_inspect = {
|
|
# image / tag come back empty for name:tag lookups (live DSM behavior).
|
|
"image": "",
|
|
"tag": "",
|
|
"id": "sha256:dddd",
|
|
"digest": "sha256:gdigest",
|
|
"size": 100 * 1024 * 1024,
|
|
"virtual_size": 100 * 1024 * 1024,
|
|
"cmd": ["/app"],
|
|
"entrypoint": [],
|
|
"env": [],
|
|
"ports": [],
|
|
"volumes": [],
|
|
}
|
|
client = _make_inspect_client(inspect_payload=registry_inspect)
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="ghcr.io/foo/bar:v1")
|
|
assert "ghcr.io/foo/bar" in result
|
|
assert "v1" in result
|
|
|
|
# The full identifier (with ':') is JSON-encoded into a single string —
|
|
# registry-prefixed names are not split into name + tag.
|
|
get_calls = [
|
|
c for c in client.request.call_args_list if c.args[:2] == ("SYNO.Docker.Image", "get")
|
|
]
|
|
assert get_calls
|
|
params = get_calls[0].kwargs.get("params") or {}
|
|
assert params.get("identity") == json.dumps("ghcr.io/foo/bar:v1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inspect_image_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "get":
|
|
raise SynologyError("inspect failed", code=120)
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["inspect_image"](image_id="nginx:1.24")
|
|
assert "Error" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# check_image_updates (existing tests preserved)
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_image_updates_all():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SAMPLE_IMAGES
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["check_image_updates"]()
|
|
assert "nginx:1.24" in result
|
|
assert "UPDATE AVAILABLE" in result
|
|
assert "postgres:15" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_image_updates_all_up_to_date():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {
|
|
"images": [
|
|
{
|
|
"id": "sha256:aaaa",
|
|
"repository": "nginx",
|
|
"tags": ["1.24"],
|
|
"size": 50 * 1024 * 1024,
|
|
"upgradable": False,
|
|
},
|
|
]
|
|
}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["check_image_updates"]()
|
|
assert "All images are up to date" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_image_updates_no_images():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {"images": []}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["check_image_updates"]()
|
|
assert "No images found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_image_updates_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("API unavailable", code=102)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["check_image_updates"]()
|
|
assert "Error" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_image_updates_for_project():
|
|
from mcp_synology_container.modules.images import register_images
|
|
|
|
project_list = {
|
|
"uuid-1": {
|
|
"id": "uuid-1",
|
|
"name": "myapp",
|
|
"status": "RUNNING",
|
|
"containerIds": ["abc123"],
|
|
}
|
|
}
|
|
project_detail = {
|
|
"containers": [
|
|
{"Image": "sha256:aaaa", "Config": {"Image": "nginx:1.24"}},
|
|
]
|
|
}
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image":
|
|
return SAMPLE_IMAGES
|
|
if api == "SYNO.Docker.Project" and method == "list":
|
|
return project_list
|
|
if api == "SYNO.Docker.Project" and method == "get":
|
|
return project_detail
|
|
return {}
|
|
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_images(mcp, make_config(), client)
|
|
|
|
result = await tools["check_image_updates"](project_name="myapp")
|
|
assert "myapp" in result
|
|
assert "nginx:1.24" in result
|