|
11 | 11 | from types import TracebackType |
12 | 12 | from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast |
13 | 13 |
|
| 14 | +from docker import DockerClient |
| 15 | +from docker.models.containers import Container |
14 | 16 | from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed |
15 | 17 | from testcontainers.core.waiting_utils import WaitStrategy |
16 | 18 |
|
@@ -137,9 +139,9 @@ def get_logs(self) -> tuple[bytes, bytes]: |
137 | 139 | stdout, stderr = self._docker_compose.get_logs(self.Service) |
138 | 140 | return stdout.encode(), stderr.encode() |
139 | 141 |
|
140 | | - def get_wrapped_container(self) -> "ComposeContainer": |
| 142 | + def get_wrapped_container(self) -> Container: |
141 | 143 | """Get the underlying container object for compatibility.""" |
142 | | - return self |
| 144 | + return self._docker_compose._get_docker_client().containers.get(self.ID) |
143 | 145 |
|
144 | 146 | def reload(self) -> None: |
145 | 147 | """Reload container information for compatibility with wait strategies.""" |
@@ -214,7 +216,9 @@ class DockerCompose: |
214 | 216 | services: Optional[list[str]] = None |
215 | 217 | docker_command_path: Optional[str] = None |
216 | 218 | profiles: Optional[list[str]] = None |
| 219 | + docker_client_kw: Optional[dict[str, Any]] = None |
217 | 220 | _wait_strategies: Optional[dict[str, Any]] = field(default=None, init=False, repr=False) |
| 221 | + _docker_client: Optional[DockerClient] = field(default=None, init=False, repr=False) |
218 | 222 |
|
219 | 223 | def __post_init__(self) -> None: |
220 | 224 | if isinstance(self.compose_file_name, str): |
@@ -259,6 +263,13 @@ def compose_command_property(self) -> list[str]: |
259 | 263 | docker_compose_cmd += ["--env-file", env_file] |
260 | 264 | return docker_compose_cmd |
261 | 265 |
|
| 266 | + def _get_docker_client(self) -> DockerClient: |
| 267 | + dc = self._docker_client |
| 268 | + if dc is None: |
| 269 | + dc = DockerClient(**(self.docker_client_kw or {})) |
| 270 | + self._docker_client = dc |
| 271 | + return dc |
| 272 | + |
262 | 273 | def waiting_for(self, strategies: dict[str, WaitStrategy]) -> "DockerCompose": |
263 | 274 | """ |
264 | 275 | Set wait strategies for specific services. |
|
0 commit comments