docker_utils.py

"""Provide utility functions related to the use of Docker during integration tests."""

from tests.utils.classes import ExternalAction
from tests.utils.logging_utils import format_action_failure_msg
from tests.utils.utils import get_binary_path


def list_running_services_in_compose_project(project_name: str) -> list[str]:
    """
    Lists running Docker services that belong to the given Docker Compose project.

    :param project_name:
    :return: List of the running services that belong to the specified Docker Compose project.
    :raise RuntimeError: if `docker compose ps` returns a non-zero exit code.
    """
    docker_bin = get_binary_path("docker")

    # fmt: off
    compose_ps_cmd = [
        docker_bin,
        "compose",
        "--project-name", project_name,
        "ps",
        "--format", "{{.Service}}",
    ]
    # fmt: on

    compose_ps_action = ExternalAction.from_cmd(compose_ps_cmd)
    if compose_ps_action.completed_proc.returncode != 0:
        err_msg = format_action_failure_msg(
            "`docker compose ps` failed with exit code"
            f" {compose_ps_action.completed_proc.returncode} for project `{project_name}`.",
            compose_ps_action,
        )
        raise RuntimeError(err_msg)

    service_names: list[str] = []
    for line in compose_ps_action.completed_proc.stdout.splitlines():
        service_name_candidate = line.strip()
        if service_name_candidate:
            service_names.append(service_name_candidate)

    return service_names