feat(new): add a new Docker Registry test container

index.rst

@@ -40,6 +40,7 @@ testcontainers-python facilitates the use of Docker containers for functional an
     modules/qdrant/README
     modules/rabbitmq/README
     modules/redis/README
+    modules/registry/README
     modules/selenium/README
     modules/weaviate/README
 

modules/registry/README.rst

@@ -0,0 +1,8 @@
+.. autoclass:: testcontainers.registry.DockerRegistryContainer
+
+When building Docker containers with Docker Buildx there is currently no option to test your containers locally without
+a local registry. Otherwise Buildx pushes your image to Docker Hub, which is not what you want in a test case. More
+and more you need to use Buildx for efficiently building images and especially multi arch images.
+
+When you use Docker Python libraries like docker-py or python-on-whales to build and test Docker images, what a lot of
+persons and DevOps engineers like me nowadays do, a test container comes in very handy.

modules/registry/testcontainers/registry/__init__.py

@@ -0,0 +1,77 @@
+import time
+from io import BytesIO
+from tarfile import TarFile, TarInfo
+from typing import TYPE_CHECKING, Optional
+
+import bcrypt
+from requests import get
+from requests.auth import HTTPBasicAuth
+from requests.exceptions import ConnectionError, ReadTimeout
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.waiting_utils import wait_container_is_ready
+
+if TYPE_CHECKING:
+    from requests import Response
+
+
+class DockerRegistryContainer(DockerContainer):
+    # https://docs.docker.com/registry/
+    credentials_path: str = "/htpasswd/credentials.txt"
+
+    def __init__(
+        self,
+        image: str = "registry:2",
+        port: int = 5000,
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(image=image, **kwargs)
+        self.port: int = port
+        self.username: Optional[str] = username
+        self.password: Optional[str] = password
+        self.with_exposed_ports(self.port)
+
+    def _copy_credentials(self) -> None:
+        # Create credentials and write them to the container
+        hashed_password: str = bcrypt.hashpw(
+            self.password.encode("utf-8"),
+            bcrypt.gensalt(rounds=12, prefix=b"2a"),
+        ).decode("utf-8")
+        content: bytes = f"{self.username}:{hashed_password}".encode("utf-8")  # noqa: UP012
+
+        with BytesIO() as tar_archive_object, TarFile(fileobj=tar_archive_object, mode="w") as tmp_tarfile:
+            tarinfo: TarInfo = TarInfo(name=self.credentials_path)
+            tarinfo.size = len(content)
+            tarinfo.mtime = time.time()
+
+            tmp_tarfile.addfile(tarinfo, BytesIO(content))
+            tar_archive_object.seek(0)
+            self.get_wrapped_container().put_archive("/", tar_archive_object)
+
+    @wait_container_is_ready(ConnectionError, ReadTimeout)
+    def _readiness_probe(self) -> None:
+        url: str = f"http://{self.get_registry()}/v2"
+        if self.username and self.password:
+            response: Response = get(url, auth=HTTPBasicAuth(self.username, self.password), timeout=1)
+        else:
+            response: Response = get(url, timeout=1)
+        response.raise_for_status()
+
+    def start(self):
+        if self.username and self.password:
+            self.with_env("REGISTRY_AUTH_HTPASSWD_REALM", "local-registry")
+            self.with_env("REGISTRY_AUTH_HTPASSWD_PATH", self.credentials_path)
+            super().start()
+            self._copy_credentials()
+        else:
+            super().start()
+
+        self._readiness_probe()
+        return self
+
+    def get_registry(self) -> str:
+        host: str = self.get_container_host_ip()
+        port: str = self.get_exposed_port(self.port)
+        return f"{host}:{port}"

modules/registry/tests/test_registry.py

@@ -0,0 +1,27 @@
+from requests import Response, get
+from requests.auth import HTTPBasicAuth
+from testcontainers.registry import DockerRegistryContainer
+
+
+REGISTRY_USERNAME: str = "foo"
+REGISTRY_PASSWORD: str = "bar"
+
+
+def test_registry():
+    with DockerRegistryContainer().with_bind_ports(5000, 5000) as registry_container:
+        url: str = f"http://{registry_container.get_registry()}/v2/_catalog"
+
+        response: Response = get(url)
+
+        assert response.status_code == 200
+
+
+def test_registry_with_authentication():
+    with DockerRegistryContainer(username=REGISTRY_USERNAME, password=REGISTRY_PASSWORD).with_bind_ports(
+        5000, 5000
+    ) as registry_container:
+        url: str = f"http://{registry_container.get_registry()}/v2/_catalog"
+
+        response: Response = get(url, auth=HTTPBasicAuth(REGISTRY_USERNAME, REGISTRY_PASSWORD))
+
+        assert response.status_code == 200

pyproject.toml

@@ -53,6 +53,7 @@ packages = [
     { include = "testcontainers", from = "modules/qdrant" },
     { include = "testcontainers", from = "modules/rabbitmq" },
     { include = "testcontainers", from = "modules/redis" },
+    { include = "testcontainers", from = "modules/registry" },
     { include = "testcontainers", from = "modules/selenium" },
     { include = "testcontainers", from = "modules/weaviate" }
 ]
@@ -94,6 +95,7 @@ selenium = { version = "*", optional = true }
 weaviate-client = { version = "^4.5.4", optional = true }
 chromadb-client = { version = "*", optional = true }
 qdrant-client = { version = "*", optional = true }
+bcrypt = { version = "*", optional = true }
 
 [tool.poetry.extras]
 arangodb = ["python-arango"]
@@ -120,6 +122,7 @@ postgres = []
 qdrant = ["qdrant-client"]
 rabbitmq = ["pika"]
 redis = ["redis"]
+registry = ["bcrypt"]
 selenium = ["selenium"]
 weaviate = ["weaviate-client"]
 chroma = ["chromadb-client"]