thePlug/backend/tests/integrations/test_docker_compose.py

import asyncio
from pathlib import Path

from app.integrations.docker.compose import get_all_services_in_dir
from app.integrations.docker.editor import change_port


def test_get_all_services_in_dir_scans_immediate_children(tmp_path: Path):
    root = tmp_path / "workspace"
    project = root / "playground"
    project.mkdir(parents=True)
    (project / "docker-compose.yml").write_text(
        """
services:
  echo:
    image: hashicorp/http-echo:1.0
    ports:
      - "9081:5678"
""".strip(),
        encoding="utf-8",
    )

    services = get_all_services_in_dir(root)

    assert services == [
        {
            "name": "echo",
            "ports": [9081],
            "bindings": [{"published": 9081, "host": None}],
            "folder": project,
        }
    ]


def test_get_all_services_in_dir_skips_nested_grandchildren(tmp_path: Path):
    root = tmp_path / "workspace"
    nested = root / "group" / "playground"
    nested.mkdir(parents=True)
    (nested / "docker-compose.yml").write_text(
        """
services:
  echo:
    image: hashicorp/http-echo:1.0
    ports:
      - "9081:5678"
""".strip(),
        encoding="utf-8",
    )

    services = get_all_services_in_dir(root)

    assert services == []


def test_get_all_services_in_dir_returns_empty_when_root_is_inaccessible(
    tmp_path: Path,
    monkeypatch,
):
    root = tmp_path / "workspace"
    root.mkdir()
    original_iterdir = Path.iterdir

    def fake_iterdir(self: Path):
        if self == root:
            raise PermissionError("denied")
        return original_iterdir(self)

    monkeypatch.setattr(Path, "iterdir", fake_iterdir)

    assert get_all_services_in_dir(root) == []


def test_get_all_services_in_dir_skips_inaccessible_child_folder(
    tmp_path: Path,
    monkeypatch,
):
    root = tmp_path / "workspace"
    allowed = root / "allowed"
    blocked = root / "blocked"
    allowed.mkdir(parents=True)
    blocked.mkdir(parents=True)
    (allowed / "docker-compose.yml").write_text(
        """
services:
  echo:
    image: hashicorp/http-echo:1.0
    ports:
      - "9081:5678"
""".strip(),
        encoding="utf-8",
    )
    original_is_dir = Path.is_dir

    def fake_is_dir(self: Path):
        if self == blocked:
            raise PermissionError("denied")
        return original_is_dir(self)

    monkeypatch.setattr(Path, "is_dir", fake_is_dir)

    services = get_all_services_in_dir(root)

    assert services == [
        {
            "name": "echo",
            "ports": [9081],
            "bindings": [{"published": 9081, "host": None}],
            "folder": allowed,
        }
    ]


def test_change_port_restores_original_compose_when_reload_fails(
    tmp_path: Path,
    monkeypatch,
):
    folder = tmp_path / "workspace"
    folder.mkdir()
    compose_path = folder / "docker-compose.yml"
    original = """
services:
  echo:
    image: hashicorp/http-echo:1.0
    ports:
      - "9081:5678"
""".strip()
    compose_path.write_text(original, encoding="utf-8")

    async def fake_run_docker_command(
        _file: Path,
        _command: list[str],
    ) -> dict | None:
        return {"status": "error", "reason": "boom"}

    monkeypatch.setattr(
        "app.integrations.docker.editor.run_docker_command",
        fake_run_docker_command,
    )

    changed = asyncio.run(change_port("echo", folder, 9081, 54001))

    assert changed is False
    assert compose_path.read_text(encoding="utf-8") == original
    assert (folder / "docker-compose.bak.yml").exists() is False


def test_change_port_restores_original_compose_when_reload_raises(
    tmp_path: Path,
    monkeypatch,
):
    folder = tmp_path / "workspace"
    folder.mkdir()
    compose_path = folder / "docker-compose.yml"
    original = """
services:
  echo:
    image: hashicorp/http-echo:1.0
    ports:
      - "127.0.0.1:54001:5678"
""".strip()
    compose_path.write_text(original, encoding="utf-8")

    async def fake_run_docker_command(
        _file: Path,
        _command: list[str],
    ) -> dict | None:
        raise RuntimeError("docker unavailable")

    monkeypatch.setattr(
        "app.integrations.docker.editor.run_docker_command",
        fake_run_docker_command,
    )

    changed = asyncio.run(
        change_port(
            "echo",
            folder,
            54001,
            9081,
            host_to=None,
            rewrite_host=True,
        )
    )

    assert changed is False
    assert compose_path.read_text(encoding="utf-8") == original