marcus
18fe063691
The 0.5.0 prompt mis-attributed the API: pull_start lives on SYNO.Docker.Image, not SYNO.Docker.Registry (live DSM capture). search and tags ARE correctly on Registry; only pull_start belongs to Image. Registry/pull_start returns "Method does not exist". Parameters are unchanged (repository + tag both JSON-encoded), and the Image/list polling for completion detection is untouched. Tests updated to assert SYNO.Docker.Image/pull_start. CLAUDE.md DSM-quirks section consolidates the Image vs. Registry split so this trap is documented for future surface additions. References #3 (already closed). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
388 lines
13 KiB
Python
388 lines
13 KiB
Python
"""Tests for modules/registry.py."""
|
|
|
|
import json
|
|
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),
|
|
)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# search_registry — Issue #5
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
SEARCH_RESPONSE = {
|
|
"data": [
|
|
{
|
|
"name": "caddy",
|
|
"description": (
|
|
"Caddy 2 is a powerful, enterprise-ready, open source web "
|
|
"server with automatic HTTPS written in Go."
|
|
),
|
|
"downloads": 650989718,
|
|
"is_automated": False,
|
|
"is_official": True,
|
|
"star_count": 881,
|
|
"registry": "https://registry.hub.docker.com",
|
|
},
|
|
{
|
|
"name": "abiosoft/caddy",
|
|
"description": "Caddy is a lightweight, general-purpose web server.",
|
|
"downloads": 111974910,
|
|
"is_automated": True,
|
|
"is_official": False,
|
|
"star_count": 289,
|
|
"registry": "https://registry.hub.docker.com",
|
|
},
|
|
],
|
|
"limit": 20,
|
|
"offset": 0,
|
|
"page_size": 20,
|
|
"total": 6647,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_registry_renders_hits():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SEARCH_RESPONSE
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["search_registry"](query="caddy")
|
|
|
|
assert "caddy" in result
|
|
assert "abiosoft/caddy" in result
|
|
assert "[official]" in result
|
|
assert "881" in result # stars
|
|
assert "650989718" in result # downloads
|
|
assert "2 of 6647" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_registry_uses_json_encoded_query():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = SEARCH_RESPONSE
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
await tools["search_registry"](query="nginx", limit=10)
|
|
|
|
call = client.request.call_args
|
|
assert call.args[0] == "SYNO.Docker.Registry"
|
|
assert call.args[1] == "search"
|
|
assert call.kwargs.get("version") == 1
|
|
params = call.kwargs.get("params", {})
|
|
assert json.loads(params["q"]) == "nginx"
|
|
assert params["offset"] == "0"
|
|
assert params["limit"] == "10"
|
|
assert params["page_size"] == "10"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_registry_empty():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = {"data": [], "total": 0}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["search_registry"](query="does-not-exist")
|
|
assert "No results" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_registry_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("Auth failure", code=119)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["search_registry"](query="caddy")
|
|
assert "Error" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# list_image_tags — bonus
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_image_tags_array_response():
|
|
"""DSM returns the envelope's data field directly as a list."""
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = [
|
|
{"tag": "latest"},
|
|
{"tag": "3.20"},
|
|
{"tag": "3.19"},
|
|
]
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["list_image_tags"](repository="alpine")
|
|
|
|
assert "alpine" in result
|
|
assert "latest" in result
|
|
assert "3.20" in result
|
|
assert "3.19" in result
|
|
assert "3 total" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_image_tags_uses_json_repo_param():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.return_value = [{"tag": "latest"}]
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
await tools["list_image_tags"](repository="grafana/grafana")
|
|
|
|
call = client.request.call_args
|
|
assert call.args[0] == "SYNO.Docker.Registry"
|
|
assert call.args[1] == "tags"
|
|
assert call.kwargs.get("version") == 1
|
|
params = call.kwargs.get("params", {})
|
|
assert json.loads(params["repo"]) == "grafana/grafana"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_image_tags_limit_truncates():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
tags = [{"tag": f"v{i}"} for i in range(120)]
|
|
client = AsyncMock()
|
|
client.request.return_value = tags
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["list_image_tags"](repository="alpine", limit=10)
|
|
assert "120 total" in result
|
|
assert "Showing first 10 of 120" in result
|
|
# Tag #10 must not appear (only v0..v9)
|
|
assert " v9" in result
|
|
assert " v10" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_image_tags_empty():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
# Empty list → DsmClient.request coerces to {} via `or {}`
|
|
client.request.return_value = {}
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["list_image_tags"](repository="nonexistent/image")
|
|
assert "No tags found" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_image_tags_api_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = SynologyError("Boom", code=100)
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["list_image_tags"](repository="alpine")
|
|
assert "Error" in result
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# pull_image — Issue #3
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_image_preview():
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["pull_image"](repository="nginx", tag="1.24")
|
|
assert "Preview" in result
|
|
assert "nginx:1.24" in result
|
|
assert "confirmed=True" in result
|
|
client.request.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_image_already_present():
|
|
"""If image is already in Image/list, no pull_start call is made."""
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
client = AsyncMock()
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return {
|
|
"images": [
|
|
{"id": "sha256:aaaa", "repository": "nginx", "tags": ["1.24"]},
|
|
]
|
|
}
|
|
raise AssertionError(f"Unexpected call: {api}/{method}")
|
|
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["pull_image"](repository="nginx", tag="1.24", confirmed=True)
|
|
assert "already present" in result
|
|
|
|
# Only the Image/list pre-check was called; no pull_start of any kind.
|
|
calls = client.request.call_args_list
|
|
assert all(c.args[1] != "pull_start" for c in calls)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_image_confirmed_success(monkeypatch):
|
|
"""pull_start succeeds, then Image/list shows the new tag on first poll."""
|
|
import mcp_synology_container.modules.registry as registry_mod
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
# Make polling instant
|
|
async def fake_sleep(_):
|
|
return None
|
|
|
|
monkeypatch.setattr(registry_mod.asyncio, "sleep", fake_sleep)
|
|
|
|
state = {"pulled": False}
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
if state["pulled"]:
|
|
return {
|
|
"images": [
|
|
{"id": "sha256:aaaa", "repository": "nginx", "tags": ["1.24"]},
|
|
]
|
|
}
|
|
return {"images": []}
|
|
if api == "SYNO.Docker.Image" and method == "pull_start":
|
|
state["pulled"] = True
|
|
return {}
|
|
raise AssertionError(f"Unexpected call: {api}/{method}")
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["pull_image"](repository="nginx", tag="1.24", confirmed=True)
|
|
assert "Pulled" in result
|
|
assert "nginx:1.24" in result
|
|
|
|
# Verify pull_start was invoked on SYNO.Docker.Image (not Registry)
|
|
# with JSON-encoded params.
|
|
pull_calls = [c for c in client.request.call_args_list if c.args[1] == "pull_start"]
|
|
assert len(pull_calls) == 1
|
|
assert pull_calls[0].args[0] == "SYNO.Docker.Image"
|
|
params = pull_calls[0].kwargs.get("params", {})
|
|
assert json.loads(params["repository"]) == "nginx"
|
|
assert json.loads(params["tag"]) == "1.24"
|
|
assert pull_calls[0].kwargs.get("version") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_image_timeout(monkeypatch):
|
|
"""If the image never appears, the tool returns a still-running hint."""
|
|
import mcp_synology_container.modules.registry as registry_mod
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
# Make polling instant
|
|
async def fake_sleep(_):
|
|
return None
|
|
|
|
monkeypatch.setattr(registry_mod.asyncio, "sleep", fake_sleep)
|
|
# Shrink the budget so the loop exits quickly
|
|
monkeypatch.setattr(registry_mod, "_PULL_POLL_TIMEOUT", 0.05)
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return {"images": []}
|
|
if api == "SYNO.Docker.Image" and method == "pull_start":
|
|
return {}
|
|
raise AssertionError(f"Unexpected call: {api}/{method}")
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["pull_image"](repository="nginx", tag="1.24", confirmed=True)
|
|
assert "still running" in result
|
|
assert "nginx:1.24" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_image_start_error():
|
|
from mcp_synology_container.dsm_client import SynologyError
|
|
from mcp_synology_container.modules.registry import register_registry
|
|
|
|
async def mock_request(api, method, **kwargs):
|
|
if api == "SYNO.Docker.Image" and method == "list":
|
|
return {"images": []}
|
|
if api == "SYNO.Docker.Image" and method == "pull_start":
|
|
raise SynologyError("Permission denied", code=105)
|
|
raise AssertionError(f"Unexpected call: {api}/{method}")
|
|
|
|
client = AsyncMock()
|
|
client.request.side_effect = mock_request
|
|
|
|
mcp, tools = make_mock_mcp()
|
|
register_registry(mcp, make_config(), client)
|
|
|
|
result = await tools["pull_image"](repository="nginx", tag="1.24", confirmed=True)
|
|
assert "Error starting pull" in result
|