-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlambda_nonauth_test_harness.py
More file actions
144 lines (117 loc) · 3.8 KB
/
lambda_nonauth_test_harness.py
File metadata and controls
144 lines (117 loc) · 3.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
"""
Test the lambda_functions non-authorizers.
Local test harness for non-authorizer Lambda functions
invoked via API Gateway proxy integration.
Run this file directly in PyCharm to debug Lambdas
outside of AWS.
"""
import json
import os
import sys
import importlib
from dataclasses import dataclass
# Ensure repo root is on sys.path so lambda_functions imports work
REPO_ROOT = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, REPO_ROOT)
@dataclass
class MockLambdaContext:
"""
Minimal stand-in for AWS Lambda context object.
"""
function_name: str
memory_limit_in_mb: int = 128
aws_request_id: str = "local-debug-request"
def build_proxy_event(
*,
path: str,
http_method: str,
api_id: str,
stage: str,
region: str,
account_id: str,
) -> dict:
"""
Construct a representative API Gateway *proxy* event.
NOTE ON ARN DATA
----------------
Non-authorizer Lambdas NEVER receive `methodArn`.
If your Lambda logic needs to infer ARN-like information,
it must come from `requestContext`, which is where API
Gateway exposes:
- apiId
- stage
- resourcePath
- region
- accountId
THIS IS THE ONLY PLACE YOU PROVIDE ARN-RELATED INFORMATION.
"""
return {
"resource": "/{proxy+}",
"path": path,
"httpMethod": http_method,
"headers": {},
"queryStringParameters": None,
"pathParameters": {
"proxy": path.lstrip("/")
},
"requestContext": {
"accountId": account_id,
"region": region,
"apiId": api_id,
"stage": stage,
"resourcePath": "/{proxy+}",
"httpMethod": http_method,
"path": path,
# -----------------------------------------------------------
},
"body": None,
"isBase64Encoded": False,
}
def run_lambda(module_name: str, event: dict) -> dict:
"""
Import and invoke a Lambda module by name from lambda_functions.
"""
print(f"\n=== Running {module_name}.lambda_handler ===")
# print(f"Event:\n{json.dumps(event, indent=2)}")
module = importlib.import_module(f"lambda_functions.{module_name}")
handler = getattr(module, "lambda_handler")
context = MockLambdaContext(function_name=module_name)
result = handler(event, context)
# print(f"\nResult:\n{json.dumps(result, indent=2)}")
return result
def assert_status(result: dict, expected_status: int) -> None:
assert "statusCode" in result
assert result["statusCode"] == expected_status
print(f"✔ statusCode == {expected_status}")
def assert_body_message(result: dict, expected_substring: str) -> None:
assert "body" in result
body = json.loads(result["body"])
assert expected_substring in body.get("message", "")
print(f"✔ body contains '{expected_substring}'")
def main():
# ------------------------------------------------------------------
# API Gateway identity (EDIT THESE IF YOU NEED REALISTIC ARN VALUES)
# ------------------------------------------------------------------
REGION = "us-east-1"
ACCOUNT_ID = "450834107946"
API_ID = "0gwixh7ht0"
STAGE = "prod"
# ------------------------------------------------------------------
# ---- No search_api versioning redirect Lambda in SenNet ----
# ---- 404 catch-all Lambda ----
not_found_event = build_proxy_event(
path="/this/endpoint/does/not/exist",
http_method="GET",
region=REGION,
account_id=ACCOUNT_ID,
api_id=API_ID,
stage=STAGE,
)
result = run_lambda(
module_name="404",
event=not_found_event,
)
assert_status(result, 404)
assert_body_message(result, "Unable to find the requested resource")
if __name__ == "__main__":
main()