fix(gcs): add fake-gcs-server container

docs/community/gcs.rst

@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.community.gcs.GcsContainer
+.. title:: testcontainers.community.gcs.GcsContainer

docs/modules/gcs.md

@@ -0,0 +1,23 @@
+# Google Cloud Storage
+
+Since testcontainers-python <a href="https://github.com/testcontainers/testcontainers-python/releases/tag/v4.16.0"><span class="tc-version">:material-tag: v4.16.0</span></a>
+
+## Introduction
+
+The Testcontainers module for Google Cloud Storage, based on [fake-gcs-server](https://github.com/fsouza/fake-gcs-server).
+
+## Adding this module to your project dependencies
+
+Please run the following command to add the Google Cloud Storage module to your python dependencies:
+
+```bash
+pip install testcontainers[gcs] google-cloud-storage
+```
+
+## Usage example
+
+<!--codeinclude-->
+
+[Creating a Google Cloud Storage container](gcs_example.py)
+
+<!--/codeinclude-->

docs/modules/gcs_example.py

@@ -0,0 +1,30 @@
+from testcontainers.community.gcs import GcsContainer
+
+
+def basic_example():
+    with GcsContainer() as gcs:
+        # Get a client connected to the emulator
+        client = gcs.get_client()
+
+        # Create a bucket
+        bucket = client.create_bucket("test-bucket")
+        print(f"Created bucket: {bucket.name}")
+
+        # Upload objects
+        for i in range(1, 4):
+            blob = bucket.blob(f"test{i}.txt")
+            blob.upload_from_string(f"Hello from object {i}!")
+            print(f"Uploaded object: {blob.name}")
+
+        # List objects
+        print("\nObjects in bucket:")
+        for blob in client.list_blobs(bucket.name):
+            print(f"Name: {blob.name}, Size: {blob.size} bytes")
+
+        # Download an object
+        content = bucket.blob("test1.txt").download_as_text()
+        print(f"\nContent of test1.txt: {content}")
+
+
+if __name__ == "__main__":
+    basic_example()

mkdocs.yml

@@ -77,6 +77,7 @@ nav:
           - modules/weaviate.md
       - modules/aws.md
       - modules/azurite.md
+      - modules/gcs.md
       - modules/generic.md
       - modules/google.md
       - modules/k3s.md

pyproject.toml

@@ -63,6 +63,7 @@ db2 = [
   "ibm_db_sa; platform_machine != 'aarch64' and platform_machine != 'arm64'",
 ]
 elasticsearch = []
+gcs = []
 generic = [
   "httpx",
   "redis>=7",
@@ -135,6 +136,7 @@ test = [
   "sqlalchemy-cockroachdb>=2",
   "sqlalchemy-cratedb",
   "paramiko>=4",
+  "google-cloud-storage>=2",
   "twine>=7.0.0",
   "anyio>=4",
   "tomli>=2.0; python_version < '3.11'", # required for pyproject.toml tests, TODO: remove once we drop py3.10 support

src/testcontainers/community/gcs/__init__.py

@@ -0,0 +1,94 @@
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+import json
+from typing import TYPE_CHECKING
+from urllib.request import Request, urlopen
+
+from typing_extensions import Self
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.wait_strategies import HttpWaitStrategy
+
+if TYPE_CHECKING:
+    from google.cloud import storage
+
+__all__ = ["GcsContainer"]
+
+
+class GcsContainer(DockerContainer):
+    """
+    Google Cloud Storage emulator container based on `fake-gcs-server <https://github.com/fsouza/fake-gcs-server>`_.
+
+    Example:
+
+        .. doctest::
+
+            >>> from testcontainers.community.gcs import GcsContainer
+
+            >>> with GcsContainer() as gcs:
+            ...    client = gcs.get_client()
+            ...    bucket = client.create_bucket("my-bucket")
+            ...    bucket.blob("hello.txt").upload_from_string("Hello, GCS!")
+            ...    content = bucket.blob("hello.txt").download_as_text()
+            >>> content
+            'Hello, GCS!'
+    """
+
+    def __init__(
+        self,
+        image: str = "fsouza/fake-gcs-server:1.56.1",
+        port: int = 4443,
+        project: str = "test-project",
+        **kwargs: object,
+    ) -> None:
+        super().__init__(image=image, **kwargs)
+        self.port = port
+        self.project = project
+        self.with_exposed_ports(self.port)
+        self.with_command(f"-scheme http -port {self.port} -backend memory")
+        self.waiting_for(HttpWaitStrategy(self.port, "/storage/v1/b"))
+
+    def get_url(self) -> str:
+        return f"http://{self.get_container_host_ip()}:{self.get_exposed_port(self.port)}"
+
+    def start(self) -> Self:
+        super().start()
+        self._update_external_url()
+        return self
+
+    def _update_external_url(self) -> None:
+        # fake-gcs-server builds object links (e.g. mediaLink) from its external URL,
+        # and the host-mapped port is only known once the container is running.
+        request = Request(
+            f"{self.get_url()}/_internal/config",
+            data=json.dumps({"externalUrl": self.get_url()}).encode(),
+            headers={"Content-Type": "application/json"},
+            method="PUT",
+        )
+        with urlopen(request, timeout=10):
+            pass
+
+    def get_client(self, **kwargs: object) -> "storage.Client":
+        """Get a ``google.cloud.storage.Client`` connected to the emulator.
+
+        Requires the ``google-cloud-storage`` package to be installed.
+        """
+        from google.auth.credentials import AnonymousCredentials
+        from google.cloud import storage
+
+        return storage.Client(
+            project=self.project,
+            credentials=AnonymousCredentials(),
+            client_options={"api_endpoint": self.get_url()},
+            **kwargs,
+        )

tests/community/gcs/test_gcs.py

@@ -0,0 +1,37 @@
+import os
+
+from testcontainers.community.gcs import GcsContainer
+
+
+def test_gcs_upload_and_download():
+    with GcsContainer() as gcs:
+        client = gcs.get_client()
+        bucket = client.create_bucket("test-bucket")
+        bucket.blob("hello.txt").upload_from_string("Hello, GCS!")
+
+        assert bucket.blob("hello.txt").download_as_text() == "Hello, GCS!"
+        assert [blob.name for blob in client.list_blobs("test-bucket")] == ["hello.txt"]
+
+
+def test_gcs_object_links_use_mapped_port():
+    with GcsContainer() as gcs:
+        blob = gcs.get_client().create_bucket("test-bucket").blob("hello.txt")
+        blob.upload_from_string("Hello, GCS!")
+        blob.reload()
+
+        assert blob.media_link.startswith(gcs.get_url())
+
+
+def test_gcs_resumable_upload():
+    # larger than the client's 8 MiB multipart limit, so a resumable upload is used
+    data = os.urandom(10 * 1024 * 1024)
+    with GcsContainer() as gcs:
+        blob = gcs.get_client().create_bucket("test-bucket").blob("big.bin")
+        blob.upload_from_string(data)
+
+        assert blob.download_as_bytes() == data
+
+
+def test_gcs_custom_project():
+    with GcsContainer(project="my-project") as gcs:
+        assert gcs.get_client().project == "my-project"