marcus
24c3059e83
partition(":") split at the first colon, so registry-prefixed images
like "ghcr.io/open-webui/open-webui:v0.8.10" produced name="ghcr"
instead of the correct repository name, causing DSM error 114.
rpartition(":") always splits at the last colon, correctly handling:
- plain images: "nginx:1.24" → name="nginx", tag="1.24"
- namespaced: "nouchka/sqlite3:latest" → correct split
- registry URLs: "ghcr.io/foo/bar:v1" → name="ghcr.io/foo/bar", tag="v1"
- no tag: "nginx" → name="nginx", tag="latest" (fallback)
Added regression test verifying correct params sent to DSM delete API.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
475 lines
15 KiB
Python
475 lines
15 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():
|
|
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": []}
|
|
if api == "SYNO.Docker.Image" and method == "delete":
|
|
return {}
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_not_found():
|
|
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
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_in_use_blocked():
|
|
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 SAMPLE_CONTAINERS # nginx is in use
|
|
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 "my-nginx" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_image_by_hash():
|
|
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": []}
|
|
if api == "SYNO.Docker.Image" and method == "delete":
|
|
return {}
|
|
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
|
|
|
|
|
|
@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 ':'."""
|
|
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()
|
|
|
|
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": []}
|
|
if api == "SYNO.Docker.Image" and method == "delete":
|
|
return {}
|
|
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
|
|
|
|
# Verify delete was called with correct split params (not "ghcr" as name)
|
|
delete_call = next(c for c in client.request.call_args_list if c.args[1] == "delete")
|
|
params = delete_call.kwargs.get("params") or {}
|
|
assert params.get("name") == "ghcr.io/open-webui/open-webui"
|
|
assert params.get("tag") == "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()
|
|
|
|
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": []}
|
|
if api == "SYNO.Docker.Image" and method == "delete":
|
|
raise SynologyError("delete failed", code=1)
|
|
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
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# 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
|