mcp-synology-container/tests/test_modules/test_registry.py

"""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_starts_background():
    """pull_start is invoked, then the tool returns immediately with a background hint —
    it does NOT poll Image/list until the tag appears (fire-and-return)."""
    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": []}  # not present locally → proceed to pull
        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 "started in the background" in result
    assert "nginx:1.24" in result
    assert "check_image_updates" in result

    # Verify pull_start was invoked on SYNO.Docker.Image (not Registry)
    # with JSON-encoded params, exactly once.
    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

    # Exactly one Image/list pre-check — no post-pull polling loop.
    list_calls = [c for c in client.request.call_args_list if c.args[1] == "list"]
    assert len(list_calls) == 1


@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