Coverage for tests/test_dependency_after_yield_raise.py: 100%

1from typing import Any 6 ctx1abcdef

2 

3import pytest 6 ctx1abcdef

4from fastapi import Depends, FastAPI, HTTPException 6 ctx1abcdef

5from fastapi.testclient import TestClient 6 ctx1abcdef

6from typing_extensions import Annotated 6 ctx1abcdef

7 

8 

9class CustomError(Exception): 6 ctx1abcdef

10 pass 6 ctx1abcdef

11 

12 

13def catching_dep() -> Any: 6 ctx1abcdef

14 try: 6 ctx1ghijkl

15 yield "s" 6 ctx1ghijkl

16 except CustomError as err: 6 ctx1ghijkl

17 raise HTTPException(status_code=418, detail="Session error") from err 6 ctx1ghijkl

18 

19 

20def broken_dep() -> Any: 6 ctx1abcdef

21 yield "s" 18 ctx1mynozpqArsBtuCvwDx

22 raise ValueError("Broken after yield") 18 ctx1mynozpqArsBtuCvwDx

23 

24 

25app = FastAPI() 6 ctx1abcdef

26 

27 

28@app.get("/catching") 6 ctx1abcdef

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

30 raise CustomError("Simulated error during streaming") 6 ctx1ghijkl

31 

32 

33@app.get("/broken") 6 ctx1abcdef

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

35 return {"message": "all good?"} 18 ctx1mynozpqArsBtuCvwDx

36 

37 

38client = TestClient(app) 6 ctx1abcdef

39 

40 

41def test_catching(): 6 ctx1abcdef

42 response = client.get("/catching") 6 ctx1ghijkl

43 assert response.status_code == 418 6 ctx1ghijkl

44 assert response.json() == {"detail": "Session error"} 6 ctx1ghijkl

45 

46 

47def test_broken_raise(): 6 ctx1abcdef

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

49 client.get("/broken") 6 ctx1yzABCD

50 

51 

52def test_broken_no_raise(): 6 ctx1abcdef

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: 6 ctx1moqsuw

60 response = client.get("/broken") 6 ctx1moqsuw

61 assert response.status_code == 200 6 ctx1moqsuw

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

63 

64 

65def test_broken_return_finishes(): 6 ctx1abcdef

66 client = TestClient(app, raise_server_exceptions=False) 6 ctx1nprtvx

67 response = client.get("/broken") 6 ctx1nprtvx

68 assert response.status_code == 200 6 ctx1nprtvx

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