Added a new Docker registry test container and tests

.github/workflows/main.yml

@@ -44,6 +44,7 @@ jobs:
         - rabbitmq
         - redis
         - selenium
+        - registry
     runs-on: ${{ matrix.runtime.machine }}
     steps:
       - uses: actions/checkout@v3

README.rst

@@ -34,6 +34,7 @@ testcontainers-python facilitates the use of Docker containers for functional an
     rabbitmq/README
     redis/README
     selenium/README
+    registry/README
 
 Getting Started
 ---------------

registry/README.rst

@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.registry.DockerRegistryContainer
+

registry/setup.py

@@ -0,0 +1,18 @@
+from setuptools import setup, find_namespace_packages
+
+description = "Redis component of testcontainers-python."
+
+setup(
+    name="testcontainers-registry",
+    version="0.0.1rc1",
+    packages=find_namespace_packages(),
+    description=description,
+    long_description=description,
+    long_description_content_type="text/x-rst",
+    url="https://github.com/testcontainers/testcontainers-python",
+    install_requires=[
+        "testcontainers-core",
+        "bcrypt",
+    ],
+    python_requires=">=3.7",
+)

registry/testcontainers/__init__.py

@@ -0,0 +1,60 @@
+import time
+from io import BytesIO
+from tarfile import TarFile, TarInfo
+from typing import Optional
+
+import bcrypt
+from testcontainers.core.container import DockerContainer
+
+class DockerRegistryContainer(DockerContainer):
+    # https://docs.docker.com/registry/
+    credentials_path: str = "/htpasswd/credentials.txt"
+
+    def __init__(
+        self,
+        image: str = "registry:latest",
+        port: int = 5000,
+        username: str = None,
+        password: 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):
+        # 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 = f"{self.username}:{hashed_password}".encode("utf-8")
+
+        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)
+
+    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()
+            return self
+        else:
+            super().start()
+            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}"

registry/tests/test_registry.py

@@ -0,0 +1,26 @@
+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

requirements.in

@@ -21,6 +21,7 @@
 -e file:rabbitmq
 -e file:redis
 -e file:selenium
+-e file:registry
 cryptography<37
 flake8<3.8.0  # 3.8.0 adds a dependency on importlib-metadata which conflicts with other packages.
 pg8000