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