compose/tests/test_compose_manager.py

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING

import pytest
import yaml

from compose import DockerComposeManager

if TYPE_CHECKING:
    from pathlib import Path


def test_create_and_save_service(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    manager.create_service(
        name="web",
        image="nginx:latest",
        ports=["80:80"],
        environment={"ENV_VAR": "value"},
    ).save()
    # Check file exists and content
    assert compose_file.exists()
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert "web" in data["services"]
    assert data["services"]["web"]["image"] == "nginx:latest"
    assert data["services"]["web"]["ports"] == ["80:80"]
    assert data["services"]["web"]["environment"] == {"ENV_VAR": "value"}


def test_modify_service(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    manager.create_service(name="web", image="nginx:latest").save()
    manager.modify_service(name="web", image="nginx:1.19", ports=["8080:80"]).save()
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert data["services"]["web"]["image"] == "nginx:1.19"
    assert data["services"]["web"]["ports"] == ["8080:80"]


def test_remove_service(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    manager.create_service(name="web", image="nginx:latest").save()
    manager.remove_service("web").save()
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert "web" not in data["services"]


def test_context_manager(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    with DockerComposeManager(str(compose_file)) as manager:
        manager.create_service(name="web", image="nginx:latest")
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert "web" in data["services"]


def test_modify_nonexistent_service(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    with pytest.raises(KeyError):
        manager.modify_service("notfound", image="nginx:latest")


def test_remove_nonexistent_service(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    # Should not raise
    manager.remove_service("notfound").save()
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert "services" in data
    assert data["services"] == {}


def test_create_service_with_extra_kwargs(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    manager.create_service(
        name="db",
        image="postgres:latest",
        environment={"POSTGRES_PASSWORD": "example"},
        volumes=["db_data:/var/lib/postgresql/data"],
        depends_on=["web"],
    ).save()
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert "db" in data["services"]
    assert data["services"]["db"]["volumes"] == ["db_data:/var/lib/postgresql/data"]
    assert data["services"]["db"]["depends_on"] == ["web"]


def test_create_service_minimal(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    manager.create_service(name="worker", image="busybox").save()
    with compose_file.open() as f:
        data = yaml.safe_load(f)
    assert "worker" in data["services"]
    assert data["services"]["worker"]["image"] == "busybox"
    assert "ports" not in data["services"]["worker"]
    assert "environment" not in data["services"]["worker"]


def test_create_service_all_fields(tmp_path: Path) -> None:
    compose_file: Path = tmp_path / "docker-compose.yml"
    manager = DockerComposeManager(str(compose_file))
    manager.create_service(
        name="full",
        image="alpine:latest",
        ports=["1234:1234"],
        environment={"FOO": "bar"},
        volumes=["data:/data"],
        networks=["default", "custom"],
        command=["echo", "hello"],
        entrypoint=["/bin/sh", "-c"],
        build={"context": ".", "dockerfile": "Dockerfile"},
        healthcheck={"test": ["CMD", "true"], "interval": "1m"},
        restart="always",
        labels={"com.example": "label"},
        depends_on=["db"],
        configs=[{"source": "my_config", "target": "/etc/config"}],
        secrets=[{"source": "my_secret", "target": "/run/secret"}],
        deploy={"replicas": 2},
        resources={"limits": {"cpus": "0.5", "memory": "50M"}},
        extra_field="extra_value",
    ).save()

    with compose_file.open() as f:
        data = yaml.safe_load(f)
    svc = data["services"]["full"]
    assert svc["image"] == "alpine:latest"
    assert svc["ports"] == ["1234:1234"]
    assert svc["environment"] == {"FOO": "bar"}
    assert svc["volumes"] == ["data:/data"]
    assert svc["networks"] == ["default", "custom"]
    assert svc["command"] == ["echo", "hello"]
    assert svc["entrypoint"] == ["/bin/sh", "-c"]
    assert svc["build"] == {"context": ".", "dockerfile": "Dockerfile"}
    assert svc["healthcheck"] == {"test": ["CMD", "true"], "interval": "1m"}
    assert svc["restart"] == "always"
    assert svc["labels"] == {"com.example": "label"}
    assert svc["depends_on"] == ["db"]
    assert svc["configs"] == [{"source": "my_config", "target": "/etc/config"}]
    assert svc["secrets"] == [{"source": "my_secret", "target": "/run/secret"}]
    assert svc["deploy"] == {"replicas": 2}
    assert svc["resources"] == {"limits": {"cpus": "0.5", "memory": "50M"}}
    assert svc["extra_field"] == "extra_value"