wip

modules/cosmosdb/README.rst

@@ -1,3 +1,3 @@
-.. autoclass:: testcontainers.cosmosdb.CosmosDBEmulatorContainer
-.. autoclass:: testcontainers.cosmosdb.Endpoints
+.. autoclass:: testcontainers.cosmosdb.MongoDBEmulatorContainer
+.. autoclass:: testcontainers.cosmosdb.NoSQLEmulatorContainer
 .. title:: testcontainers.cosmosdb.CosmosDBEmulatorContainer

modules/cosmosdb/testcontainers/cosmosdb/__init__.py

@@ -1,183 +1,4 @@
-import os
-import socket
-import ssl
-from collections.abc import Iterable
-from enum import Enum, auto
-from typing import Callable, Optional
-from urllib.error import HTTPError, URLError
-from urllib.request import urlopen
+from .mongodb import MongoDBEmulatorContainer
+from .nosql import NoSQLEmulatorContainer
 
-from azure.core.exceptions import ServiceRequestError
-from azure.cosmos import CosmosClient as SyncCosmosClient
-from azure.cosmos.aio import CosmosClient as AsyncCosmosClient
-from typing_extensions import Self
-
-from testcontainers.core.container import DockerContainer
-from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
-
-__all__ = ["CosmosDBEmulatorContainer", "Endpoints"]
-
-
-class Endpoints(Enum):
-    MongoDB = auto()
-
-
-# Ports mostly derived from https://docs.microsoft.com/en-us/azure/cosmos-db/emulator-command-line-parameters
-EMULATOR_PORT = 8081
-endpoint_ports = {
-    Endpoints.MongoDB: frozenset([10255]),
-}
-
-
-def is_truthy_string(s: str):
-    return s.lower().strip() in {"true", "yes", "y", "1"}
-
-
-class CosmosDBEmulatorContainer(DockerContainer):
-    """
-    CosmosDB Emulator container.
-
-    Example:
-        .. doctest::
-                >>> from testcontainers.cosmosdb import CosmosDBEmulatorContainer
-                >>> with CosmosDBEmulatorContainer() as cosmosdb:
-                ...    db = cosmosdb.insecure_sync_client().create_database_if_not_exists("test")
-
-        .. doctest::
-                >>> from testcontainers.cosmosdb import CosmosDBEmulatorContainer
-                >>> with CosmosDBEmulatorContainer() as emulator:
-                ...    cosmosdb = CosmosClient(url=emulator.url, credential=emulator.key, connection_verify=False)
-                ...    db = cosmosdb.create_database_if_not_exists("test")
-
-        .. doctest::
-                >>> from testcontainers.cosmosdb import CosmosDBEmulatorContainer, Endpoints
-                >>> with CosmosDBEmulatorContainer(endpoints=[Endpoints.MongoDB], mongodb_version="4.0") as emulator:
-                ...    print(f"Point yout MongoDB client to {emulator.host}:{next(iter(emulator.ports(Endpoints.MongoDB)))}")
-    """
-
-    def __init__(
-        self,
-        image: str = os.getenv(
-            "AZURE_COSMOS_EMULATOR_IMAGE", "mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:latest"
-        ),
-        partition_count: int = os.getenv("AZURE_COSMOS_EMULATOR_PARTITION_COUNT", None),
-        enable_data_persistence: bool = is_truthy_string(
-            os.getenv("AZURE_COSMOS_EMULATOR_ENABLE_DATA_PERSISTENCE", "false")
-        ),
-        bind_ports: bool = is_truthy_string(os.getenv("AZURE_COSMOS_EMULATOR_BIND_PORTS", "true")),
-        key: str = os.getenv(
-            "AZURE_COSMOS_EMULATOR_KEY",
-            "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==",
-        ),
-        endpoints: Iterable[Endpoints] = [],  # the emulator image does not support host-container port mapping
-        mongodb_version: Optional[str] = None,
-        **docker_client_kw,
-    ):
-        super().__init__(image=image, **docker_client_kw)
-        self.partition_count = partition_count
-        self.key = key
-        self.enable_data_persistence = enable_data_persistence
-        self.endpoints = frozenset(endpoints)
-        self.bind_ports = bind_ports
-        assert (Endpoints.MongoDB not in self.endpoints) or (
-            mongodb_version is not None
-        ), "A MongoDB version is required to use the MongoDB Endpoint"
-        self.mongodb_version = mongodb_version
-
-    @property
-    def url(self) -> str:
-        """
-        The url to the CosmosDB server
-        """
-        return f"https://{self.host}:{self.get_exposed_port(EMULATOR_PORT)}"
-
-    @property
-    def host(self) -> str:
-        return self.get_container_host_ip()
-
-    @property
-    def certificate_pem(self) -> bytes:
-        """
-        PEM-encoded certificate of the CosmosDB server
-        """
-        return self._cert_pem_bytes
-
-    def ports(self, endpoint: Endpoints) -> Iterable[int]:
-        """
-        Returns the set of exposed ports for a given endpoint.
-        If bind_ports is True, the returned ports will be the NAT-ed ports reachable from the host.
-        """
-        assert endpoint in self.endpoints, f"Endpoint {endpoint} is not exposed"
-        return {self.get_exposed_port(p) for p in endpoint_ports[endpoint]}
-
-    def insecure_async_client(self) -> AsyncCosmosClient:
-        """
-        Returns an asynchronous CosmosClient instance
-        """
-        return AsyncCosmosClient(url=self.url, credential=self.key, connection_verify=False)
-
-    def insecure_sync_client(self) -> SyncCosmosClient:
-        """
-        Returns a synchronous CosmosClient instance
-        """
-        return SyncCosmosClient(url=self.url, credential=self.key, connection_verify=False)
-
-    def start(self) -> Self:
-        self._configure()
-        super().start()
-        self._wait_until_ready()
-        self._cert_pem_bytes = self._download_cert()
-        return self
-
-    def _configure(self) -> None:
-        self.with_bind_ports(EMULATOR_PORT, EMULATOR_PORT)
-
-        endpoints_ports = []
-        for endpoint in self.endpoints:
-            endpoints_ports.extend(endpoint_ports[endpoint])
-
-        if self.bind_ports:
-            [self.with_bind_ports(port, port) for port in endpoints_ports]
-        else:
-            self.with_exposed_ports(*endpoints_ports)
-
-        (
-            self.with_env("AZURE_COSMOS_EMULATOR_PARTITION_COUNT", str(self.partition_count))
-            .with_env("AZURE_COSMOS_EMULATOR_IP_ADDRESS_OVERRIDE", socket.gethostbyname(socket.gethostname()))
-            .with_env("AZURE_COSMOS_EMULATOR_ENABLE_DATA_PERSISTENCE", str(self.enable_data_persistence))
-            .with_env("AZURE_COSMOS_EMULATOR_KEY", str(self.key))
-        )
-
-        if Endpoints.MongoDB in self.endpoints:
-            self.with_env("AZURE_COSMOS_EMULATOR_ENABLE_MONGODB_ENDPOINT", self.mongodb_version)
-
-    def _wait_until_ready(self) -> Self:
-        """
-        Waits until the CosmosDB Emulator image is ready to be used.
-        """
-        (
-            self._wait_for_logs(container=self, predicate="Started\\s*$")
-            ._wait_for_url(f"{self.url}/_explorer/index.html")
-            ._wait_for_query_success(lambda sync_client: list(sync_client.list_databases()))
-        )
-        return self
-
-    @wait_container_is_ready(HTTPError, URLError)
-    def _wait_for_url(self, url: str) -> Self:
-        with urlopen(url, context=ssl._create_unverified_context()) as response:
-            response.read()
-        return self
-
-    def _wait_for_logs(self, *args, **kwargs) -> Self:
-        wait_for_logs(*args, **kwargs)
-        return self
-
-    @wait_container_is_ready(ServiceRequestError)
-    def _wait_for_query_success(self, query: Callable[[SyncCosmosClient], None]) -> Self:
-        with self.insecure_sync_client() as c:
-            query(c)
-        return self
-
-    def _download_cert(self) -> bytes:
-        with urlopen(f"{self.url}/_explorer/emulator.pem", context=ssl._create_unverified_context()) as response:
-            return response.read()
+__all__ = ["MongoDBEmulatorContainer", "NoSQLEmulatorContainer"]

modules/cosmosdb/testcontainers/cosmosdb/_emulator.py

@@ -0,0 +1,102 @@
+import os
+import socket
+import ssl
+from collections.abc import Iterable
+from typing_extensions import Self
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
+from . import _grab as grab
+from distutils.util import strtobool
+from urllib.error import HTTPError, URLError
+from urllib.request import urlopen
+
+__all__ = ["CosmosDBEmulatorContainer"]
+
+EMULATOR_PORT = 8081
+
+class CosmosDBEmulatorContainer(DockerContainer):
+    """
+    CosmosDB Emulator container.
+    """
+
+    def __init__(
+        self,
+        image: str = os.getenv(
+            "AZURE_COSMOS_EMULATOR_IMAGE", "mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:latest"
+        ),
+        partition_count: int = os.getenv("AZURE_COSMOS_EMULATOR_PARTITION_COUNT", None),
+        enable_data_persistence: bool = strtobool(
+            os.getenv("AZURE_COSMOS_EMULATOR_ENABLE_DATA_PERSISTENCE", "false")
+        ),
+        key: str = os.getenv(
+            "AZURE_COSMOS_EMULATOR_KEY",
+            "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==",
+        ),
+        bind_ports: bool = strtobool(os.getenv("AZURE_COSMOS_EMULATOR_BIND_PORTS", "true")),
+        endpoint_ports: Iterable[int] = [],
+        **other_kwargs,
+    ):
+        super().__init__(image=image, **other_kwargs)
+        self.endpoint_ports = endpoint_ports
+        self.partition_count = partition_count
+        self.key = key
+        self.enable_data_persistence = enable_data_persistence
+        self.bind_ports = bind_ports
+
+    @property
+    def host(self) -> str:
+        return self.get_container_host_ip()
+
+    @property
+    def server_certificate_pem(self) -> bytes:
+        """
+        PEM-encoded server certificate
+        """
+        return self._cert_pem_bytes
+
+    def start(self) -> Self:
+        self._configure()
+        super().start()
+        self._wait_until_ready()
+        self._cert_pem_bytes = self._download_cert()
+        return self
+
+    def _configure(self) -> None:
+        all_ports = set([EMULATOR_PORT] + self.endpoint_ports)
+        if self.bind_ports:
+            for port in all_ports:
+                self.with_bind_ports(port, port)
+        else:
+            self.with_exposed_ports(*all_ports)
+
+        (
+            self
+						.with_env("AZURE_COSMOS_EMULATOR_PARTITION_COUNT", str(self.partition_count))
+            .with_env("AZURE_COSMOS_EMULATOR_IP_ADDRESS_OVERRIDE", socket.gethostbyname(socket.gethostname()))
+            .with_env("AZURE_COSMOS_EMULATOR_ENABLE_DATA_PERSISTENCE", str(self.enable_data_persistence))
+            .with_env("AZURE_COSMOS_EMULATOR_KEY", str(self.key))
+        )
+
+    def _wait_until_ready(self) -> Self:
+        wait_for_logs(container=self, predicate="Started\\s*$")
+
+        if self.bind_ports:
+            self._wait_for_url(f"https://{self.host}:{EMULATOR_PORT}/_explorer/index.html")
+            self._wait_for_query_success()
+
+        return self
+
+    def _download_cert(self) -> bytes:
+        with grab.file(
+            self._container, "/tmp/cosmos/appdata/.system/profiles/Client/AppData/Local/CosmosDBEmulator/emulator.pem"
+        ) as cert:
+            return cert.read()
+
+    @wait_container_is_ready(HTTPError, URLError)
+    def _wait_for_url(self, url: str) -> Self:
+        with urlopen(url, context=ssl._create_unverified_context()) as response:
+            response.read()
+        return self
+
+    def _wait_for_query_success(self) -> None:
+        pass

modules/cosmosdb/testcontainers/cosmosdb/_grab.py

@@ -0,0 +1,25 @@
+
+from pathlib import Path
+from os import path
+import tarfile
+import tempfile
+from contextlib import contextmanager
+
+@contextmanager
+def file(container, target):
+	target_path = Path(target)
+	assert target_path.is_absolute(), "target must be an absolute path"
+
+	with tempfile.TemporaryDirectory() as tmpdirname:
+		archive = Path(tmpdirname) / 'grabbed.tar'
+
+		# download from container as tar archive
+		with open(archive, 'wb') as f:
+			tar_bits, _ = container.get_archive(target)
+			for chunk in tar_bits:
+				f.write(chunk)
+
+		# extract target file from tar archive
+		with tarfile.TarFile(archive) as tar:
+			yield tar.extractfile(path.basename(target))
+

modules/cosmosdb/testcontainers/cosmosdb/mongodb.py

@@ -0,0 +1,38 @@
+import os
+from ._emulator import CosmosDBEmulatorContainer
+
+__all__ = ["MongoDBEmulatorContainer"]
+
+ENDPOINT_PORT = 10255
+
+class MongoDBEmulatorContainer(CosmosDBEmulatorContainer):
+    """
+    CosmosDB MongoDB enpoint Emulator.
+
+    Example:
+
+        .. doctest::
+                >>> from testcontainers.cosmosdb import MongoDBEmulatorContainer
+                >>> with CosmosDBEmulatorContainer(mongodb_version="4.0") as emulator:
+                ...    print(f"Point yout MongoDB client to {emulator.host}:{emulator.port}}")
+    """
+
+    def __init__(
+        self,
+        mongodb_version: str = None,
+        image: str = os.getenv(
+            "AZURE_COSMOS_EMULATOR_IMAGE", "mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:mongodb"
+        ),
+        **other_kwargs,
+    ):
+        super().__init__(image=image, endpoint_ports=[ENDPOINT_PORT], **other_kwargs)
+        assert mongodb_version is not None, "A MongoDB version is required to use the MongoDB Endpoint"
+        self.mongodb_version = mongodb_version
+
+    @property
+    def port(self) -> str:
+        return self.get_exposed_port(ENDPOINT_PORT)
+
+    def _configure(self) -> None:
+        super()._configure()
+        self.with_env("AZURE_COSMOS_EMULATOR_ENABLE_MONGODB_ENDPOINT", self.mongodb_version)