more fixes

Author: Tranquility2

core/testcontainers/compose/compose.py

@@ -293,7 +293,7 @@ def get_config(
             config_cmd.append("--no-interpolate")
 
         cmd_output = self._run_command(cmd=config_cmd).stdout
-        return cast(dict[str, Any], loads(cmd_output))
+        return cast(dict[str, Any], loads(cmd_output))  # noqa: TC006
 
     def get_containers(self, include_all=False) -> list[ComposeContainer]:
         """

core/testcontainers/core/config.py

@@ -17,13 +17,11 @@ class ConnectionMode(Enum):
     @property
     def use_mapped_port(self) -> bool:
         """
-        Return true if we need to use mapped port for this connection
+        Return True if mapped ports should be used for this connection mode.
 
-        This is true for everything but bridge mode.
+        Mapped ports are used for all connection modes except 'bridge_ip'.
         """
-        if self == self.bridge_ip:  # type: ignore[comparison-overlap]
-            return False
-        return True
+        return self != ConnectionMode.bridge_ip
 
 
 def get_docker_socket() -> str:
@@ -137,15 +135,15 @@ def timeout(self) -> int:
 testcontainers_config = TestcontainersConfiguration()
 
 __all__ = [
-    # the public API of this module
-    "testcontainers_config",
-    # and all the legacy things that are deprecated:
+    # Legacy things that are deprecated:
     "MAX_TRIES",
-    "SLEEP_TIME",
-    "TIMEOUT",
-    "RYUK_IMAGE",
-    "RYUK_PRIVILEGED",
     "RYUK_DISABLED",
     "RYUK_DOCKER_SOCKET",
+    "RYUK_IMAGE",
+    "RYUK_PRIVILEGED",
     "RYUK_RECONNECTION_TIMEOUT",
+    "SLEEP_TIME",
+    "TIMEOUT",
+    # Public API of this module:
+    "testcontainers_config",
 ]

core/testcontainers/core/docker_client.py

@@ -162,7 +162,7 @@ def port(self, container_id: str, port: int) -> str:
         """
         port_mappings = self.client.api.port(container_id, port)
         if not port_mappings:
-            raise ConnectionError(f"Port mapping for container {container_id} and port {port} is " "not available")
+            raise ConnectionError(f"Port mapping for container {container_id} and port {port} is not available")
         return cast(str, port_mappings[0]["HostPort"])
 
     def get_container(self, container_id: str) -> dict[str, Any]:
@@ -233,7 +233,10 @@ def host(self) -> str:
             url = urllib.parse.urlparse(self.client.api.base_url)
         except ValueError:
             return "localhost"
-        if "http" in url.scheme or "tcp" in url.scheme and url.hostname:
+
+        is_http_scheme = "http" in url.scheme
+        is_tcp_scheme_with_hostname = "tcp" in url.scheme and url.hostname
+        if is_http_scheme or is_tcp_scheme_with_hostname:
             # see https://github.com/testcontainers/testcontainers-python/issues/415
             hostname = url.hostname
             if not hostname or (hostname == "localnpipe" and utils.is_windows()):

core/testcontainers/socat/__init__.py

@@ -1,2 +1,2 @@
-# flake8: noqa
+# flake8: noqa: F401
 from testcontainers.socat.socat import SocatContainer

modules/mqtt/testcontainers/mqtt/__init__.py

@@ -50,7 +50,7 @@ def __init__(
         super().__init__(image, **kwargs)
         # self.password = password
         # reusable client context:
-        self.client: Optional["Client"] = None
+        self.client: Optional["Client"] = None  # noqa: UP037
 
     @wait_container_is_ready()
     def get_client(self) -> "Client":

modules/opensearch/testcontainers/opensearch/__init__.py

@@ -8,6 +8,8 @@
 from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
+MIN_REQUIRED_INITIAL_ADMIN_PASSWORD = [2, 12, 0]
+
 
 class OpenSearchContainer(DockerContainer):
     """
@@ -65,7 +67,7 @@ def __init__(
 
     def _supports_initial_admin_password(self, image: str) -> bool:
         with suppress(Exception):
-            return [int(n) for n in image.split(":")[-1].split(".")] >= [int(n) for n in "2.12.0".split(".")]
+            return [int(n) for n in image.split(":")[-1].split(".")] >= MIN_REQUIRED_INITIAL_ADMIN_PASSWORD
         return False
 
     def get_config(self) -> dict: