Coverage for tests / test_dependency_after_yield_raise.py: 100%

1from typing import Annotated, Any 4 ctx1abcd

2 

3import pytest 4 ctx1abcd

4from fastapi import Depends, FastAPI, HTTPException 4 ctx1abcd

5from fastapi.testclient import TestClient 4 ctx1abcd

6 

7 

8class CustomError(Exception): 4 ctx1abcd

9 pass 4 ctx1abcd

10 

11 

12def catching_dep() -> Any: 4 ctx1abcd

13 try: 4 ctx1efgh

14 yield "s" 4 ctx1efgh

15 except CustomError as err: 4 ctx1efgh

16 raise HTTPException(status_code=418, detail="Session error") from err 4 ctx1efgh

17 

18 

19def broken_dep() -> Any: 4 ctx1abcd

20 yield "s" 11 ctx1ipjkqlrmnso

21 raise ValueError("Broken after yield") 11 ctx1ipjkqlrmnso

22 

23 

24app = FastAPI() 4 ctx1abcd

25 

26 

27@app.get("/catching") 4 ctx1abcd

28def catching(d: Annotated[str, Depends(catching_dep)]) -> Any: 4 ctx1abcd

29 raise CustomError("Simulated error during streaming") 4 ctx1efgh

30 

31 

32@app.get("/broken") 4 ctx1abcd

33def broken(d: Annotated[str, Depends(broken_dep)]) -> Any: 4 ctx1abcd

34 return {"message": "all good?"} 11 ctx1ipjkqlrmnso

35 

36 

37client = TestClient(app) 4 ctx1abcd

38 

39 

40def test_catching(): 4 ctx1abcd

41 response = client.get("/catching") 4 ctx1efgh

42 assert response.status_code == 418 4 ctx1efgh

43 assert response.json() == {"detail": "Session error"} 4 ctx1efgh

44 

45 

46def test_broken_raise(): 4 ctx1abcd

47 with pytest.raises(ValueError, match="Broken after yield"): 4 ctx1pqrs

48 client.get("/broken") 4 ctx1pqrs

49 

50 

51def test_broken_no_raise(): 4 ctx1abcd

52 """ 

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

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

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

56 in the server logs. 

57 """ 

58 with TestClient(app, raise_server_exceptions=False) as client: 4 ctx1ikln

59 response = client.get("/broken") 4 ctx1ikln

60 assert response.status_code == 200 4 ctx1ikln

61 assert response.json() == {"message": "all good?"} 4 ctx1ikln

62 

63 

64def test_broken_return_finishes(): 4 ctx1abcd

65 client = TestClient(app, raise_server_exceptions=False) 4 ctx1jtmo

66 response = client.get("/broken") 4 ctx1jtmo

67 assert response.status_code == 200 4 ctx1jtmo

68 assert response.json() == {"message": "all good?"} 4 ctx1jtmo