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