Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 28 additions & 15 deletions core/testcontainers/compose/compose.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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:
"""
Expand Down Expand Up @@ -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)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we need to care about invalidating?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is something I was thinking about, and I guess the answer depends, of course, network conditions, container status, and statistics can change along the way. But in a test scenario, you would usually create the container, run your tests, and then destroy it. I see that this new functionality would be used to collect the status at the end of the test, so this simple cache helps avoid extra overhead during test execution. What do you think?

Otherwise, we have a few options here. The easiest one would be to always execute the docker inspect command and return fresh information, or we could define a default TTL for the data so that after this period, the next call retrieves updated data or something like this.


def __post_init__(self) -> None:
if self.Publishers:
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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
20 changes: 19 additions & 1 deletion core/testcontainers/core/container.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading