test(core): replace registry import with a private _LocalRegistryContainer copy

Author: Tranquility2

core/tests/_local_registry_container.py

@@ -0,0 +1,99 @@
+"""Slim, test-private copy of ``DockerRegistryContainer``.
+
+``core/tests`` should not import from sibling modules (``testcontainers.registry`` in this case).
+This keeps ``core`` self-contained as the package layout evolves into a separate ``core`` / ``community`` split.
+
+We keep the integration coverage by inlining only the bits needed to spin up an authenticated ``registry:2``.
+That means: htpasswd creds setup, an HTTP readiness probe, and an accessor for the published address.
+
+This is intentional duplication of the production class.
+Behavioural changes here are not expected to track ``testcontainers.registry`` and vice versa.
+"""
+
+from __future__ import annotations
+
+import time
+from io import BytesIO
+from tarfile import TarFile, TarInfo
+from typing import TYPE_CHECKING, Any, Optional
+
+import bcrypt
+from requests import get
+from requests.auth import HTTPBasicAuth
+from requests.exceptions import ConnectionError as RequestsConnectionError
+from requests.exceptions import 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 _LocalRegistryContainer(DockerContainer):
+    """Private ``registry:2`` fixture used by ``core/tests`` only.
+
+    Mirrors the public ``DockerRegistryContainer`` API surface used by tests.
+    That is: ``__init__``, ``start``, and ``get_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: Any,
+    ) -> 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:
+        if self.password is None:
+            raise ValueError("Password cannot be None")
+        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()
+
+        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 = int(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(RequestsConnectionError, ReadTimeout)
+    def _readiness_probe(self) -> None:
+        url: str = f"http://{self.get_registry()}/v2"
+        if self.username and self.password:
+            auth_response: Response = get(url, auth=HTTPBasicAuth(self.username, self.password), timeout=1)
+            auth_response.raise_for_status()
+        else:
+            response: Response = get(url, timeout=1)
+            response.raise_for_status()
+
+    def start(self) -> "_LocalRegistryContainer":
+        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: int = self.get_exposed_port(self.port)
+        return f"{host}:{port}"

core/tests/test_core_registry.py

@@ -1,29 +1,25 @@
-"""Integration test using login to a private registry.
+"""
+Integration tests for private-registry support in ``DockerClient``.
 
-Note: Using the testcontainers-python library to test the Docker registry.
-This could be considered a bad practice as it is not recommended to use the same library to test itself.
-However, it is a very good use case for DockerRegistryContainer and allows us to test it thoroughly.
+These tests spin up a real ``registry:2`` container and exercise the wiring that turns ``DOCKER_AUTH_CONFIG`` into a successful login + image pull.
+The registry is provided by the local ``_LocalRegistryContainer`` helper, kept as a private copy in ``core/tests`` so ``core`` does not import from a sibling module.
 
-Note2: These tests are skipped on macOS and SSH-based Docker hosts because they rely on insecure HTTP registries,
-which are not supported in those environments without additional configuration.
+Skipped where insecure HTTP registries cannot be reached without daemon reconfiguration.
+That includes macOS, Podman, and SSH-based Docker hosts.
 """
 
-import json
-import os
 import base64
-import pytest
+import json
 
+import pytest
+from _local_registry_container import _LocalRegistryContainer  # type: ignore[import-not-found]
 from docker.errors import NotFound
 
 from testcontainers.core.config import testcontainers_config
 from testcontainers.core.container import DockerContainer
-from testcontainers.core.docker_client import DockerClient, is_ssh_docker_host
-from testcontainers.core.waiting_utils import wait_for_logs
-
-from testcontainers.registry import DockerRegistryContainer
+from testcontainers.core.docker_client import DockerClient, is_podman, is_ssh_docker_host
 from testcontainers.core.utils import is_mac
-from testcontainers.core.docker_client import is_podman
-
+from testcontainers.core.waiting_utils import wait_for_logs
 
 _skip_insecure_registry = pytest.mark.skipif(
     is_mac() or is_podman() or is_ssh_docker_host(),
@@ -38,7 +34,7 @@ def test_missing_on_private_registry(monkeypatch):
     image = "hello-world"
     tag = "test"
 
-    with DockerRegistryContainer(username=username, password=password) as registry:
+    with _LocalRegistryContainer(username=username, password=password) as registry:
         registry_url = registry.get_registry()
 
         # prepare auth config
@@ -65,7 +61,7 @@ def test_missing_on_private_registry(monkeypatch):
 def test_with_private_registry(image, tag, username, password, expected_output, monkeypatch):
     client = DockerClient().client
 
-    with DockerRegistryContainer(username=username, password=password) as registry:
+    with _LocalRegistryContainer(username=username, password=password) as registry:
         registry_url = registry.get_registry()
 
         # prepare image