Full working example with layer and sentry server collecting envelopes

Author: antonpirker

tests/integrations/aws_lambda/lambda_functions/hello_world/index.py

@@ -1,2 +1,6 @@
+from somelayer import layer_function
+
+
 def handler(event, context):
-    return {"message": f"Hello, {event['name']}!"}
+    message = f"Hello, {event['name']}! ({layer_function()})"
+    return {"message": message}

tests/integrations/aws_lambda/lambda_layer/python/somelayer.py

@@ -0,0 +1,14 @@
+import urllib.request
+import urllib.parse
+import json
+
+
+def layer_function():
+    data = json.dumps({"event": "test from layer"}).encode("utf-8")
+    req = urllib.request.Request(
+        "http://host.docker.internal:9999/api/0/envelope/",
+        data=data,
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req) as response:
+        return "This is from the Layer"

tests/integrations/aws_lambda/test_aws_lambda.py

@@ -4,6 +4,7 @@
 import requests
 import subprocess
 import time
+import threading
 import yaml
 from aws_cdk import (
     App,
@@ -26,6 +27,25 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
         self.template_options.template_format_version = "2010-09-09"
         self.template_options.transforms = ["AWS::Serverless-2016-10-31"]
 
+        # Create Sentry Lambda Layer
+        layer = CfnResource(
+            self,
+            "SentryPythonServerlessSDK",
+            type="AWS::Serverless::LayerVersion",
+            properties={
+                "ContentUri": "./tests/integrations/aws_lambda/lambda_layer",
+                "CompatibleRuntimes": [
+                    "python3.7",
+                    "python3.8",
+                    "python3.9",
+                    "python3.10",
+                    "python3.11",
+                    "python3.12",
+                    "python3.13",
+                ],
+            },
+        )
+
         # Add the function using SAM format
         CfnResource(
             self,
@@ -35,6 +55,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
                 "CodeUri": "./tests/integrations/aws_lambda/lambda_functions/hello_world",
                 "Handler": "index.handler",
                 "Runtime": "python3.12",
+                "Layers": [{"Ref": layer.logical_id}],
             },
         )
 
@@ -59,11 +80,55 @@ def wait_for_sam(timeout=30, port=SAM_PORT):
             continue
 
 
+class SentryTestServer:
+    def __init__(self, port=9999):
+        self.envelopes = []
+        self.port = port
+
+        from fastapi import FastAPI, Request
+
+        self.app = FastAPI()
+
+        @self.app.get("/")
+        async def root():
+            return {
+                "message": "Sentry Test Server. Use DSN http://123@localhost:9999/0 in your SDK."
+            }
+
+        @self.app.post("/api/0/envelope/")
+        async def envelope(request: Request):
+            self.envelopes.append(await request.json())
+            return {"status": "ok"}
+
+    def run_server(self):
+        import uvicorn
+
+        uvicorn.run(self.app, host="0.0.0.0", port=self.port)
+
+    def start(self):
+        server_thread = threading.Thread(target=self.run_server, daemon=True)
+        server_thread.start()
+
+    def clear_envelopes(self):
+        self.envelopes = []
+
+
+@pytest.fixture(scope="session")
+def sentry_test_server():
+    server = SentryTestServer()
+    server.start()
+
+    time.sleep(1)  # Give it a moment to start up
+
+    yield server
+
+
 @pytest.fixture(scope="session")
 def sam_stack():
     """
     Create and deploy the SAM stack once for all tests
     """
+    # Create the SAM stack
     app = App()
     stack = DummyLambdaStack(app, "DummyLambdaStack", env={"region": SAM_REGION})
 
@@ -92,6 +157,7 @@ def sam_stack():
         # Wait for SAM to be ready
         wait_for_sam()
         yield stack
+
     finally:
         process.terminate()
         process.wait(timeout=5)  # Give it time to shut down gracefully
@@ -115,13 +181,25 @@ def lambda_client():
     )
 
 
-def test_basic(lambda_client, sam_stack):
-    region = boto3.Session().region_name
-    print("Region: ", region)
+def test_basic(lambda_client, sam_stack, sentry_test_server):
+    sentry_test_server.clear_envelopes()
+
+    response = lambda_client.invoke(
+        FunctionName="BasicTestFunction", Payload=json.dumps({"name": "Anton"})
+    )
+    result = json.loads(response["Payload"].read().decode())
+
+    print(sentry_test_server.envelopes)
+    assert result == {"message": "Hello, Anton!"}
+
+
+def test_basic_2(lambda_client, sam_stack, sentry_test_server):
+    sentry_test_server.clear_envelopes()
 
     response = lambda_client.invoke(
         FunctionName="BasicTestFunction", Payload=json.dumps({"name": "Anton"})
     )
     result = json.loads(response["Payload"].read().decode())
 
+    print(sentry_test_server.envelopes)
     assert result == {"message": "Hello, Anton!"}