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: 3 ctx1efg

14 yield "s" 3 ctx1efg

15 except CustomError as err: 3 ctx1efg

16 raise HTTPException(status_code=418, detail="Session error") from err 3 ctx1efg

17 

18 

19def broken_dep() -> Any: 4 ctx1abcd

20 yield "s" 9 ctx1hnijoklpm

21 raise ValueError("Broken after yield") 9 ctx1hnijoklpm

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") 3 ctx1efg

30 

31 

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

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

34 return {"message": "all good?"} 9 ctx1hnijoklpm

35 

36 

37client = TestClient(app) 4 ctx1abcd

38 

39 

40def test_catching(): 4 ctx1abcd

41 response = client.get("/catching") 3 ctx1efg

42 assert response.status_code == 418 3 ctx1efg

43 assert response.json() == {"detail": "Session error"} 3 ctx1efg

44 

45 

46def test_broken_raise(): 4 ctx1abcd

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

48 client.get("/broken") 3 ctx1nop

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: 3 ctx1hjl

59 response = client.get("/broken") 3 ctx1hjl

60 assert response.status_code == 200 3 ctx1hjl

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

62 

63 

64def test_broken_return_finishes(): 4 ctx1abcd

65 client = TestClient(app, raise_server_exceptions=False) 3 ctx1ikm

66 response = client.get("/broken") 3 ctx1ikm

67 assert response.status_code == 200 3 ctx1ikm

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