fix(core): make config editable to avoid monkeypatching.1

core/testcontainers/core/config.py

@@ -1,4 +1,7 @@
+from dataclasses import dataclass, field
 from os import environ
+from os.path import exists
+from pathlib import Path
 
 MAX_TRIES = int(environ.get("TC_MAX_TRIES", 120))
 SLEEP_TIME = int(environ.get("TC_POOLING_INTERVAL", 1))
@@ -9,3 +12,61 @@
 RYUK_DISABLED: bool = environ.get("TESTCONTAINERS_RYUK_DISABLED", "false") == "true"
 RYUK_DOCKER_SOCKET: str = environ.get("TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE", "/var/run/docker.sock")
 RYUK_RECONNECTION_TIMEOUT: str = environ.get("RYUK_RECONNECTION_TIMEOUT", "10s")
+
+TC_FILE = ".testcontainers.properties"
+TC_GLOBAL = Path.home() / TC_FILE
+
+
+def read_tc_properties() -> dict[str, str]:
+    """
+    Read the .testcontainers.properties for settings. (see the Java implementation for details)
+    Currently we only support the ~/.testcontainers.properties but may extend to per-project variables later.
+
+    :return: the merged properties from the sources.
+    """
+    tc_files = [item for item in [TC_GLOBAL] if exists(item)]
+    if not tc_files:
+        return {}
+    settings = {}
+
+    for file in tc_files:
+        with open(file) as contents:
+            tuples = [line.split("=") for line in contents.readlines() if "=" in line]
+            settings = {**settings, **{item[0].strip(): item[1].strip() for item in tuples}}
+    return settings
+
+
+@dataclass
+class TestcontainersConfiguration:
+    max_tries: int = MAX_TRIES
+    sleep_time: int = SLEEP_TIME
+    ryuk_image: str = RYUK_IMAGE
+    ryuk_privileged: bool = RYUK_PRIVILEGED
+    ryuk_disabled: bool = RYUK_DISABLED
+    ryuk_docker_socket: str = RYUK_DOCKER_SOCKET
+    ryuk_reconnection_timeout: str = RYUK_RECONNECTION_TIMEOUT
+    tc_properties: dict[str, str] = field(default_factory=read_tc_properties)
+
+    def tc_properties_get_tc_host(self):
+        return self.tc_properties.get("tc.host")
+
+    @property
+    def timeout(self):
+        return self.max_tries * self.sleep_time
+
+
+testcontainers_config = TestcontainersConfiguration()
+
+__all__ = [
+    # the public API of this module
+    "testcontainers_config",
+    # and all the legacy things that are deprecated:
+    "MAX_TRIES",
+    "SLEEP_TIME",
+    "TIMEOUT",
+    "RYUK_IMAGE",
+    "RYUK_PRIVILEGED",
+    "RYUK_DISABLED",
+    "RYUK_DOCKER_SOCKET",
+    "RYUK_RECONNECTION_TIMEOUT",
+]

core/testcontainers/core/container.py

@@ -6,13 +6,7 @@
 import docker.errors
 from typing_extensions import Self
 
-from testcontainers.core.config import (
-    RYUK_DISABLED,
-    RYUK_DOCKER_SOCKET,
-    RYUK_IMAGE,
-    RYUK_PRIVILEGED,
-    RYUK_RECONNECTION_TIMEOUT,
-)
+from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.docker_client import DockerClient
 from testcontainers.core.exceptions import ContainerStartException
 from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID
@@ -77,7 +71,7 @@ def maybe_emulate_amd64(self) -> Self:
         return self
 
     def start(self) -> Self:
-        if not RYUK_DISABLED and self.image != RYUK_IMAGE:
+        if not c.ryuk_disabled and self.image != c.ryuk_image:
             logger.debug("Creating Ryuk container")
             Reaper.get_instance()
         logger.info("Pulling image %s", self.image)
@@ -201,12 +195,12 @@ def _create_instance(cls) -> "Reaper":
         logger.debug(f"Creating new Reaper for session: {SESSION_ID}")
 
         Reaper._container = (
-            DockerContainer(RYUK_IMAGE)
+            DockerContainer(c.ryuk_image)
             .with_name(f"testcontainers-ryuk-{SESSION_ID}")
             .with_exposed_ports(8080)
-            .with_volume_mapping(RYUK_DOCKER_SOCKET, "/var/run/docker.sock", "rw")
-            .with_kwargs(privileged=RYUK_PRIVILEGED, auto_remove=True)
-            .with_env("RYUK_RECONNECTION_TIMEOUT", RYUK_RECONNECTION_TIMEOUT)
+            .with_volume_mapping(c.ryuk_docker_socket, "/var/run/docker.sock", "rw")
+            .with_kwargs(privileged=c.ryuk_privileged, auto_remove=True)
+            .with_env("RYUK_RECONNECTION_TIMEOUT", c.ryuk_reconnection_timeout)
             .start()
         )
         wait_for_logs(Reaper._container, r".* Started!")

core/testcontainers/core/docker_client.py

@@ -16,20 +16,17 @@
 import os
 import urllib
 import urllib.parse
-from os.path import exists
-from pathlib import Path
 from typing import Callable, Optional, TypeVar, Union
 
 import docker
 from docker.models.containers import Container, ContainerCollection
 from typing_extensions import ParamSpec
 
+from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.labels import SESSION_ID, create_labels
 from testcontainers.core.utils import default_gateway_ip, inside_container, setup_logger
 
 LOGGER = setup_logger(__name__)
-TC_FILE = ".testcontainers.properties"
-TC_GLOBAL = Path.home() / TC_FILE
 
 _P = ParamSpec("_P")
 _T = TypeVar("_T")
@@ -185,26 +182,5 @@ def host(self) -> str:
         return "localhost"
 
 
-@ft.cache
-def read_tc_properties() -> dict[str, str]:
-    """
-    Read the .testcontainers.properties for settings. (see the Java implementation for details)
-    Currently we only support the ~/.testcontainers.properties but may extend to per-project variables later.
-
-    :return: the merged properties from the sources.
-    """
-    tc_files = [item for item in [TC_GLOBAL] if exists(item)]
-    if not tc_files:
-        return {}
-    settings = {}
-
-    for file in tc_files:
-        tuples = []
-        with open(file) as contents:
-            tuples = [line.split("=") for line in contents.readlines() if "=" in line]
-            settings = {**settings, **{item[0].strip(): item[1].strip() for item in tuples}}
-    return settings
-
-
 def get_docker_host() -> Optional[str]:
-    return read_tc_properties().get("tc.host") or os.getenv("DOCKER_HOST")
+    return c.tc_properties_get_tc_host() or os.getenv("DOCKER_HOST")

core/testcontainers/core/labels.py

@@ -1,7 +1,7 @@
 from typing import Optional
 from uuid import uuid4
 
-from testcontainers.core.config import RYUK_IMAGE
+from testcontainers.core.config import testcontainers_config as c
 
 SESSION_ID: str = str(uuid4())
 LABEL_SESSION_ID = "org.testcontainers.session-id"
@@ -13,7 +13,7 @@ def create_labels(image: str, labels: Optional[dict[str, str]]) -> dict[str, str
         labels = {}
     labels[LABEL_LANG] = "python"
 
-    if image == RYUK_IMAGE:
+    if image == c.ryuk_image:
         return labels
 
     labels[LABEL_SESSION_ID] = SESSION_ID

core/testcontainers/core/waiting_utils.py

@@ -19,7 +19,7 @@
 
 import wrapt
 
-from testcontainers.core import config
+from testcontainers.core.config import testcontainers_config as config
 from testcontainers.core.utils import setup_logger
 
 if TYPE_CHECKING:
@@ -54,18 +54,18 @@ def wrapper(wrapped: Callable, instance: Any, args: list, kwargs: dict) -> Any:
             logger.info("Waiting for %s to be ready ...", instance)
 
         exception = None
-        for attempt_no in range(config.MAX_TRIES):
+        for attempt_no in range(config.max_tries):
             try:
                 return wrapped(*args, **kwargs)
             except transient_exceptions as e:
                 logger.debug(
-                    f"Connection attempt '{attempt_no + 1}' of '{config.MAX_TRIES + 1}' "
+                    f"Connection attempt '{attempt_no + 1}' of '{config.max_tries + 1}' "
                     f"failed: {traceback.format_exc()}"
                 )
-                time.sleep(config.SLEEP_TIME)
+                time.sleep(config.sleep_time)
                 exception = e
         raise TimeoutError(
-            f"Wait time ({config.TIMEOUT}s) exceeded for {wrapped.__name__}(args: {args}, kwargs: "
+            f"Wait time ({config.timeout}s) exceeded for {wrapped.__name__}(args: {args}, kwargs: "
             f"{kwargs}). Exception: {exception}"
         )
 

core/tests/test_ryuk.py

@@ -5,15 +5,15 @@
 from docker import DockerClient
 from docker.errors import NotFound
 
-from testcontainers.core import container as container_module
+from testcontainers.core.config import testcontainers_config
 from testcontainers.core.container import Reaper
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.waiting_utils import wait_for_logs
 
 
 def test_wait_for_reaper(monkeypatch: MonkeyPatch):
     Reaper.delete_instance()
-    monkeypatch.setattr(container_module, "RYUK_RECONNECTION_TIMEOUT", "0.1s")
+    monkeypatch.setattr(testcontainers_config, "ryuk_reconnection_timeout", "0.1s")
     docker_client = DockerClient()
     container = DockerContainer("hello-world").start()
 
@@ -40,7 +40,7 @@ def test_wait_for_reaper(monkeypatch: MonkeyPatch):
 
 def test_container_without_ryuk(monkeypatch: MonkeyPatch):
     Reaper.delete_instance()
-    monkeypatch.setattr(container_module, "RYUK_DISABLED", True)
+    monkeypatch.setattr(testcontainers_config, "ryuk_disabled", True)
     with DockerContainer("hello-world") as container:
         wait_for_logs(container, "Hello from Docker!")
         assert Reaper._instance is None

modules/arangodb/testcontainers/arangodb/__init__.py

@@ -5,7 +5,7 @@
 import typing
 from os import environ
 
-from testcontainers.core.config import TIMEOUT
+from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.generic import DbContainer
 from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_for_logs
@@ -90,4 +90,4 @@ def get_connection_url(self) -> str:
         return f"http://{self.get_container_host_ip()}:{port}"
 
     def _connect(self) -> None:
-        wait_for_logs(self, predicate="is ready for business", timeout=TIMEOUT)
+        wait_for_logs(self, predicate="is ready for business", timeout=c.timeout)

modules/k3s/testcontainers/k3s/__init__.py

@@ -11,7 +11,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from testcontainers.core.config import MAX_TRIES
+from testcontainers.core.config import testcontainers_config
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.waiting_utils import wait_for_logs
 
@@ -46,7 +46,7 @@ def __init__(self, image="rancher/k3s:latest", **kwargs) -> None:
         self.with_volume_mapping("/sys/fs/cgroup", "/sys/fs/cgroup", "rw")
 
     def _connect(self) -> None:
-        wait_for_logs(self, predicate="Node controller sync successful", timeout=MAX_TRIES)
+        wait_for_logs(self, predicate="Node controller sync successful", timeout=testcontainers_config.timeout)
 
     def start(self) -> "K3SContainer":
         super().start()

modules/neo4j/testcontainers/neo4j/__init__.py

@@ -15,7 +15,7 @@
 from typing import Optional
 
 from neo4j import Driver, GraphDatabase
-from testcontainers.core.config import TIMEOUT
+from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.generic import DbContainer
 from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
@@ -62,7 +62,7 @@ def get_connection_url(self) -> str:
 
     @wait_container_is_ready()
     def _connect(self) -> None:
-        wait_for_logs(self, "Remote interface available at", TIMEOUT)
+        wait_for_logs(self, "Remote interface available at", c.timeout)
 
         # Then we actually check that the container really is listening
         with self.get_driver() as driver:

modules/postgres/testcontainers/postgres/__init__.py

@@ -14,7 +14,7 @@
 from time import sleep
 from typing import Optional
 
-from testcontainers.core.config import MAX_TRIES, SLEEP_TIME
+from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.generic import DbContainer
 from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
@@ -91,15 +91,15 @@ def get_connection_url(self, host: Optional[str] = None, driver: Optional[str] =
 
     @wait_container_is_ready()
     def _connect(self) -> None:
-        wait_for_logs(self, ".*database system is ready to accept connections.*", MAX_TRIES, SLEEP_TIME)
+        wait_for_logs(self, ".*database system is ready to accept connections.*", c.max_tries, c.sleep_time)
 
         count = 0
-        while count < MAX_TRIES:
+        while count < c.max_tries:
             status, _ = self.exec(f"pg_isready -hlocalhost -p{self.port} -U{self.username}")
             if status == 0:
                 return
 
-            sleep(SLEEP_TIME)
+            sleep(c.sleep_time)
             count += 1
 
         raise RuntimeError("Postgres could not get into a ready state")

modules/qdrant/testcontainers/qdrant/__init__.py

@@ -15,7 +15,7 @@
 from pathlib import Path
 from typing import Optional
 
-from testcontainers.core.config import TIMEOUT
+from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.generic import DbContainer
 from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
 
@@ -61,7 +61,7 @@ def _configure(self) -> None:
 
     @wait_container_is_ready()
     def _connect(self) -> None:
-        wait_for_logs(self, ".*Actix runtime found; starting in Actix runtime.*", TIMEOUT)
+        wait_for_logs(self, ".*Actix runtime found; starting in Actix runtime.*", c.timeout)
 
     def get_client(self, **kwargs):
         """