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

"""Tests for modules/containers.py."""

import pytest
from unittest.mock import AsyncMock


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_CONTAINERS_DATA = {
    "containers": [
        {
            "name": "myapp_web",
            "status": "running",
            "image": "nginx:alpine",
            "project_name": "myapp",
        },
        {
            "name": "myapp_db",
            "status": "running",
            "image": "postgres:15",
            "project_name": "myapp",
        },
        {
            "name": "other_svc",
            "status": "stopped",
            "image": "redis:7",
            "project_name": "other",
        },
    ]
}

SAMPLE_LOGS_DATA = {
    "logs": [
        {
            "created": "2025-01-01T10:00:00Z",
            "stream": "stdout",
            "text": "Server started",
            "docid": "1",
        },
        {
            "created": "2025-01-01T10:00:01Z",
            "stream": "stderr",
            "text": "Warning: deprecated option",
            "docid": "2",
        },
    ],
    "total": 2,
}


@pytest.mark.asyncio
async def test_list_containers_all():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()
    client.request.return_value = SAMPLE_CONTAINERS_DATA

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

    result = await tools["list_containers"]()
    assert "myapp_web" in result
    assert "myapp_db" in result
    assert "other_svc" in result


@pytest.mark.asyncio
async def test_list_containers_filtered_by_project():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()
    client.request.return_value = SAMPLE_CONTAINERS_DATA

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

    result = await tools["list_containers"](project_name="myapp")
    assert "myapp_web" in result
    assert "myapp_db" in result
    assert "other_svc" not in result


@pytest.mark.asyncio
async def test_list_containers_empty():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()
    client.request.return_value = {"containers": []}

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

    result = await tools["list_containers"]()
    assert "No containers found" in result


@pytest.mark.asyncio
async def test_get_container_logs():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()
    client.request.return_value = SAMPLE_LOGS_DATA

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

    result = await tools["get_container_logs"]("myapp_web", tail=50)
    assert "myapp_web" in result
    assert "Server started" in result
    assert "Warning: deprecated option" in result


@pytest.mark.asyncio
async def test_get_container_logs_with_keyword():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()
    client.request.return_value = SAMPLE_LOGS_DATA

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

    await tools["get_container_logs"]("myapp_web", tail=100, keyword="error")
    call_params = client.request.call_args[1]["params"]
    assert call_params["keyword"] == "error"


@pytest.mark.asyncio
async def test_exec_in_container_requires_confirmation():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()

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

    result = await tools["exec_in_container"]("myapp_web", "ls /app", confirmed=False)
    assert "confirmed=True" in result
    client.request.assert_not_called()


@pytest.mark.asyncio
async def test_exec_in_container_confirmed():
    from mcp_synology_container.modules.containers import register_containers

    client = AsyncMock()
    client.request.return_value = {"output": "file1.py\nfile2.py", "exit_code": 0}

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

    result = await tools["exec_in_container"]("myapp_web", "ls /app", confirmed=True)
    assert "file1.py" in result
    assert "Exit code: 0" in result