Coverage for tests/test_router_events.py: 100%

1from contextlib import asynccontextmanager 5 ctx1abcde

2from typing import AsyncGenerator, Dict 5 ctx1abcde

3 

4import pytest 5 ctx1abcde

5from fastapi import APIRouter, FastAPI 5 ctx1abcde

6from fastapi.testclient import TestClient 5 ctx1abcde

7from pydantic import BaseModel 5 ctx1abcde

8 

9 

10class State(BaseModel): 5 ctx1abcde

11 app_startup: bool = False 5 ctx1abcde

12 app_shutdown: bool = False 5 ctx1abcde

13 router_startup: bool = False 5 ctx1abcde

14 router_shutdown: bool = False 5 ctx1abcde

15 sub_router_startup: bool = False 5 ctx1abcde

16 sub_router_shutdown: bool = False 5 ctx1abcde

17 

18 

19@pytest.fixture 5 ctx1abcde

20def state() -> State: 5 ctx1abcde

21 return State() 5 ctx1abcde

22 

23 

24@pytest.mark.filterwarnings( 5 ctx1abcde

25 r"ignore:\s*on_event is deprecated, use lifespan event handlers instead.*:DeprecationWarning" 

26) 

27def test_router_events(state: State) -> None: 5 ctx1abcde

28 app = FastAPI() 5 ctx1abcde

29 

30 @app.get("/") 5 ctx1abcde

31 def main() -> Dict[str, str]: 5 ctx1abcde

32 return {"message": "Hello World"} 5 ctx1abcde

33 

34 @app.on_event("startup") 5 ctx1abcde

35 def app_startup() -> None: 5 ctx1abcde

36 state.app_startup = True 5 ctx1abcde

37 

38 @app.on_event("shutdown") 5 ctx1abcde

39 def app_shutdown() -> None: 5 ctx1abcde

40 state.app_shutdown = True 5 ctx1abcde

41 

42 router = APIRouter() 5 ctx1abcde

43 

44 @router.on_event("startup") 5 ctx1abcde

45 def router_startup() -> None: 5 ctx1abcde

46 state.router_startup = True 5 ctx1abcde

47 

48 @router.on_event("shutdown") 5 ctx1abcde

49 def router_shutdown() -> None: 5 ctx1abcde

50 state.router_shutdown = True 5 ctx1abcde

51 

52 sub_router = APIRouter() 5 ctx1abcde

53 

54 @sub_router.on_event("startup") 5 ctx1abcde

55 def sub_router_startup() -> None: 5 ctx1abcde

56 state.sub_router_startup = True 5 ctx1abcde

57 

58 @sub_router.on_event("shutdown") 5 ctx1abcde

59 def sub_router_shutdown() -> None: 5 ctx1abcde

60 state.sub_router_shutdown = True 5 ctx1abcde

61 

62 router.include_router(sub_router) 5 ctx1abcde

63 app.include_router(router) 5 ctx1abcde

64 

65 assert state.app_startup is False 5 ctx1abcde

66 assert state.router_startup is False 5 ctx1abcde

67 assert state.sub_router_startup is False 5 ctx1abcde

68 assert state.app_shutdown is False 5 ctx1abcde

69 assert state.router_shutdown is False 5 ctx1abcde

70 assert state.sub_router_shutdown is False 5 ctx1abcde

71 with TestClient(app) as client: 5 ctx1abcde

72 assert state.app_startup is True 5 ctx1abcde

73 assert state.router_startup is True 5 ctx1abcde

74 assert state.sub_router_startup is True 5 ctx1abcde

75 assert state.app_shutdown is False 5 ctx1abcde

76 assert state.router_shutdown is False 5 ctx1abcde

77 assert state.sub_router_shutdown is False 5 ctx1abcde

78 response = client.get("/") 5 ctx1abcde

79 assert response.status_code == 200, response.text 5 ctx1abcde

80 assert response.json() == {"message": "Hello World"} 5 ctx1abcde

81 assert state.app_startup is True 5 ctx1abcde

82 assert state.router_startup is True 5 ctx1abcde

83 assert state.sub_router_startup is True 5 ctx1abcde

84 assert state.app_shutdown is True 5 ctx1abcde

85 assert state.router_shutdown is True 5 ctx1abcde

86 assert state.sub_router_shutdown is True 5 ctx1abcde

87 

88 

89def test_app_lifespan_state(state: State) -> None: 5 ctx1abcde

90 @asynccontextmanager 5 ctx1abcde

91 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 5 ctx1abcde

92 state.app_startup = True 5 ctx1abcde

93 yield 5 ctx1abcde

94 state.app_shutdown = True 5 ctx1abcde

95 

96 app = FastAPI(lifespan=lifespan) 5 ctx1abcde

97 

98 @app.get("/") 5 ctx1abcde

99 def main() -> Dict[str, str]: 5 ctx1abcde

100 return {"message": "Hello World"} 5 ctx1abcde

101 

102 assert state.app_startup is False 5 ctx1abcde

103 assert state.app_shutdown is False 5 ctx1abcde

104 with TestClient(app) as client: 5 ctx1abcde

105 assert state.app_startup is True 5 ctx1abcde

106 assert state.app_shutdown is False 5 ctx1abcde

107 response = client.get("/") 5 ctx1abcde

108 assert response.status_code == 200, response.text 5 ctx1abcde

109 assert response.json() == {"message": "Hello World"} 5 ctx1abcde

110 assert state.app_startup is True 5 ctx1abcde

111 assert state.app_shutdown is True 5 ctx1abcde