feat(core): Extend container support

core/README.rst

@@ -7,6 +7,8 @@ testcontainers-core
 
 .. autoclass:: testcontainers.core.image.DockerImage
 
+.. autoclass:: testcontainers.core.generic.SrvContainer
+
 Using `DockerContainer` and `DockerImage` directly:
 
 .. doctest::

core/testcontainers/core/generic.py

@@ -11,10 +11,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 from typing import Optional
+from urllib.error import HTTPError
 from urllib.parse import quote
+from urllib.request import urlopen
 
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.exceptions import ContainerStartException
+from testcontainers.core.image import DockerImage
 from testcontainers.core.utils import raise_for_deprecated_parameter
 from testcontainers.core.waiting_utils import wait_container_is_ready
 
@@ -79,3 +82,67 @@ def _configure(self) -> None:
 
     def _transfer_seed(self) -> None:
         pass
+
+
+class SrvContainer(DockerContainer):
+    """
+    Container for a generic server that is based on a custom image.
+
+    Example:
+
+    .. doctest::
+
+        >>> import httpx
+        >>> from testcontainers.core.generic import SrvContainer
+        >>> from testcontainers.core.waiting_utils import wait_for_logs
+
+        >>> with SrvContainer(path="./core/tests/image_fixtures/python_server", port=9000, tag="test-srv:latest") as srv:
+        ...     url = srv._create_connection_url()
+        ...     response = httpx.get(f"{url}", timeout=5)
+        ...     assert response.status_code == 200, "Response status code is not 200"
+        ...     delay = wait_for_logs(srv, "GET / HTTP/1.1")
+
+
+    :param path: Path to the Dockerfile to build the image
+    :param tag: Tag for the image to be built (default: None)
+    """
+
+    def __init__(self, path: str, port: int, tag: Optional[str], image_cleanup: bool = True) -> None:
+        self.docker_image = DockerImage(path=path, tag=tag, clean_up=image_cleanup).build()
+        super().__init__(str(self.docker_image))
+        self.internal_port = port
+        self.with_exposed_ports(self.internal_port)
+
+    @wait_container_is_ready(HTTPError)
+    def _connect(self) -> None:
+        # noinspection HttpUrlsUsage
+        url = self._create_connection_url()
+        try:
+            with urlopen(url) as r:
+                assert b"" in r.read()
+        except HTTPError as e:
+            # 404 is expected, as the server may not have the specific endpoint we are looking for
+            if e.code == 404:
+                pass
+            else:
+                raise
+
+    def get_api_url(self) -> str:
+        raise NotImplementedError
+
+    def _create_connection_url(self) -> str:
+        if self._container is None:
+            raise ContainerStartException("container has not been started")
+        host = self.get_container_host_ip()
+        exposed_port = self.get_exposed_port(self.internal_port)
+        url = f"http://{host}:{exposed_port}"
+        return url
+
+    def start(self) -> "SrvContainer":
+        super().start()
+        self._connect()
+        return self
+
+    def stop(self, force=True, delete_volume=True) -> None:
+        super().stop(force, delete_volume)
+        self.docker_image.remove()

core/tests/image_fixtures/python_server/Dockerfile

@@ -0,0 +1,3 @@
+FROM python:3
+EXPOSE 9000
+CMD ["python", "-m", "http.server", "9000"]

core/tests/test_generics.py

@@ -0,0 +1,27 @@
+import pytest
+from typing import Optional
+from testcontainers.core.generic import SrvContainer
+
+import re
+
+
+@pytest.mark.parametrize("test_image_cleanup", [True, False])
+@pytest.mark.parametrize("test_image_tag", [None, "custom-image:test"])
+def test_srv_container(test_image_tag: Optional[str], test_image_cleanup: bool, check_for_image, port=9000):
+    with SrvContainer(
+        path="./core/tests/image_fixtures/python_server",
+        port=port,
+        tag=test_image_tag,
+        image_cleanup=test_image_cleanup,
+    ) as srv:
+        image_short_id = srv.docker_image.short_id
+        image_build_logs = srv.docker_image.get_logs()
+        # check if dict is in any of the logs
+        assert {"stream": f"Step 2/3 : EXPOSE {port}"} in image_build_logs, "Image logs mismatch"
+        assert (port, None) in srv.ports.items(), "Port mismatch"
+        with pytest.raises(NotImplementedError):
+            srv.get_api_url()
+        test_url = srv._create_connection_url()
+        assert re.match(r"http://localhost:\d+", test_url), "Connection URL mismatch"
+
+    check_for_image(image_short_id, test_image_cleanup)

index.rst

@@ -16,11 +16,13 @@ testcontainers-python facilitates the use of Docker containers for functional an
 
     core/README
     modules/arangodb/README
+    modules/aws/README
     modules/azurite/README
     modules/cassandra/README
     modules/chroma/README
     modules/clickhouse/README
     modules/elasticsearch/README
+    modules/fastapi/README
     modules/google/README
     modules/influxdb/README
     modules/k3s/README

modules/aws/README.rst

@@ -0,0 +1,6 @@
+.. autoclass:: testcontainers.aws.AWSLambdaContainer
+.. title:: testcontainers.aws.AWSLambdaContainer
+
+Make sure you are using an image based on `public.ecr.aws/lambda/python`
+
+Please checkout https://docs.aws.amazon.com/lambda/latest/dg/python-image.html for more information on how to run AWS Lambda functions locally.

modules/aws/testcontainers/aws/__init__.py

@@ -0,0 +1 @@
+from .aws_lambda import AWSLambdaContainer  # noqa: F401

modules/aws/testcontainers/aws/aws_lambda.py

@@ -0,0 +1,65 @@
+import os
+from typing import Optional
+
+import httpx
+
+from testcontainers.core.generic import SrvContainer
+
+RIE_PATH = "/2015-03-31/functions/function/invocations"
+# AWS OS-only base images contain an Amazon Linux distribution and the runtime interface emulator.
+
+
+class AWSLambdaContainer(SrvContainer):
+    """
+    AWS Lambda container that is based on a custom image.
+
+    Example:
+
+        .. doctest::
+
+            >>> from testcontainers.aws import AWSLambdaContainer
+            >>> from testcontainers.core.waiting_utils import wait_for_logs
+
+            >>> with AWSLambdaContainer(path="./modules/aws/tests/lambda_sample", port=8080, tag="lambda_func:latest") as func:
+            ...     response = func.send_request(data={'payload': 'some data'})
+            ...     assert response.status_code == 200
+            ...     assert "Hello from AWS Lambda using Python" in response.json()
+            ...     delay = wait_for_logs(func, "START RequestId:")
+    """
+
+    def __init__(
+        self,
+        path: str,
+        port: int = 8080,
+        region_name: Optional[str] = None,
+        tag: Optional[str] = None,
+        image_cleanup: bool = True,
+    ) -> None:
+        """
+        :param path: Path to the AWS Lambda dockerfile.
+        :param port: Port to be exposed on the container (default: 8080).
+        :param region_name: AWS region name (default: None).
+        :param tag: Tag for the image to be built (default: None).
+        :param image_cleanup: Clean up the image after the container is stopped (default: True).
+        """
+        super().__init__(path, port, tag, image_cleanup)
+        self.region_name = region_name or os.environ.get("AWS_DEFAULT_REGION", "us-west-1")
+        self.with_env("AWS_DEFAULT_REGION", self.region_name)
+        self.with_env("AWS_ACCESS_KEY_ID", "testcontainers-aws")
+        self.with_env("AWS_SECRET_ACCESS_KEY", "testcontainers-aws")
+
+    def get_api_url(self) -> str:
+        return self._create_connection_url() + RIE_PATH
+
+    def send_request(self, data: dict) -> httpx.Response:
+        """
+        Send a request to the AWS Lambda function.
+
+        :param data: Data to be sent to the AWS Lambda function.
+        :return: Response from the AWS Lambda function.
+        """
+        client = httpx.Client()
+        return client.post(self.get_api_url(), json=data)
+
+    def get_stdout(self) -> str:
+        return self.get_logs()[0].decode("utf-8")

modules/aws/tests/lambda_sample/Dockerfile

@@ -0,0 +1,10 @@
+FROM public.ecr.aws/lambda/python:3.9
+
+RUN pip install boto3
+
+COPY lambda_function.py ${LAMBDA_TASK_ROOT}
+
+EXPOSE 8080
+
+# Set the CMD to your handler (could also be done as a parameter override outside of the Dockerfile)
+CMD [ "lambda_function.handler" ]

modules/aws/tests/lambda_sample/lambda_function.py

@@ -0,0 +1,5 @@
+import sys
+
+
+def handler(event, context):
+    return "Hello from AWS Lambda using Python" + sys.version + "!"

modules/aws/tests/test_aws.py

@@ -0,0 +1,35 @@
+import re
+import pytest
+
+from testcontainers.aws import AWSLambdaContainer
+from testcontainers.aws.aws_lambda import RIE_PATH
+
+DOCKER_FILE_PATH = "./modules/aws/tests/lambda_sample"
+IMAGE_TAG = "lambda:test"
+
+
+def test_aws_lambda_container():
+    with AWSLambdaContainer(path=DOCKER_FILE_PATH, port=8080, tag=IMAGE_TAG, image_cleanup=False) as func:
+        assert func.get_container_host_ip() == "localhost"
+        assert func.internal_port == 8080
+        assert func.env["AWS_DEFAULT_REGION"] == "us-west-1"
+        assert func.env["AWS_ACCESS_KEY_ID"] == "testcontainers-aws"
+        assert func.env["AWS_SECRET_ACCESS_KEY"] == "testcontainers-aws"
+        assert re.match(rf"http://localhost:\d+{RIE_PATH}", func.get_api_url())
+        response = func.send_request(data={"payload": "test"})
+        assert response.status_code == 200
+        assert "Hello from AWS Lambda using Python" in response.json()
+        for log_str in ["START RequestId", "END RequestId", "REPORT RequestId"]:
+            assert log_str in func.get_stdout()
+
+
+def test_aws_lambda_container_no_tag():
+    with AWSLambdaContainer(path=DOCKER_FILE_PATH, image_cleanup=True) as func:
+        response = func.send_request(data={"payload": "test"})
+        assert response.status_code == 200
+
+
+def test_aws_lambda_container_no_path():
+    with pytest.raises(TypeError):
+        with AWSLambdaContainer(port=8080, tag=IMAGE_TAG, image_cleanup=True):
+            pass

modules/fastapi/README.rst

@@ -0,0 +1,2 @@
+.. autoclass:: testcontainers.fastapi.FastAPIContainer
+.. title:: testcontainers.fastapi.FastAPIContainer

modules/fastapi/testcontainers/fastapi/__init__.py

@@ -0,0 +1,43 @@
+from typing import Optional
+
+import httpx
+
+from testcontainers.core.generic import SrvContainer
+
+
+class FastAPIContainer(SrvContainer):
+    """
+    FastAPI container that is based on a custom image.
+
+    Example:
+
+        .. doctest::
+
+            >>> from testcontainers.fastapi import FastAPIContainer
+            >>> from testcontainers.core.waiting_utils import wait_for_logs
+
+            >>> with FastAPIContainer(path="./modules/fastapi/tests/sample", port=80, tag="fastapi:latest") as fastapi:
+            ...     delay = wait_for_logs(fastapi, "Uvicorn running on http://0.0.0.0:80")
+            ...     client = fastapi.get_client()
+            ...     response = client.get("/")
+            ...     assert response.status_code == 200
+            ...     assert response.json() == {"Status": "Working"}
+    """
+
+    def __init__(self, path: str, port: int, tag: Optional[str] = None, image_cleanup: bool = True) -> None:
+        """
+        :param path: Path to the FastAPI application.
+        :param port: Port to expose the FastAPI application.
+        :param tag: Tag for the image to be built (default: None).
+        :param image_cleanup: Clean up the image after the container is stopped (default: True).
+        """
+        super().__init__(path, port, tag, image_cleanup)
+
+    def get_api_url(self) -> str:
+        return self._create_connection_url() + "/api/v1/"
+
+    def get_client(self) -> httpx.Client:
+        return httpx.Client(base_url=self.get_api_url())
+
+    def get_stdout(self) -> str:
+        return self.get_logs()[0].decode("utf-8")

modules/fastapi/tests/sample/Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.9
+
+WORKDIR /app
+
+RUN pip install fastapi
+
+COPY ./app /app
+
+EXPOSE 80
+
+CMD ["fastapi", "run", "main.py", "--port", "80"]

modules/fastapi/tests/sample/app/main.py

@@ -0,0 +1,8 @@
+from fastapi import FastAPI
+
+app = FastAPI()
+
+
+@app.get("/api/v1/")
+def read_root():
+    return {"Status": "Working"}

modules/fastapi/tests/test_fastapi.py

@@ -0,0 +1,33 @@
+import re
+import pytest
+
+from testcontainers.fastapi import FastAPIContainer
+
+
+def test_fastapi_container():
+    with FastAPIContainer(
+        path="./modules/fastapi/tests/sample", port=80, tag="fastapi:test", image_cleanup=False
+    ) as fastapi:
+        assert fastapi.get_container_host_ip() == "localhost"
+        assert fastapi.internal_port == 80
+        assert re.match(r"http://localhost:\d+/api/v1/", fastapi.get_api_url())
+        assert fastapi.get_client().get("/").status_code == 200
+        assert fastapi.get_client().get("/").json() == {"Status": "Working"}
+
+
+def test_fastapi_container_no_tag():
+    with FastAPIContainer(path="./modules/fastapi/tests/sample", port=80, image_cleanup=False) as fastapi:
+        assert fastapi.get_client().get("/").status_code == 200
+        assert fastapi.get_client().get("/").json() == {"Status": "Working"}
+
+
+def test_fastapi_container_no_port():
+    with pytest.raises(TypeError):
+        with FastAPIContainer(path="./modules/fastapi/tests/sample", tag="fastapi:test", image_cleanup=False):
+            pass
+
+
+def test_fastapi_container_no_path():
+    with pytest.raises(TypeError):
+        with FastAPIContainer(port=80, tag="fastapi:test", image_cleanup=True):
+            pass