diff --git a/core/testcontainers/compose/compose.py b/core/testcontainers/compose/compose.py index 61961ce0d..e424ab720 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,28 +11,15 @@ from types import TracebackType from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast +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__) -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 PublishedPortModel: """ @@ -81,6 +68,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 +135,24 @@ 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: + 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}") + self._cached_container_info = None + + return self._cached_container_info + @property def status(self) -> str: """Get container status for compatibility with wait strategies.""" @@ -215,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): @@ -575,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 4bb4eec48..6bf20239f 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,23 @@ 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: + 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}") + 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..4250f8ef4 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]: @@ -264,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") @@ -271,3 +278,619 @@ 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 + + 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: + """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 __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 + + +@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=_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, + 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=_ignore_properties(ContainerNetworkSettings, 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 + 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 + return _ignore_properties(ContainerNetworkSettings, data) + + 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 755b8b17b..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") @@ -378,3 +378,59 @@ 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""" + + container = ComposeContainer() + info = container.get_container_info() + assert info is None diff --git a/core/tests/test_container.py b/core/tests/test_container.py index 30b80f79d..32d35f2a9 100644 --- a/core/tests/test_container.py +++ b/core/tests/test_container.py @@ -1,13 +1,18 @@ +from typing import Any + 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" class FakeContainer: + def __init__(self) -> None: + self.attrs: dict[str, Any] = {} + @property def id(self) -> str: return FAKE_ID @@ -96,3 +101,220 @@ 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_data = {"Id": "test123", "Name": "/test-container", "Image": "nginx:alpine"} + fake_info = ContainerInspectInfo.from_dict(fake_data) + + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) + + info1 = container.get_container_info() + assert info1 is not None + assert info1.Id == "test123" + assert info1.Name == "/test-container" + assert info1.Image == "nginx:alpine" + + info2 = container.get_container_info() + 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_data = { + "Id": "abc123def456", + "Name": "/my-test-container", + "Image": "sha256:nginx123", + "Created": "2023-01-01T00:00:00Z", + "State": { + "Status": "running", + "Running": True, + "Pid": 5678, + "ExitCode": 0, + "Health": {"Status": "healthy", "FailingStreak": 0, "Log": [{"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_info = ContainerInspectInfo.from_dict(fake_data) + + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) + + info = container.get_container_info() + assert info is not None + + assert info.Id == "abc123def456" + assert info.Name == "/my-test-container" + assert info.Image == "sha256:nginx123" + assert info.Created == "2023-01-01T00:00:00Z" + + 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" + + 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": {}} + + 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" + + 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"] + + 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.""" + + def mock_exception(_): + raise Exception("Docker API error") + + monkeypatch.setattr(container._docker, "get_container_inspect_info", mock_exception) + + 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_data = { + "Id": "test-none-values", + "Name": "/test-none", + "Image": "nginx:alpine", + "NetworkSettings": {"IPAddress": "172.17.0.2", "Networks": None, "Ports": None}, + "HostConfig": {"Memory": 0, "NetworkMode": "bridge", "PortBindings": None}, + } + fake_info = ContainerInspectInfo.from_dict(fake_data) + + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) + + 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.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_data = { + "Id": "test-port-bindings", + "Name": "/test-ports", + "Image": "nginx:alpine", + "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) + + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) + + 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_data = { + "Id": "regression-test", + "Name": "/regression-container", + "Image": "nginx:alpine", + "NetworkSettings": {"IPAddress": "172.17.0.2", "Networks": None, "Ports": None}, + "HostConfig": {"Memory": 0, "NetworkMode": "bridge", "PortBindings": None}, + } + fake_info = ContainerInspectInfo.from_dict(fake_data) + + monkeypatch.setattr(container._docker, "get_container_inspect_info", lambda _: fake_info) + + 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.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