Merge branch 'main' into copy_file_tests

Author: Tranquility2

core/testcontainers/core/config.py

@@ -11,6 +11,8 @@
 
 import docker
 
+ENABLE_FLAGS = ("yes", "true", "t", "y", "1")
+
 
 class ConnectionMode(Enum):
     bridge_ip = "bridge_ip"
@@ -45,16 +47,6 @@ def get_docker_socket() -> str:
         return "/var/run/docker.sock"
 
 
-def get_bool_env(name: str) -> bool:
-    """
-    Get environment variable named `name` and convert it to bool.
-
-    Defaults to False.
-    """
-    value = environ.get(name, "")
-    return value.lower() in ("yes", "true", "t", "y", "1")
-
-
 TC_FILE = ".testcontainers.properties"
 TC_GLOBAL = Path.home() / TC_FILE
 
@@ -96,11 +88,20 @@ def read_tc_properties() -> dict[str, str]:
 
 @dataclass
 class TestcontainersConfiguration:
+    def _render_bool(self, env_name: str, prop_name: str) -> bool:
+        env_val = environ.get(env_name, None)
+        if env_val is not None:
+            return env_val.lower() in ENABLE_FLAGS
+        prop_val = self.tc_properties.get(prop_name, None)
+        if prop_val is not None:
+            return prop_val.lower() in ENABLE_FLAGS
+        return False
+
     max_tries: int = int(environ.get("TC_MAX_TRIES", "120"))
     sleep_time: float = float(environ.get("TC_POOLING_INTERVAL", "1"))
     ryuk_image: str = environ.get("RYUK_CONTAINER_IMAGE", "testcontainers/ryuk:0.8.1")
-    ryuk_privileged: bool = get_bool_env("TESTCONTAINERS_RYUK_PRIVILEGED")
-    ryuk_disabled: bool = get_bool_env("TESTCONTAINERS_RYUK_DISABLED")
+    _ryuk_privileged: Optional[bool] = None
+    _ryuk_disabled: Optional[bool] = None
     _ryuk_docker_socket: str = ""
     ryuk_reconnection_timeout: str = environ.get("RYUK_RECONNECTION_TIMEOUT", "10s")
     tc_properties: dict[str, str] = field(default_factory=read_tc_properties)
@@ -129,6 +130,28 @@ def docker_auth_config(self, value: str) -> None:
     def tc_properties_get_tc_host(self) -> Union[str, None]:
         return self.tc_properties.get("tc.host")
 
+    @property
+    def ryuk_privileged(self) -> bool:
+        if self._ryuk_privileged is not None:
+            return bool(self._ryuk_privileged)
+        self._ryuk_privileged = self._render_bool("TESTCONTAINERS_RYUK_PRIVILEGED", "ryuk.container.privileged")
+        return self._ryuk_privileged
+
+    @ryuk_privileged.setter
+    def ryuk_privileged(self, value: bool) -> None:
+        self._ryuk_privileged = value
+
+    @property
+    def ryuk_disabled(self) -> bool:
+        if self._ryuk_disabled is not None:
+            return bool(self._ryuk_disabled)
+        self._ryuk_disabled = self._render_bool("TESTCONTAINERS_RYUK_DISABLED", "ryuk.disabled")
+        return self._ryuk_disabled
+
+    @ryuk_disabled.setter
+    def ryuk_disabled(self, value: bool) -> None:
+        self._ryuk_disabled = value
+
     @property
     def timeout(self) -> float:
         return self.max_tries * self.sleep_time

core/testcontainers/core/container.py

@@ -236,15 +236,17 @@ def get_container_host_ip(self) -> str:
         connection_mode: ConnectionMode
         connection_mode = self.get_docker_client().get_connection_mode()
 
-        # mypy:
-        container = self._container
-        assert container is not None
-
         if connection_mode == ConnectionMode.docker_host:
             return self.get_docker_client().host()
         elif connection_mode == ConnectionMode.gateway_ip:
+            # mypy:
+            container = self._container
+            assert container is not None
             return self.get_docker_client().gateway_ip(container.id)
         elif connection_mode == ConnectionMode.bridge_ip:
+            # mypy:
+            container = self._container
+            assert container is not None
             return self.get_docker_client().bridge_ip(container.id)
         else:
             # ensure that we covered all possible connection_modes