Coverage for tests / test_response_dependency.py: 100%

1"""Test using special types (Response, Request, BackgroundTasks) as dependency annotations. 

2 

3These tests verify that special FastAPI types can be used with Depends() annotations 

4and that the dependency injection system properly handles them. 

5""" 

6 

7from typing import Annotated 4 ctx1uvwx

8 

9from fastapi import BackgroundTasks, Depends, FastAPI, Request, Response 4 ctx1uvwx

10from fastapi.responses import JSONResponse 4 ctx1uvwx

11from fastapi.testclient import TestClient 4 ctx1uvwx

12 

13 

14def test_response_with_depends_annotated(): 4 ctx1uvwx

15 """Response type hint should work with Annotated[Response, Depends(...)].""" 

16 app = FastAPI() 4 ctx1mnop

17 

18 def modify_response(response: Response) -> Response: 4 ctx1mnop

19 response.headers["X-Custom"] = "modified" 4 ctx1mnop

20 return response 4 ctx1mnop

21 

22 @app.get("/") 4 ctx1mnop

23 def endpoint(response: Annotated[Response, Depends(modify_response)]): 4 ctx1mnop

24 return {"status": "ok"} 4 ctx1mnop

25 

26 client = TestClient(app) 4 ctx1mnop

27 resp = client.get("/") 4 ctx1mnop

28 

29 assert resp.status_code == 200 4 ctx1mnop

30 assert resp.json() == {"status": "ok"} 4 ctx1mnop

31 assert resp.headers.get("X-Custom") == "modified" 4 ctx1mnop

32 

33 

34def test_response_with_depends_default(): 4 ctx1uvwx

35 """Response type hint should work with Response = Depends(...).""" 

36 app = FastAPI() 4 ctx1qrst

37 

38 def modify_response(response: Response) -> Response: 4 ctx1qrst

39 response.headers["X-Custom"] = "modified" 4 ctx1qrst

40 return response 4 ctx1qrst

41 

42 @app.get("/") 4 ctx1qrst

43 def endpoint(response: Response = Depends(modify_response)): 4 ctx1qrst

44 return {"status": "ok"} 4 ctx1qrst

45 

46 client = TestClient(app) 4 ctx1qrst

47 resp = client.get("/") 4 ctx1qrst

48 

49 assert resp.status_code == 200 4 ctx1qrst

50 assert resp.json() == {"status": "ok"} 4 ctx1qrst

51 assert resp.headers.get("X-Custom") == "modified" 4 ctx1qrst

52 

53 

54def test_response_without_depends(): 4 ctx1uvwx

55 """Regular Response injection should still work.""" 

56 app = FastAPI() 4 ctx1yzAB

57 

58 @app.get("/") 4 ctx1yzAB

59 def endpoint(response: Response): 4 ctx1yzAB

60 response.headers["X-Direct"] = "set" 4 ctx1yzAB

61 return {"status": "ok"} 4 ctx1yzAB

62 

63 client = TestClient(app) 4 ctx1yzAB

64 resp = client.get("/") 4 ctx1yzAB

65 

66 assert resp.status_code == 200 4 ctx1yzAB

67 assert resp.json() == {"status": "ok"} 4 ctx1yzAB

68 assert resp.headers.get("X-Direct") == "set" 4 ctx1yzAB

69 

70 

71def test_response_dependency_chain(): 4 ctx1uvwx

72 """Response dependency should work in a chain of dependencies.""" 

73 app = FastAPI() 4 ctx1efgh

74 

75 def first_modifier(response: Response) -> Response: 4 ctx1efgh

76 response.headers["X-First"] = "1" 4 ctx1efgh

77 return response 4 ctx1efgh

78 

79 def second_modifier( 4 ctx1efgh

80 response: Annotated[Response, Depends(first_modifier)], 

81 ) -> Response: 

82 response.headers["X-Second"] = "2" 4 ctx1efgh

83 return response 4 ctx1efgh

84 

85 @app.get("/") 4 ctx1efgh

86 def endpoint(response: Annotated[Response, Depends(second_modifier)]): 4 ctx1efgh

87 return {"status": "ok"} 4 ctx1efgh

88 

89 client = TestClient(app) 4 ctx1efgh

90 resp = client.get("/") 4 ctx1efgh

91 

92 assert resp.status_code == 200 4 ctx1efgh

93 assert resp.headers.get("X-First") == "1" 4 ctx1efgh

94 assert resp.headers.get("X-Second") == "2" 4 ctx1efgh

95 

96 

97def test_response_dependency_returns_different_response_instance(): 4 ctx1uvwx

98 """Dependency that returns a different Response instance should work. 

99 

100 When a dependency returns a new Response object (e.g., JSONResponse) instead 

101 of modifying the injected one, the returned response should be used and any 

102 modifications to it in the endpoint should be preserved. 

103 """ 

104 app = FastAPI() 4 ctx1ijkl

105 

106 def default_response() -> Response: 4 ctx1ijkl

107 response = JSONResponse(content={"status": "ok"}) 4 ctx1ijkl

108 response.headers["X-Custom"] = "initial" 4 ctx1ijkl

109 return response 4 ctx1ijkl

110 

111 @app.get("/") 4 ctx1ijkl

112 def endpoint(response: Annotated[Response, Depends(default_response)]): 4 ctx1ijkl

113 response.headers["X-Custom"] = "modified" 4 ctx1ijkl

114 return response 4 ctx1ijkl

115 

116 client = TestClient(app) 4 ctx1ijkl

117 resp = client.get("/") 4 ctx1ijkl

118 

119 assert resp.status_code == 200 4 ctx1ijkl

120 assert resp.json() == {"status": "ok"} 4 ctx1ijkl

121 assert resp.headers.get("X-Custom") == "modified" 4 ctx1ijkl

122 

123 

124# Tests for Request type hint with Depends 

125def test_request_with_depends_annotated(): 4 ctx1uvwx

126 """Request type hint should work in dependency chain.""" 

127 app = FastAPI() 4 ctx1CDEF

128 

129 def extract_request_info(request: Request) -> dict: 4 ctx1CDEF

130 return { 4 ctx1CDEF

131 "path": request.url.path, 

132 "user_agent": request.headers.get("user-agent", "unknown"), 

133 } 

134 

135 @app.get("/") 4 ctx1CDEF

136 def endpoint( 4 ctx1CDEF

137 info: Annotated[dict, Depends(extract_request_info)], 

138 ): 

139 return info 4 ctx1CDEF

140 

141 client = TestClient(app) 4 ctx1CDEF

142 resp = client.get("/", headers={"user-agent": "test-agent"}) 4 ctx1CDEF

143 

144 assert resp.status_code == 200 4 ctx1CDEF

145 assert resp.json() == {"path": "/", "user_agent": "test-agent"} 4 ctx1CDEF

146 

147 

148# Tests for BackgroundTasks type hint with Depends 

149def test_background_tasks_with_depends_annotated(): 4 ctx1uvwx

150 """BackgroundTasks type hint should work with Annotated[BackgroundTasks, Depends(...)].""" 

151 app = FastAPI() 4 ctx1abcd

152 task_results = [] 4 ctx1abcd

153 

154 def background_task(message: str): 4 ctx1abcd

155 task_results.append(message) 4 ctx1abcd

156 

157 def add_background_task(background_tasks: BackgroundTasks) -> BackgroundTasks: 4 ctx1abcd

158 background_tasks.add_task(background_task, "from dependency") 4 ctx1abcd

159 return background_tasks 4 ctx1abcd

160 

161 @app.get("/") 4 ctx1abcd

162 def endpoint( 4 ctx1abcd

163 background_tasks: Annotated[BackgroundTasks, Depends(add_background_task)], 

164 ): 

165 background_tasks.add_task(background_task, "from endpoint") 4 ctx1abcd

166 return {"status": "ok"} 4 ctx1abcd

167 

168 client = TestClient(app) 4 ctx1abcd

169 resp = client.get("/") 4 ctx1abcd

170 

171 assert resp.status_code == 200 4 ctx1abcd

172 assert "from dependency" in task_results 4 ctx1abcd

173 assert "from endpoint" in task_results 4 ctx1abcd