Coverage for tests/test_dependency_after_yield_raise.py: 100%

1from typing import Any 7 ctx1abcdefg

2 

3import pytest 7 ctx1abcdefg

4from fastapi import Depends, FastAPI, HTTPException 7 ctx1abcdefg

5from fastapi.testclient import TestClient 7 ctx1abcdefg

6from typing_extensions import Annotated 7 ctx1abcdefg

7 

8 

9class CustomError(Exception): 7 ctx1abcdefg

10 pass 7 ctx1abcdefg

11 

12 

13def catching_dep() -> Any: 7 ctx1abcdefg

14 try: 7 ctx1hijklmn

15 yield "s" 7 ctx1hijklmn

16 except CustomError as err: 7 ctx1hijklmn

17 raise HTTPException(status_code=418, detail="Session error") from err 7 ctx1hijklmn

18 

19 

20def broken_dep() -> Any: 7 ctx1abcdefg

21 yield "s" 21 ctx1oCpqDrsEtuFvwGxyHzAIB

22 raise ValueError("Broken after yield") 21 ctx1oCpqDrsEtuFvwGxyHzAIB

23 

24 

25app = FastAPI() 7 ctx1abcdefg

26 

27 

28@app.get("/catching") 7 ctx1abcdefg

29def catching(d: Annotated[str, Depends(catching_dep)]) -> Any: 7 ctx1abcdefg

30 raise CustomError("Simulated error during streaming") 7 ctx1hijklmn

31 

32 

33@app.get("/broken") 7 ctx1abcdefg

34def broken(d: Annotated[str, Depends(broken_dep)]) -> Any: 7 ctx1abcdefg

35 return {"message": "all good?"} 21 ctx1oCpqDrsEtuFvwGxyHzAIB

36 

37 

38client = TestClient(app) 7 ctx1abcdefg

39 

40 

41def test_catching(): 7 ctx1abcdefg

42 response = client.get("/catching") 7 ctx1hijklmn

43 assert response.status_code == 418 7 ctx1hijklmn

44 assert response.json() == {"detail": "Session error"} 7 ctx1hijklmn

45 

46 

47def test_broken_raise(): 7 ctx1abcdefg

48 with pytest.raises(ValueError, match="Broken after yield"): 7 ctx1CDEFGHI

49 client.get("/broken") 7 ctx1CDEFGHI

50 

51 

52def test_broken_no_raise(): 7 ctx1abcdefg

53 """ 

54 When a dependency with yield raises after the yield (not in an except), the 

55 response is already "successfully" sent back to the client, but there's still 

56 an error in the server afterwards, an exception is raised and captured or shown 

57 in the server logs. 

58 """ 

59 with TestClient(app, raise_server_exceptions=False) as client: 7 ctx1oqsuwyA

60 response = client.get("/broken") 7 ctx1oqsuwyA

61 assert response.status_code == 200 7 ctx1oqsuwyA

62 assert response.json() == {"message": "all good?"} 7 ctx1oqsuwyA

63 

64 

65def test_broken_return_finishes(): 7 ctx1abcdefg

66 client = TestClient(app, raise_server_exceptions=False) 7 ctx1prtvxzB

67 response = client.get("/broken") 7 ctx1prtvxzB

68 assert response.status_code == 200 7 ctx1prtvxzB

69 assert response.json() == {"message": "all good?"} 7 ctx1prtvxzB