Skip to content

Commit 2ef4d25

Browse files
committed
test(event-handler): add tests for OpenAPI schema bleed fix
Relates to #7711 Add comprehensive tests to verify that when multiple routes share the same response dictionary, each route gets its own correct OpenAPI schema without bleeding return types between routes. Tests cover: - Different return types (list vs single object) with shared responses - Verification that shared dictionaries are not mutated - Regression testing for standard behavior
1 parent 6ec11b8 commit 2ef4d25

File tree

1 file changed

+173
-0
lines changed

1 file changed

+173
-0
lines changed
Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
"""Test for bug #7711: OpenAPI schema return types bleed across routes when reusing response dictionaries"""
2+
3+
from __future__ import annotations
4+
5+
from pydantic import BaseModel
6+
7+
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, Response
8+
from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse
9+
10+
11+
class ExamSummary(BaseModel):
12+
"""Summary information about an exam"""
13+
14+
id: str
15+
name: str
16+
duration_minutes: int
17+
18+
19+
class ExamConfig(BaseModel):
20+
"""Detailed configuration for an exam"""
21+
22+
id: str
23+
name: str
24+
duration_minutes: int
25+
max_attempts: int
26+
passing_score: int
27+
28+
29+
class Responses:
30+
"""Pre-configured OpenAPI response schemas."""
31+
32+
# Base responses
33+
OK = {200: OpenAPIResponse(description="Successful operation")}
34+
NOT_FOUND = {404: OpenAPIResponse(description="Resource not found")}
35+
VALIDATION_ERROR = {422: OpenAPIResponse(description="Validation error")}
36+
SERVER_ERROR = {500: OpenAPIResponse(description="Internal server error")}
37+
38+
# Common combinations
39+
STANDARD_ERRORS = {**NOT_FOUND, **VALIDATION_ERROR, **SERVER_ERROR}
40+
41+
@classmethod
42+
def combine(cls, *response_dicts: dict[int, OpenAPIResponse]) -> dict[int, OpenAPIResponse]:
43+
"""Combine multiple response dictionaries."""
44+
result = {}
45+
for response_dict in response_dicts:
46+
result.update(response_dict)
47+
return result
48+
49+
50+
def test_openapi_shared_response_no_bleed():
51+
"""
52+
Test that when reusing the same response dictionary across multiple routes,
53+
each route gets the correct return type in its OpenAPI schema.
54+
55+
This reproduces bug #7711 where the schema from one route bleeds into another
56+
when they share the same response dictionary object.
57+
"""
58+
app = APIGatewayRestResolver(enable_validation=True)
59+
60+
@app.get(
61+
"/exams",
62+
tags=["Exams"],
63+
responses=Responses.combine(Responses.OK, Responses.STANDARD_ERRORS),
64+
)
65+
def list_exams() -> Response[list[ExamSummary]]:
66+
"""Lists all available exams."""
67+
return Response(
68+
status_code=200,
69+
body=[
70+
ExamSummary(id="1", name="Math", duration_minutes=60),
71+
ExamSummary(id="2", name="Science", duration_minutes=90),
72+
],
73+
)
74+
75+
@app.get(
76+
"/exams/<exam_id>",
77+
tags=["Exams"],
78+
responses=Responses.combine(Responses.OK, Responses.STANDARD_ERRORS), # Reusing the shared Responses.OK
79+
)
80+
def get_exam_config(exam_id: str) -> Response[ExamConfig]:
81+
"""Get the configuration for a specific exam"""
82+
return Response(
83+
status_code=200,
84+
body=ExamConfig(
85+
id=exam_id,
86+
name="Math",
87+
duration_minutes=60,
88+
max_attempts=3,
89+
passing_score=70,
90+
),
91+
)
92+
93+
# Generate the OpenAPI schema
94+
schema = app.get_openapi_schema()
95+
96+
# Verify /exams endpoint has the correct list[ExamSummary] schema
97+
exams_response = schema.paths["/exams"].get.responses[200]
98+
exams_schema = exams_response.content["application/json"].schema_
99+
100+
# The schema should be an array type
101+
assert exams_schema.type == "array", f"/exams should return an array, got {exams_schema.type}"
102+
assert exams_schema.items is not None, "/exams should have items definition"
103+
104+
# The items should reference ExamSummary
105+
if hasattr(exams_schema.items, "ref"):
106+
assert "ExamSummary" in exams_schema.items.ref, (
107+
f"/exams should return list[ExamSummary], got {exams_schema.items.ref}"
108+
)
109+
elif hasattr(exams_schema.items, "title"):
110+
assert exams_schema.items.title == "ExamSummary", (
111+
f"/exams should return list[ExamSummary], got {exams_schema.items.title}"
112+
)
113+
114+
# Verify /exams/{exam_id} endpoint has the correct ExamConfig schema
115+
exam_detail_response = schema.paths["/exams/{exam_id}"].get.responses[200]
116+
exam_detail_schema = exam_detail_response.content["application/json"].schema_
117+
118+
# The schema should NOT be an array - it should be an object
119+
assert exam_detail_schema.type != "array", "/exams/{exam_id} should not return an array (bug #7711 - schema bleed)"
120+
121+
# The schema should reference ExamConfig
122+
if hasattr(exam_detail_schema, "ref"):
123+
assert "ExamConfig" in exam_detail_schema.ref, (
124+
f"/exams/{{exam_id}} should return ExamConfig, got {exam_detail_schema.ref}"
125+
)
126+
elif hasattr(exam_detail_schema, "title"):
127+
assert exam_detail_schema.title == "ExamConfig", (
128+
f"/exams/{{exam_id}} should return ExamConfig, got {exam_detail_schema.title}"
129+
)
130+
131+
132+
def test_openapi_shared_response_dict_not_mutated():
133+
"""
134+
Test that the original shared response dictionary is not mutated
135+
when generating OpenAPI schemas.
136+
"""
137+
app = APIGatewayRestResolver(enable_validation=True)
138+
139+
# Create a shared response dictionary
140+
shared_responses = Responses.combine(Responses.OK, Responses.STANDARD_ERRORS)
141+
142+
# Store the original state - the 200 response should not have 'content' key
143+
original_200_response = shared_responses[200].copy()
144+
assert "content" not in original_200_response, "200 response should not have content initially"
145+
146+
@app.get("/route1", responses=shared_responses)
147+
def route1() -> Response[ExamSummary]:
148+
return Response(
149+
status_code=200,
150+
body=ExamSummary(id="1", name="Test", duration_minutes=60),
151+
)
152+
153+
@app.get("/route2", responses=shared_responses)
154+
def route2() -> Response[ExamConfig]:
155+
return Response(
156+
status_code=200,
157+
body=ExamConfig(
158+
id="1",
159+
name="Test",
160+
duration_minutes=60,
161+
max_attempts=3,
162+
passing_score=70,
163+
),
164+
)
165+
166+
# Generate the OpenAPI schema
167+
app.get_openapi_schema()
168+
169+
# Verify the shared dictionary was not mutated
170+
# The original 200 response should still not have 'content' key
171+
assert "content" not in shared_responses[200], (
172+
"Shared response dictionary should not be mutated during OpenAPI schema generation (bug #7711)"
173+
)

0 commit comments

Comments
 (0)