Coverage for tests/test_router_events.py: 100%

1from contextlib import asynccontextmanager 7 ctx1opqrstu

2from typing import AsyncGenerator, Dict, Union 7 ctx1opqrstu

3 

4import pytest 7 ctx1opqrstu

5from fastapi import APIRouter, FastAPI, Request 7 ctx1opqrstu

6from fastapi.testclient import TestClient 7 ctx1opqrstu

7from pydantic import BaseModel 7 ctx1opqrstu

8 

9 

10class State(BaseModel): 7 ctx1opqrstu

11 app_startup: bool = False 7 ctx1opqrstu

12 app_shutdown: bool = False 7 ctx1opqrstu

13 router_startup: bool = False 7 ctx1opqrstu

14 router_shutdown: bool = False 7 ctx1opqrstu

15 sub_router_startup: bool = False 7 ctx1opqrstu

16 sub_router_shutdown: bool = False 7 ctx1opqrstu

17 

18 

19@pytest.fixture 7 ctx1opqrstu

20def state() -> State: 7 ctx1opqrstu

21 return State() 7 ctx1opqrstu

22 

23 

24@pytest.mark.filterwarnings( 7 ctx1opqrstu

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

26) 

27def test_router_events(state: State) -> None: 7 ctx1opqrstu

28 app = FastAPI() 7 ctx1abcdefg

29 

30 @app.get("/") 7 ctx1abcdefg

31 def main() -> Dict[str, str]: 7 ctx1abcdefg

32 return {"message": "Hello World"} 7 ctx1abcdefg

33 

34 @app.on_event("startup") 7 ctx1abcdefg

35 def app_startup() -> None: 7 ctx1abcdefg

36 state.app_startup = True 7 ctx1abcdefg

37 

38 @app.on_event("shutdown") 7 ctx1abcdefg

39 def app_shutdown() -> None: 7 ctx1abcdefg

40 state.app_shutdown = True 7 ctx1abcdefg

41 

42 router = APIRouter() 7 ctx1abcdefg

43 

44 @router.on_event("startup") 7 ctx1abcdefg

45 def router_startup() -> None: 7 ctx1abcdefg

46 state.router_startup = True 7 ctx1abcdefg

47 

48 @router.on_event("shutdown") 7 ctx1abcdefg

49 def router_shutdown() -> None: 7 ctx1abcdefg

50 state.router_shutdown = True 7 ctx1abcdefg

51 

52 sub_router = APIRouter() 7 ctx1abcdefg

53 

54 @sub_router.on_event("startup") 7 ctx1abcdefg

55 def sub_router_startup() -> None: 7 ctx1abcdefg

56 state.sub_router_startup = True 7 ctx1abcdefg

57 

58 @sub_router.on_event("shutdown") 7 ctx1abcdefg

59 def sub_router_shutdown() -> None: 7 ctx1abcdefg

60 state.sub_router_shutdown = True 7 ctx1abcdefg

61 

62 router.include_router(sub_router) 7 ctx1abcdefg

63 app.include_router(router) 7 ctx1abcdefg

64 

65 assert state.app_startup is False 7 ctx1abcdefg

66 assert state.router_startup is False 7 ctx1abcdefg

67 assert state.sub_router_startup is False 7 ctx1abcdefg

68 assert state.app_shutdown is False 7 ctx1abcdefg

69 assert state.router_shutdown is False 7 ctx1abcdefg

70 assert state.sub_router_shutdown is False 7 ctx1abcdefg

71 with TestClient(app) as client: 7 ctx1abcdefg

72 assert state.app_startup is True 7 ctx1abcdefg

73 assert state.router_startup is True 7 ctx1abcdefg

74 assert state.sub_router_startup is True 7 ctx1abcdefg

75 assert state.app_shutdown is False 7 ctx1abcdefg

76 assert state.router_shutdown is False 7 ctx1abcdefg

77 assert state.sub_router_shutdown is False 7 ctx1abcdefg

78 response = client.get("/") 7 ctx1abcdefg

79 assert response.status_code == 200, response.text 7 ctx1abcdefg

80 assert response.json() == {"message": "Hello World"} 7 ctx1abcdefg

81 assert state.app_startup is True 7 ctx1abcdefg

82 assert state.router_startup is True 7 ctx1abcdefg

83 assert state.sub_router_startup is True 7 ctx1abcdefg

84 assert state.app_shutdown is True 7 ctx1abcdefg

85 assert state.router_shutdown is True 7 ctx1abcdefg

86 assert state.sub_router_shutdown is True 7 ctx1abcdefg

87 

88 

89def test_app_lifespan_state(state: State) -> None: 7 ctx1opqrstu

90 @asynccontextmanager 7 ctx1vwxyzAB

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

92 state.app_startup = True 7 ctx1vwxyzAB

93 yield 7 ctx1vwxyzAB

94 state.app_shutdown = True 7 ctx1vwxyzAB

95 

96 app = FastAPI(lifespan=lifespan) 7 ctx1vwxyzAB

97 

98 @app.get("/") 7 ctx1vwxyzAB

99 def main() -> Dict[str, str]: 7 ctx1vwxyzAB

100 return {"message": "Hello World"} 7 ctx1vwxyzAB

101 

102 assert state.app_startup is False 7 ctx1vwxyzAB

103 assert state.app_shutdown is False 7 ctx1vwxyzAB

104 with TestClient(app) as client: 7 ctx1vwxyzAB

105 assert state.app_startup is True 7 ctx1vwxyzAB

106 assert state.app_shutdown is False 7 ctx1vwxyzAB

107 response = client.get("/") 7 ctx1vwxyzAB

108 assert response.status_code == 200, response.text 7 ctx1vwxyzAB

109 assert response.json() == {"message": "Hello World"} 7 ctx1vwxyzAB

110 assert state.app_startup is True 7 ctx1vwxyzAB

111 assert state.app_shutdown is True 7 ctx1vwxyzAB

112 

113 

114def test_router_nested_lifespan_state(state: State) -> None: 7 ctx1opqrstu

115 @asynccontextmanager 7 ctx1hijklmn

116 async def lifespan(app: FastAPI) -> AsyncGenerator[Dict[str, bool], None]: 7 ctx1hijklmn

117 state.app_startup = True 7 ctx1hijklmn

118 yield {"app": True} 7 ctx1hijklmn

119 state.app_shutdown = True 7 ctx1hijklmn

120 

121 @asynccontextmanager 7 ctx1hijklmn

122 async def router_lifespan(app: FastAPI) -> AsyncGenerator[Dict[str, bool], None]: 7 ctx1hijklmn

123 state.router_startup = True 7 ctx1hijklmn

124 yield {"router": True} 7 ctx1hijklmn

125 state.router_shutdown = True 7 ctx1hijklmn

126 

127 @asynccontextmanager 7 ctx1hijklmn

128 async def subrouter_lifespan(app: FastAPI) -> AsyncGenerator[Dict[str, bool], None]: 7 ctx1hijklmn

129 state.sub_router_startup = True 7 ctx1hijklmn

130 yield {"sub_router": True} 7 ctx1hijklmn

131 state.sub_router_shutdown = True 7 ctx1hijklmn

132 

133 sub_router = APIRouter(lifespan=subrouter_lifespan) 7 ctx1hijklmn

134 

135 router = APIRouter(lifespan=router_lifespan) 7 ctx1hijklmn

136 router.include_router(sub_router) 7 ctx1hijklmn

137 

138 app = FastAPI(lifespan=lifespan) 7 ctx1hijklmn

139 app.include_router(router) 7 ctx1hijklmn

140 

141 @app.get("/") 7 ctx1hijklmn

142 def main(request: Request) -> Dict[str, str]: 7 ctx1hijklmn

143 assert request.state.app 7 ctx1hijklmn

144 assert request.state.router 7 ctx1hijklmn

145 assert request.state.sub_router 7 ctx1hijklmn

146 return {"message": "Hello World"} 7 ctx1hijklmn

147 

148 assert state.app_startup is False 7 ctx1hijklmn

149 assert state.router_startup is False 7 ctx1hijklmn

150 assert state.sub_router_startup is False 7 ctx1hijklmn

151 assert state.app_shutdown is False 7 ctx1hijklmn

152 assert state.router_shutdown is False 7 ctx1hijklmn

153 assert state.sub_router_shutdown is False 7 ctx1hijklmn

154 

155 with TestClient(app) as client: 7 ctx1hijklmn

156 assert state.app_startup is True 7 ctx1hijklmn

157 assert state.router_startup is True 7 ctx1hijklmn

158 assert state.sub_router_startup is True 7 ctx1hijklmn

159 assert state.app_shutdown is False 7 ctx1hijklmn

160 assert state.router_shutdown is False 7 ctx1hijklmn

161 assert state.sub_router_shutdown is False 7 ctx1hijklmn

162 response = client.get("/") 7 ctx1hijklmn

163 assert response.status_code == 200, response.text 7 ctx1hijklmn

164 assert response.json() == {"message": "Hello World"} 7 ctx1hijklmn

165 

166 assert state.app_startup is True 7 ctx1hijklmn

167 assert state.router_startup is True 7 ctx1hijklmn

168 assert state.sub_router_startup is True 7 ctx1hijklmn

169 assert state.app_shutdown is True 7 ctx1hijklmn

170 assert state.router_shutdown is True 7 ctx1hijklmn

171 assert state.sub_router_shutdown is True 7 ctx1hijklmn

172 

173 

174def test_router_nested_lifespan_state_overriding_by_parent() -> None: 7 ctx1opqrstu

175 @asynccontextmanager 7 ctx1JKLMNOP

176 async def lifespan( 7 ctx1JKLMNOP

177 app: FastAPI, 

178 ) -> AsyncGenerator[Dict[str, Union[str, bool]], None]: 

179 yield { 7 ctx1JKLMNOP

180 "app_specific": True, 

181 "overridden": "app", 

182 } 

183 

184 @asynccontextmanager 7 ctx1JKLMNOP

185 async def router_lifespan( 7 ctx1JKLMNOP

186 app: FastAPI, 

187 ) -> AsyncGenerator[Dict[str, Union[str, bool]], None]: 

188 yield { 7 ctx1JKLMNOP

189 "router_specific": True, 

190 "overridden": "router", # should override parent 

191 } 

192 

193 router = APIRouter(lifespan=router_lifespan) 7 ctx1JKLMNOP

194 app = FastAPI(lifespan=lifespan) 7 ctx1JKLMNOP

195 app.include_router(router) 7 ctx1JKLMNOP

196 

197 with TestClient(app) as client: 7 ctx1JKLMNOP

198 assert client.app_state == { 7 ctx1JKLMNOP

199 "app_specific": True, 

200 "router_specific": True, 

201 "overridden": "app", 

202 } 

203 

204 

205def test_merged_no_return_lifespans_return_none() -> None: 7 ctx1opqrstu

206 @asynccontextmanager 7 ctx1QRSTUVW

207 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 7 ctx1QRSTUVW

208 yield 7 ctx1QRSTUVW

209 

210 @asynccontextmanager 7 ctx1QRSTUVW

211 async def router_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 7 ctx1QRSTUVW

212 yield 7 ctx1QRSTUVW

213 

214 router = APIRouter(lifespan=router_lifespan) 7 ctx1QRSTUVW

215 app = FastAPI(lifespan=lifespan) 7 ctx1QRSTUVW

216 app.include_router(router) 7 ctx1QRSTUVW

217 

218 with TestClient(app) as client: 7 ctx1QRSTUVW

219 assert not client.app_state 7 ctx1QRSTUVW

220 

221 

222def test_merged_mixed_state_lifespans() -> None: 7 ctx1opqrstu

223 @asynccontextmanager 7 ctx1CDEFGHI

224 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 7 ctx1CDEFGHI

225 yield 7 ctx1CDEFGHI

226 

227 @asynccontextmanager 7 ctx1CDEFGHI

228 async def router_lifespan(app: FastAPI) -> AsyncGenerator[Dict[str, bool], None]: 7 ctx1CDEFGHI

229 yield {"router": True} 7 ctx1CDEFGHI

230 

231 @asynccontextmanager 7 ctx1CDEFGHI

232 async def sub_router_lifespan(app: FastAPI) -> AsyncGenerator[None, None]: 7 ctx1CDEFGHI

233 yield 7 ctx1CDEFGHI

234 

235 sub_router = APIRouter(lifespan=sub_router_lifespan) 7 ctx1CDEFGHI

236 router = APIRouter(lifespan=router_lifespan) 7 ctx1CDEFGHI

237 app = FastAPI(lifespan=lifespan) 7 ctx1CDEFGHI

238 router.include_router(sub_router) 7 ctx1CDEFGHI

239 app.include_router(router) 7 ctx1CDEFGHI

240 

241 with TestClient(app) as client: 7 ctx1CDEFGHI

242 assert client.app_state == {"router": True} 7 ctx1CDEFGHI