Skip to content

Commit 2cca44b

Browse files
feat(compose): add structured container inspect information
1 parent bb646e9 commit 2cca44b

File tree

2 files changed

+199
-0
lines changed

2 files changed

+199
-0
lines changed

core/testcontainers/compose/compose.py

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,122 @@ def _ignore_properties(cls: type[_IPT], dict_: Any) -> _IPT:
3333
return cast("_IPT", cls(**filtered))
3434

3535

36+
@dataclass
37+
class ContainerState:
38+
"""Container state from docker inspect."""
39+
40+
Status: Optional[str] = None
41+
Running: Optional[bool] = None
42+
Paused: Optional[bool] = None
43+
Restarting: Optional[bool] = None
44+
OOMKilled: Optional[bool] = None
45+
Dead: Optional[bool] = None
46+
Pid: Optional[int] = None
47+
ExitCode: Optional[int] = None
48+
Error: Optional[str] = None
49+
StartedAt: Optional[str] = None
50+
FinishedAt: Optional[str] = None
51+
52+
53+
@dataclass
54+
class ContainerConfig:
55+
"""Container config from docker inspect."""
56+
57+
Hostname: Optional[str] = None
58+
User: Optional[str] = None
59+
Env: Optional[list[str]] = None
60+
Cmd: Optional[list[str]] = None
61+
Image: Optional[str] = None
62+
WorkingDir: Optional[str] = None
63+
Entrypoint: Optional[list[str]] = None
64+
ExposedPorts: Optional[dict[str, Any]] = None
65+
Labels: Optional[dict[str, str]] = None
66+
67+
68+
@dataclass
69+
class Network:
70+
"""Individual network from docker inspect."""
71+
72+
IPAddress: Optional[str] = None
73+
Gateway: Optional[str] = None
74+
NetworkID: Optional[str] = None
75+
EndpointID: Optional[str] = None
76+
MacAddress: Optional[str] = None
77+
Aliases: Optional[list[str]] = None
78+
79+
80+
@dataclass
81+
class NetworkSettings:
82+
"""Network settings from docker inspect."""
83+
84+
Bridge: Optional[str] = None
85+
IPAddress: Optional[str] = None
86+
Gateway: Optional[str] = None
87+
Ports: Optional[dict[str, Any]] = None
88+
Networks: Optional[dict[str, Network]] = None
89+
90+
def get_networks(self) -> Optional[dict[str, Network]]:
91+
"""Get networks for the container."""
92+
return self.Networks
93+
94+
95+
@dataclass
96+
class ContainerInspectInfo:
97+
"""Container information from docker inspect."""
98+
99+
Id: Optional[str] = None
100+
Name: Optional[str] = None
101+
Created: Optional[str] = None
102+
Path: Optional[str] = None
103+
Args: Optional[list[str]] = None
104+
Image: Optional[str] = None
105+
State: Optional[ContainerState] = None
106+
Config: Optional[ContainerConfig] = None
107+
network_settings: Optional[NetworkSettings] = None
108+
Mounts: Optional[list[dict[str, Any]]] = None
109+
110+
@classmethod
111+
def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo":
112+
"""Create from docker inspect JSON."""
113+
return cls(
114+
Id=data.get("Id"),
115+
Name=data.get("Name"),
116+
Created=data.get("Created"),
117+
Path=data.get("Path"),
118+
Args=data.get("Args"),
119+
Image=data.get("Image"),
120+
State=_ignore_properties(ContainerState, data.get("State", {})) if data.get("State") else None,
121+
Config=_ignore_properties(ContainerConfig, data.get("Config", {})) if data.get("Config") else None,
122+
network_settings=cls._parse_network_settings(data.get("NetworkSettings", {}))
123+
if data.get("NetworkSettings")
124+
else None,
125+
Mounts=data.get("Mounts"),
126+
)
127+
128+
@classmethod
129+
def _parse_network_settings(cls, data: dict[str, Any]) -> Optional[NetworkSettings]:
130+
"""Parse NetworkSettings with Networks as Network objects."""
131+
if not data:
132+
return None
133+
134+
networks_data = data.get("Networks", {})
135+
networks = {}
136+
for name, net_data in networks_data.items():
137+
networks[name] = _ignore_properties(Network, net_data)
138+
139+
return NetworkSettings(
140+
Bridge=data.get("Bridge"),
141+
IPAddress=data.get("IPAddress"),
142+
Gateway=data.get("Gateway"),
143+
Ports=data.get("Ports"),
144+
Networks=networks,
145+
)
146+
147+
def get_network_settings(self) -> Optional[NetworkSettings]:
148+
"""Get network settings for the container."""
149+
return self.network_settings
150+
151+
36152
@dataclass
37153
class PublishedPortModel:
38154
"""
@@ -81,6 +197,7 @@ class ComposeContainer:
81197
ExitCode: Optional[int] = None
82198
Publishers: list[PublishedPortModel] = field(default_factory=list)
83199
_docker_compose: Optional["DockerCompose"] = field(default=None, init=False, repr=False)
200+
_cached_container_info: Optional[ContainerInspectInfo] = field(default=None, init=False, repr=False)
84201

85202
def __post_init__(self) -> None:
86203
if self.Publishers:
@@ -147,6 +264,31 @@ def reload(self) -> None:
147264
# each time through get_container(), but we need this method for compatibility
148265
pass
149266

267+
def get_container_info(self) -> Optional[ContainerInspectInfo]:
268+
"""Get container information via docker inspect (lazy loaded)."""
269+
if self._cached_container_info is not None:
270+
return self._cached_container_info
271+
272+
if not self._docker_compose or not self.ID:
273+
return None
274+
275+
try:
276+
inspect_command = ["docker", "inspect", self.ID]
277+
result = self._docker_compose._run_command(cmd=inspect_command)
278+
inspect_output = result.stdout.decode("utf-8").strip()
279+
280+
if inspect_output:
281+
raw_data = loads(inspect_output)[0]
282+
self._cached_container_info = ContainerInspectInfo.from_dict(raw_data)
283+
else:
284+
self._cached_container_info = None
285+
286+
except Exception as e:
287+
logger.warning(f"Failed to get container info for {self.ID}: {e}")
288+
self._cached_container_info = None
289+
290+
return self._cached_container_info
291+
150292
@property
151293
def status(self) -> str:
152294
"""Get container status for compatibility with wait strategies."""

core/tests/test_compose.py

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -378,3 +378,60 @@ def test_compose_profile_support(profiles: Optional[list[str]], running: list[st
378378
for service in not_running:
379379
with pytest.raises(ContainerIsNotRunning):
380380
compose.get_container(service)
381+
382+
383+
def test_container_info():
384+
"""Test get_container_info functionality"""
385+
basic = DockerCompose(context=FIXTURES / "basic")
386+
with basic:
387+
container = basic.get_container("alpine")
388+
389+
info = container.get_container_info()
390+
assert info is not None
391+
assert info.Id is not None
392+
assert info.Name is not None
393+
assert info.Image is not None
394+
395+
assert info.State is not None
396+
assert info.State.Status == "running"
397+
assert info.State.Running is True
398+
assert info.State.Pid is not None
399+
400+
assert info.Config is not None
401+
assert info.Config.Image is not None
402+
assert info.Config.Hostname is not None
403+
404+
network_settings = info.get_network_settings()
405+
assert network_settings is not None
406+
assert network_settings.Networks is not None
407+
408+
info2 = container.get_container_info()
409+
assert info is info2
410+
411+
412+
def test_container_info_network_details():
413+
"""Test network details in container info"""
414+
single = DockerCompose(context=FIXTURES / "port_single")
415+
with single:
416+
container = single.get_container()
417+
info = container.get_container_info()
418+
assert info is not None
419+
420+
network_settings = info.get_network_settings()
421+
assert network_settings is not None
422+
423+
if network_settings.Networks:
424+
# Test first network
425+
network_name, network = next(iter(network_settings.Networks.items()))
426+
assert network.IPAddress is not None
427+
assert network.Gateway is not None
428+
assert network.NetworkID is not None
429+
430+
431+
def test_container_info_none_when_no_docker_compose():
432+
"""Test get_container_info returns None when docker_compose reference is missing"""
433+
from testcontainers.compose.compose import ComposeContainer
434+
435+
container = ComposeContainer()
436+
info = container.get_container_info()
437+
assert info is None

0 commit comments

Comments
 (0)