Coverage for tests / test_sub_callbacks.py: 100%
38 statements
« prev ^ index » next coverage.py v7.13.3, created at 2026-02-21 17:29 +0000
1from fastapi import APIRouter, FastAPI 1abcd
2from fastapi.testclient import TestClient 1abcd
3from inline_snapshot import snapshot 1abcd
4from pydantic import BaseModel, HttpUrl 1abcd
6app = FastAPI() 1abcd
9class Invoice(BaseModel): 1abcd
10 id: str 1abcd
11 title: str | None = None 1abcd
12 customer: str 1abcd
13 total: float 1abcd
16class InvoiceEvent(BaseModel): 1abcd
17 description: str 1abcd
18 paid: bool 1abcd
21class InvoiceEventReceived(BaseModel): 1abcd
22 ok: bool 1abcd
25invoices_callback_router = APIRouter() 1abcd
28@invoices_callback_router.post( 1abcd
29 "{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived
30)
31def invoice_notification(body: InvoiceEvent): 1abcd
32 pass # pragma: nocover
35class Event(BaseModel): 1abcd
36 name: str 1abcd
37 total: float 1abcd
40events_callback_router = APIRouter() 1abcd
43@events_callback_router.get("{$callback_url}/events/{$request.body.title}") 1abcd
44def event_callback(event: Event): 1abcd
45 pass # pragma: nocover
48subrouter = APIRouter() 1abcd
51@subrouter.post("/invoices/", callbacks=invoices_callback_router.routes) 1abcd
52def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None): 1abcd
53 """
54 Create an invoice.
56 This will (let's imagine) let the API user (some external developer) create an
57 invoice.
59 And this path operation will:
61 * Send the invoice to the client.
62 * Collect the money from the client.
63 * Send a notification back to the API user (the external developer), as a callback.
64 * At this point is that the API will somehow send a POST request to the
65 external API with the notification of the invoice event
66 (e.g. "payment successful").
67 """
68 # Send the invoice, collect the money, send the notification (the callback)
69 return {"msg": "Invoice received"} 1efg
72app.include_router(subrouter, callbacks=events_callback_router.routes) 1abcd
74client = TestClient(app) 1abcd
77def test_get(): 1abcd
78 response = client.post( 1efg
79 "/invoices/", json={"id": "fooinvoice", "customer": "John", "total": 5.3}
80 )
81 assert response.status_code == 200, response.text 1efg
82 assert response.json() == {"msg": "Invoice received"} 1efg
85def test_openapi_schema(): 1abcd
86 with client: 1hij
87 response = client.get("/openapi.json") 1hij
88 assert response.json() == snapshot( 1hij
89 {
90 "openapi": "3.1.0",
91 "info": {"title": "FastAPI", "version": "0.1.0"},
92 "paths": {
93 "/invoices/": {
94 "post": {
95 "summary": "Create Invoice",
96 "description": 'Create an invoice.\n\nThis will (let\'s imagine) let the API user (some external developer) create an\ninvoice.\n\nAnd this path operation will:\n\n* Send the invoice to the client.\n* Collect the money from the client.\n* Send a notification back to the API user (the external developer), as a callback.\n * At this point is that the API will somehow send a POST request to the\n external API with the notification of the invoice event\n (e.g. "payment successful").',
97 "operationId": "create_invoice_invoices__post",
98 "parameters": [
99 {
100 "required": False,
101 "schema": {
102 "title": "Callback Url",
103 "anyOf": [
104 {
105 "type": "string",
106 "format": "uri",
107 "minLength": 1,
108 "maxLength": 2083,
109 },
110 {"type": "null"},
111 ],
112 },
113 "name": "callback_url",
114 "in": "query",
115 }
116 ],
117 "requestBody": {
118 "content": {
119 "application/json": {
120 "schema": {
121 "$ref": "#/components/schemas/Invoice"
122 }
123 }
124 },
125 "required": True,
126 },
127 "responses": {
128 "200": {
129 "description": "Successful Response",
130 "content": {"application/json": {"schema": {}}},
131 },
132 "422": {
133 "description": "Validation Error",
134 "content": {
135 "application/json": {
136 "schema": {
137 "$ref": "#/components/schemas/HTTPValidationError"
138 }
139 }
140 },
141 },
142 },
143 "callbacks": {
144 "event_callback": {
145 "{$callback_url}/events/{$request.body.title}": {
146 "get": {
147 "summary": "Event Callback",
148 "operationId": "event_callback__callback_url__events___request_body_title__get",
149 "requestBody": {
150 "required": True,
151 "content": {
152 "application/json": {
153 "schema": {
154 "$ref": "#/components/schemas/Event"
155 }
156 }
157 },
158 },
159 "responses": {
160 "200": {
161 "description": "Successful Response",
162 "content": {
163 "application/json": {
164 "schema": {}
165 }
166 },
167 },
168 "422": {
169 "description": "Validation Error",
170 "content": {
171 "application/json": {
172 "schema": {
173 "$ref": "#/components/schemas/HTTPValidationError"
174 }
175 }
176 },
177 },
178 },
179 }
180 }
181 },
182 "invoice_notification": {
183 "{$callback_url}/invoices/{$request.body.id}": {
184 "post": {
185 "summary": "Invoice Notification",
186 "operationId": "invoice_notification__callback_url__invoices___request_body_id__post",
187 "requestBody": {
188 "required": True,
189 "content": {
190 "application/json": {
191 "schema": {
192 "$ref": "#/components/schemas/InvoiceEvent"
193 }
194 }
195 },
196 },
197 "responses": {
198 "200": {
199 "description": "Successful Response",
200 "content": {
201 "application/json": {
202 "schema": {
203 "$ref": "#/components/schemas/InvoiceEventReceived"
204 }
205 }
206 },
207 },
208 "422": {
209 "description": "Validation Error",
210 "content": {
211 "application/json": {
212 "schema": {
213 "$ref": "#/components/schemas/HTTPValidationError"
214 }
215 }
216 },
217 },
218 },
219 }
220 }
221 },
222 },
223 }
224 }
225 },
226 "components": {
227 "schemas": {
228 "Event": {
229 "title": "Event",
230 "required": ["name", "total"],
231 "type": "object",
232 "properties": {
233 "name": {"title": "Name", "type": "string"},
234 "total": {"title": "Total", "type": "number"},
235 },
236 },
237 "HTTPValidationError": {
238 "title": "HTTPValidationError",
239 "type": "object",
240 "properties": {
241 "detail": {
242 "title": "Detail",
243 "type": "array",
244 "items": {
245 "$ref": "#/components/schemas/ValidationError"
246 },
247 }
248 },
249 },
250 "Invoice": {
251 "title": "Invoice",
252 "required": ["id", "customer", "total"],
253 "type": "object",
254 "properties": {
255 "id": {"title": "Id", "type": "string"},
256 "title": {
257 "title": "Title",
258 "anyOf": [{"type": "string"}, {"type": "null"}],
259 },
260 "customer": {"title": "Customer", "type": "string"},
261 "total": {"title": "Total", "type": "number"},
262 },
263 },
264 "InvoiceEvent": {
265 "title": "InvoiceEvent",
266 "required": ["description", "paid"],
267 "type": "object",
268 "properties": {
269 "description": {
270 "title": "Description",
271 "type": "string",
272 },
273 "paid": {"title": "Paid", "type": "boolean"},
274 },
275 },
276 "InvoiceEventReceived": {
277 "title": "InvoiceEventReceived",
278 "required": ["ok"],
279 "type": "object",
280 "properties": {"ok": {"title": "Ok", "type": "boolean"}},
281 },
282 "ValidationError": {
283 "title": "ValidationError",
284 "required": ["loc", "msg", "type"],
285 "type": "object",
286 "properties": {
287 "ctx": {"title": "Context", "type": "object"},
288 "input": {"title": "Input"},
289 "loc": {
290 "title": "Location",
291 "type": "array",
292 "items": {
293 "anyOf": [
294 {"type": "string"},
295 {"type": "integer"},
296 ]
297 },
298 },
299 "msg": {"title": "Message", "type": "string"},
300 "type": {"title": "Error Type", "type": "string"},
301 },
302 },
303 }
304 },
305 }
306 )