test_docker_in_docker.py

import contextlib
import json
import logging
import os
import time
import socket
import sys
from textwrap import indent
from pathlib import Path
from typing import Final, Any, Generator

import pytest
from docker.models.containers import Container

from testcontainers.core import utils
from testcontainers.core.config import testcontainers_config as tcc
from testcontainers.core.labels import SESSION_ID
from testcontainers.core.network import Network
from testcontainers.core.container import DockerContainer
from testcontainers.core.docker_client import DockerClient, LOGGER
from testcontainers.core.utils import inside_container
from testcontainers.core.utils import is_mac
from testcontainers.core.waiting_utils import wait_for_logs

_DIND_PYTHON_VERSION = (3, 10)
logger = logging.getLogger(__name__)


def _should_skip_dind() -> bool:
    if os.getenv("DIND"):
        return False
    # todo refine macos check -> run in ci but not locally
    return is_mac() or tuple([*sys.version_info][:2]) != _DIND_PYTHON_VERSION


def _wait_for_dind_return_ip(client: DockerClient, dind: Container):
    # get ip address for DOCKER_HOST
    # avoiding DockerContainer class here to prevent code changes affecting the test
    docker_host_ip = client.bridge_ip(dind.id)
    # Wait for startup
    timeout = 10
    start_wait = time.perf_counter()
    while True:
        try:
            with socket.create_connection((docker_host_ip, 2375), timeout=timeout):
                break
        except ConnectionRefusedError:
            if time.perf_counter() - start_wait > timeout:
                raise RuntimeError("Docker in docker took longer than 10 seconds to start")
            time.sleep(0.01)
    return docker_host_ip


@pytest.mark.skipif(_should_skip_dind(), reason="Docker socket forwarding (socat) is unsupported on Docker Desktop for macOS")
def test_wait_for_logs_docker_in_docker():
    # real dind isn't possible (AFAIK) in CI
    # forwarding the socket to a container port is at least somewhat the same
    logger.info("starting test_wait_for_logs_docker_in_docker")
    client = DockerClient()
    not_really_dind = client.run(
        image="alpine/socat",
        command="tcp-listen:2375,fork,reuseaddr unix-connect:/var/run/docker.sock",
        volumes={"/var/run/docker.sock": {"bind": "/var/run/docker.sock"}},
        detach=True,
    )

    logger.info("starting not_really_dind")
    not_really_dind.start()
    logger.info("started not_really_dind")
    docker_host_ip = _wait_for_dind_return_ip(client, not_really_dind)
    logger.info("waited for not_really_dind: '_wait_for_dind_return_ip'")
    docker_host = f"tcp://{docker_host_ip}:2375"

    try:
        with DockerContainer(
            image="hello-world",
            docker_client_kw={"environment": {"DOCKER_HOST": docker_host, "DOCKER_CERT_PATH": "", "DOCKER_TLS_VERIFY": ""}},
        ) as container:
            logger.info("started hello-world container")
            assert container.get_container_host_ip() == docker_host_ip
            wait_for_logs(container, "Hello from Docker!")
            stdout, stderr = container.get_logs()
            assert stdout, "There should be something on stdout"
    finally:
        not_really_dind.stop()
        not_really_dind.remove()


@pytest.mark.skipif(
    _should_skip_dind(), reason="Bridge networking and Docker socket forwarding are not supported on Docker Desktop for macOS"
)
def test_dind_inherits_network():
    client = DockerClient()
    try:
        custom_network = client.client.networks.create("custom_network", driver="bridge", check_duplicate=True)
    except Exception:
        custom_network = client.client.networks.list(names=["custom_network"])[0]
    not_really_dind = client.run(
        image="alpine/socat",
        command="tcp-listen:2375,fork,reuseaddr unix-connect:/var/run/docker.sock",
        volumes={"/var/run/docker.sock": {"bind": "/var/run/docker.sock"}},
        detach=True,
    )

    not_really_dind.start()

    docker_host_ip = _wait_for_dind_return_ip(client, not_really_dind)
    docker_host = f"tcp://{docker_host_ip}:2375"

    with DockerContainer(
        image="hello-world",
        docker_client_kw={"environment": {"DOCKER_HOST": docker_host, "DOCKER_CERT_PATH": "", "DOCKER_TLS_VERIFY": ""}},
    ) as container:
        assert container.get_container_host_ip() == docker_host_ip
        # Check the gateways are the same, so they can talk to each other
        assert container.get_docker_client().gateway_ip(container.get_wrapped_container().id) == client.gateway_ip(
            not_really_dind.id
        )
        wait_for_logs(container, "Hello from Docker!")
        stdout, stderr = container.get_logs()
        assert stdout, "There should be something on stdout"

    not_really_dind.stop()
    not_really_dind.remove()
    custom_network.remove()


@contextlib.contextmanager
def print_surround_header(what: str, header_len: int = 80) -> Generator[None, None, None]:
    """
    Helper to visually mark a block with headers
    """
    start = f"#   Beginning of {what}"
    end = f"#   End of {what}"

    logger.info("\n")
    logger.info("#" * header_len)
    logger.info(start + " " * (header_len - len(start) - 1) + "#")
    logger.info("#" * header_len)
    logger.info("\n")

    yield

    logger.info("\n")
    logger.info("#" * header_len)
    logger.info(end + " " * (header_len - len(end) - 1) + "#")
    logger.info("#" * header_len)
    logger.info("\n")


EXPECTED_NETWORK_VAR: Final[str] = "TCC_EXPECTED_NETWORK"


def get_docker_info() -> dict[str, Any]:
    client = DockerClient().client

    # Get Docker version info
    version_info = client.version()

    # Get Docker system info
    system_info = client.info()

    # Get container inspections
    containers = client.containers.list(all=True)  # List all containers (running or not)
    container_inspections = {container.name: container.attrs for container in containers}

    # Return as a dictionary
    return {"version_info": version_info, "system_info": system_info, "container_inspections": container_inspections}


# see https://forums.docker.com/t/get-a-containers-full-id-from-inside-of-itself
@pytest.mark.xfail(reason="Does not work in rootles docker i.e. github actions")
@pytest.mark.inside_docker_check
@pytest.mark.skipif(_should_skip_dind() or not os.environ.get(EXPECTED_NETWORK_VAR), reason="No expected network given")
def test_find_host_network_in_dood() -> None:
    """
    Check that the correct host network is found for DooD
    """
    logger.info(f"Running container id={utils.get_running_in_container_id()}")
    # Get some debug information in the hope this helps to find
    logger.info(f"hostname: {socket.gethostname()}")
    logger.info(f"docker info: {json.dumps(get_docker_info(), indent=2)}")
    assert DockerClient().find_host_network() == os.environ[EXPECTED_NETWORK_VAR]


@pytest.mark.skipif(
    _should_skip_dind(), reason="Docker socket mounting and container networking do not work reliably on Docker Desktop for macOS"
)
@pytest.mark.skipif(not Path(tcc.ryuk_docker_socket).exists(), reason="No docker socket available")
def test_dood(python_testcontainer_image: str) -> None:
    """
    Run tests marked as inside_docker_check inside docker out of docker
    """

    docker_sock = tcc.ryuk_docker_socket
    with Network() as network:
        logger.info("test_dood - created network")
        with (
            DockerContainer(
                image=python_testcontainer_image,
            )
            .with_command("poetry run pytest -m inside_docker_check")
            .with_volume_mapping(docker_sock, docker_sock, "rw")
            # test also that the correct network was found
            # but only do this if not already inside a container
            # as there for some reason this doesn't work
            .with_env(EXPECTED_NETWORK_VAR, "" if inside_container() else network.name)
            .with_env("RYUK_RECONNECTION_TIMEOUT", "1s")
            .with_network(network)
        ) as container:
            logger.info("test_dood - created container")
            status = container.get_wrapped_container().wait()
            logger.info("test_dood - container returned status %s", status)
            stdout, stderr = container.get_logs()
            # ensure ryuk removed the containers created inside container
            # because they are bound our network the deletion of the network
            # would fail otherwise
            time.sleep(1.1)

    # Show what was done inside test
    with print_surround_header("test_dood results"):
        logger.info(indent(stdout.decode("utf-8", errors="replace"), prefix=(" " * 4) + "container log: "))
        logger.info(indent(stderr.decode("utf-8", errors="replace"), prefix=(" " * 4) + "container log: "))
    assert status["StatusCode"] == 0


@pytest.mark.skipif(
    _should_skip_dind(), reason="Docker socket mounting and container networking do not work reliably on Docker Desktop for macOS"
)
def test_dind(python_testcontainer_image: str, tmp_path: Path) -> None:
    """
    Run selected tests in Docker in Docker
    """
    cert_dir = tmp_path / "certs"
    dind_name = f"docker_{SESSION_ID}"
    with Network() as network:
        logger.info("test_dind - created network")
        with (
            DockerContainer(image="docker:dind", privileged=True)
            .with_name(dind_name)
            .with_volume_mapping(str(cert_dir), "/certs", "rw")
            .with_env("DOCKER_TLS_CERTDIR", "/certs/docker")
            .with_env("DOCKER_TLS_VERIFY", "1")
            .with_network(network)
            .with_network_aliases("docker")
        ) as dind_container:
            logger.info("test_dind - created docker:dind container")
            wait_for_logs(dind_container, "API listen on")
            logger.info("test_dind - waited for ready message in logs")
            client_dir = cert_dir / "docker" / "client"
            ca_file = client_dir / "ca.pem"
            assert ca_file.is_file()
            try:
                with (
                    DockerContainer(image=python_testcontainer_image)
                    .with_command("poetry run pytest -m inside_docker_check")
                    .with_volume_mapping(str(cert_dir), "/certs")
                    # for some reason the docker client does not respect
                    # DOCKER_TLS_CERTDIR and looks in /root/.docker instead
                    .with_volume_mapping(str(client_dir), "/root/.docker")
                    .with_env("DOCKER_TLS_CERTDIR", "/certs/docker/client")
                    .with_env("DOCKER_TLS_VERIFY", "1")
                    # docker port is 2376 for https, 2375 for http
                    .with_env("DOCKER_HOST", "tcp://docker:2376")
                    .with_network(network)
                ) as test_container:
                    logger.info("test_dind - created test suite container")
                    status = test_container.get_wrapped_container().wait()
                    logger.info("test_dind - test suite container returned status %s", status)
                    stdout, stderr = test_container.get_logs()
            finally:
                # ensure the certs are deleted from inside the container
                # as they might be owned by root it otherwise could lead to problems
                # with pytest cleanup
                dind_container.exec("rm -rf /certs/docker")
                dind_container.exec("chmod -R a+rwX /certs")

    # Show what was done inside test
    with print_surround_header("test_dood results"):
        logger.info(indent(stdout.decode("utf-8", errors="replace"), prefix=(" " * 4) + "container log: "))
        logger.info(indent(stderr.decode("utf-8", errors="replace"), prefix=(" " * 4) + "container log: "))
    assert status["StatusCode"] == 0