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

"""Tests for modules/images.py."""

from unittest.mock import AsyncMock, patch

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


# ──────────────────────────────────────────────────────────────────────────────
#  pull_image
# ──────────────────────────────────────────────────────────────────────────────

PULLED_IMAGE = {
    "id": "sha256:dddd",
    "repository": "postgres",
    "tags": ["17.8"],
    "size": 80 * 1024 * 1024,
    "created": 1700000000,
    "upgradable": False,
}


@pytest.mark.asyncio
async def test_pull_image_success():
    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 == "pull":
            return {}
        if api == "SYNO.Docker.Image" and method == "list":
            return {"images": [PULLED_IMAGE]}
        return {}

    client.request.side_effect = mock_request
    mcp, tools = make_mock_mcp()
    register_images(mcp, make_config(), client)

    with patch("asyncio.sleep", new=AsyncMock()):
        result = await tools["pull_image"]("postgres:17.8")

    assert "Pulled postgres:17.8" in result
    assert "MiB" in result


@pytest.mark.asyncio
async def test_pull_image_no_tag_defaults_to_latest():
    from mcp_synology_container.modules.images import register_images

    client = AsyncMock()
    found_image = {**PULLED_IMAGE, "repository": "nginx", "tags": ["latest"]}

    async def mock_request(api, method, **kwargs):
        if api == "SYNO.Docker.Image" and method == "pull":
            # Verify tag defaults to "latest"
            assert kwargs.get("params", {}).get("tag") == "latest"
            return {}
        if api == "SYNO.Docker.Image" and method == "list":
            return {"images": [found_image]}
        return {}

    client.request.side_effect = mock_request
    mcp, tools = make_mock_mcp()
    register_images(mcp, make_config(), client)

    with patch("asyncio.sleep", new=AsyncMock()):
        result = await tools["pull_image"]("nginx")

    assert "Pulled nginx:latest" in result


@pytest.mark.asyncio
async def test_pull_image_registry_prefixed():
    """Registry-prefixed images (e.g. ghcr.io/foo/bar:v1) split at last ':'."""
    from mcp_synology_container.modules.images import register_images

    client = AsyncMock()
    found_image = {
        **PULLED_IMAGE,
        "repository": "ghcr.io/open-webui/open-webui",
        "tags": ["v0.9.0"],
    }

    async def mock_request(api, method, **kwargs):
        if api == "SYNO.Docker.Image" and method == "pull":
            params = kwargs.get("params", {})
            assert params["repository"] == "ghcr.io/open-webui/open-webui"
            assert params["tag"] == "v0.9.0"
            return {}
        if api == "SYNO.Docker.Image" and method == "list":
            return {"images": [found_image]}
        return {}

    client.request.side_effect = mock_request
    mcp, tools = make_mock_mcp()
    register_images(mcp, make_config(), client)

    with patch("asyncio.sleep", new=AsyncMock()):
        result = await tools["pull_image"]("ghcr.io/open-webui/open-webui:v0.9.0")

    assert "Pulled ghcr.io/open-webui/open-webui:v0.9.0" in result


@pytest.mark.asyncio
async def test_pull_image_timeout():
    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 == "pull":
            return {}
        # image never appears in list
        return {"images": []}

    client.request.side_effect = mock_request
    mcp, tools = make_mock_mcp()
    register_images(mcp, make_config(), client)

    with patch("asyncio.sleep", new=AsyncMock()):
        result = await tools["pull_image"]("nonexistent:latest")

    assert "did not complete" in result
    assert "120" in result


@pytest.mark.asyncio
async def test_pull_image_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("pull failed", code=400)

    mcp, tools = make_mock_mcp()
    register_images(mcp, make_config(), client)

    with patch("asyncio.sleep", new=AsyncMock()):
        result = await tools["pull_image"]("bad/image:tag")

    assert "Error" in result
    assert "bad/image:tag" in result