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 ctx1pqrs

8 

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

10from fastapi.responses import JSONResponse 4 ctx1pqrs

11from fastapi.testclient import TestClient 4 ctx1pqrs

12 

13 

14def test_response_with_depends_annotated(): 4 ctx1pqrs

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

16 app = FastAPI() 3 ctx1jkl

17 

18 def modify_response(response: Response) -> Response: 3 ctx1jkl

19 response.headers["X-Custom"] = "modified" 3 ctx1jkl

20 return response 3 ctx1jkl

21 

22 @app.get("/") 3 ctx1jkl

23 def endpoint(response: Annotated[Response, Depends(modify_response)]): 3 ctx1jkl

24 return {"status": "ok"} 3 ctx1jkl

25 

26 client = TestClient(app) 3 ctx1jkl

27 resp = client.get("/") 3 ctx1jkl

28 

29 assert resp.status_code == 200 3 ctx1jkl

30 assert resp.json() == {"status": "ok"} 3 ctx1jkl

31 assert resp.headers.get("X-Custom") == "modified" 3 ctx1jkl

32 

33 

34def test_response_with_depends_default(): 4 ctx1pqrs

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

36 app = FastAPI() 3 ctx1mno

37 

38 def modify_response(response: Response) -> Response: 3 ctx1mno

39 response.headers["X-Custom"] = "modified" 3 ctx1mno

40 return response 3 ctx1mno

41 

42 @app.get("/") 3 ctx1mno

43 def endpoint(response: Response = Depends(modify_response)): 3 ctx1mno

44 return {"status": "ok"} 3 ctx1mno

45 

46 client = TestClient(app) 3 ctx1mno

47 resp = client.get("/") 3 ctx1mno

48 

49 assert resp.status_code == 200 3 ctx1mno

50 assert resp.json() == {"status": "ok"} 3 ctx1mno

51 assert resp.headers.get("X-Custom") == "modified" 3 ctx1mno

52 

53 

54def test_response_without_depends(): 4 ctx1pqrs

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

56 app = FastAPI() 3 ctx1tuv

57 

58 @app.get("/") 3 ctx1tuv

59 def endpoint(response: Response): 3 ctx1tuv

60 response.headers["X-Direct"] = "set" 3 ctx1tuv

61 return {"status": "ok"} 3 ctx1tuv

62 

63 client = TestClient(app) 3 ctx1tuv

64 resp = client.get("/") 3 ctx1tuv

65 

66 assert resp.status_code == 200 3 ctx1tuv

67 assert resp.json() == {"status": "ok"} 3 ctx1tuv

68 assert resp.headers.get("X-Direct") == "set" 3 ctx1tuv

69 

70 

71def test_response_dependency_chain(): 4 ctx1pqrs

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

73 app = FastAPI() 3 ctx1def

74 

75 def first_modifier(response: Response) -> Response: 3 ctx1def

76 response.headers["X-First"] = "1" 3 ctx1def

77 return response 3 ctx1def

78 

79 def second_modifier( 3 ctx1def

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

81 ) -> Response: 

82 response.headers["X-Second"] = "2" 3 ctx1def

83 return response 3 ctx1def

84 

85 @app.get("/") 3 ctx1def

86 def endpoint(response: Annotated[Response, Depends(second_modifier)]): 3 ctx1def

87 return {"status": "ok"} 3 ctx1def

88 

89 client = TestClient(app) 3 ctx1def

90 resp = client.get("/") 3 ctx1def

91 

92 assert resp.status_code == 200 3 ctx1def

93 assert resp.headers.get("X-First") == "1" 3 ctx1def

94 assert resp.headers.get("X-Second") == "2" 3 ctx1def

95 

96 

97def test_response_dependency_returns_different_response_instance(): 4 ctx1pqrs

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() 3 ctx1ghi

105 

106 def default_response() -> Response: 3 ctx1ghi

107 response = JSONResponse(content={"status": "ok"}) 3 ctx1ghi

108 response.headers["X-Custom"] = "initial" 3 ctx1ghi

109 return response 3 ctx1ghi

110 

111 @app.get("/") 3 ctx1ghi

112 def endpoint(response: Annotated[Response, Depends(default_response)]): 3 ctx1ghi

113 response.headers["X-Custom"] = "modified" 3 ctx1ghi

114 return response 3 ctx1ghi

115 

116 client = TestClient(app) 3 ctx1ghi

117 resp = client.get("/") 3 ctx1ghi

118 

119 assert resp.status_code == 200 3 ctx1ghi

120 assert resp.json() == {"status": "ok"} 3 ctx1ghi

121 assert resp.headers.get("X-Custom") == "modified" 3 ctx1ghi

122 

123 

124# Tests for Request type hint with Depends 

125def test_request_with_depends_annotated(): 4 ctx1pqrs

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

127 app = FastAPI() 3 ctx1wxy

128 

129 def extract_request_info(request: Request) -> dict: 3 ctx1wxy

130 return { 3 ctx1wxy

131 "path": request.url.path, 

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

133 } 

134 

135 @app.get("/") 3 ctx1wxy

136 def endpoint( 3 ctx1wxy

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

138 ): 

139 return info 3 ctx1wxy

140 

141 client = TestClient(app) 3 ctx1wxy

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

143 

144 assert resp.status_code == 200 3 ctx1wxy

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

146 

147 

148# Tests for BackgroundTasks type hint with Depends 

149def test_background_tasks_with_depends_annotated(): 4 ctx1pqrs

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

151 app = FastAPI() 3 ctx1abc

152 task_results = [] 3 ctx1abc

153 

154 def background_task(message: str): 3 ctx1abc

155 task_results.append(message) 3 ctx1abc

156 

157 def add_background_task(background_tasks: BackgroundTasks) -> BackgroundTasks: 3 ctx1abc

158 background_tasks.add_task(background_task, "from dependency") 3 ctx1abc

159 return background_tasks 3 ctx1abc

160 

161 @app.get("/") 3 ctx1abc

162 def endpoint( 3 ctx1abc

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

164 ): 

165 background_tasks.add_task(background_task, "from endpoint") 3 ctx1abc

166 return {"status": "ok"} 3 ctx1abc

167 

168 client = TestClient(app) 3 ctx1abc

169 resp = client.get("/") 3 ctx1abc

170 

171 assert resp.status_code == 200 3 ctx1abc

172 assert "from dependency" in task_results 3 ctx1abc

173 assert "from endpoint" in task_results 3 ctx1abc