From f0ee0bcac6fca8aadc007f572adcddc1d4638092 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Hells=C3=A9n?= Date: Wed, 18 Jun 2025 04:14:51 +0200 Subject: [PATCH] Implement volume and network management methods in DockerComposeManager and enhance service creation with additional parameters --- src/compose/__init__.py | 91 ++++++++++++++++++++++++++++++++++- tests/test_compose_manager.py | 87 +++++++++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 2 deletions(-) diff --git a/src/compose/__init__.py b/src/compose/__init__.py index 810543f..acbf5e0 100644 --- a/src/compose/__init__.py +++ b/src/compose/__init__.py @@ -20,6 +20,52 @@ if TYPE_CHECKING: class DockerComposeManager: + def add_volume(self, name: str, config: dict[str, Any] | None = None) -> DockerComposeManager: + """Add a top-level volume definition. + + Returns: + DockerComposeManager: self (for chaining) + """ + if "volumes" not in self._data or not isinstance(self._data["volumes"], dict): + self._data["volumes"] = {} + self._data["volumes"][name] = config or {} + self._dirty = True + return self + + def remove_volume(self, name: str) -> DockerComposeManager: + """Remove a top-level volume definition. + + Returns: + DockerComposeManager: self (for chaining) + """ + if "volumes" in self._data and name in self._data["volumes"]: + del self._data["volumes"][name] + self._dirty = True + return self + + def add_network(self, name: str, config: dict[str, Any] | None = None) -> DockerComposeManager: + """Add a top-level network definition. + + Returns: + DockerComposeManager: self (for chaining) + """ + if "networks" not in self._data or not isinstance(self._data["networks"], dict): + self._data["networks"] = {} + self._data["networks"][name] = config or {} + self._dirty = True + return self + + def remove_network(self, name: str) -> DockerComposeManager: + """Remove a top-level network definition. + + Returns: + DockerComposeManager: self (for chaining) + """ + if "networks" in self._data and name in self._data["networks"]: + del self._data["networks"][name] + self._dirty = True + return self + """A class to create and modify Docker Compose YAML files programmatically. Supports context manager usage for auto-saving. @@ -47,9 +93,22 @@ class DockerComposeManager: def create_service( self, name: str, - image: str, + image: str = "", ports: list[str] | None = None, environment: dict[str, str] | None = None, + volumes: list[str] | None = None, + networks: list[str] | None = None, + command: str | list[str] | None = None, + entrypoint: str | list[str] | None = None, + build: dict[str, Any] | str | None = None, + healthcheck: dict[str, Any] | None = None, + restart: str | None = None, + labels: dict[str, str] | list[str] | None = None, + depends_on: list[str] | None = None, + configs: list[dict[str, Any]] | None = None, + secrets: list[dict[str, Any]] | None = None, + deploy: dict[str, Any] | None = None, + resources: dict[str, Any] | None = None, **kwargs: object, ) -> DockerComposeManager: """Create a new service in the compose file. @@ -59,11 +118,39 @@ class DockerComposeManager: """ services = self._data["services"] - service: dict[str, Any] = {"image": image} + service: dict[str, Any] = {} + if image: + service["image"] = image if ports is not None: service["ports"] = ports if environment is not None: service["environment"] = environment + if volumes is not None: + service["volumes"] = volumes + if networks is not None: + service["networks"] = networks + if command is not None: + service["command"] = command + if entrypoint is not None: + service["entrypoint"] = entrypoint + if build is not None: + service["build"] = build + if healthcheck is not None: + service["healthcheck"] = healthcheck + if restart is not None: + service["restart"] = restart + if labels is not None: + service["labels"] = labels + if depends_on is not None: + service["depends_on"] = depends_on + if configs is not None: + service["configs"] = configs + if secrets is not None: + service["secrets"] = secrets + if deploy is not None: + service["deploy"] = deploy + if resources is not None: + service["resources"] = resources service.update(kwargs) services[name] = service self._dirty = True diff --git a/tests/test_compose_manager.py b/tests/test_compose_manager.py index c626de3..72c695c 100644 --- a/tests/test_compose_manager.py +++ b/tests/test_compose_manager.py @@ -1,5 +1,6 @@ from __future__ import annotations +from pathlib import Path from typing import TYPE_CHECKING import pytest @@ -65,3 +66,89 @@ def test_modify_nonexistent_service(tmp_path: Path) -> None: 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"