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

"""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


# ──────────────────────────────────────────────────────────────────────────────
#  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