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

"""Tests for modules/compose.py."""

from unittest.mock import AsyncMock

import pytest
import yaml

from mcp_synology_container.modules.compose import _validate_project_name


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),
        compose_base_path="/volume1/docker",
    )


SAMPLE_COMPOSE = """
services:
  web:
    image: nginx:1.24
    ports:
      - "80:80"
    environment:
      - APP_ENV=production
      - LOG_LEVEL=info
  db:
    image: postgres:15
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: admin
"""


def make_compose_client(content: str, filename: str = "docker-compose.yml") -> AsyncMock:
    """Create an AsyncMock client pre-configured for compose tests.

    FileStation.List returns a single file entry so that _find_compose_path
    can locate the compose file.  All other requests return {}.
    download_text returns the provided content.
    """
    client = AsyncMock()

    async def _request(api, method, **kwargs):
        if api == "SYNO.FileStation.List":
            return {"files": [{"name": filename}]}
        return {}

    client.request.side_effect = _request
    client.download_text.return_value = content
    return client


@pytest.mark.asyncio
async def test_read_compose():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)

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

    result = await tools["read_compose"]("myapp")
    assert "nginx:1.24" in result
    assert "postgres:15" in result


@pytest.mark.asyncio
async def test_read_compose_not_found():
    from mcp_synology_container.modules.compose import register_compose

    client = AsyncMock()
    # Simulate all FileStation.Info calls failing
    from mcp_synology_container.dsm_client import SynologyError

    client.request.side_effect = SynologyError("not found", code=408)

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

    result = await tools["read_compose"]("nonexistent")
    assert "No compose file found" in result


@pytest.mark.asyncio
async def test_update_image_tag_requires_confirmation():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "web", "1.25", confirmed=False)
    assert "confirmed=True" in result
    assert "nginx:1.24" in result
    assert "nginx:1.25" in result
    client.upload_text.assert_not_called()


@pytest.mark.asyncio
async def test_update_image_tag_confirmed():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "web", "1.25", confirmed=True)
    assert "nginx:1.24 → nginx:1.25" in result
    assert "redeploy_project" in result
    client.upload_text.assert_called_once()

    # Verify the uploaded content has the new tag
    uploaded_content = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded_content)
    assert parsed["services"]["web"]["image"] == "nginx:1.25"


@pytest.mark.asyncio
async def test_update_image_tag_service_not_found():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "nonexistent", "1.25", confirmed=True)
    assert "not found" in result
    assert "web" in result  # should list available services


@pytest.mark.asyncio
async def test_update_env_var_new_var_list_format():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_env_var"]("myapp", "web", "NEW_VAR", "value123", confirmed=True)
    assert "NEW_VAR=value123" in result

    uploaded_content = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded_content)
    env = parsed["services"]["web"]["environment"]
    assert any("NEW_VAR=value123" in str(e) for e in env)


@pytest.mark.asyncio
async def test_update_env_var_update_existing_list():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_env_var"]("myapp", "web", "LOG_LEVEL", "debug", confirmed=True)
    assert "LOG_LEVEL=debug" in result

    uploaded_content = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded_content)
    env = parsed["services"]["web"]["environment"]
    assert "LOG_LEVEL=debug" in env
    assert "LOG_LEVEL=info" not in env


@pytest.mark.asyncio
async def test_update_env_var_dict_format():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    # db service has dict-format environment
    result = await tools["update_env_var"]("myapp", "db", "POSTGRES_DB", "newdb", confirmed=True)
    assert "POSTGRES_DB=newdb" in result

    uploaded_content = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded_content)
    assert parsed["services"]["db"]["environment"]["POSTGRES_DB"] == "newdb"


@pytest.mark.asyncio
async def test_update_compose_invalid_yaml():
    from mcp_synology_container.modules.compose import register_compose

    # YAML validation happens before any file I/O — no compose file needed
    client = AsyncMock()
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_compose"]("myapp", "not: valid: yaml: {{{{", confirmed=True)
    assert "Invalid YAML" in result


@pytest.mark.asyncio
async def test_update_compose_missing_services_key():
    from mcp_synology_container.modules.compose import register_compose

    client = AsyncMock()
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_compose"]("myapp", "version: '3'\n", confirmed=True)
    assert "services" in result


@pytest.mark.asyncio
async def test_update_compose_requires_confirmation():
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_compose"]("myapp", SAMPLE_COMPOSE, confirmed=False)
    assert "confirmed=True" in result
    client.upload_text.assert_not_called()


# ──────────────────────────────────────────────────────────────────────────────
# Auto-version-update in update_image_tag
# ──────────────────────────────────────────────────────────────────────────────

SAMPLE_COMPOSE_VERSIONED = """
services:
  jenkins:
    image: jenkins/jenkins:2.558-jdk21
    environment:
      - JENKINS_VERSION=2.558
      - JAVA_OPTS=-Xmx512m
"""

SAMPLE_COMPOSE_VERSIONED_DICT_ENV = """
services:
  jenkins:
    image: jenkins/jenkins:2.558-jdk21
    environment:
      JENKINS_VERSION: "2.558"
      JAVA_OPTS: -Xmx512m
"""


@pytest.mark.asyncio
async def test_update_image_tag_auto_updates_version_env_var_list():
    """Tag 2.558-jdk21 → 2.560-jdk21 must also update JENKINS_VERSION=2.558 → 2.560."""
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE_VERSIONED)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "jenkins", "2.560-jdk21", confirmed=True)
    assert "jenkins/jenkins:2.558-jdk21 → jenkins/jenkins:2.560-jdk21" in result
    assert "JENKINS_VERSION=2.560" in result
    client.upload_text.assert_called_once()

    uploaded = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded)
    assert parsed["services"]["jenkins"]["image"] == "jenkins/jenkins:2.560-jdk21"
    env = parsed["services"]["jenkins"]["environment"]
    assert "JENKINS_VERSION=2.560" in env
    assert "JENKINS_VERSION=2.558" not in env


@pytest.mark.asyncio
async def test_update_image_tag_auto_updates_version_env_var_dict():
    """Dict-format env: JENKINS_VERSION value matching old prefix must be updated."""
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE_VERSIONED_DICT_ENV)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "jenkins", "2.560-jdk21", confirmed=True)
    assert "JENKINS_VERSION=2.560" in result

    uploaded = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded)
    assert parsed["services"]["jenkins"]["environment"]["JENKINS_VERSION"] == "2.560"


@pytest.mark.asyncio
async def test_update_image_tag_no_auto_update_without_version_suffix():
    """Tag without numeric-prefix pattern (e.g. 'latest') must not touch env vars."""
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE_VERSIONED)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "jenkins", "latest", confirmed=True)
    assert "jenkins/jenkins:2.558-jdk21 → jenkins/jenkins:latest" in result
    # No auto-update mention expected
    assert "Auto-updated" not in result

    uploaded = client.upload_text.call_args[0][2]
    parsed = yaml.safe_load(uploaded)
    env = parsed["services"]["jenkins"]["environment"]
    # JENKINS_VERSION must be unchanged
    assert "JENKINS_VERSION=2.558" in env


@pytest.mark.asyncio
async def test_update_image_tag_preview_shows_auto_update():
    """Unconfirmed call must preview auto-update of matching env vars."""
    from mcp_synology_container.modules.compose import register_compose

    client = make_compose_client(SAMPLE_COMPOSE_VERSIONED)
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("myapp", "jenkins", "2.560-jdk21", confirmed=False)
    assert "confirmed=True" in result
    assert "JENKINS_VERSION" in result
    assert "2.558" in result
    assert "2.560" in result
    client.upload_text.assert_not_called()


def test_extract_version_prefix():
    """Unit tests for _extract_version_prefix helper."""
    from mcp_synology_container.modules.compose import _extract_version_prefix

    assert _extract_version_prefix("2.558-jdk21") == "2.558"
    assert _extract_version_prefix("1.2.3-alpine") == "1.2.3"
    assert _extract_version_prefix("2-slim") == "2"
    assert _extract_version_prefix("latest") is None
    assert _extract_version_prefix("1.24") is None  # no suffix
    assert _extract_version_prefix("") is None
    assert _extract_version_prefix("v2.0-rc1") is None  # starts with 'v'


# ──────────────────────────────────────────────────────────────────────
#  _validate_project_name — path-traversal guard
# ──────────────────────────────────────────────────────────────────────


@pytest.mark.parametrize(
    "name",
    [
        "myapp",
        "MyApp",
        "my-app",
        "my_app",
        "app123",
        "A",
        "1",
        "snake_case-and-dash_42",
    ],
)
def test_validate_project_name_accepts_safe_names(name: str) -> None:
    assert _validate_project_name(name) is None


@pytest.mark.parametrize(
    "name",
    [
        "",  # empty
        "../etc",  # parent traversal
        "../../etc/passwd",  # multi-level traversal
        "foo/../bar",  # embedded traversal
        "foo/bar",  # forward slash
        "foo\\bar",  # backslash
        ".",  # bare dot
        "..",  # bare dotdot
        ".hidden",  # leading dot
        "foo.bar",  # dot inside (.yaml extensions, etc.)
        "foo bar",  # whitespace
        " foo",  # leading space
        "foo ",  # trailing space
        "foo\tbar",  # tab
        "foo\nbar",  # newline
        "foo;rm",  # shell metachar
        "foo|bar",
        "foo&bar",
        "foo*bar",
        "foo?bar",
        "foo:bar",
        "foo'bar",
        'foo"bar',
        "foo$bar",
        "foo`bar",
        "café",  # non-ASCII letter
    ],
)
def test_validate_project_name_rejects_unsafe_names(name: str) -> None:
    msg = _validate_project_name(name)
    assert msg is not None
    assert "invalid project name" in msg


@pytest.mark.asyncio
async def test_read_compose_rejects_traversal_name() -> None:
    """Traversal name must be rejected before any DSM call."""
    from mcp_synology_container.modules.compose import register_compose

    client = AsyncMock()
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["read_compose"]("../../etc")
    assert "invalid project name" in result
    client.request.assert_not_called()
    client.download_text.assert_not_called()


@pytest.mark.asyncio
async def test_update_compose_rejects_traversal_name() -> None:
    """update_compose with a traversal name must not validate YAML or upload."""
    from mcp_synology_container.modules.compose import register_compose

    client = AsyncMock()
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_compose"](
        "foo/../bar", "services:\n  web:\n    image: nginx\n", confirmed=True
    )
    assert "invalid project name" in result
    client.upload_text.assert_not_called()


@pytest.mark.asyncio
async def test_update_image_tag_rejects_traversal_name() -> None:
    from mcp_synology_container.modules.compose import register_compose

    client = AsyncMock()
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_image_tag"]("foo/bar", "web", "1.25", confirmed=True)
    assert "invalid project name" in result
    client.upload_text.assert_not_called()
    client.download_text.assert_not_called()


@pytest.mark.asyncio
async def test_update_env_var_rejects_traversal_name() -> None:
    from mcp_synology_container.modules.compose import register_compose

    client = AsyncMock()
    mcp, tools = make_mock_mcp()
    register_compose(mcp, make_config(), client)

    result = await tools["update_env_var"]("..", "web", "FOO", "bar", confirmed=True)
    assert "invalid project name" in result
    client.upload_text.assert_not_called()
    client.download_text.assert_not_called()