feat(compose): add ability to get docker compose config

core/testcontainers/compose/compose.py

@@ -1,19 +1,21 @@
 from dataclasses import asdict, dataclass, field, fields
 from functools import cached_property
 from json import loads
+from logging import warning
 from os import PathLike
 from platform import system
 from re import split
 from subprocess import CompletedProcess
 from subprocess import run as subprocess_run
-from typing import Callable, Literal, Optional, TypeVar, Union
+from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast
 from urllib.error import HTTPError, URLError
 from urllib.request import urlopen
 
 from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
 _IPT = TypeVar("_IPT")
+_WARNINGS = {"DOCKER_COMPOSE_GET_CONFIG": "get_config is experimental, see testcontainers/testcontainers-python#669"}
 
 
 def _ignore_properties(cls: type[_IPT], dict_: any) -> _IPT:
@@ -258,6 +260,36 @@ def get_logs(self, *services: str) -> tuple[str, str]:
         result = self._run_command(cmd=logs_cmd)
         return result.stdout.decode("utf-8"), result.stderr.decode("utf-8")
 
+    def get_config(
+        self, *, path_resolution: bool = True, normalize: bool = True, interpolate: bool = True
+    ) -> dict[str, Any]:
+        """
+        Parse, resolve and returns compose file via `docker config --format json`.
+        In case of multiple compose files, the returned value will be a merge of all files.
+
+        See: https://docs.docker.com/reference/cli/docker/compose/config/ for more details
+
+        :param path_resolution: whether to resolve file paths
+        :param normalize: whether to normalize compose model
+        :param interpolate: whether to interpolate environment variables
+
+        Returns:
+            Compose file
+
+        """
+        if "DOCKER_COMPOSE_GET_CONFIG" in _WARNINGS:
+            warning(_WARNINGS.pop("DOCKER_COMPOSE_GET_CONFIG"))
+        config_cmd = [*self.compose_command_property, "config", "--format", "json"]
+        if not path_resolution:
+            config_cmd.append("--no-path-resolution")
+        if not normalize:
+            config_cmd.append("--no-normalize")
+        if not interpolate:
+            config_cmd.append("--no-interpolate")
+
+        cmd_output = self._run_command(cmd=config_cmd).stdout
+        return cast(dict[str, Any], loads(cmd_output))
+
     def get_containers(self, include_all=False) -> list[ComposeContainer]:
         """
         Fetch information about running containers via `docker compose ps --format json`.

core/tests/test_compose.py

@@ -6,6 +6,7 @@
 from urllib.request import urlopen, Request
 
 import pytest
+from pytest_mock import MockerFixture
 
 from testcontainers.compose import DockerCompose, ContainerIsNotRunning, NoSuchPortExposed
 
@@ -304,6 +305,45 @@ def test_exec_in_container_multiple():
         assert "test_exec_in_container" in body
 
 
+CONTEXT_FIXTURES = [pytest.param(ctx, id=ctx.name) for ctx in FIXTURES.iterdir()]
+
+
+@pytest.mark.parametrize("context", CONTEXT_FIXTURES)
+def test_compose_config(context: Path, mocker: MockerFixture) -> None:
+    compose = DockerCompose(context)
+    run_command = mocker.spy(compose, "_run_command")
+    expected_cmd = [*compose.compose_command_property, "config", "--format", "json"]
+
+    received_config = compose.get_config()
+
+    assert received_config
+    assert isinstance(received_config, dict)
+    assert "services" in received_config
+    assert run_command.call_args.kwargs["cmd"] == expected_cmd
+
+
+@pytest.mark.parametrize("context", CONTEXT_FIXTURES)
+def test_compose_config_raw(context: Path, mocker: MockerFixture) -> None:
+    compose = DockerCompose(context)
+    run_command = mocker.spy(compose, "_run_command")
+    expected_cmd = [
+        *compose.compose_command_property,
+        "config",
+        "--format",
+        "json",
+        "--no-path-resolution",
+        "--no-normalize",
+        "--no-interpolate",
+    ]
+
+    received_config = compose.get_config(path_resolution=False, normalize=False, interpolate=False)
+
+    assert received_config
+    assert isinstance(received_config, dict)
+    assert "services" in received_config
+    assert run_command.call_args.kwargs["cmd"] == expected_cmd
+
+
 def fetch(req: Union[Request, str]):
     if isinstance(req, str):
         req = Request(method="GET", url=req)

pyproject.toml

@@ -182,6 +182,7 @@ paho-mqtt = "2.1.0"
 sqlalchemy-cockroachdb = "2.0.2"
 paramiko = "^3.4.0"
 types-paramiko = "^3.4.0.20240423"
+pytest-mock = "^3.14.0"
 
 [[tool.poetry.source]]
 name = "PyPI"