refactor(core): Standardize configuration and readiness steps in container lifecycle

core/testcontainers/core/container.py

@@ -75,7 +75,14 @@ def maybe_emulate_amd64(self) -> "DockerContainer":
  return self.with_kwargs(platform="linux/amd64")
  return self
 
+ def _configure(self) -> None:
+ pass
+
+ def _wait_until_ready(self) -> None:
+ pass
+
  def start(self):
+ self._configure()
  if not RYUK_DISABLED and self.image != RYUK_IMAGE:
  logger.debug("Creating Ryuk container")
  Reaper.get_instance()
@@ -92,6 +99,10 @@ def start(self):
  **self._kwargs,
  )
  logger.info("Container started: %s", self._container.short_id)
+
+ self._wait_until_ready()
+ logger.info("Container ready: %s", self._container.short_id)
+
  return self
 
  def stop(self, force=True, delete_volume=True) -> None:

core/testcontainers/core/generic.py

@@ -32,7 +32,7 @@ class DbContainer(DockerContainer):
  """
 
  @wait_container_is_ready(*ADDITIONAL_TRANSIENT_ERRORS)
- def _connect(self) -> None:
+ def _wait_until_ready(self):
  import sqlalchemy
 
  engine = sqlalchemy.create_engine(self.get_connection_url())
@@ -64,12 +64,3 @@ def _create_connection_url(
  if dbname:
  url = f"{url}/{dbname}"
  return url
-
- def start(self) -> "DbContainer":
- self._configure()
- super().start()
- self._connect()
- return self
-
- def _configure(self) -> None:
- raise NotImplementedError

core/testcontainers/core/utils.py

@@ -3,6 +3,7 @@
 import platform
 import subprocess
 import sys
+from typing import Optional, Union
 
 LINUX = "linux"
 MAC = "mac"
@@ -53,6 +54,43 @@ def inside_container() -> bool:
  return os.path.exists("/.dockerenv")
 
 
+def create_connection_string(
+ dialect: str,
+ host: str,
+ port: Union[str, int, None] = None,
+ username: Optional[str] = None,
+ password: Optional[str] = None,
+ driver: Optional[str] = None,
+ dbname: Optional[str] = None,
+) -> str:
+ """
+ Returns a connection URL following the RFC-1738 format.
+ Compatible with database clients such as SQLAlchemy and other popular database client libraries.
+
+ Example: postgres+psycopg2://myuser:mypassword@localhost:5432/mytestdb
+ """
+ dialect_driver = dialect
+ if driver:
+ dialect_driver += f"+{driver}"
+
+ username_password = username if username else ""
+ if password:
+ username_password += f":{password}"
+
+ if username_password:
+ username_password += "@"
+
+ host_port = host
+ if port:
+ host_port += f":{port}"
+
+ connection_string = f"{dialect_driver}://{username_password}{host_port}"
+ if dbname:
+ connection_string += f"/{dbname}"
+
+ return connection_string
+
+
 def default_gateway_ip() -> str:
  """
  Returns gateway IP address of the host that testcontainer process is

core/testcontainers/core/waiting_utils.py

@@ -12,10 +12,13 @@
 # under the License.
 
 
+import contextlib
 import re
 import time
 import traceback
+from http.client import HTTPException
 from typing import TYPE_CHECKING, Any, Callable, Optional, Union
+from urllib.request import urlopen
 
 import wrapt
 
@@ -77,6 +80,18 @@ def wait_for(condition: Callable[..., bool]) -> bool:
  return condition()
 
 
+def wait_for_http(url: str, timeout: Optional[float] = config.TIMEOUT, interval: float = 1) -> float:
+ start = time.time()
+ while True:
+ duration = time.time() - start
+ with contextlib.suppress(HTTPException), urlopen(url) as r:
+ if r.status < 300:
+ return duration
+ if timeout and duration > timeout:
+ raise TimeoutError(f"Container did not respond to URL check {url} in {timeout:.3f} seconds")
+ time.sleep(interval)
+
+
 def wait_for_logs(
  container: "DockerContainer", predicate: Union[Callable, str], timeout: Optional[float] = None, interval: float = 1
 ) -> float:

modules/arangodb/testcontainers/arangodb/__init__.py

@@ -2,16 +2,18 @@
 ArangoDB container support.
 """
 
-import typing
 from os import environ
+from typing import Optional
+
+from typing_extensions import override
 
 from testcontainers.core.config import TIMEOUT
-from testcontainers.core.generic import DbContainer
-from testcontainers.core.utils import raise_for_deprecated_parameter
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.utils import create_connection_string, raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_for_logs
 
 
-class ArangoDbContainer(DbContainer):
+class ArangoDbContainer(DockerContainer):
  """
  ArangoDB container.
 
@@ -26,7 +28,7 @@ class ArangoDbContainer(DbContainer):
  >>> from testcontainers.arangodb import ArangoDbContainer
  >>> from arango import ArangoClient
 
- >>> with ArangoDbContainer("arangodb:3.11.8") as arango:
+ >>> with ArangoDbContainer("arangodb:3.12.0") as arango:
  ... client = ArangoClient(hosts=arango.get_connection_url())
  ...
  ... # Connect
@@ -42,8 +44,8 @@ def __init__(
  image: str = "arangodb:latest",
  port: int = 8529,
  arango_root_password: str = "passwd",
- arango_no_auth: typing.Optional[bool] = None,
- arango_random_root_password: typing.Optional[bool] = None,
+ arango_no_auth: Optional[bool] = None,
+ arango_random_root_password: Optional[bool] = None,
  **kwargs,
  ) -> None:
  """
@@ -78,16 +80,21 @@ def __init__(
  )
  )
 
+ @override
  def _configure(self) -> None:
  self.with_env("ARANGO_ROOT_PASSWORD", self.arango_root_password)
  if self.arango_no_auth:
  self.with_env("ARANGO_NO_AUTH", "1")
  if self.arango_random_root_password:
  self.with_env("ARANGO_RANDOM_ROOT_PASSWORD", "1")
 
- def get_connection_url(self) -> str:
- port = self.get_exposed_port(self.port)
- return f"http://{self.get_container_host_ip()}:{port}"
-
- def _connect(self) -> None:
+ @override
+ def _wait_until_ready(self) -> None:
  wait_for_logs(self, predicate="is ready for business", timeout=TIMEOUT)
+
+ def get_connection_url(self) -> str:
+ return create_connection_string(
+ dialect="http",
+ host=self.get_container_host_ip(),
+ port=self.get_exposed_port(self.port),
+ )

modules/arangodb/tests/test_arangodb.py

@@ -48,22 +48,17 @@ def arango_test_ops(arango_client, expeced_version, username="root", password=""
  assert len(student_names) == students_to_insert_cnt
 
 
-def test_docker_run_arango():
- """
- Test ArangoDB container with default settings.
- """
- image = f"{ARANGODB_IMAGE_NAME}:{IMAGE_VERSION}"
- arango_root_password = "passwd"
-
- with ArangoDbContainer(image) as arango:
+@pytest.mark.parametrize("version", ["3.12.0", "3.11.8", "3.10.13"])
+def test_docker_run_arango(version: str):
+ with ArangoDbContainer(f"arangodb:{version}") as arango:
  client = ArangoClient(hosts=arango.get_connection_url())
 
  # Test invalid auth
  sys_db = client.db("_system", username="root", password="notTheRightPass")
  with pytest.raises(DatabaseCreateError):
  sys_db.create_database("test")
 
- arango_test_ops(arango_client=client, expeced_version=IMAGE_VERSION, password=arango_root_password)
+ arango_test_ops(arango_client=client, expeced_version=version, password="passwd")
 
 
 def test_docker_run_arango_without_auth():

modules/azurite/testcontainers/azurite/__init__.py

@@ -33,10 +33,9 @@ class AzuriteContainer(DockerContainer):
  >>> from testcontainers.azurite import AzuriteContainer
  >>> from azure.storage.blob import BlobServiceClient
 
- >>> with AzuriteContainer() as azurite_container:
- ... connection_string = azurite_container.get_connection_string()
+ >>> with AzuriteContainer("mcr.microsoft.com/azure-storage/azurite:3.29.0") as azurite_container:
  ... client = BlobServiceClient.from_connection_string(
- ... connection_string,
+ ... azurite_container.get_connection_string(),
  ... api_version="2019-12-12"
  ... )
  """
@@ -102,12 +101,7 @@ def get_connection_string(self) -> str:
 
  return connection_string
 
- def start(self) -> "AzuriteContainer":
- super().start()
- self._connect()
- return self
-
  @wait_container_is_ready(OSError)
- def _connect(self) -> None:
+ def _wait_until_ready(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))))))

modules/azurite/tests/test_azurite.py

@@ -4,9 +4,9 @@
 
 
 def test_docker_run_azurite():
- with AzuriteContainer() as azurite_container:
+ with AzuriteContainer("mcr.microsoft.com/azure-storage/azurite:3.29.0") as azurite_container:
  blob_service_client = BlobServiceClient.from_connection_string(
- azurite_container.get_connection_string(), api_version="2019-12-12"
+ azurite_container.get_connection_string(), api_version="2023-11-03"
  )
 
  blob_service_client.create_container("test-container")

modules/cassandra/testcontainers/cassandra/__init__.py

@@ -10,6 +10,8 @@
 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 # License for the specific language governing permissions and limitations
 # under the License.
+from typing_extensions import override
+
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.waiting_utils import wait_for_logs
 
@@ -47,14 +49,10 @@ def __init__(self, image: str = "cassandra:latest", **kwargs) -> None:
  self.with_env("CASSANDRA_ENDPOINT_SNITCH", "GossipingPropertyFileSnitch")
  self.with_env("CASSANDRA_DC", self.DEFAULT_LOCAL_DATACENTER)
 
- def _connect(self):
+ @override
+ def _wait_until_ready(self):
  wait_for_logs(self, "Startup complete")
 
- def start(self) -> "CassandraContainer":
- super().start()
- self._connect()
- return self
-
  def get_contact_points(self) -> list[tuple[str, int]]:
  return [(self.get_container_host_ip(), int(self.get_exposed_port(self.CQL_PORT)))]
 

modules/cassandra/tests/test_cassandra.py

@@ -1,14 +1,17 @@
+import pytest
+
 from cassandra.cluster import Cluster, DCAwareRoundRobinPolicy
 
 from testcontainers.cassandra import CassandraContainer
 
 
-def test_docker_run_cassandra():
- with CassandraContainer("cassandra:4.1.4") as cassandra:
+@pytest.mark.parametrize("version", ["4.1.4", "3.11.16"])
+def test_docker_run_cassandra(version: str):
+ with CassandraContainer(f"cassandra:{version}") as cassandra:
  cluster = Cluster(
  cassandra.get_contact_points(),
  load_balancing_policy=DCAwareRoundRobinPolicy(cassandra.get_local_datacenter()),
  )
  session = cluster.connect()
  result = session.execute("SELECT release_version FROM system.local;")
- assert result.one().release_version == "4.1.4"
+ assert result.one().release_version == version

modules/chroma/testcontainers/chroma/__init__.py

@@ -1,13 +1,8 @@
-from typing import TYPE_CHECKING
-
-from requests import ConnectionError, get
+from typing_extensions import override
 
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.utils import raise_for_deprecated_parameter
-from testcontainers.core.waiting_utils import wait_container_is_ready
-
-if TYPE_CHECKING:
- from requests import Response
+from testcontainers.core.waiting_utils import wait_for_http
 
 
 class ChromaContainer(DockerContainer):
@@ -22,7 +17,7 @@ class ChromaContainer(DockerContainer):
  >>> import chromadb
  >>> from testcontainers.chroma import ChromaContainer
 
- >>> with ChromaContainer() as chroma:
+ >>> with ChromaContainer("chromadb/chroma:0.4.24") as chroma:
  ... config = chroma.get_config()
  ... client = chromadb.HttpClient(host=config["host"], port=config["port"])
  ... col = client.get_or_create_collection("test")
@@ -48,7 +43,6 @@ def __init__(
  self.port = port
 
  self.with_exposed_ports(self.port)
- # self.with_command(f"server /data --address :{self.port}")
 
  def get_config(self) -> dict:
  """This method returns the configuration of the Chroma container,
@@ -65,17 +59,6 @@ def get_config(self) -> dict:
  "port": exposed_port,
  }
 
- @wait_container_is_ready(ConnectionError)
- def _healthcheck(self) -> None:
- """This is an internal method used to check if the Chroma container
- is healthy and ready to receive requests."""
- url = f"http://{self.get_config()['endpoint']}/api/v1/heartbeat"
- response: Response = get(url)
- response.raise_for_status()
-
- def start(self) -> "ChromaContainer":
- """This method starts the Chroma container and runs the healthcheck
- to verify that the container is ready to use."""
- super().start()
- self._healthcheck()
- return self
+ @override
+ def _wait_until_ready(self) -> None:
+ wait_for_http(f"http://{self.get_config()['endpoint']}/api/v1/heartbeat")

modules/chroma/tests/test_chroma.py

@@ -5,5 +5,5 @@
 def test_docker_run_chroma():
  with ChromaContainer(image="chromadb/chroma:0.4.24") as chroma:
  client = chromadb.HttpClient(host=chroma.get_config()["host"], port=chroma.get_config()["port"])
-col = client.get_or_create_collection("test")
- assert col.name == "test"
+collection = client.get_or_create_collection("test")
+ assert collection.name == "test"