use docker compose profiling

Author: Karthikeyannhs

poetry.lock

@@ -62,7 +62,7 @@ behave = "^1.3.3"
 python-dotenv = "^1.2.1"
 openapi-spec-validator = "^0.7.2"
 pip-licenses = "^5.5.0"
-
+cachetools = "^7.0.1"
 
 [tool.poetry-plugin-lambda-build]
 docker-image = "public.ecr.aws/sam/build-python3.13:1.139-x86_64"  # See https://gallery.ecr.aws/search?searchTerm=%22python%22&architecture=x86-64&popularRegistries=amazon&verified=verified&operatingSystems=Linux

pyproject.toml

@@ -1,5 +1,6 @@
 services:
   moto-server:
+    #used for s3, dynamodb, kinesis, secret manager
     image: motoserver/moto:latest
     container_name: moto-server
     ports:
@@ -8,6 +9,8 @@ services:
       - test-network
 
   lambda-api:
+    #used for lambda simulation
+    profiles: ["lambda-test"]
     image: public.ecr.aws/lambda/python:3.13
     container_name: lambda-api
     ports:
@@ -39,6 +42,8 @@ services:
       - test-network
 
   api-gateway-mock:
+    #used for api gateway simulation
+    profiles: ["lambda-test"]
     image: openresty/openresty:alpine
     ports:
       - "9123:9123"

tests/docker-compose.yml

@@ -2,10 +2,7 @@
 import json
 import logging
 import os
-import subprocess
 from collections.abc import Callable, Generator
-from pathlib import Path
-from typing import List
 from typing import TYPE_CHECKING, Any
 
 import httpx

tests/integration/conftest.py

@@ -1,3 +1,4 @@
+import os
 import subprocess
 from pathlib import Path
 from typing import Callable
@@ -10,10 +11,17 @@
 from tests.integration.conftest import is_responsive
 
 
+def get_project_root() -> Path:
+    """Finds the project root by looking for the 'dist' directory or a git folder."""
+    current = Path(__file__).resolve()
+    for parent in current.parents:
+        if (parent / "dist").exists() or (parent / ".git").exists():
+            return parent
+    return Path(__file__).resolve().parents[3]
+
 @pytest.fixture(scope="session")
 def lambda_zip() -> Path:
-    # Determine project root (directory containing this conftest.py)
-    project_root = Path(__file__).resolve().parents[3]
+    project_root = get_project_root()
 
     build_result = subprocess.run(
         ["make", "build"],
@@ -37,29 +45,51 @@ def lambda_zip() -> Path:
     return zip_path
 
 @pytest.fixture(scope="session")
-def lambda_runtime_url(request: pytest.FixtureRequest, lambda_zip: Path) -> URL:
+def lambda_runtime_url(request, lambda_zip):
+    """
+    kick-starts the lambda simulation
+    """
     docker_services = request.getfixturevalue("docker_services")
     docker_ip = request.getfixturevalue("docker_ip")
-
-    wait_for_zip_in_container(docker_services, "lambda-api", "/tmp/lambda.zip")
-    force_lambda_reload(docker_services)
+    project_root = get_project_root()
+    compose_file = project_root / "tests/docker-compose.yml"
+
+    env = os.environ.copy()
+    env["COMPOSE_PROFILES"] = "lambda-test"
+
+    subprocess.run(
+        [
+            "docker", "compose",
+            "-f", str(compose_file),
+            "up", "-d", "--build", "--force-recreate",
+            "lambda-api", "api-gateway-mock",
+        ],
+        env=env,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
 
     port = docker_services.port_for("lambda-api", 8080)
     base_url = URL(f"http://{docker_ip}:{port}")
 
     docker_services.wait_until_responsive(
-        timeout=30.0,
-        pause=1.0,
-        check=lambda: is_responsive(base_url)
+        timeout=30.0, pause=0.5, check=lambda: is_responsive(base_url)
     )
+
     return base_url
 
+
+
 @pytest.fixture(scope="session")
 def lambda_client(boto3_session: Session, lambda_runtime_url: URL) -> BaseClient:
     return boto3_session.client("lambda", endpoint_url=str(lambda_runtime_url))
 
 @pytest.fixture(scope="session")
 def api_gateway_endpoint(request: pytest.FixtureRequest, lambda_runtime_url) -> URL:
+    """
+        kick-starts the api-gateway lambda simulation
+    """
     docker_services = request.getfixturevalue("docker_services")
     docker_ip = request.getfixturevalue("docker_ip")
 
@@ -75,41 +105,21 @@ def api_gateway_endpoint(request: pytest.FixtureRequest, lambda_runtime_url) ->
     )
     return base_url
 
-def wait_for_zip_in_container(docker_services, service: str, path: str) -> None:
-    def _check() -> bool:
-        try:
-            docker_services._docker_compose.execute(
-                f"exec {service} test -f {path}"
-            )
-            return True
-        except Exception:
-            return False
-
-    docker_services.wait_until_responsive(
-        timeout=30,
-        pause=1,
-        check=_check,
-    )
-
-
-def force_lambda_reload(docker_services):
-    # Stop and remove lambda-api
-    docker_services._docker_compose.execute(f"stop lambda-api")
-    docker_services._docker_compose.execute(f"rm -f lambda-api")
-
-    # Start lambda-api fresh so it re-reads the mounted ZIP
-    docker_services._docker_compose.execute(f"up --build -d lambda-api")
-
 
 @pytest.fixture
 def lambda_logs(docker_services) -> Callable[[], list[str]]:
+    """Returns a callable that fetches the latest lambda-api logs,
+    allowing tests to inspect runtime output on demand."""
+
     def _get_messages() -> list[str]:
         return get_lambda_logs(docker_services)
 
     return _get_messages
 
 
 def get_lambda_logs(docker_services) -> list[str]:
+    """returns logs from lambda-api container"""
+
     result: bytes = docker_services._docker_compose.execute("logs --no-color lambda-api")
     raw_lines = result.decode("utf-8").splitlines()
     return [line.partition("|")[-1].strip() for line in raw_lines]