lint

Author: alexanderankin

core/testcontainers/core/container.py

@@ -285,7 +285,7 @@ def status(self) -> str:
         """Get container status for compatibility with wait strategies."""
         if not self._container:
             return "not_started"
-        return self._container.status
+        return cast("str", self._container.status)
 
     def exec(self, command: Union[str, list[str]]) -> ExecResult:
         if not self._container:

core/testcontainers/core/generic.py

@@ -62,6 +62,7 @@ def _create_connection_url(
         if self._container is None:
             raise ContainerStartException("container has not been started")
         host = host or self.get_container_host_ip()
+        assert port is not None
         port = self.get_exposed_port(port)
         quoted_password = quote(password, safe=" +")
         url = f"{dialect}://{username}:{quoted_password}@{host}:{port}"

core/testcontainers/core/wait_strategies.py

@@ -29,13 +29,16 @@
 import re
 import time
 from datetime import timedelta
-from typing import Union
+from typing import TYPE_CHECKING, Union
 
 from testcontainers.core.utils import setup_logger
 
 # Import base classes from waiting_utils to make them available for tests
 from .waiting_utils import WaitStrategy
 
+if TYPE_CHECKING:
+    from .waiting_utils import WaitStrategyTarget
+
 logger = setup_logger(__name__)
 
 
@@ -89,7 +92,7 @@ def with_poll_interval(self, interval: Union[float, timedelta]) -> "LogMessageWa
             self._poll_interval = interval
         return self
 
-    def wait_until_ready(self, container) -> None:
+    def wait_until_ready(self, container: "WaitStrategyTarget") -> None:
         """
         Wait until the specified message appears in the container logs.
 

core/testcontainers/core/waiting_utils.py

@@ -14,20 +14,16 @@
 
 import re
 import time
-import traceback
 import warnings
 from abc import ABC, abstractmethod
 from datetime import timedelta
-from typing import Any, Callable, Protocol, TypeVar, TYPE_CHECKING, Union, cast
+from typing import Any, Callable, Optional, Protocol, TypeVar, Union, cast
 
 import wrapt
 
 from testcontainers.core.config import testcontainers_config as config
 from testcontainers.core.utils import setup_logger
 
-if TYPE_CHECKING:
-    from testcontainers.core.container import DockerContainer
-
 logger = setup_logger(__name__)
 
 # Get a tuple of transient exceptions for which we'll retry. Other exceptions will be raised.
@@ -36,6 +32,7 @@
 # Type variables for generic functions
 F = TypeVar("F", bound=Callable[..., Any])
 
+
 class WaitStrategyTarget(Protocol):
     """
     Protocol defining the interface that containers must implement for wait strategies.
@@ -163,7 +160,7 @@ def wait_until_ready(self, container: WaitStrategyTarget) -> Any:
                     logger.debug(f"Connection attempt failed: {e!s}")
                     time.sleep(self._poll_interval)
 
-    @wrapt.decorator
+    @wrapt.decorator  # type: ignore[misc]
     def wrapper(wrapped: Callable[..., Any], instance: Any, args: list[Any], kwargs: dict[str, Any]) -> Any:
         # Use the LegacyWaitStrategy to handle retries with proper timeout
         strategy = LegacyWaitStrategy(wrapped, instance, args, kwargs)
@@ -177,6 +174,7 @@ def wrapper(wrapped: Callable[..., Any], instance: Any, args: list[Any], kwargs:
 
     return cast("Callable[[F], F]", wrapper)
 
+
 @wait_container_is_ready()
 def wait_for(condition: Callable[..., bool]) -> bool:
     warnings.warn(
@@ -244,8 +242,13 @@ def wait_for_logs(
         # For more complex scenarios, use structured wait strategies directly:
         container.waiting_for(LogMessageWaitStrategy("ready"))
     """
-    # Only warn for legacy usage (string or callable predicates, not WaitStrategy objects)
-    if not isinstance(predicate, WaitStrategy):
+    if isinstance(predicate, WaitStrategy):
+        start = time.time()
+        predicate.with_startup_timeout(int(timeout)).with_poll_interval(interval)
+        predicate.wait_until_ready(container)
+        return time.time() - start
+    else:
+        # Only warn for legacy usage (string or callable predicates, not WaitStrategy objects)
         warnings.warn(
             "The wait_for_logs function with string or callable predicates is deprecated and will be removed in a future version. "
             "Use structured wait strategies instead: "
@@ -254,11 +257,6 @@ def wait_for_logs(
             DeprecationWarning,
             stacklevel=2,
         )
-    if isinstance(predicate, WaitStrategy):
-        start = time.time()
-        predicate.with_startup_timeout(int(timeout)).with_poll_interval(interval)
-        predicate.wait_until_ready(container)
-        return time.time() - start
 
     # Original implementation for backwards compatibility
     re_predicate: Optional[Callable[[str], Any]] = None

core/testcontainers/socat/socat.py

@@ -85,4 +85,7 @@ def start(self) -> "SocatContainer":
     @wait_container_is_ready(OSError)
     def _connect(self) -> None:
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            s.connect((self.get_container_host_ip(), int(self.get_exposed_port(next(iter(self.ports))))))
+            next_port = next(iter(self.ports))
+            # todo remove this limitation
+            assert isinstance(next_port, int)
+            s.connect((self.get_container_host_ip(), int(self.get_exposed_port(next_port))))

core/tests/test_wait_strategies.py

@@ -1,15 +1,17 @@
+import itertools
 import re
 import time
+import typing
 from datetime import timedelta
-from unittest.mock import Mock, patch, MagicMock
+from unittest.mock import Mock, patch
+
 import pytest
-import itertools
 
-from testcontainers.core.container import DockerContainer
-from testcontainers.core.wait_strategies import (
-    LogMessageWaitStrategy,
-    WaitStrategy,
-)
+from testcontainers.core.wait_strategies import LogMessageWaitStrategy
+from testcontainers.core.waiting_utils import WaitStrategy
+
+if typing.TYPE_CHECKING:
+    from testcontainers.core.waiting_utils import WaitStrategyTarget
 
 
 class ConcreteWaitStrategy(WaitStrategy):

core/tests/test_waiting_utils.py

@@ -18,7 +18,7 @@ def test_wait_container_is_ready_decorator_basic() -> None:
     """Test the basic wait_container_is_ready decorator functionality."""
 
     @wait_container_is_ready()
-    def simple_check():
+    def simple_check() -> bool:
         return True
 
     result = simple_check()
@@ -29,7 +29,7 @@ def test_wait_container_is_ready_decorator_with_container() -> None:
     """Test wait_container_is_ready decorator with a real container."""
 
     @wait_container_is_ready()
-    def check_container_logs(container):
+    def check_container_logs(container: DockerContainer) -> bool:
         stdout, stderr = container.get_logs()
         return b"Hello from Docker!" in stdout or b"Hello from Docker!" in stderr