add status wait strategy

Author: alexanderankin

core/testcontainers/core/container.py

@@ -247,8 +247,11 @@ def get_container_host_ip(self) -> str:
             # ensure that we covered all possible connection_modes
             assert_never(connection_mode)
 
-    @wait_container_is_ready()
     def get_exposed_port(self, port: int) -> int:
+        from testcontainers.core.wait_strategies import ContainerStatusWaitStrategy as C
+
+        C().wait_until_ready(self)
+
         if self.get_docker_client().get_connection_mode().use_mapped_port:
             c = self._container
             assert c is not None

core/testcontainers/core/docker_client.py

@@ -174,7 +174,7 @@ def get_container(self, container_id: str) -> dict[str, Any]:
         """
         Get the container with a given identifier.
         """
-        containers = self.client.api.containers(filters={"id": container_id})
+        containers = self.client.api.containers(all=True, filters={"id": container_id})
         if not containers:
             raise RuntimeError(f"Could not get container with id {container_id}")
         return cast("dict[str, Any]", containers[0])

core/testcontainers/core/wait_strategies.py

@@ -31,14 +31,18 @@
 import time
 from datetime import timedelta
 from pathlib import Path
-from typing import Any, Callable, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
 from testcontainers.core.utils import setup_logger
-
+from . import testcontainers_config
 # Import base classes from waiting_utils to make them available for tests
-from .waiting_utils import WaitStrategy, WaitStrategyTarget
+from testcontainers.core.waiting_utils import WaitStrategy, WaitStrategyTarget
+from testcontainers.compose import DockerCompose
+
+if TYPE_CHECKING:
+    from testcontainers.core.container import DockerContainer
 
 logger = setup_logger(__name__)
 
@@ -718,6 +722,60 @@ def wait_until_ready(self, container: WaitStrategyTarget) -> None:
             time.sleep(self._poll_interval)
 
 
+class ContainerStatusWaitStrategy(WaitStrategy):
+    """
+    The possible values for the container status are:
+        created
+        running
+        paused
+        restarting
+        exited
+        removing
+        dead
+    https://docs.docker.com/reference/cli/docker/container/ls/#status
+    """
+    CONTINUE_STATUSES = {"created", "restarting"}
+
+    def __init__(self):
+        super().__init__()
+
+    def wait_until_ready(self, container: WaitStrategyTarget) -> None:
+        result = self._poll(lambda: self.running(self.get_status(container)))
+        if not result:
+            raise TimeoutError("container did not become running")
+
+    @staticmethod
+    def running(status: str) -> bool:
+        if status == "running":
+            logger.debug("status is now running")
+            return True
+        if status in ContainerStatusWaitStrategy.CONTINUE_STATUSES:
+            logger.debug("status is %s, which is valid for continuing (%s)", status, ContainerStatusWaitStrategy.CONTINUE_STATUSES)
+            return False
+        raise StopIteration(f"container status not valid for continuing: {status}")
+
+    def get_status(self, container: Any) -> str:
+        from testcontainers.core.container import DockerContainer
+
+        if isinstance(container, DockerContainer):
+            return self._get_status_tc_container(container)
+        if isinstance(container, DockerCompose):
+            return self._get_status_compose_container(container)
+        raise TypeError(f"not supported operation: 'get_status' for type: {type(container)}")
+
+    @staticmethod
+    def _get_status_tc_container(container: "DockerContainer") -> str:
+        logger.debug("fetching status of container %s", container)
+        wrapped = container.get_wrapped_container()
+        wrapped.reload()
+        return wrapped.status
+
+    @staticmethod
+    def _get_status_compose_container(container: DockerCompose) -> str:
+        logger.debug("fetching status of compose container %s", container)
+        raise NotImplementedError
+
+
 class CompositeWaitStrategy(WaitStrategy):
     """
     Wait for multiple conditions to be satisfied in sequence.
@@ -816,6 +874,7 @@ def wait_until_ready(self, container: WaitStrategyTarget) -> None:
 
 __all__ = [
     "CompositeWaitStrategy",
+    "ContainerStatusWaitStrategy",
     "FileExistsWaitStrategy",
     "HealthcheckWaitStrategy",
     "HttpWaitStrategy",

core/testcontainers/core/waiting_utils.py

@@ -21,7 +21,7 @@
 
 import wrapt
 
-from testcontainers.core.config import testcontainers_config as config
+from testcontainers.core.config import testcontainers_config
 from testcontainers.core.utils import setup_logger
 
 logger = setup_logger(__name__)
@@ -73,8 +73,8 @@ class WaitStrategy(ABC):
     """Base class for all wait strategies."""
 
     def __init__(self) -> None:
-        self._startup_timeout: float = config.timeout
-        self._poll_interval: float = config.sleep_time
+        self._startup_timeout: float = testcontainers_config.timeout
+        self._poll_interval: float = testcontainers_config.sleep_time
 
     def with_startup_timeout(self, timeout: Union[int, timedelta]) -> "WaitStrategy":
         """Set the maximum time to wait for the container to be ready."""
@@ -97,6 +97,27 @@ def wait_until_ready(self, container: WaitStrategyTarget) -> None:
         """Wait until the container is ready."""
         pass
 
+    def _poll(self, check: Callable[[], bool]) -> bool:
+        start = time.time()
+        while True:
+            start_attempt = time.time()
+            duration = start_attempt - start
+            if duration > self._startup_timeout:
+                return False
+
+            # noinspection PyBroadException
+            try:
+                result = check()
+                if result:
+                    return result
+            except StopIteration:
+                return False
+            except:  # noqa: E722, RUF100
+                pass
+
+            seconds_left_until_next = self._poll_interval - (time.time() - start_attempt)
+            time.sleep(max(0.0, seconds_left_until_next))
+
 
 # Keep existing wait_container_is_ready but make it use the new system internally
 def wait_container_is_ready(*transient_exceptions: type[Exception]) -> Callable[[F], F]:
@@ -194,7 +215,7 @@ def wait_for(condition: Callable[..., bool]) -> bool:
 def wait_for_logs(
     container: WaitStrategyTarget,
     predicate: Union[Callable[[str], bool], str, WaitStrategy],
-    timeout: float = config.timeout,
+    timeout: float = testcontainers_config.timeout,
     interval: float = 1,
     predicate_streams_and: bool = False,
     raise_on_exit: bool = False,
@@ -261,7 +282,7 @@ def wait_for_logs(
     # Original implementation for backwards compatibility
     re_predicate: Optional[Callable[[str], Any]] = None
     if timeout is None:
-        timeout = config.timeout
+        timeout = testcontainers_config.timeout
     if isinstance(predicate, str):
         re_predicate = re.compile(predicate, re.MULTILINE).search
     elif callable(predicate):