Coverage for tests/test_dependency_contextmanager.py: 100%
271 statements
« prev ^ index » next coverage.py v7.6.1, created at 2025-01-13 13:38 +0000
1import json 1abcde
2from typing import Dict 1abcde
4import pytest 1abcde
5from fastapi import BackgroundTasks, Depends, FastAPI 1abcde
6from fastapi.responses import StreamingResponse 1abcde
7from fastapi.testclient import TestClient 1abcde
9app = FastAPI() 1abcde
10state = { 1abcde
11 "/async": "asyncgen not started",
12 "/sync": "generator not started",
13 "/async_raise": "asyncgen raise not started",
14 "/sync_raise": "generator raise not started",
15 "context_a": "not started a",
16 "context_b": "not started b",
17 "bg": "not set",
18 "sync_bg": "not set",
19}
21errors = [] 1abcde
24async def get_state(): 1abcde
25 return state 2: Z z { f p A | 0 B qbk q C ; 1 D } ~ 2 E rb= , 3 abg r F bb- 4 sbl s G ? . 5 cbdb/ 6 tb@ 7 H ebh t I fb8 J ubm u K [ 9 L gbhb! M vb] # N ibi v O jb$ P wbn w Q ^ % R kblb' S xb_ ( T mbj x U nb) V ybo y W ` * X obpb+ Y zb
28class AsyncDependencyError(Exception): 1abcde
29 pass 1abcde
32class SyncDependencyError(Exception): 1abcde
33 pass 1abcde
36class OtherDependencyError(Exception): 1abcde
37 pass 1abcde
40async def asyncgen_state(state: Dict[str, str] = Depends(get_state)): 1abcde
41 state["/async"] = "asyncgen started" 2{ qbabsbebubibwbmbyb
42 yield state["/async"] 2{ qbabsbebubibwbmbyb
43 state["/async"] = "asyncgen completed" 2{ qbabsbebubibwbmbyb
46def generator_state(state: Dict[str, str] = Depends(get_state)): 1abcde
47 state["/sync"] = "generator started" 2} rbcbtbgbvbkbxbobzb
48 yield state["/sync"] 2} rbcbtbgbvbkbxbobzb
49 state["/sync"] = "generator completed" 2} rbcbtbgbvbkbxbobzb
52async def asyncgen_state_try(state: Dict[str, str] = Depends(get_state)): 1abcde
53 state["/async_raise"] = "asyncgen raise started" 2: Z z | 0 B = , 3 bb- 4 @ 7 H fb8 J ] # N jb$ P _ ( T nb) V
54 try: 2: Z z | 0 B = , 3 bb- 4 @ 7 H fb8 J ] # N jb$ P _ ( T nb) V
55 yield state["/async_raise"] 2: Z z | 0 B = , 3 bb- 4 @ 7 H fb8 J ] # N jb$ P _ ( T nb) V
56 except AsyncDependencyError: 2: Z z | 0 B = , 3 bb- 4 @ 7 H fb8 J ] # N jb$ P _ ( T nb) V
57 errors.append("/async_raise") 1Zz0B,3-47H8J#N$P(T)V
58 raise 1Zz0B,3-47H8J#N$P(T)V
59 finally:
60 state["/async_raise"] = "asyncgen raise finalized" 2: Z z | 0 B = bb@ 7 H fb8 J ] # N jb$ P _ ( T nb) V
63def generator_state_try(state: Dict[str, str] = Depends(get_state)): 1abcde
64 state["/sync_raise"] = "generator raise started" 2; 1 D ~ 2 E ? . 5 db/ 6 [ 9 L hb! M ^ % R lb' S ` * X pb+ Y
65 try: 2; 1 D ~ 2 E ? . 5 db/ 6 [ 9 L hb! M ^ % R lb' S ` * X pb+ Y
66 yield state["/sync_raise"] 2; 1 D ~ 2 E ? . 5 db/ 6 [ 9 L hb! M ^ % R lb' S ` * X pb+ Y
67 except SyncDependencyError: 2; 1 D ~ 2 E ? . 5 db/ 6 [ 9 L hb! M ^ % R lb' S ` * X pb+ Y
68 errors.append("/sync_raise") 11D2E.5/69L!M%R'S*X+Y
69 raise 11D2E.5/69L!M%R'S*X+Y
70 finally:
71 state["/sync_raise"] = "generator raise finalized" 2; 1 D ~ 2 E ? db[ 9 L hb! M ^ % R lb' S ` * X pb+ Y
74async def context_a(state: dict = Depends(get_state)): 1abcde
75 state["context_a"] = "started a" 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
76 try: 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
77 yield state 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
78 finally:
79 state["context_a"] = "finished a" 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
82async def context_b(state: dict = Depends(context_a)): 1abcde
83 state["context_b"] = "started b" 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
84 try: 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
85 yield state 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
86 finally:
87 state["context_b"] = f"finished b with a: {state['context_a']}" 1fpAkqCgrFlsGhtImuKivOnwQjxUoyW
90@app.get("/async") 1abcde
91async def get_async(state: str = Depends(asyncgen_state)): 1abcde
92 return state 2{ abebibmb
95@app.get("/sync") 1abcde
96async def get_sync(state: str = Depends(generator_state)): 1abcde
97 return state 2} cbgbkbob
100@app.get("/async_raise") 1abcde
101async def get_async_raise(state: str = Depends(asyncgen_state_try)): 1abcde
102 assert state == "asyncgen raise started" 1Zz,37H#N(T
103 raise AsyncDependencyError() 1Zz,37H#N(T
106@app.get("/sync_raise") 1abcde
107async def get_sync_raise(state: str = Depends(generator_state_try)): 1abcde
108 assert state == "generator raise started" 11D.59L%R*X
109 raise SyncDependencyError() 11D.59L%R*X
112@app.get("/async_raise_other") 1abcde
113async def get_async_raise_other(state: str = Depends(asyncgen_state_try)): 1abcde
114 assert state == "asyncgen raise started" 1:=@]_
115 raise OtherDependencyError() 1:=@]_
118@app.get("/sync_raise_other") 1abcde
119async def get_sync_raise_other(state: str = Depends(generator_state_try)): 1abcde
120 assert state == "generator raise started" 1;?[^`
121 raise OtherDependencyError() 1;?[^`
124@app.get("/context_b") 1abcde
125async def get_context_b(state: dict = Depends(context_b)): 1abcde
126 return state 1prtvx
129@app.get("/context_b_raise") 1abcde
130async def get_context_b_raise(state: dict = Depends(context_b)): 1abcde
131 assert state["context_b"] == "started b" 1AFIOU
132 assert state["context_a"] == "started a" 1AFIOU
133 raise OtherDependencyError() 1AFIOU
136@app.get("/context_b_bg") 1abcde
137async def get_context_b_bg(tasks: BackgroundTasks, state: dict = Depends(context_b)): 1abcde
138 async def bg(state: dict): 1fghij
139 state["bg"] = f"bg set - b: {state['context_b']} - a: {state['context_a']}" 1fghij
141 tasks.add_task(bg, state) 1fghij
142 return state 1fghij
145# Sync versions
148@app.get("/sync_async") 1abcde
149def get_sync_async(state: str = Depends(asyncgen_state)): 1abcde
150 return state 2qbsbubwbyb
153@app.get("/sync_sync") 1abcde
154def get_sync_sync(state: str = Depends(generator_state)): 1abcde
155 return state 2rbtbvbxbzb
158@app.get("/sync_async_raise") 1abcde
159def get_sync_async_raise(state: str = Depends(asyncgen_state_try)): 1abcde
160 assert state == "asyncgen raise started" 10B-48J$P)V
161 raise AsyncDependencyError() 10B-48J$P)V
164@app.get("/sync_sync_raise") 1abcde
165def get_sync_sync_raise(state: str = Depends(generator_state_try)): 1abcde
166 assert state == "generator raise started" 12E/6!M'S+Y
167 raise SyncDependencyError() 12E/6!M'S+Y
170@app.get("/sync_async_raise_other") 1abcde
171def get_sync_async_raise_other(state: str = Depends(asyncgen_state_try)): 1abcde
172 assert state == "asyncgen raise started" 2| bbfbjbnb
173 raise OtherDependencyError() 2| bbfbjbnb
176@app.get("/sync_sync_raise_other") 1abcde
177def get_sync_sync_raise_other(state: str = Depends(generator_state_try)): 1abcde
178 assert state == "generator raise started" 2~ dbhblbpb
179 raise OtherDependencyError() 2~ dbhblbpb
182@app.get("/sync_context_b") 1abcde
183def get_sync_context_b(state: dict = Depends(context_b)): 1abcde
184 return state 1qsuwy
187@app.get("/sync_context_b_raise") 1abcde
188def get_sync_context_b_raise(state: dict = Depends(context_b)): 1abcde
189 assert state["context_b"] == "started b" 1CGKQW
190 assert state["context_a"] == "started a" 1CGKQW
191 raise OtherDependencyError() 1CGKQW
194@app.get("/sync_context_b_bg") 1abcde
195async def get_sync_context_b_bg( 1abcde
196 tasks: BackgroundTasks, state: dict = Depends(context_b)
197):
198 async def bg(state: dict): 1klmno
199 state["sync_bg"] = ( 1klmno
200 f"sync_bg set - b: {state['context_b']} - a: {state['context_a']}"
201 )
203 tasks.add_task(bg, state) 1klmno
204 return state 1klmno
207@app.middleware("http") 1abcde
208async def middleware(request, call_next): 1abcde
209 response: StreamingResponse = await call_next(request) 2: Z z { f p A | 0 B qbk q C ; 1 D } ~ 2 E rb= , 3 abg r F bb- 4 sbl s G ? . 5 cbdb/ 6 tb@ 7 H ebh t I fb8 J ubm u K [ 9 L gbhb! M vb] # N ibi v O jb$ P wbn w Q ^ % R kblb' S xb_ ( T mbj x U nb) V ybo y W ` * X obpb+ Y zb
210 response.headers["x-state"] = json.dumps(state.copy()) 2{ f p qbk q } rbabg r sbl s cbtbebh t ubm u gbvbibi v wbn w kbxbmbj x ybo y obzb
211 return response 2{ f p qbk q } rbabg r sbl s cbtbebh t ubm u gbvbibi v wbn w kbxbmbj x ybo y obzb
214client = TestClient(app) 1abcde
217def test_async_state(): 1abcde
218 assert state["/async"] == "asyncgen not started" 2{ abebibmb
219 response = client.get("/async") 2{ abebibmb
220 assert response.status_code == 200, response.text 2{ abebibmb
221 assert response.json() == "asyncgen started" 2{ abebibmb
222 assert state["/async"] == "asyncgen completed" 2{ abebibmb
225def test_sync_state(): 1abcde
226 assert state["/sync"] == "generator not started" 2} cbgbkbob
227 response = client.get("/sync") 2} cbgbkbob
228 assert response.status_code == 200, response.text 2} cbgbkbob
229 assert response.json() == "generator started" 2} cbgbkbob
230 assert state["/sync"] == "generator completed" 2} cbgbkbob
233def test_async_raise_other(): 1abcde
234 assert state["/async_raise"] == "asyncgen raise not started" 1:=@]_
235 with pytest.raises(OtherDependencyError): 1:=@]_
236 client.get("/async_raise_other") 1:=@]_
237 assert state["/async_raise"] == "asyncgen raise finalized" 1:=@]_
238 assert "/async_raise" not in errors 1:=@]_
241def test_sync_raise_other(): 1abcde
242 assert state["/sync_raise"] == "generator raise not started" 1;?[^`
243 with pytest.raises(OtherDependencyError): 1;?[^`
244 client.get("/sync_raise_other") 1;?[^`
245 assert state["/sync_raise"] == "generator raise finalized" 1;?[^`
246 assert "/sync_raise" not in errors 1;?[^`
249def test_async_raise_raises(): 1abcde
250 with pytest.raises(AsyncDependencyError): 1Z,7#(
251 client.get("/async_raise") 1Z,7#(
252 assert state["/async_raise"] == "asyncgen raise finalized" 1Z,7#(
253 assert "/async_raise" in errors 1Z,7#(
254 errors.clear() 1Z,7#(
257def test_async_raise_server_error(): 1abcde
258 client = TestClient(app, raise_server_exceptions=False) 1z3HNT
259 response = client.get("/async_raise") 1z3HNT
260 assert response.status_code == 500, response.text 1z3HNT
261 assert state["/async_raise"] == "asyncgen raise finalized" 1z3HNT
262 assert "/async_raise" in errors 1z3HNT
263 errors.clear() 1z3HNT
266def test_context_b(): 1abcde
267 response = client.get("/context_b") 1prtvx
268 data = response.json() 1prtvx
269 assert data["context_b"] == "started b" 1prtvx
270 assert data["context_a"] == "started a" 1prtvx
271 assert state["context_b"] == "finished b with a: started a" 1prtvx
272 assert state["context_a"] == "finished a" 1prtvx
275def test_context_b_raise(): 1abcde
276 with pytest.raises(OtherDependencyError): 1AFIOU
277 client.get("/context_b_raise") 1AFIOU
278 assert state["context_b"] == "finished b with a: started a" 1AFIOU
279 assert state["context_a"] == "finished a" 1AFIOU
282def test_background_tasks(): 1abcde
283 response = client.get("/context_b_bg") 1fghij
284 data = response.json() 1fghij
285 assert data["context_b"] == "started b" 1fghij
286 assert data["context_a"] == "started a" 1fghij
287 assert data["bg"] == "not set" 1fghij
288 middleware_state = json.loads(response.headers["x-state"]) 1fghij
289 assert middleware_state["context_b"] == "finished b with a: started a" 1fghij
290 assert middleware_state["context_a"] == "finished a" 1fghij
291 assert middleware_state["bg"] == "not set" 1fghij
292 assert state["context_b"] == "finished b with a: started a" 1fghij
293 assert state["context_a"] == "finished a" 1fghij
294 assert state["bg"] == "bg set - b: finished b with a: started a - a: finished a" 1fghij
297def test_sync_raise_raises(): 1abcde
298 with pytest.raises(SyncDependencyError): 11.9%*
299 client.get("/sync_raise") 11.9%*
300 assert state["/sync_raise"] == "generator raise finalized" 11.9%*
301 assert "/sync_raise" in errors 11.9%*
302 errors.clear() 11.9%*
305def test_sync_raise_server_error(): 1abcde
306 client = TestClient(app, raise_server_exceptions=False) 1D5LRX
307 response = client.get("/sync_raise") 1D5LRX
308 assert response.status_code == 500, response.text 1D5LRX
309 assert state["/sync_raise"] == "generator raise finalized" 1D5LRX
310 assert "/sync_raise" in errors 1D5LRX
311 errors.clear() 1D5LRX
314def test_sync_async_state(): 1abcde
315 response = client.get("/sync_async") 2qbsbubwbyb
316 assert response.status_code == 200, response.text 2qbsbubwbyb
317 assert response.json() == "asyncgen started" 2qbsbubwbyb
318 assert state["/async"] == "asyncgen completed" 2qbsbubwbyb
321def test_sync_sync_state(): 1abcde
322 response = client.get("/sync_sync") 2rbtbvbxbzb
323 assert response.status_code == 200, response.text 2rbtbvbxbzb
324 assert response.json() == "generator started" 2rbtbvbxbzb
325 assert state["/sync"] == "generator completed" 2rbtbvbxbzb
328def test_sync_async_raise_other(): 1abcde
329 with pytest.raises(OtherDependencyError): 2| bbfbjbnb
330 client.get("/sync_async_raise_other") 2| bbfbjbnb
331 assert state["/async_raise"] == "asyncgen raise finalized" 2| bbfbjbnb
332 assert "/async_raise" not in errors 2| bbfbjbnb
335def test_sync_sync_raise_other(): 1abcde
336 with pytest.raises(OtherDependencyError): 2~ dbhblbpb
337 client.get("/sync_sync_raise_other") 2~ dbhblbpb
338 assert state["/sync_raise"] == "generator raise finalized" 2~ dbhblbpb
339 assert "/sync_raise" not in errors 2~ dbhblbpb
342def test_sync_async_raise_raises(): 1abcde
343 with pytest.raises(AsyncDependencyError): 10-8$)
344 client.get("/sync_async_raise") 10-8$)
345 assert state["/async_raise"] == "asyncgen raise finalized" 10-8$)
346 assert "/async_raise" in errors 10-8$)
347 errors.clear() 10-8$)
350def test_sync_async_raise_server_error(): 1abcde
351 client = TestClient(app, raise_server_exceptions=False) 1B4JPV
352 response = client.get("/sync_async_raise") 1B4JPV
353 assert response.status_code == 500, response.text 1B4JPV
354 assert state["/async_raise"] == "asyncgen raise finalized" 1B4JPV
355 assert "/async_raise" in errors 1B4JPV
356 errors.clear() 1B4JPV
359def test_sync_sync_raise_raises(): 1abcde
360 with pytest.raises(SyncDependencyError): 12/!'+
361 client.get("/sync_sync_raise") 12/!'+
362 assert state["/sync_raise"] == "generator raise finalized" 12/!'+
363 assert "/sync_raise" in errors 12/!'+
364 errors.clear() 12/!'+
367def test_sync_sync_raise_server_error(): 1abcde
368 client = TestClient(app, raise_server_exceptions=False) 1E6MSY
369 response = client.get("/sync_sync_raise") 1E6MSY
370 assert response.status_code == 500, response.text 1E6MSY
371 assert state["/sync_raise"] == "generator raise finalized" 1E6MSY
372 assert "/sync_raise" in errors 1E6MSY
373 errors.clear() 1E6MSY
376def test_sync_context_b(): 1abcde
377 response = client.get("/sync_context_b") 1qsuwy
378 data = response.json() 1qsuwy
379 assert data["context_b"] == "started b" 1qsuwy
380 assert data["context_a"] == "started a" 1qsuwy
381 assert state["context_b"] == "finished b with a: started a" 1qsuwy
382 assert state["context_a"] == "finished a" 1qsuwy
385def test_sync_context_b_raise(): 1abcde
386 with pytest.raises(OtherDependencyError): 1CGKQW
387 client.get("/sync_context_b_raise") 1CGKQW
388 assert state["context_b"] == "finished b with a: started a" 1CGKQW
389 assert state["context_a"] == "finished a" 1CGKQW
392def test_sync_background_tasks(): 1abcde
393 response = client.get("/sync_context_b_bg") 1klmno
394 data = response.json() 1klmno
395 assert data["context_b"] == "started b" 1klmno
396 assert data["context_a"] == "started a" 1klmno
397 assert data["sync_bg"] == "not set" 1klmno
398 assert state["context_b"] == "finished b with a: started a" 1klmno
399 assert state["context_a"] == "finished a" 1klmno
400 assert ( 1klmno
401 state["sync_bg"]
402 == "sync_bg set - b: finished b with a: started a - a: finished a"
403 )