feat(compose): add ability to get docker compose config (testcontainers#669)

This PR add a new function to the
`testcontainers.compose.DockerComposer` class, `get_config` which use
`docker compose config` command for resolving and returning the actual
docker compose configuration.

This can be useful for example if you want to retrieve a connection
string you pass to your app in your docker compose in order to connect
to your database service instead of copy pasting it from your compose
file into your tests.

Also note thats its way easier to rely on docker compose config to get
you the config than trying to manually find, read and merge compose
files in specified context (I tried it first ...).

About the tests I mostly ensured the docker compose command was as
expected. This is because the config produced by the docker compose can
not always reflect exactly what is in the file. There is some
normalization/resolving which is done (even when you pass all flags to
disable them). But anyway, I'm not sure its a good idea to actually test
the behavior of the docker config command itself.

Let me know what you think of it!

---------

Co-authored-by: David Ankin <daveankin@gmail.com>

Author: g0di

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)

poetry.lock

@@ -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"