From 2cca44b64e62ec96d9560a942b282f3d1978a92f Mon Sep 17 00:00:00 2001 From: VictorCavichioli Date: Mon, 6 Oct 2025 15:04:01 -0300 Subject: [PATCH 1/5] feat(compose): add structured container inspect information --- core/testcontainers/compose/compose.py | 142 +++++++++++++++++++++++++ core/tests/test_compose.py | 57 ++++++++++ 2 files changed, 199 insertions(+) diff --git a/core/testcontainers/compose/compose.py b/core/testcontainers/compose/compose.py index 61961ce0d..7dd5f1c5d 100644 --- a/core/testcontainers/compose/compose.py +++ b/core/testcontainers/compose/compose.py @@ -33,6 +33,122 @@ def _ignore_properties(cls: type[_IPT], dict_: Any) -> _IPT: return cast("_IPT", cls(**filtered)) +@dataclass +class ContainerState: + """Container state from docker inspect.""" + + Status: Optional[str] = None + Running: Optional[bool] = None + Paused: Optional[bool] = None + Restarting: Optional[bool] = None + OOMKilled: Optional[bool] = None + Dead: Optional[bool] = None + Pid: Optional[int] = None + ExitCode: Optional[int] = None + Error: Optional[str] = None + StartedAt: Optional[str] = None + FinishedAt: Optional[str] = None + + +@dataclass +class ContainerConfig: + """Container config from docker inspect.""" + + Hostname: Optional[str] = None + User: Optional[str] = None + Env: Optional[list[str]] = None + Cmd: Optional[list[str]] = None + Image: Optional[str] = None + WorkingDir: Optional[str] = None + Entrypoint: Optional[list[str]] = None + ExposedPorts: Optional[dict[str, Any]] = None + Labels: Optional[dict[str, str]] = None + + +@dataclass +class Network: + """Individual network from docker inspect.""" + + IPAddress: Optional[str] = None + Gateway: Optional[str] = None + NetworkID: Optional[str] = None + EndpointID: Optional[str] = None + MacAddress: Optional[str] = None + Aliases: Optional[list[str]] = None + + +@dataclass +class NetworkSettings: + """Network settings from docker inspect.""" + + Bridge: Optional[str] = None + IPAddress: Optional[str] = None + Gateway: Optional[str] = None + Ports: Optional[dict[str, Any]] = None + Networks: Optional[dict[str, Network]] = None + + def get_networks(self) -> Optional[dict[str, Network]]: + """Get networks for the container.""" + return self.Networks + + +@dataclass +class ContainerInspectInfo: + """Container information from docker inspect.""" + + Id: Optional[str] = None + Name: Optional[str] = None + Created: Optional[str] = None + Path: Optional[str] = None + Args: Optional[list[str]] = None + Image: Optional[str] = None + State: Optional[ContainerState] = None + Config: Optional[ContainerConfig] = None + network_settings: Optional[NetworkSettings] = None + Mounts: Optional[list[dict[str, Any]]] = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo": + """Create from docker inspect JSON.""" + return cls( + Id=data.get("Id"), + Name=data.get("Name"), + Created=data.get("Created"), + Path=data.get("Path"), + Args=data.get("Args"), + Image=data.get("Image"), + State=_ignore_properties(ContainerState, data.get("State", {})) if data.get("State") else None, + Config=_ignore_properties(ContainerConfig, data.get("Config", {})) if data.get("Config") else None, + network_settings=cls._parse_network_settings(data.get("NetworkSettings", {})) + if data.get("NetworkSettings") + else None, + Mounts=data.get("Mounts"), + ) + + @classmethod + def _parse_network_settings(cls, data: dict[str, Any]) -> Optional[NetworkSettings]: + """Parse NetworkSettings with Networks as Network objects.""" + if not data: + return None + + networks_data = data.get("Networks", {}) + networks = {} + for name, net_data in networks_data.items(): + networks[name] = _ignore_properties(Network, net_data) + + return NetworkSettings( + Bridge=data.get("Bridge"), + IPAddress=data.get("IPAddress"), + Gateway=data.get("Gateway"), + Ports=data.get("Ports"), + Networks=networks, + ) + + def get_network_settings(self) -> Optional[NetworkSettings]: + """Get network settings for the container.""" + return self.network_settings + + @dataclass class PublishedPortModel: """ @@ -81,6 +197,7 @@ class ComposeContainer: ExitCode: Optional[int] = None Publishers: list[PublishedPortModel] = field(default_factory=list) _docker_compose: Optional["DockerCompose"] = field(default=None, init=False, repr=False) + _cached_container_info: Optional[ContainerInspectInfo] = field(default=None, init=False, repr=False) def __post_init__(self) -> None: if self.Publishers: @@ -147,6 +264,31 @@ def reload(self) -> None: # each time through get_container(), but we need this method for compatibility pass + def get_container_info(self) -> Optional[ContainerInspectInfo]: + """Get container information via docker inspect (lazy loaded).""" + if self._cached_container_info is not None: + return self._cached_container_info + + if not self._docker_compose or not self.ID: + return None + + try: + inspect_command = ["docker", "inspect", self.ID] + result = self._docker_compose._run_command(cmd=inspect_command) + inspect_output = result.stdout.decode("utf-8").strip() + + if inspect_output: + raw_data = loads(inspect_output)[0] + self._cached_container_info = ContainerInspectInfo.from_dict(raw_data) + else: + self._cached_container_info = None + + except Exception as e: + logger.warning(f"Failed to get container info for {self.ID}: {e}") + self._cached_container_info = None + + return self._cached_container_info + @property def status(self) -> str: """Get container status for compatibility with wait strategies.""" diff --git a/core/tests/test_compose.py b/core/tests/test_compose.py index 755b8b17b..8d16d4756 100644 --- a/core/tests/test_compose.py +++ b/core/tests/test_compose.py @@ -378,3 +378,60 @@ def test_compose_profile_support(profiles: Optional[list[str]], running: list[st for service in not_running: with pytest.raises(ContainerIsNotRunning): compose.get_container(service) + + +def test_container_info(): + """Test get_container_info functionality""" + basic = DockerCompose(context=FIXTURES / "basic") + with basic: + container = basic.get_container("alpine") + + info = container.get_container_info() + assert info is not None + assert info.Id is not None + assert info.Name is not None + assert info.Image is not None + + assert info.State is not None + assert info.State.Status == "running" + assert info.State.Running is True + assert info.State.Pid is not None + + assert info.Config is not None + assert info.Config.Image is not None + assert info.Config.Hostname is not None + + network_settings = info.get_network_settings() + assert network_settings is not None + assert network_settings.Networks is not None + + info2 = container.get_container_info() + assert info is info2 + + +def test_container_info_network_details(): + """Test network details in container info""" + single = DockerCompose(context=FIXTURES / "port_single") + with single: + container = single.get_container() + info = container.get_container_info() + assert info is not None + + network_settings = info.get_network_settings() + assert network_settings is not None + + if network_settings.Networks: + # Test first network + network_name, network = next(iter(network_settings.Networks.items())) + assert network.IPAddress is not None + assert network.Gateway is not None + assert network.NetworkID is not None + + +def test_container_info_none_when_no_docker_compose(): + """Test get_container_info returns None when docker_compose reference is missing""" + from testcontainers.compose.compose import ComposeContainer + + container = ComposeContainer() + info = container.get_container_info() + assert info is None From 11a8f5d3de8ea42907e58384038e82e216db1c78 Mon Sep 17 00:00:00 2001 From: VictorCavichioli Date: Tue, 7 Oct 2025 10:42:44 -0300 Subject: [PATCH 2/5] feat(container): move docker response dataclasses to docker_client.py --- core/testcontainers/compose/compose.py | 132 +--- core/testcontainers/core/container.py | 21 +- core/testcontainers/core/docker_client.py | 703 ++++++++++++++++++++++ core/tests/test_compose.py | 24 + core/tests/test_container.py | 298 +++++++++ docs/features/creating_container.md | 56 ++ docs/features/docker_compose.md | 56 ++ 7 files changed, 1159 insertions(+), 131 deletions(-) diff --git a/core/testcontainers/compose/compose.py b/core/testcontainers/compose/compose.py index 7dd5f1c5d..8163d8324 100644 --- a/core/testcontainers/compose/compose.py +++ b/core/testcontainers/compose/compose.py @@ -1,5 +1,5 @@ import sys -from dataclasses import asdict, dataclass, field, fields, is_dataclass +from dataclasses import asdict, dataclass, field from functools import cached_property from json import loads from logging import getLogger, warning @@ -11,6 +11,7 @@ from types import TracebackType from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast +from testcontainers.core.docker_client import ContainerInspectInfo, _ignore_properties from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed from testcontainers.core.waiting_utils import WaitStrategy @@ -20,135 +21,6 @@ logger = getLogger(__name__) -def _ignore_properties(cls: type[_IPT], dict_: Any) -> _IPT: - """omits extra fields like @JsonIgnoreProperties(ignoreUnknown = true) - - https://gist.github.com/alexanderankin/2a4549ac03554a31bef6eaaf2eaf7fd5""" - if isinstance(dict_, cls): - return dict_ - if not is_dataclass(cls): - raise TypeError(f"Expected a dataclass type, got {cls}") - class_fields = {f.name for f in fields(cls)} - filtered = {k: v for k, v in dict_.items() if k in class_fields} - return cast("_IPT", cls(**filtered)) - - -@dataclass -class ContainerState: - """Container state from docker inspect.""" - - Status: Optional[str] = None - Running: Optional[bool] = None - Paused: Optional[bool] = None - Restarting: Optional[bool] = None - OOMKilled: Optional[bool] = None - Dead: Optional[bool] = None - Pid: Optional[int] = None - ExitCode: Optional[int] = None - Error: Optional[str] = None - StartedAt: Optional[str] = None - FinishedAt: Optional[str] = None - - -@dataclass -class ContainerConfig: - """Container config from docker inspect.""" - - Hostname: Optional[str] = None - User: Optional[str] = None - Env: Optional[list[str]] = None - Cmd: Optional[list[str]] = None - Image: Optional[str] = None - WorkingDir: Optional[str] = None - Entrypoint: Optional[list[str]] = None - ExposedPorts: Optional[dict[str, Any]] = None - Labels: Optional[dict[str, str]] = None - - -@dataclass -class Network: - """Individual network from docker inspect.""" - - IPAddress: Optional[str] = None - Gateway: Optional[str] = None - NetworkID: Optional[str] = None - EndpointID: Optional[str] = None - MacAddress: Optional[str] = None - Aliases: Optional[list[str]] = None - - -@dataclass -class NetworkSettings: - """Network settings from docker inspect.""" - - Bridge: Optional[str] = None - IPAddress: Optional[str] = None - Gateway: Optional[str] = None - Ports: Optional[dict[str, Any]] = None - Networks: Optional[dict[str, Network]] = None - - def get_networks(self) -> Optional[dict[str, Network]]: - """Get networks for the container.""" - return self.Networks - - -@dataclass -class ContainerInspectInfo: - """Container information from docker inspect.""" - - Id: Optional[str] = None - Name: Optional[str] = None - Created: Optional[str] = None - Path: Optional[str] = None - Args: Optional[list[str]] = None - Image: Optional[str] = None - State: Optional[ContainerState] = None - Config: Optional[ContainerConfig] = None - network_settings: Optional[NetworkSettings] = None - Mounts: Optional[list[dict[str, Any]]] = None - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo": - """Create from docker inspect JSON.""" - return cls( - Id=data.get("Id"), - Name=data.get("Name"), - Created=data.get("Created"), - Path=data.get("Path"), - Args=data.get("Args"), - Image=data.get("Image"), - State=_ignore_properties(ContainerState, data.get("State", {})) if data.get("State") else None, - Config=_ignore_properties(ContainerConfig, data.get("Config", {})) if data.get("Config") else None, - network_settings=cls._parse_network_settings(data.get("NetworkSettings", {})) - if data.get("NetworkSettings") - else None, - Mounts=data.get("Mounts"), - ) - - @classmethod - def _parse_network_settings(cls, data: dict[str, Any]) -> Optional[NetworkSettings]: - """Parse NetworkSettings with Networks as Network objects.""" - if not data: - return None - - networks_data = data.get("Networks", {}) - networks = {} - for name, net_data in networks_data.items(): - networks[name] = _ignore_properties(Network, net_data) - - return NetworkSettings( - Bridge=data.get("Bridge"), - IPAddress=data.get("IPAddress"), - Gateway=data.get("Gateway"), - Ports=data.get("Ports"), - Networks=networks, - ) - - def get_network_settings(self) -> Optional[NetworkSettings]: - """Get network settings for the container.""" - return self.network_settings - - @dataclass class PublishedPortModel: """ diff --git a/core/testcontainers/core/container.py b/core/testcontainers/core/container.py index 4bb4eec48..64902e4dc 100644 --- a/core/testcontainers/core/container.py +++ b/core/testcontainers/core/container.py @@ -14,7 +14,7 @@ from testcontainers.core.config import ConnectionMode from testcontainers.core.config import testcontainers_config as c -from testcontainers.core.docker_client import DockerClient +from testcontainers.core.docker_client import ContainerInspectInfo, DockerClient from testcontainers.core.exceptions import ContainerConnectException, ContainerStartException from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID from testcontainers.core.network import Network @@ -96,6 +96,7 @@ def __init__( self._kwargs = kwargs self._wait_strategy: Optional[WaitStrategy] = _wait_strategy + self._cached_container_info: Optional[ContainerInspectInfo] = None def with_env(self, key: str, value: str) -> Self: self.env[key] = value @@ -300,6 +301,24 @@ def exec(self, command: Union[str, list[str]]) -> ExecResult: raise ContainerStartException("Container should be started before executing a command") return self._container.exec_run(command) + def get_container_info(self) -> Optional[ContainerInspectInfo]: + """Get container information via docker inspect (lazy loaded).""" + if self._cached_container_info is not None: + return self._cached_container_info + + if not self._container: + return None + + try: + raw_data = self._container.attrs + self._cached_container_info = ContainerInspectInfo.from_dict(raw_data) + + except Exception as e: + logger.warning(f"Failed to get container info for {self._container.id}: {e}") + self._cached_container_info = None + + return self._cached_container_info + def _configure(self) -> None: # placeholder if subclasses want to define this and use the default start method pass diff --git a/core/testcontainers/core/docker_client.py b/core/testcontainers/core/docker_client.py index 12384c94c..901bcf208 100644 --- a/core/testcontainers/core/docker_client.py +++ b/core/testcontainers/core/docker_client.py @@ -19,6 +19,7 @@ import urllib import urllib.parse from collections.abc import Iterable +from dataclasses import dataclass, fields, is_dataclass from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar, Union, cast import docker @@ -39,6 +40,7 @@ _P = ParamSpec("_P") _T = TypeVar("_T") +_IPT = TypeVar("_IPT") def _wrapped_container_collection(function: Callable[_P, _T]) -> Callable[_P, _T]: @@ -271,3 +273,704 @@ def get_docker_host() -> Optional[str]: def get_docker_auth_config() -> Optional[str]: return c.docker_auth_config + + +# Docker Engine API data structures + + +@dataclass +class ContainerLog: + """Container health check log entry.""" + + Start: Optional[str] = None + End: Optional[str] = None + ExitCode: Optional[int] = None + Output: Optional[str] = None + + +@dataclass +class ContainerHealth: + """Container health check information.""" + + Status: Optional[str] = None + FailingStreak: Optional[int] = None + Log: Optional[list[ContainerLog]] = None + + +@dataclass +class ContainerState: + """Container state information.""" + + Status: Optional[str] = None + Running: Optional[bool] = None + Paused: Optional[bool] = None + Restarting: Optional[bool] = None + OOMKilled: Optional[bool] = None + Dead: Optional[bool] = None + Pid: Optional[int] = None + ExitCode: Optional[int] = None + Error: Optional[str] = None + StartedAt: Optional[str] = None + FinishedAt: Optional[str] = None + Health: Optional[ContainerHealth] = None + + +@dataclass +class ContainerPlatform: + """Platform information for image manifest.""" + + architecture: Optional[str] = None + os: Optional[str] = None + variant: Optional[str] = None + + +@dataclass +class ContainerImageManifestDescriptor: + """Image manifest descriptor.""" + + mediaType: Optional[str] = None + digest: Optional[str] = None + size: Optional[int] = None + urls: Optional[list[str]] = None + annotations: Optional[dict[str, str]] = None + data: Optional[Any] = None + platform: Optional[ContainerPlatform] = None + artifactType: Optional[str] = None + + +@dataclass +class ContainerBlkioWeightDevice: + """Block IO weight device configuration.""" + + Path: Optional[str] = None + Weight: Optional[int] = None + + +@dataclass +class ContainerBlkioDeviceRate: + """Block IO device rate configuration.""" + + Path: Optional[str] = None + Rate: Optional[int] = None + + +@dataclass +class ContainerDeviceMapping: + """Device mapping configuration.""" + + PathOnHost: Optional[str] = None + PathInContainer: Optional[str] = None + CgroupPermissions: Optional[str] = None + + +@dataclass +class ContainerDeviceRequest: + """Device request configuration.""" + + Driver: Optional[str] = None + Count: Optional[int] = None + DeviceIDs: Optional[list[str]] = None + Capabilities: Optional[list[list[str]]] = None + Options: Optional[dict[str, str]] = None + + +@dataclass +class ContainerUlimit: + """Ulimit configuration.""" + + Name: Optional[str] = None + Soft: Optional[int] = None + Hard: Optional[int] = None + + +@dataclass +class ContainerLogConfig: + """Logging configuration.""" + + Type: Optional[str] = None + Config: Optional[dict[str, str]] = None + + +@dataclass +class ContainerPortBinding: + """Port binding configuration.""" + + HostIp: Optional[str] = None + HostPort: Optional[str] = None + + +@dataclass +class ContainerRestartPolicy: + """Restart policy configuration.""" + + Name: Optional[str] = None + MaximumRetryCount: Optional[int] = None + + +@dataclass +class ContainerBindOptions: + """Bind mount options.""" + + Propagation: Optional[str] = None + NonRecursive: Optional[bool] = None + CreateMountpoint: Optional[bool] = None + ReadOnlyNonRecursive: Optional[bool] = None + ReadOnlyForceRecursive: Optional[bool] = None + + +@dataclass +class ContainerVolumeDriverConfig: + """Volume driver configuration.""" + + Name: Optional[str] = None + Options: Optional[dict[str, str]] = None + + +@dataclass +class ContainerVolumeOptions: + """Volume mount options.""" + + NoCopy: Optional[bool] = None + Labels: Optional[dict[str, str]] = None + DriverConfig: Optional[ContainerVolumeDriverConfig] = None + Subpath: Optional[str] = None + + +@dataclass +class ContainerImageOptions: + """Image mount options.""" + + Subpath: Optional[str] = None + + +@dataclass +class ContainerTmpfsOptions: + """Tmpfs mount options.""" + + SizeBytes: Optional[int] = None + Mode: Optional[int] = None + Options: Optional[list[list[str]]] = None + + +@dataclass +class ContainerMountPoint: + """Mount point configuration.""" + + Target: Optional[str] = None + Source: Optional[str] = None + Type: Optional[str] = None + ReadOnly: Optional[bool] = None + Consistency: Optional[str] = None + BindOptions: Optional[ContainerBindOptions] = None + VolumeOptions: Optional[ContainerVolumeOptions] = None + ImageOptions: Optional[ContainerImageOptions] = None + TmpfsOptions: Optional[ContainerTmpfsOptions] = None + + +@dataclass +class ContainerHostConfig: + """Host configuration for container.""" + + CpuShares: Optional[int] = None + Memory: Optional[int] = None + CgroupParent: Optional[str] = None + BlkioWeight: Optional[int] = None + BlkioWeightDevice: Optional[list[ContainerBlkioWeightDevice]] = None + BlkioDeviceReadBps: Optional[list[ContainerBlkioDeviceRate]] = None + BlkioDeviceWriteBps: Optional[list[ContainerBlkioDeviceRate]] = None + BlkioDeviceReadIOps: Optional[list[ContainerBlkioDeviceRate]] = None + BlkioDeviceWriteIOps: Optional[list[ContainerBlkioDeviceRate]] = None + CpuPeriod: Optional[int] = None + CpuQuota: Optional[int] = None + CpuRealtimePeriod: Optional[int] = None + CpuRealtimeRuntime: Optional[int] = None + CpusetCpus: Optional[str] = None + CpusetMems: Optional[str] = None + Devices: Optional[list[ContainerDeviceMapping]] = None + DeviceCgroupRules: Optional[list[str]] = None + DeviceRequests: Optional[list[ContainerDeviceRequest]] = None + KernelMemoryTCP: Optional[int] = None + MemoryReservation: Optional[int] = None + MemorySwap: Optional[int] = None + MemorySwappiness: Optional[int] = None + NanoCpus: Optional[int] = None + OomKillDisable: Optional[bool] = None + Init: Optional[bool] = None + PidsLimit: Optional[int] = None + Ulimits: Optional[list[ContainerUlimit]] = None + CpuCount: Optional[int] = None + CpuPercent: Optional[int] = None + IOMaximumIOps: Optional[int] = None + IOMaximumBandwidth: Optional[int] = None + Binds: Optional[list[str]] = None + ContainerIDFile: Optional[str] = None + LogConfig: Optional[ContainerLogConfig] = None + NetworkMode: Optional[str] = None + PortBindings: Optional[dict[str, Optional[list[ContainerPortBinding]]]] = None + RestartPolicy: Optional[ContainerRestartPolicy] = None + AutoRemove: Optional[bool] = None + VolumeDriver: Optional[str] = None + VolumesFrom: Optional[list[str]] = None + Mounts: Optional[list[ContainerMountPoint]] = None + ConsoleSize: Optional[list[int]] = None + Annotations: Optional[dict[str, str]] = None + CapAdd: Optional[list[str]] = None + CapDrop: Optional[list[str]] = None + CgroupnsMode: Optional[str] = None + Dns: Optional[list[str]] = None + DnsOptions: Optional[list[str]] = None + DnsSearch: Optional[list[str]] = None + ExtraHosts: Optional[list[str]] = None + GroupAdd: Optional[list[str]] = None + IpcMode: Optional[str] = None + Cgroup: Optional[str] = None + Links: Optional[list[str]] = None + OomScoreAdj: Optional[int] = None + PidMode: Optional[str] = None + Privileged: Optional[bool] = None + PublishAllPorts: Optional[bool] = None + ReadonlyRootfs: Optional[bool] = None + SecurityOpt: Optional[list[str]] = None + StorageOpt: Optional[dict[str, str]] = None + Tmpfs: Optional[dict[str, str]] = None + UTSMode: Optional[str] = None + UsernsMode: Optional[str] = None + ShmSize: Optional[int] = None + Sysctls: Optional[dict[str, str]] = None + Runtime: Optional[str] = None + Isolation: Optional[str] = None + MaskedPaths: Optional[list[str]] = None + ReadonlyPaths: Optional[list[str]] = None + + +@dataclass +class ContainerGraphDriver: + """Graph driver information.""" + + Name: Optional[str] = None + Data: Optional[dict[str, str]] = None + + +@dataclass +class ContainerMount: + """Mount information.""" + + Type: Optional[str] = None + Name: Optional[str] = None + Source: Optional[str] = None + Destination: Optional[str] = None + Driver: Optional[str] = None + Mode: Optional[str] = None + RW: Optional[bool] = None + Propagation: Optional[str] = None + + +@dataclass +class ContainerHealthcheck: + """Container healthcheck configuration.""" + + Test: Optional[list[str]] = None + Interval: Optional[int] = None + Timeout: Optional[int] = None + Retries: Optional[int] = None + StartPeriod: Optional[int] = None + StartInterval: Optional[int] = None + + +@dataclass +class ContainerConfig: + """Container configuration.""" + + Hostname: Optional[str] = None + Domainname: Optional[str] = None + User: Optional[str] = None + AttachStdin: Optional[bool] = None + AttachStdout: Optional[bool] = None + AttachStderr: Optional[bool] = None + ExposedPorts: Optional[dict[str, dict[str, Any]]] = None + Tty: Optional[bool] = None + OpenStdin: Optional[bool] = None + StdinOnce: Optional[bool] = None + Env: Optional[list[str]] = None + Cmd: Optional[list[str]] = None + Healthcheck: Optional[ContainerHealthcheck] = None + ArgsEscaped: Optional[bool] = None + Image: Optional[str] = None + Volumes: Optional[dict[str, dict[str, Any]]] = None + WorkingDir: Optional[str] = None + Entrypoint: Optional[list[str]] = None + NetworkDisabled: Optional[bool] = None + MacAddress: Optional[str] = None + OnBuild: Optional[list[str]] = None + Labels: Optional[dict[str, str]] = None + StopSignal: Optional[str] = None + StopTimeout: Optional[int] = None + Shell: Optional[list[str]] = None + + +@dataclass +class ContainerIPAMConfig: + """IPAM configuration for network.""" + + IPv4Address: Optional[str] = None + IPv6Address: Optional[str] = None + LinkLocalIPs: Optional[list[str]] = None + + +@dataclass +class ContainerNetworkEndpoint: + """Network endpoint information.""" + + IPAMConfig: Optional[ContainerIPAMConfig] = None + Links: Optional[list[str]] = None + MacAddress: Optional[str] = None + Aliases: Optional[list[str]] = None + DriverOpts: Optional[dict[str, str]] = None + GwPriority: Optional[list[int]] = None + NetworkID: Optional[str] = None + EndpointID: Optional[str] = None + Gateway: Optional[str] = None + IPAddress: Optional[str] = None + IPPrefixLen: Optional[int] = None + IPv6Gateway: Optional[str] = None + GlobalIPv6Address: Optional[str] = None + GlobalIPv6PrefixLen: Optional[int] = None + DNSNames: Optional[list[str]] = None + + +@dataclass +class ContainerAddress: + """IP address information.""" + + Addr: Optional[str] = None + PrefixLen: Optional[int] = None + + +@dataclass +class ContainerNetworkSettings: + """Network settings for container.""" + + Bridge: Optional[str] = None + SandboxID: Optional[str] = None + HairpinMode: Optional[bool] = None + LinkLocalIPv6Address: Optional[str] = None + LinkLocalIPv6PrefixLen: Optional[str] = None + Ports: Optional[dict[str, Optional[list[ContainerPortBinding]]]] = None + SandboxKey: Optional[str] = None + SecondaryIPAddresses: Optional[list[ContainerAddress]] = None + SecondaryIPv6Addresses: Optional[list[ContainerAddress]] = None + EndpointID: Optional[str] = None + Gateway: Optional[str] = None + GlobalIPv6Address: Optional[str] = None + GlobalIPv6PrefixLen: Optional[int] = None + IPAddress: Optional[str] = None + IPPrefixLen: Optional[int] = None + IPv6Gateway: Optional[str] = None + MacAddress: Optional[str] = None + Networks: Optional[dict[str, ContainerNetworkEndpoint]] = None + + def get_networks(self) -> Optional[dict[str, ContainerNetworkEndpoint]]: + """Get networks for the container.""" + return self.Networks + + +@dataclass +class ContainerInspectInfo: + """Complete container information from docker inspect.""" + + Id: Optional[str] = None + Created: Optional[str] = None + Path: Optional[str] = None + Args: Optional[list[str]] = None + State: Optional[ContainerState] = None + Image: Optional[str] = None + ResolvConfPath: Optional[str] = None + HostnamePath: Optional[str] = None + HostsPath: Optional[str] = None + LogPath: Optional[str] = None + Name: Optional[str] = None + RestartCount: Optional[int] = None + Driver: Optional[str] = None + Platform: Optional[str] = None + ImageManifestDescriptor: Optional[ContainerImageManifestDescriptor] = None + MountLabel: Optional[str] = None + ProcessLabel: Optional[str] = None + AppArmorProfile: Optional[str] = None + ExecIDs: Optional[list[str]] = None + HostConfig: Optional[ContainerHostConfig] = None + GraphDriver: Optional[ContainerGraphDriver] = None + SizeRw: Optional[str] = None + SizeRootFs: Optional[str] = None + Mounts: Optional[list[ContainerMount]] = None + Config: Optional[ContainerConfig] = None + NetworkSettings: Optional[ContainerNetworkSettings] = None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo": + """Create from docker inspect JSON.""" + return cls( + Id=data.get("Id"), + Created=data.get("Created"), + Path=data.get("Path"), + Args=data.get("Args"), + State=cls._parse_state(data.get("State", {})) if data.get("State") else None, + Image=data.get("Image"), + ResolvConfPath=data.get("ResolvConfPath"), + HostnamePath=data.get("HostnamePath"), + HostsPath=data.get("HostsPath"), + LogPath=data.get("LogPath"), + Name=data.get("Name"), + RestartCount=data.get("RestartCount"), + Driver=data.get("Driver"), + Platform=data.get("Platform"), + ImageManifestDescriptor=cls._parse_image_manifest(data.get("ImageManifestDescriptor", {})) + if data.get("ImageManifestDescriptor") + else None, + MountLabel=data.get("MountLabel"), + ProcessLabel=data.get("ProcessLabel"), + AppArmorProfile=data.get("AppArmorProfile"), + ExecIDs=data.get("ExecIDs"), + HostConfig=cls._parse_host_config(data.get("HostConfig", {})) if data.get("HostConfig") else None, + GraphDriver=_ignore_properties(ContainerGraphDriver, data.get("GraphDriver", {})) + if data.get("GraphDriver") + else None, + SizeRw=data.get("SizeRw"), + SizeRootFs=data.get("SizeRootFs"), + Mounts=[_ignore_properties(ContainerMount, mount) for mount in data.get("Mounts", [])], + Config=_ignore_properties(ContainerConfig, data.get("Config", {})) if data.get("Config") else None, + NetworkSettings=cls._parse_network_settings(data.get("NetworkSettings", {})) + if data.get("NetworkSettings") + else None, + ) + + @classmethod + def _parse_state(cls, data: dict[str, Any]) -> Optional[ContainerState]: + """Parse State with nested Health object.""" + if not data: + return None + + health_data = data.get("Health", {}) + health = None + if health_data: + logs = [_ignore_properties(ContainerLog, log) for log in health_data.get("Log", [])] + health = ContainerHealth( + Status=health_data.get("Status"), + FailingStreak=health_data.get("FailingStreak"), + Log=logs if logs else None, + ) + + return ContainerState( + Status=data.get("Status"), + Running=data.get("Running"), + Paused=data.get("Paused"), + Restarting=data.get("Restarting"), + OOMKilled=data.get("OOMKilled"), + Dead=data.get("Dead"), + Pid=data.get("Pid"), + ExitCode=data.get("ExitCode"), + Error=data.get("Error"), + StartedAt=data.get("StartedAt"), + FinishedAt=data.get("FinishedAt"), + Health=health, + ) + + @classmethod + def _parse_image_manifest(cls, data: dict[str, Any]) -> Optional[ContainerImageManifestDescriptor]: + """Parse ImageManifestDescriptor with nested Platform.""" + if not data: + return None + + platform_data = data.get("platform", {}) + platform = _ignore_properties(ContainerPlatform, platform_data) if platform_data else None + + return ContainerImageManifestDescriptor( + mediaType=data.get("mediaType"), + digest=data.get("digest"), + size=data.get("size"), + urls=data.get("urls"), + annotations=data.get("annotations"), + data=data.get("data"), + platform=platform, + artifactType=data.get("artifactType"), + ) + + @classmethod + def _parse_host_config(cls, data: dict[str, Any]) -> Optional[ContainerHostConfig]: + """Parse HostConfig with all nested objects.""" + if not data: + return None + + blkio_weight_devices = [ + _ignore_properties(ContainerBlkioWeightDevice, d) for d in (data.get("BlkioWeightDevice") or []) + ] + blkio_read_bps = [ + _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceReadBps") or []) + ] + blkio_write_bps = [ + _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceWriteBps") or []) + ] + blkio_read_iops = [ + _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceReadIOps") or []) + ] + blkio_write_iops = [ + _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceWriteIOps") or []) + ] + devices = [_ignore_properties(ContainerDeviceMapping, d) for d in (data.get("Devices") or [])] + device_requests = [_ignore_properties(ContainerDeviceRequest, d) for d in (data.get("DeviceRequests") or [])] + ulimits = [_ignore_properties(ContainerUlimit, d) for d in (data.get("Ulimits") or [])] + mounts = [_ignore_properties(ContainerMountPoint, d) for d in (data.get("Mounts") or [])] + + port_bindings: dict[str, Optional[list[ContainerPortBinding]]] = {} + port_bindings_data = data.get("PortBindings") + if port_bindings_data is not None: + for port, bindings in port_bindings_data.items(): + if bindings is None: + port_bindings[port] = None + else: + port_bindings[port] = [_ignore_properties(ContainerPortBinding, b) for b in bindings] + + return ContainerHostConfig( + CpuShares=data.get("CpuShares"), + Memory=data.get("Memory"), + CgroupParent=data.get("CgroupParent"), + BlkioWeight=data.get("BlkioWeight"), + BlkioWeightDevice=blkio_weight_devices if blkio_weight_devices else None, + BlkioDeviceReadBps=blkio_read_bps if blkio_read_bps else None, + BlkioDeviceWriteBps=blkio_write_bps if blkio_write_bps else None, + BlkioDeviceReadIOps=blkio_read_iops if blkio_read_iops else None, + BlkioDeviceWriteIOps=blkio_write_iops if blkio_write_iops else None, + CpuPeriod=data.get("CpuPeriod"), + CpuQuota=data.get("CpuQuota"), + CpuRealtimePeriod=data.get("CpuRealtimePeriod"), + CpuRealtimeRuntime=data.get("CpuRealtimeRuntime"), + CpusetCpus=data.get("CpusetCpus"), + CpusetMems=data.get("CpusetMems"), + Devices=devices if devices else None, + DeviceCgroupRules=data.get("DeviceCgroupRules"), + DeviceRequests=device_requests if device_requests else None, + KernelMemoryTCP=data.get("KernelMemoryTCP"), + MemoryReservation=data.get("MemoryReservation"), + MemorySwap=data.get("MemorySwap"), + MemorySwappiness=data.get("MemorySwappiness"), + NanoCpus=data.get("NanoCpus"), + OomKillDisable=data.get("OomKillDisable"), + Init=data.get("Init"), + PidsLimit=data.get("PidsLimit"), + Ulimits=ulimits if ulimits else None, + CpuCount=data.get("CpuCount"), + CpuPercent=data.get("CpuPercent"), + IOMaximumIOps=data.get("IOMaximumIOps"), + IOMaximumBandwidth=data.get("IOMaximumBandwidth"), + Binds=data.get("Binds"), + ContainerIDFile=data.get("ContainerIDFile"), + LogConfig=_ignore_properties(ContainerLogConfig, data.get("LogConfig", {})) + if data.get("LogConfig") + else None, + NetworkMode=data.get("NetworkMode"), + PortBindings=port_bindings if port_bindings else None, + RestartPolicy=_ignore_properties(ContainerRestartPolicy, data.get("RestartPolicy", {})) + if data.get("RestartPolicy") + else None, + AutoRemove=data.get("AutoRemove"), + VolumeDriver=data.get("VolumeDriver"), + VolumesFrom=data.get("VolumesFrom"), + Mounts=mounts if mounts else None, + ConsoleSize=data.get("ConsoleSize"), + Annotations=data.get("Annotations"), + CapAdd=data.get("CapAdd"), + CapDrop=data.get("CapDrop"), + CgroupnsMode=data.get("CgroupnsMode"), + Dns=data.get("Dns"), + DnsOptions=data.get("DnsOptions"), + DnsSearch=data.get("DnsSearch"), + ExtraHosts=data.get("ExtraHosts"), + GroupAdd=data.get("GroupAdd"), + IpcMode=data.get("IpcMode"), + Cgroup=data.get("Cgroup"), + Links=data.get("Links"), + OomScoreAdj=data.get("OomScoreAdj"), + PidMode=data.get("PidMode"), + Privileged=data.get("Privileged"), + PublishAllPorts=data.get("PublishAllPorts"), + ReadonlyRootfs=data.get("ReadonlyRootfs"), + SecurityOpt=data.get("SecurityOpt"), + StorageOpt=data.get("StorageOpt"), + Tmpfs=data.get("Tmpfs"), + UTSMode=data.get("UTSMode"), + UsernsMode=data.get("UsernsMode"), + ShmSize=data.get("ShmSize"), + Sysctls=data.get("Sysctls"), + Runtime=data.get("Runtime"), + Isolation=data.get("Isolation"), + MaskedPaths=data.get("MaskedPaths"), + ReadonlyPaths=data.get("ReadonlyPaths"), + ) + + @classmethod + def _parse_network_settings(cls, data: dict[str, Any]) -> Optional[ContainerNetworkSettings]: + """Parse NetworkSettings with nested Networks and Ports.""" + if not data: + return None + + ports: dict[str, Optional[list[ContainerPortBinding]]] = {} + ports_data = data.get("Ports") + if ports_data is not None: + for port, bindings in ports_data.items(): + if bindings is None: + ports[port] = None + else: + ports[port] = [_ignore_properties(ContainerPortBinding, b) for b in bindings] + + networks = {} + networks_data = data.get("Networks") + if networks_data is not None: + for name, network_data in networks_data.items(): + networks[name] = _ignore_properties(ContainerNetworkEndpoint, network_data) + + secondary_ipv4 = [] + secondary_ipv4_data = data.get("SecondaryIPAddresses") + if secondary_ipv4_data is not None: + secondary_ipv4 = [_ignore_properties(ContainerAddress, addr) for addr in secondary_ipv4_data] + + secondary_ipv6 = [] + secondary_ipv6_data = data.get("SecondaryIPv6Addresses") + if secondary_ipv6_data is not None: + secondary_ipv6 = [_ignore_properties(ContainerAddress, addr) for addr in secondary_ipv6_data] + + return ContainerNetworkSettings( + Bridge=data.get("Bridge"), + SandboxID=data.get("SandboxID"), + HairpinMode=data.get("HairpinMode"), + LinkLocalIPv6Address=data.get("LinkLocalIPv6Address"), + LinkLocalIPv6PrefixLen=data.get("LinkLocalIPv6PrefixLen"), + Ports=ports if ports else None, + SandboxKey=data.get("SandboxKey"), + SecondaryIPAddresses=secondary_ipv4 if secondary_ipv4 else None, + SecondaryIPv6Addresses=secondary_ipv6 if secondary_ipv6 else None, + EndpointID=data.get("EndpointID"), + Gateway=data.get("Gateway"), + GlobalIPv6Address=data.get("GlobalIPv6Address"), + GlobalIPv6PrefixLen=data.get("GlobalIPv6PrefixLen"), + IPAddress=data.get("IPAddress"), + IPPrefixLen=data.get("IPPrefixLen"), + IPv6Gateway=data.get("IPv6Gateway"), + MacAddress=data.get("MacAddress"), + Networks=networks if networks else None, + ) + + def get_network_settings(self) -> Optional[ContainerNetworkSettings]: + """Get network settings for the container.""" + return self.NetworkSettings + + +def _ignore_properties(cls: type[_IPT], dict_: Any) -> _IPT: + """omits extra fields like @JsonIgnoreProperties(ignoreUnknown = true) + + https://gist.github.com/alexanderankin/2a4549ac03554a31bef6eaaf2eaf7fd5""" + if isinstance(dict_, cls): + return dict_ + if not is_dataclass(cls): + raise TypeError(f"Expected a dataclass type, got {cls}") + class_fields = {f.name for f in fields(cls)} + filtered = {k: v for k, v in dict_.items() if k in class_fields} + return cast("_IPT", cls(**filtered)) diff --git a/core/tests/test_compose.py b/core/tests/test_compose.py index 8d16d4756..c723ccde5 100644 --- a/core/tests/test_compose.py +++ b/core/tests/test_compose.py @@ -435,3 +435,27 @@ def test_container_info_none_when_no_docker_compose(): container = ComposeContainer() info = container.get_container_info() assert info is None + + +def test_docker_container_info(): + """Test get_container_info works with DockerContainer too""" + from testcontainers.core.container import DockerContainer + + with DockerContainer("hello-world") as container: + info = container.get_container_info() + assert info is not None + assert info.Id is not None + assert info.Image is not None + + if info.State: + assert hasattr(info.State, "Status") + assert hasattr(info.State, "Running") + + if info.Config: + assert hasattr(info.Config, "Image") + assert hasattr(info.Config, "Hostname") + + network_settings = info.get_network_settings() + if network_settings: + assert hasattr(network_settings, "IPAddress") + assert hasattr(network_settings, "Networks") diff --git a/core/tests/test_container.py b/core/tests/test_container.py index 30b80f79d..5e4387065 100644 --- a/core/tests/test_container.py +++ b/core/tests/test_container.py @@ -1,3 +1,5 @@ +from typing import Any + import pytest from testcontainers.core.container import DockerContainer @@ -8,6 +10,9 @@ class FakeContainer: + def __init__(self) -> None: + self.attrs: dict[str, Any] = {} + @property def id(self) -> str: return FAKE_ID @@ -96,3 +101,296 @@ def test_attribute(init_attr, init_value, class_attr, stored_value): """Test that the attributes set through the __init__ function are properly stored.""" with DockerContainer("ubuntu", **{init_attr: init_value}) as container: assert getattr(container, class_attr) == stored_value + + +def test_get_container_info_returns_none_when_no_container( + container: DockerContainer, monkeypatch: pytest.MonkeyPatch +) -> None: + """Test get_container_info returns None when container is not started.""" + monkeypatch.setattr(container, "_container", None) + info = container.get_container_info() + assert info is None + + +def test_get_container_info_lazy_loading(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: + """Test get_container_info lazy loading and caching.""" + fake_attrs = { + "Id": "test123", + "Name": "/test-container", + "Image": "nginx:alpine", + "State": {"Status": "running", "Running": True, "Pid": 1234}, + "Config": {"Image": "nginx:alpine", "Hostname": "test-host"}, + "NetworkSettings": {"IPAddress": "172.17.0.2"}, + } + + fake_container = FakeContainer() + fake_container.attrs = fake_attrs + monkeypatch.setattr(container, "_container", fake_container) + + # First call should populate cache + info1 = container.get_container_info() + assert info1 is not None + assert info1.Id == "test123" + assert info1.Name == "/test-container" + assert info1.Image == "nginx:alpine" + + # Second call should return cached result + info2 = container.get_container_info() + assert info1 is info2 # Same object reference + + +def test_get_container_info_structure(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: + """Test get_container_info returns properly structured data.""" + fake_attrs = { + "Id": "abc123def456", + "Name": "/my-test-container", + "Image": "sha256:nginx123", + "Created": "2023-01-01T00:00:00Z", + "State": { + "Status": "running", + "Running": True, + "Pid": 5678, + "ExitCode": 0, + "StartedAt": "2023-01-01T00:01:00Z", + "Health": { + "Status": "healthy", + "FailingStreak": 0, + "Log": [ + {"Start": "2023-01-01T00:00:00Z", "End": "2023-01-01T00:00:01Z", "ExitCode": 0, "Output": "healthy"} + ], + }, + }, + "Config": { + "Image": "nginx:alpine", + "Hostname": "my-hostname", + "Env": ["PATH=/usr/bin", "HOME=/root"], + "Cmd": ["nginx", "-g", "daemon off;"], + "ExposedPorts": {"80/tcp": {}}, + }, + "NetworkSettings": { + "IPAddress": "172.17.0.3", + "Gateway": "172.17.0.1", + "Networks": { + "bridge": { + "IPAddress": "172.17.0.3", + "Gateway": "172.17.0.1", + "NetworkID": "net123", + "MacAddress": "02:42:ac:11:00:03", + "Aliases": ["container-alias"], + } + }, + }, + "HostConfig": {"Memory": 1073741824, "CpuShares": 1024, "NetworkMode": "bridge"}, + } + + fake_container = FakeContainer() + fake_container.attrs = fake_attrs + monkeypatch.setattr(container, "_container", fake_container) + + info = container.get_container_info() + assert info is not None + + # Test basic fields + assert info.Id == "abc123def456" + assert info.Name == "/my-test-container" + assert info.Image == "sha256:nginx123" + assert info.Created == "2023-01-01T00:00:00Z" + + # Test State with Health + assert info.State is not None + assert info.State.Status == "running" + assert info.State.Running is True + assert info.State.Pid == 5678 + assert info.State.ExitCode == 0 + assert info.State.Health is not None + assert info.State.Health.Status == "healthy" + assert info.State.Health.FailingStreak == 0 + assert info.State.Health.Log is not None + assert len(info.State.Health.Log) == 1 + assert info.State.Health.Log[0].Output == "healthy" + + # Test Config + assert info.Config is not None + assert info.Config.Image == "nginx:alpine" + assert info.Config.Hostname == "my-hostname" + assert info.Config.Env == ["PATH=/usr/bin", "HOME=/root"] + assert info.Config.Cmd == ["nginx", "-g", "daemon off;"] + assert info.Config.ExposedPorts == {"80/tcp": {}} + + # Test NetworkSettings + network_settings = info.get_network_settings() + assert network_settings is not None + assert network_settings.IPAddress == "172.17.0.3" + assert network_settings.Gateway == "172.17.0.1" + + # Test Networks + assert network_settings.Networks is not None + assert "bridge" in network_settings.Networks + bridge_network = network_settings.Networks["bridge"] + assert bridge_network.IPAddress == "172.17.0.3" + assert bridge_network.Gateway == "172.17.0.1" + assert bridge_network.NetworkID == "net123" + assert bridge_network.MacAddress == "02:42:ac:11:00:03" + assert bridge_network.Aliases == ["container-alias"] + + # Test HostConfig + assert info.HostConfig is not None + assert info.HostConfig.Memory == 1073741824 + assert info.HostConfig.CpuShares == 1024 + assert info.HostConfig.NetworkMode == "bridge" + + +def test_get_container_info_handles_exceptions(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: + """Test get_container_info handles exceptions gracefully.""" + fake_container = FakeContainer() + + # Simulate an exception when accessing attrs + def raise_exception() -> None: + raise Exception("Docker API error") + + fake_container.attrs = property(lambda self: raise_exception()) # type: ignore[assignment] + monkeypatch.setattr(container, "_container", fake_container) + + info = container.get_container_info() + assert info is None + + +def test_get_container_info_with_none_values(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: + """Test get_container_info handles None values in HostConfig and NetworkSettings.""" + fake_attrs = { + "Id": "test-none-values", + "Name": "/test-none", + "Image": "nginx:alpine", + "State": {"Status": "running", "Running": True}, + "Config": {"Image": "nginx:alpine"}, + "NetworkSettings": { + "IPAddress": "172.17.0.2", + "Networks": None, + "Ports": None, + "SecondaryIPAddresses": None, + "SecondaryIPv6Addresses": None, + }, + "HostConfig": { + "Memory": 0, + "NetworkMode": "bridge", + "BlkioWeightDevice": None, + "BlkioDeviceReadBps": None, + "Devices": None, + "PortBindings": None, + }, + } + + fake_container = FakeContainer() + fake_container.attrs = fake_attrs + monkeypatch.setattr(container, "_container", fake_container) + + info = container.get_container_info() + assert info is not None + assert info.Id == "test-none-values" + + network_settings = info.get_network_settings() + assert network_settings is not None + assert network_settings.IPAddress == "172.17.0.2" + assert network_settings.Networks is None + assert network_settings.Ports is None + + assert info.HostConfig is not None + assert info.HostConfig.Memory == 0 + assert info.HostConfig.NetworkMode == "bridge" + assert info.HostConfig.BlkioWeightDevice is None + assert info.HostConfig.Devices is None + assert info.HostConfig.PortBindings is None + + +def test_get_container_info_with_port_bindings(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: + """Test get_container_info handles port bindings correctly.""" + fake_attrs = { + "Id": "test-port-bindings", + "Name": "/test-ports", + "Image": "nginx:alpine", + "State": {"Status": "running", "Running": True}, + "Config": {"Image": "nginx:alpine"}, + "NetworkSettings": {"Ports": {"80/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8080"}], "443/tcp": None}}, + "HostConfig": { + "NetworkMode": "bridge", + "PortBindings": {"80/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8080"}], "443/tcp": None}, + }, + } + + fake_container = FakeContainer() + fake_container.attrs = fake_attrs + monkeypatch.setattr(container, "_container", fake_container) + + info = container.get_container_info() + assert info is not None + + network_settings = info.get_network_settings() + assert network_settings is not None + assert network_settings.Ports is not None + assert "80/tcp" in network_settings.Ports + port_bindings = network_settings.Ports["80/tcp"] + assert port_bindings is not None + assert len(port_bindings) == 1 + assert port_bindings[0].HostPort == "8080" + assert network_settings.Ports["443/tcp"] is None + + assert info.HostConfig is not None + assert info.HostConfig.PortBindings is not None + assert "80/tcp" in info.HostConfig.PortBindings + host_port_bindings = info.HostConfig.PortBindings["80/tcp"] + assert host_port_bindings is not None + assert len(host_port_bindings) == 1 + assert host_port_bindings[0].HostPort == "8080" + assert info.HostConfig.PortBindings["443/tcp"] is None + + +def test_get_container_info_edge_cases_regression(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test for None value handling.""" + fake_attrs = { + "Id": "regression-test", + "Name": "/regression-container", + "Image": "nginx:alpine", + "State": {"Status": "running", "Running": True}, + "Config": {"Image": "nginx:alpine"}, + "NetworkSettings": { + "IPAddress": "172.17.0.2", + "Networks": None, + "Ports": None, + "SecondaryIPAddresses": None, + "SecondaryIPv6Addresses": None, + }, + "HostConfig": { + "Memory": 0, + "NetworkMode": "bridge", + "BlkioWeightDevice": None, + "BlkioDeviceReadBps": None, + "BlkioDeviceWriteBps": None, + "BlkioDeviceReadIOps": None, + "BlkioDeviceWriteIOps": None, + "Devices": None, + "DeviceRequests": None, + "Ulimits": None, + "Mounts": None, + "PortBindings": None, + }, + } + + fake_container = FakeContainer() + fake_container.attrs = fake_attrs + monkeypatch.setattr(container, "_container", fake_container) + + info = container.get_container_info() + assert info is not None + assert info.Id == "regression-test" + + network_settings = info.get_network_settings() + assert network_settings is not None + assert network_settings.Networks is None + assert network_settings.Ports is None + + host_config = info.HostConfig + assert host_config is not None + assert host_config.BlkioWeightDevice is None + assert host_config.BlkioDeviceReadBps is None + assert host_config.Devices is None + assert host_config.PortBindings is None diff --git a/docs/features/creating_container.md b/docs/features/creating_container.md index fa3b1190b..40c632943 100644 --- a/docs/features/creating_container.md +++ b/docs/features/creating_container.md @@ -128,6 +128,62 @@ def test_with_nginx(nginx_container): For details on waiting for containers to be ready, see [Wait strategies](wait_strategies.md). +## Container Information + +You can get detailed information about containers using the `get_container_info()` method. This works with both `DockerContainer` and `ComposeContainer`: + +```python +from testcontainers.generic import GenericContainer + +def test_container_info(): + with GenericContainer("nginx:alpine") as container: + # Get detailed container information + info = container.get_container_info() + + if info: + # Basic container information + print(f"Container ID: {info.Id}") + print(f"Name: {info.Name}") + print(f"Image: {info.Image}") + + # Container state + if info.State: + print(f"Status: {info.State.Status}") + print(f"Running: {info.State.Running}") + print(f"PID: {info.State.Pid}") + print(f"Exit Code: {info.State.ExitCode}") + + # Container configuration + if info.Config: + print(f"Hostname: {info.Config.Hostname}") + print(f"Environment: {info.Config.Env}") + print(f"Command: {info.Config.Cmd}") + + # Network information + network_settings = info.get_network_settings() + if network_settings and network_settings.Networks: + for network_name, network in network_settings.Networks.items(): + print(f"Network: {network_name}") + print(f" IP Address: {network.IPAddress}") + print(f" Gateway: {network.Gateway}") + print(f" MAC Address: {network.MacAddress}") +``` + +The container information is lazy-loaded and cached, so subsequent calls to `get_container_info()` will return the same data without making additional Docker API calls. + +### Available Information + +The `ContainerInspectInfo` object provides structured access to all Docker Engine API fields: + +- **Basic Info**: Container ID, name, image, creation time, platform +- **State**: Running status, PID, exit code, start/finish times, health status +- **Config**: Environment variables, command, working directory, labels, exposed ports +- **Network**: IP addresses, port bindings, network configurations, aliases +- **Host Config**: Memory limits, CPU settings, device mappings, restart policies +- **Mounts**: Volume and bind mount information with detailed options +- **Health**: Health check status and logs (if configured) +- **Platform**: Architecture and OS information + ## Best Practices 1. Always use context managers or ensure proper cleanup diff --git a/docs/features/docker_compose.md b/docs/features/docker_compose.md index 006a12b92..6b874a348 100644 --- a/docs/features/docker_compose.md +++ b/docs/features/docker_compose.md @@ -60,6 +60,22 @@ with DockerCompose("path/to/compose/directory") as compose: # Get container logs stdout, stderr = compose.get_logs("web") + + # Get detailed container information + container = compose.get_container("web") + info = container.get_container_info() + if info: + print(f"Container ID: {info.Id}") + if info.State: + print(f"Status: {info.State.Status}") + if info.Config: + print(f"Image: {info.Config.Image}") + + # Access network settings + network_settings = info.get_network_settings() + if network_settings and network_settings.Networks: + for name, network in network_settings.Networks.items(): + print(f"Network {name}: IP {network.IPAddress}") ``` ## Waiting for Services @@ -105,6 +121,46 @@ def test_web_application(): assert exit_code == 0 ``` +## Container Information + +You can get detailed information about containers using the `get_container_info()` method: + +```python +with DockerCompose("path/to/compose/directory") as compose: + container = compose.get_container("web") + info = container.get_container_info() + + if info: + # Basic container information + print(f"Container ID: {info.Id}") + print(f"Name: {info.Name}") + print(f"Image: {info.Image}") + + # Container state + if info.State: + print(f"Status: {info.State.Status}") + print(f"Running: {info.State.Running}") + print(f"PID: {info.State.Pid}") + print(f"Exit Code: {info.State.ExitCode}") + + # Container configuration + if info.Config: + print(f"Hostname: {info.Config.Hostname}") + print(f"Environment: {info.Config.Env}") + print(f"Command: {info.Config.Cmd}") + + # Network information + network_settings = info.get_network_settings() + if network_settings and network_settings.Networks: + for network_name, network in network_settings.Networks.items(): + print(f"Network: {network_name}") + print(f" IP Address: {network.IPAddress}") + print(f" Gateway: {network.Gateway}") + print(f" MAC Address: {network.MacAddress}") +``` + +The container information is lazy-loaded and cached, so subsequent calls to `get_container_info()` will return the same data without making additional Docker API calls. + ## Best Practices 1. Use context managers (`with` statement) to ensure proper cleanup From f1866d7cd534e7b5a5f47d8cdfc442a25807a71c Mon Sep 17 00:00:00 2001 From: VictorCavichioli Date: Wed, 8 Oct 2025 15:29:02 -0300 Subject: [PATCH 3/5] refactor: simplify container info parsing using __post_init__ pattern and centralize in DockerClient --- core/testcontainers/compose/compose.py | 21 +- core/testcontainers/core/container.py | 3 +- core/testcontainers/core/docker_client.py | 228 +++++++--------------- core/tests/test_container.py | 128 +++--------- 4 files changed, 111 insertions(+), 269 deletions(-) diff --git a/core/testcontainers/compose/compose.py b/core/testcontainers/compose/compose.py index 8163d8324..e424ab720 100644 --- a/core/testcontainers/compose/compose.py +++ b/core/testcontainers/compose/compose.py @@ -11,11 +11,10 @@ from types import TracebackType from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast -from testcontainers.core.docker_client import ContainerInspectInfo, _ignore_properties +from testcontainers.core.docker_client import ContainerInspectInfo, DockerClient, _ignore_properties from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed from testcontainers.core.waiting_utils import WaitStrategy -_IPT = TypeVar("_IPT") _WARNINGS = {"DOCKER_COMPOSE_GET_CONFIG": "get_config is experimental, see testcontainers/testcontainers-python#669"} logger = getLogger(__name__) @@ -145,15 +144,8 @@ def get_container_info(self) -> Optional[ContainerInspectInfo]: return None try: - inspect_command = ["docker", "inspect", self.ID] - result = self._docker_compose._run_command(cmd=inspect_command) - inspect_output = result.stdout.decode("utf-8").strip() - - if inspect_output: - raw_data = loads(inspect_output)[0] - self._cached_container_info = ContainerInspectInfo.from_dict(raw_data) - else: - self._cached_container_info = None + docker_client = self._docker_compose._get_docker_client() + self._cached_container_info = docker_client.get_container_inspect_info(self.ID) except Exception as e: logger.warning(f"Failed to get container info for {self.ID}: {e}") @@ -229,6 +221,7 @@ class DockerCompose: docker_command_path: Optional[str] = None profiles: Optional[list[str]] = None _wait_strategies: Optional[dict[str, Any]] = field(default=None, init=False, repr=False) + _docker_client: Optional[DockerClient] = field(default=None, init=False, repr=False) def __post_init__(self) -> None: if isinstance(self.compose_file_name, str): @@ -589,3 +582,9 @@ def wait_for(self, url: str) -> "DockerCompose": with urlopen(url) as response: response.read() return self + + def _get_docker_client(self) -> DockerClient: + """Get Docker client instance.""" + if self._docker_client is None: + self._docker_client = DockerClient() + return self._docker_client diff --git a/core/testcontainers/core/container.py b/core/testcontainers/core/container.py index 64902e4dc..6bf20239f 100644 --- a/core/testcontainers/core/container.py +++ b/core/testcontainers/core/container.py @@ -310,8 +310,7 @@ def get_container_info(self) -> Optional[ContainerInspectInfo]: return None try: - raw_data = self._container.attrs - self._cached_container_info = ContainerInspectInfo.from_dict(raw_data) + self._cached_container_info = self.get_docker_client().get_container_inspect_info(self._container.id) except Exception as e: logger.warning(f"Failed to get container info for {self._container.id}: {e}") diff --git a/core/testcontainers/core/docker_client.py b/core/testcontainers/core/docker_client.py index 901bcf208..4250f8ef4 100644 --- a/core/testcontainers/core/docker_client.py +++ b/core/testcontainers/core/docker_client.py @@ -266,6 +266,11 @@ def client_networks_create(self, name: str, param: dict[str, Any]) -> "DockerNet labels = create_labels("", param.get("labels")) return self.client.networks.create(name, **{**param, "labels": labels}) + def get_container_inspect_info(self, container_id: str) -> "ContainerInspectInfo": + """Get container inspect information with fresh data.""" + container = self.client.containers.get(container_id) + return ContainerInspectInfo.from_dict(container.attrs) + def get_docker_host() -> Optional[str]: return c.tc_properties_get_tc_host() or os.getenv("DOCKER_HOST") @@ -542,6 +547,44 @@ class ContainerHostConfig: MaskedPaths: Optional[list[str]] = None ReadonlyPaths: Optional[list[str]] = None + def __post_init__(self) -> None: + list_conversions = [ + ("BlkioWeightDevice", ContainerBlkioWeightDevice), + ("BlkioDeviceReadBps", ContainerBlkioDeviceRate), + ("BlkioDeviceWriteBps", ContainerBlkioDeviceRate), + ("BlkioDeviceReadIOps", ContainerBlkioDeviceRate), + ("BlkioDeviceWriteIOps", ContainerBlkioDeviceRate), + ("Devices", ContainerDeviceMapping), + ("DeviceRequests", ContainerDeviceRequest), + ("Ulimits", ContainerUlimit), + ("Mounts", ContainerMountPoint), + ] + + for field_name, target_class in list_conversions: + field_value = getattr(self, field_name) + if field_value is not None and isinstance(field_value, list): + setattr( + self, + field_name, + [ + _ignore_properties(target_class, item) if isinstance(item, dict) else item + for item in field_value + ], + ) + + if self.LogConfig is not None and isinstance(self.LogConfig, dict): + self.LogConfig = _ignore_properties(ContainerLogConfig, self.LogConfig) + + if self.RestartPolicy is not None and isinstance(self.RestartPolicy, dict): + self.RestartPolicy = _ignore_properties(ContainerRestartPolicy, self.RestartPolicy) + + if self.PortBindings is not None and isinstance(self.PortBindings, dict): + for port, bindings in self.PortBindings.items(): + if bindings is not None and isinstance(bindings, list): + self.PortBindings[port] = [ + _ignore_properties(ContainerPortBinding, b) if isinstance(b, dict) else b for b in bindings + ] + @dataclass class ContainerGraphDriver: @@ -669,6 +712,31 @@ class ContainerNetworkSettings: MacAddress: Optional[str] = None Networks: Optional[dict[str, ContainerNetworkEndpoint]] = None + def __post_init__(self) -> None: + if self.Ports is not None and isinstance(self.Ports, dict): + for port, bindings in self.Ports.items(): + if bindings is not None and isinstance(bindings, list): + self.Ports[port] = [ + _ignore_properties(ContainerPortBinding, b) if isinstance(b, dict) else b for b in bindings + ] + + if self.Networks is not None and isinstance(self.Networks, dict): + for name, network_data in self.Networks.items(): + if isinstance(network_data, dict): + self.Networks[name] = _ignore_properties(ContainerNetworkEndpoint, network_data) + + if self.SecondaryIPAddresses is not None and isinstance(self.SecondaryIPAddresses, list): + self.SecondaryIPAddresses = [ + _ignore_properties(ContainerAddress, addr) if isinstance(addr, dict) else addr + for addr in self.SecondaryIPAddresses + ] + + if self.SecondaryIPv6Addresses is not None and isinstance(self.SecondaryIPv6Addresses, list): + self.SecondaryIPv6Addresses = [ + _ignore_properties(ContainerAddress, addr) if isinstance(addr, dict) else addr + for addr in self.SecondaryIPv6Addresses + ] + def get_networks(self) -> Optional[dict[str, ContainerNetworkEndpoint]]: """Get networks for the container.""" return self.Networks @@ -730,7 +798,9 @@ def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo": ProcessLabel=data.get("ProcessLabel"), AppArmorProfile=data.get("AppArmorProfile"), ExecIDs=data.get("ExecIDs"), - HostConfig=cls._parse_host_config(data.get("HostConfig", {})) if data.get("HostConfig") else None, + HostConfig=_ignore_properties(ContainerHostConfig, data.get("HostConfig", {})) + if data.get("HostConfig") + else None, GraphDriver=_ignore_properties(ContainerGraphDriver, data.get("GraphDriver", {})) if data.get("GraphDriver") else None, @@ -738,7 +808,7 @@ def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo": SizeRootFs=data.get("SizeRootFs"), Mounts=[_ignore_properties(ContainerMount, mount) for mount in data.get("Mounts", [])], Config=_ignore_properties(ContainerConfig, data.get("Config", {})) if data.get("Config") else None, - NetworkSettings=cls._parse_network_settings(data.get("NetworkSettings", {})) + NetworkSettings=_ignore_properties(ContainerNetworkSettings, data.get("NetworkSettings", {})) if data.get("NetworkSettings") else None, ) @@ -799,164 +869,14 @@ def _parse_host_config(cls, data: dict[str, Any]) -> Optional[ContainerHostConfi """Parse HostConfig with all nested objects.""" if not data: return None - - blkio_weight_devices = [ - _ignore_properties(ContainerBlkioWeightDevice, d) for d in (data.get("BlkioWeightDevice") or []) - ] - blkio_read_bps = [ - _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceReadBps") or []) - ] - blkio_write_bps = [ - _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceWriteBps") or []) - ] - blkio_read_iops = [ - _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceReadIOps") or []) - ] - blkio_write_iops = [ - _ignore_properties(ContainerBlkioDeviceRate, d) for d in (data.get("BlkioDeviceWriteIOps") or []) - ] - devices = [_ignore_properties(ContainerDeviceMapping, d) for d in (data.get("Devices") or [])] - device_requests = [_ignore_properties(ContainerDeviceRequest, d) for d in (data.get("DeviceRequests") or [])] - ulimits = [_ignore_properties(ContainerUlimit, d) for d in (data.get("Ulimits") or [])] - mounts = [_ignore_properties(ContainerMountPoint, d) for d in (data.get("Mounts") or [])] - - port_bindings: dict[str, Optional[list[ContainerPortBinding]]] = {} - port_bindings_data = data.get("PortBindings") - if port_bindings_data is not None: - for port, bindings in port_bindings_data.items(): - if bindings is None: - port_bindings[port] = None - else: - port_bindings[port] = [_ignore_properties(ContainerPortBinding, b) for b in bindings] - - return ContainerHostConfig( - CpuShares=data.get("CpuShares"), - Memory=data.get("Memory"), - CgroupParent=data.get("CgroupParent"), - BlkioWeight=data.get("BlkioWeight"), - BlkioWeightDevice=blkio_weight_devices if blkio_weight_devices else None, - BlkioDeviceReadBps=blkio_read_bps if blkio_read_bps else None, - BlkioDeviceWriteBps=blkio_write_bps if blkio_write_bps else None, - BlkioDeviceReadIOps=blkio_read_iops if blkio_read_iops else None, - BlkioDeviceWriteIOps=blkio_write_iops if blkio_write_iops else None, - CpuPeriod=data.get("CpuPeriod"), - CpuQuota=data.get("CpuQuota"), - CpuRealtimePeriod=data.get("CpuRealtimePeriod"), - CpuRealtimeRuntime=data.get("CpuRealtimeRuntime"), - CpusetCpus=data.get("CpusetCpus"), - CpusetMems=data.get("CpusetMems"), - Devices=devices if devices else None, - DeviceCgroupRules=data.get("DeviceCgroupRules"), - DeviceRequests=device_requests if device_requests else None, - KernelMemoryTCP=data.get("KernelMemoryTCP"), - MemoryReservation=data.get("MemoryReservation"), - MemorySwap=data.get("MemorySwap"), - MemorySwappiness=data.get("MemorySwappiness"), - NanoCpus=data.get("NanoCpus"), - OomKillDisable=data.get("OomKillDisable"), - Init=data.get("Init"), - PidsLimit=data.get("PidsLimit"), - Ulimits=ulimits if ulimits else None, - CpuCount=data.get("CpuCount"), - CpuPercent=data.get("CpuPercent"), - IOMaximumIOps=data.get("IOMaximumIOps"), - IOMaximumBandwidth=data.get("IOMaximumBandwidth"), - Binds=data.get("Binds"), - ContainerIDFile=data.get("ContainerIDFile"), - LogConfig=_ignore_properties(ContainerLogConfig, data.get("LogConfig", {})) - if data.get("LogConfig") - else None, - NetworkMode=data.get("NetworkMode"), - PortBindings=port_bindings if port_bindings else None, - RestartPolicy=_ignore_properties(ContainerRestartPolicy, data.get("RestartPolicy", {})) - if data.get("RestartPolicy") - else None, - AutoRemove=data.get("AutoRemove"), - VolumeDriver=data.get("VolumeDriver"), - VolumesFrom=data.get("VolumesFrom"), - Mounts=mounts if mounts else None, - ConsoleSize=data.get("ConsoleSize"), - Annotations=data.get("Annotations"), - CapAdd=data.get("CapAdd"), - CapDrop=data.get("CapDrop"), - CgroupnsMode=data.get("CgroupnsMode"), - Dns=data.get("Dns"), - DnsOptions=data.get("DnsOptions"), - DnsSearch=data.get("DnsSearch"), - ExtraHosts=data.get("ExtraHosts"), - GroupAdd=data.get("GroupAdd"), - IpcMode=data.get("IpcMode"), - Cgroup=data.get("Cgroup"), - Links=data.get("Links"), - OomScoreAdj=data.get("OomScoreAdj"), - PidMode=data.get("PidMode"), - Privileged=data.get("Privileged"), - PublishAllPorts=data.get("PublishAllPorts"), - ReadonlyRootfs=data.get("ReadonlyRootfs"), - SecurityOpt=data.get("SecurityOpt"), - StorageOpt=data.get("StorageOpt"), - Tmpfs=data.get("Tmpfs"), - UTSMode=data.get("UTSMode"), - UsernsMode=data.get("UsernsMode"), - ShmSize=data.get("ShmSize"), - Sysctls=data.get("Sysctls"), - Runtime=data.get("Runtime"), - Isolation=data.get("Isolation"), - MaskedPaths=data.get("MaskedPaths"), - ReadonlyPaths=data.get("ReadonlyPaths"), - ) + return _ignore_properties(ContainerHostConfig, data) @classmethod def _parse_network_settings(cls, data: dict[str, Any]) -> Optional[ContainerNetworkSettings]: """Parse NetworkSettings with nested Networks and Ports.""" if not data: return None - - ports: dict[str, Optional[list[ContainerPortBinding]]] = {} - ports_data = data.get("Ports") - if ports_data is not None: - for port, bindings in ports_data.items(): - if bindings is None: - ports[port] = None - else: - ports[port] = [_ignore_properties(ContainerPortBinding, b) for b in bindings] - - networks = {} - networks_data = data.get("Networks") - if networks_data is not None: - for name, network_data in networks_data.items(): - networks[name] = _ignore_properties(ContainerNetworkEndpoint, network_data) - - secondary_ipv4 = [] - secondary_ipv4_data = data.get("SecondaryIPAddresses") - if secondary_ipv4_data is not None: - secondary_ipv4 = [_ignore_properties(ContainerAddress, addr) for addr in secondary_ipv4_data] - - secondary_ipv6 = [] - secondary_ipv6_data = data.get("SecondaryIPv6Addresses") - if secondary_ipv6_data is not None: - secondary_ipv6 = [_ignore_properties(ContainerAddress, addr) for addr in secondary_ipv6_data] - - return ContainerNetworkSettings( - Bridge=data.get("Bridge"), - SandboxID=data.get("SandboxID"), - HairpinMode=data.get("HairpinMode"), - LinkLocalIPv6Address=data.get("LinkLocalIPv6Address"), - LinkLocalIPv6PrefixLen=data.get("LinkLocalIPv6PrefixLen"), - Ports=ports if ports else None, - SandboxKey=data.get("SandboxKey"), - SecondaryIPAddresses=secondary_ipv4 if secondary_ipv4 else None, - SecondaryIPv6Addresses=secondary_ipv6 if secondary_ipv6 else None, - EndpointID=data.get("EndpointID"), - Gateway=data.get("Gateway"), - GlobalIPv6Address=data.get("GlobalIPv6Address"), - GlobalIPv6PrefixLen=data.get("GlobalIPv6PrefixLen"), - IPAddress=data.get("IPAddress"), - IPPrefixLen=data.get("IPPrefixLen"), - IPv6Gateway=data.get("IPv6Gateway"), - MacAddress=data.get("MacAddress"), - Networks=networks if networks else None, - ) + return _ignore_properties(ContainerNetworkSettings, data) def get_network_settings(self) -> Optional[ContainerNetworkSettings]: """Get network settings for the container.""" diff --git a/core/tests/test_container.py b/core/tests/test_container.py index 5e4387065..32d35f2a9 100644 --- a/core/tests/test_container.py +++ b/core/tests/test_container.py @@ -3,7 +3,7 @@ import pytest from testcontainers.core.container import DockerContainer -from testcontainers.core.docker_client import DockerClient +from testcontainers.core.docker_client import DockerClient, ContainerInspectInfo from testcontainers.core.config import ConnectionMode FAKE_ID = "ABC123" @@ -114,34 +114,24 @@ def test_get_container_info_returns_none_when_no_container( def test_get_container_info_lazy_loading(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: """Test get_container_info lazy loading and caching.""" - fake_attrs = { - "Id": "test123", - "Name": "/test-container", - "Image": "nginx:alpine", - "State": {"Status": "running", "Running": True, "Pid": 1234}, - "Config": {"Image": "nginx:alpine", "Hostname": "test-host"}, - "NetworkSettings": {"IPAddress": "172.17.0.2"}, - } + fake_data = {"Id": "test123", "Name": "/test-container", "Image": "nginx:alpine"} + fake_info = ContainerInspectInfo.from_dict(fake_data) - fake_container = FakeContainer() - fake_container.attrs = fake_attrs - monkeypatch.setattr(container, "_container", fake_container) + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) - # First call should populate cache info1 = container.get_container_info() assert info1 is not None assert info1.Id == "test123" assert info1.Name == "/test-container" assert info1.Image == "nginx:alpine" - # Second call should return cached result info2 = container.get_container_info() - assert info1 is info2 # Same object reference + assert info1 is info2 def test_get_container_info_structure(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: """Test get_container_info returns properly structured data.""" - fake_attrs = { + fake_data = { "Id": "abc123def456", "Name": "/my-test-container", "Image": "sha256:nginx123", @@ -151,14 +141,7 @@ def test_get_container_info_structure(container: DockerContainer, monkeypatch: p "Running": True, "Pid": 5678, "ExitCode": 0, - "StartedAt": "2023-01-01T00:01:00Z", - "Health": { - "Status": "healthy", - "FailingStreak": 0, - "Log": [ - {"Start": "2023-01-01T00:00:00Z", "End": "2023-01-01T00:00:01Z", "ExitCode": 0, "Output": "healthy"} - ], - }, + "Health": {"Status": "healthy", "FailingStreak": 0, "Log": [{"Output": "healthy"}]}, }, "Config": { "Image": "nginx:alpine", @@ -182,21 +165,18 @@ def test_get_container_info_structure(container: DockerContainer, monkeypatch: p }, "HostConfig": {"Memory": 1073741824, "CpuShares": 1024, "NetworkMode": "bridge"}, } + fake_info = ContainerInspectInfo.from_dict(fake_data) - fake_container = FakeContainer() - fake_container.attrs = fake_attrs - monkeypatch.setattr(container, "_container", fake_container) + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) info = container.get_container_info() assert info is not None - # Test basic fields assert info.Id == "abc123def456" assert info.Name == "/my-test-container" assert info.Image == "sha256:nginx123" assert info.Created == "2023-01-01T00:00:00Z" - # Test State with Health assert info.State is not None assert info.State.Status == "running" assert info.State.Running is True @@ -209,7 +189,6 @@ def test_get_container_info_structure(container: DockerContainer, monkeypatch: p assert len(info.State.Health.Log) == 1 assert info.State.Health.Log[0].Output == "healthy" - # Test Config assert info.Config is not None assert info.Config.Image == "nginx:alpine" assert info.Config.Hostname == "my-hostname" @@ -217,13 +196,11 @@ def test_get_container_info_structure(container: DockerContainer, monkeypatch: p assert info.Config.Cmd == ["nginx", "-g", "daemon off;"] assert info.Config.ExposedPorts == {"80/tcp": {}} - # Test NetworkSettings network_settings = info.get_network_settings() assert network_settings is not None assert network_settings.IPAddress == "172.17.0.3" assert network_settings.Gateway == "172.17.0.1" - # Test Networks assert network_settings.Networks is not None assert "bridge" in network_settings.Networks bridge_network = network_settings.Networks["bridge"] @@ -233,7 +210,6 @@ def test_get_container_info_structure(container: DockerContainer, monkeypatch: p assert bridge_network.MacAddress == "02:42:ac:11:00:03" assert bridge_network.Aliases == ["container-alias"] - # Test HostConfig assert info.HostConfig is not None assert info.HostConfig.Memory == 1073741824 assert info.HostConfig.CpuShares == 1024 @@ -242,14 +218,11 @@ def test_get_container_info_structure(container: DockerContainer, monkeypatch: p def test_get_container_info_handles_exceptions(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: """Test get_container_info handles exceptions gracefully.""" - fake_container = FakeContainer() - # Simulate an exception when accessing attrs - def raise_exception() -> None: + def mock_exception(_): raise Exception("Docker API error") - fake_container.attrs = property(lambda self: raise_exception()) # type: ignore[assignment] - monkeypatch.setattr(container, "_container", fake_container) + monkeypatch.setattr(container._docker, "get_container_inspect_info", mock_exception) info = container.get_container_info() assert info is None @@ -257,32 +230,16 @@ def raise_exception() -> None: def test_get_container_info_with_none_values(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: """Test get_container_info handles None values in HostConfig and NetworkSettings.""" - fake_attrs = { + fake_data = { "Id": "test-none-values", "Name": "/test-none", "Image": "nginx:alpine", - "State": {"Status": "running", "Running": True}, - "Config": {"Image": "nginx:alpine"}, - "NetworkSettings": { - "IPAddress": "172.17.0.2", - "Networks": None, - "Ports": None, - "SecondaryIPAddresses": None, - "SecondaryIPv6Addresses": None, - }, - "HostConfig": { - "Memory": 0, - "NetworkMode": "bridge", - "BlkioWeightDevice": None, - "BlkioDeviceReadBps": None, - "Devices": None, - "PortBindings": None, - }, + "NetworkSettings": {"IPAddress": "172.17.0.2", "Networks": None, "Ports": None}, + "HostConfig": {"Memory": 0, "NetworkMode": "bridge", "PortBindings": None}, } + fake_info = ContainerInspectInfo.from_dict(fake_data) - fake_container = FakeContainer() - fake_container.attrs = fake_attrs - monkeypatch.setattr(container, "_container", fake_container) + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) info = container.get_container_info() assert info is not None @@ -297,29 +254,21 @@ def test_get_container_info_with_none_values(container: DockerContainer, monkeyp assert info.HostConfig is not None assert info.HostConfig.Memory == 0 assert info.HostConfig.NetworkMode == "bridge" - assert info.HostConfig.BlkioWeightDevice is None - assert info.HostConfig.Devices is None assert info.HostConfig.PortBindings is None def test_get_container_info_with_port_bindings(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: """Test get_container_info handles port bindings correctly.""" - fake_attrs = { + fake_data = { "Id": "test-port-bindings", "Name": "/test-ports", "Image": "nginx:alpine", - "State": {"Status": "running", "Running": True}, - "Config": {"Image": "nginx:alpine"}, - "NetworkSettings": {"Ports": {"80/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8080"}], "443/tcp": None}}, - "HostConfig": { - "NetworkMode": "bridge", - "PortBindings": {"80/tcp": [{"HostIp": "0.0.0.0", "HostPort": "8080"}], "443/tcp": None}, - }, + "NetworkSettings": {"Ports": {"80/tcp": [{"HostPort": "8080"}], "443/tcp": None}}, + "HostConfig": {"NetworkMode": "bridge", "PortBindings": {"80/tcp": [{"HostPort": "8080"}], "443/tcp": None}}, } + fake_info = ContainerInspectInfo.from_dict(fake_data) - fake_container = FakeContainer() - fake_container.attrs = fake_attrs - monkeypatch.setattr(container, "_container", fake_container) + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) info = container.get_container_info() assert info is not None @@ -346,38 +295,16 @@ def test_get_container_info_with_port_bindings(container: DockerContainer, monke def test_get_container_info_edge_cases_regression(container: DockerContainer, monkeypatch: pytest.MonkeyPatch) -> None: """Regression test for None value handling.""" - fake_attrs = { + fake_data = { "Id": "regression-test", "Name": "/regression-container", "Image": "nginx:alpine", - "State": {"Status": "running", "Running": True}, - "Config": {"Image": "nginx:alpine"}, - "NetworkSettings": { - "IPAddress": "172.17.0.2", - "Networks": None, - "Ports": None, - "SecondaryIPAddresses": None, - "SecondaryIPv6Addresses": None, - }, - "HostConfig": { - "Memory": 0, - "NetworkMode": "bridge", - "BlkioWeightDevice": None, - "BlkioDeviceReadBps": None, - "BlkioDeviceWriteBps": None, - "BlkioDeviceReadIOps": None, - "BlkioDeviceWriteIOps": None, - "Devices": None, - "DeviceRequests": None, - "Ulimits": None, - "Mounts": None, - "PortBindings": None, - }, + "NetworkSettings": {"IPAddress": "172.17.0.2", "Networks": None, "Ports": None}, + "HostConfig": {"Memory": 0, "NetworkMode": "bridge", "PortBindings": None}, } + fake_info = ContainerInspectInfo.from_dict(fake_data) - fake_container = FakeContainer() - fake_container.attrs = fake_attrs - monkeypatch.setattr(container, "_container", fake_container) + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) info = container.get_container_info() assert info is not None @@ -390,7 +317,4 @@ def test_get_container_info_edge_cases_regression(container: DockerContainer, mo host_config = info.HostConfig assert host_config is not None - assert host_config.BlkioWeightDevice is None - assert host_config.BlkioDeviceReadBps is None - assert host_config.Devices is None assert host_config.PortBindings is None From 66804f1ef03280890c2c3326f882c90691b314d2 Mon Sep 17 00:00:00 2001 From: VictorCavichioli Date: Wed, 8 Oct 2025 15:42:00 -0300 Subject: [PATCH 4/5] Remove redundant test in test_compose.py --- core/tests/test_compose.py | 27 +-------------------------- 1 file changed, 1 insertion(+), 26 deletions(-) diff --git a/core/tests/test_compose.py b/core/tests/test_compose.py index c723ccde5..357ecc52e 100644 --- a/core/tests/test_compose.py +++ b/core/tests/test_compose.py @@ -8,7 +8,7 @@ import pytest from pytest_mock import MockerFixture -from testcontainers.compose import DockerCompose +from testcontainers.compose import DockerCompose, ComposeContainer from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed FIXTURES = Path(__file__).parent.joinpath("compose_fixtures") @@ -430,32 +430,7 @@ def test_container_info_network_details(): def test_container_info_none_when_no_docker_compose(): """Test get_container_info returns None when docker_compose reference is missing""" - from testcontainers.compose.compose import ComposeContainer container = ComposeContainer() info = container.get_container_info() assert info is None - - -def test_docker_container_info(): - """Test get_container_info works with DockerContainer too""" - from testcontainers.core.container import DockerContainer - - with DockerContainer("hello-world") as container: - info = container.get_container_info() - assert info is not None - assert info.Id is not None - assert info.Image is not None - - if info.State: - assert hasattr(info.State, "Status") - assert hasattr(info.State, "Running") - - if info.Config: - assert hasattr(info.Config, "Image") - assert hasattr(info.Config, "Hostname") - - network_settings = info.get_network_settings() - if network_settings: - assert hasattr(network_settings, "IPAddress") - assert hasattr(network_settings, "Networks") From 6f2fdab6e30644db3e0a3de0be6d345216194b96 Mon Sep 17 00:00:00 2001 From: VictorCavichioli Date: Tue, 28 Oct 2025 15:45:17 -0300 Subject: [PATCH 5/5] Fix: make docs fails with reference target not found --- conf.py | 1 + core/testcontainers/compose/compose.py | 6 +++++- core/testcontainers/core/container.py | 6 +++++- 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/conf.py b/conf.py index 25271fd6c..3c37b2bff 100644 --- a/conf.py +++ b/conf.py @@ -168,4 +168,5 @@ nitpick_ignore = [ ("py:class", "typing_extensions.Self"), ("py:class", "docker.models.containers.ExecResult"), + ("py:class", "testcontainers.core.docker_client.ContainerInspectInfo"), ] diff --git a/core/testcontainers/compose/compose.py b/core/testcontainers/compose/compose.py index e424ab720..5f9f76066 100644 --- a/core/testcontainers/compose/compose.py +++ b/core/testcontainers/compose/compose.py @@ -136,7 +136,11 @@ def reload(self) -> None: pass def get_container_info(self) -> Optional[ContainerInspectInfo]: - """Get container information via docker inspect (lazy loaded).""" + """Get container information via docker inspect (lazy loaded). + + Returns: + Container inspect information or None if container is not started. + """ if self._cached_container_info is not None: return self._cached_container_info diff --git a/core/testcontainers/core/container.py b/core/testcontainers/core/container.py index 6bf20239f..c203f46d5 100644 --- a/core/testcontainers/core/container.py +++ b/core/testcontainers/core/container.py @@ -302,7 +302,11 @@ def exec(self, command: Union[str, list[str]]) -> ExecResult: return self._container.exec_run(command) def get_container_info(self) -> Optional[ContainerInspectInfo]: - """Get container information via docker inspect (lazy loaded).""" + """Get container information via docker inspect (lazy loaded). + + Returns: + Container inspect information or None if container is not started. + """ if self._cached_container_info is not None: return self._cached_container_info