feat: v0.5.0 — welle B Teil 1 (registry tools: search, tags, pull)

Three new SYNO.Docker.Registry tools, reverse-engineered from a live DSM API capture (n4s4 reference disagrees on param names and methods). - search_registry (#5): SYNO.Docker.Registry/search v1 with JSON-encoded q, plus offset/limit/page_size. Renders stars, downloads, official flag, truncated description, and total match count. Read-only. - list_image_tags: SYNO.Docker.Registry/tags v1 with JSON-encoded repo (not name — DSM live capture diverges from n4s4). Response shape is unusual: tag list comes back as the envelope's data field directly. Output capped by limit (default 50); accepts both list and dict response shapes defensively. Read-only. - pull_image (#3): SYNO.Docker.Registry/pull_start v1 with both repository and tag JSON-encoded. Async pull — no pull_status method confirmed on this DSM, so completion is detected by polling SYNO.Docker.Image/list (2–10 s backoff, 240 s budget under the Claude Desktop ~4 min tool-call ceiling). Timeout returns a non-fatal "still running" hint. Short-circuits when the image is already present locally. Confirmation gate required. Tool count: 31 → 34. CLAUDE.md confirmation list updated. New DSM quirks documented for pull_start (no pull_status) and tags (repo param name, top-level data array). Closes #3 Closes #5 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-18 13:25:59 +02:00
parent 82e8167f67
commit f27a5456f6
7 changed files with 639 additions and 6 deletions
@@ -0,0 +1,385 @@
+"""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; pull_start must NOT fire.
+    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.Registry" 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 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
+    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.Registry" 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.Registry" 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