Coverage for fastapi_restly / testing / _client.py: 93%

1from __future__ import annotations 

2 

3import json 

4from typing import TYPE_CHECKING, Any 

5 

6try: 

7 from fastapi.testclient import TestClient 

8except RuntimeError as exc: # pragma: no cover -- exercised via subprocess 

9 # Starlette >=1.x raises RuntimeError (not ModuleNotFoundError) at import 

10 # when neither httpx2 nor httpx is installed. Normalize it to a missing 

11 # module so the importing namespaces surface the [testing] extras hint 

12 # instead of Starlette's bare "pip install httpx2" message. 

13 if "httpx" not in str(exc): 

14 raise 

15 raise ModuleNotFoundError(name="httpx2") from exc 

16 

17if TYPE_CHECKING: 

18 # httpx2 is a drop-in rename of httpx 0.28.1 with identical stubs, so we 

19 # type-check against httpx and let runtime pick whichever is installed. 

20 import httpx 

21else: 

22 # Match Starlette's TestClient, which prefers httpx2 and only falls back to 

23 # httpx (with a deprecation warning) when httpx2 is absent. Aligning our 

24 # reference keeps httpx.Response/httpx.URL the same classes the parent 

25 # TestClient yields. 

26 try: 

27 import httpx2 as httpx 

28 except ModuleNotFoundError: 

29 import httpx 

30 

31# httpx accepts either a string or a `httpx.URL` for request URLs. The base 

32# class' `URLTypes` alias is private, so we replicate the public union here. 

33URLTypes = httpx.URL | str 

34 

35 

36class _StatusAssertions: 

37 def assert_status( 

38 self, response: httpx.Response, expected_code: int | None = None 

39 ) -> None: 

40 """Check if the response status code matches the expected code.""" 

41 __tracebackhide__ = True 

42 

43 status_code = response.status_code 

44 

45 if expected_code is not None and status_code == expected_code: 

46 return # All good 

47 

48 if expected_code is None and status_code < 400: 

49 return # Also fine 

50 

51 # Raise AssertionError with detailed error message 

52 try: 

53 response_content = response.json() 

54 except (ValueError, TypeError, json.JSONDecodeError): 

55 response_content = response.content.decode(errors="ignore") 

56 

57 content_str_raw = str(response_content) 

58 if len(content_str_raw) > 1000: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 content_str_raw = content_str_raw[:1000] + "...(truncated)" 

60 content_str = f"Response JSON: {content_str_raw}" 

61 

62 # Safe method/URL extraction 

63 try: 

64 method = response.request.method.upper() 

65 url = str(response.request.url) 

66 request_info = f"{method} {url}" 

67 except Exception: 

68 request_info = "(request info unavailable)" 

69 

70 raise AssertionError( 

71 f"Expected {request_info} to return {expected_code}, got {status_code}\n" 

72 f"{content_str}" 

73 ) 

74 

75 

76class RestlyTestClient(_StatusAssertions, TestClient): 

77 """Synchronous test client with Restly's response status assertions.""" 

78 

79 def __init__( 

80 self, app: Any, *args: Any, _transport_app: Any | None = None, **kwargs: Any 

81 ) -> None: 

82 """Build a client while optionally exposing a different public app. 

83 

84 Managed database isolation wraps the ASGI lifespan internally, but 

85 callers have always used ``client.app`` as the original ``FastAPI`` 

86 instance (including to register a view after constructing the client). 

87 ``_transport_app`` lets the fixture keep that contract while Starlette's 

88 transport and lifespan run the internal wrapper. 

89 """ 

90 self._restly_public_app = app 

91 self._restly_transport_app = app if _transport_app is None else _transport_app 

92 super().__init__(self._restly_transport_app, *args, **kwargs) 

93 self.app = self._restly_public_app 

94 

95 def __enter__(self) -> RestlyTestClient: 

96 # Starlette's lifespan coroutine reads ``self.app`` when __enter__ starts 

97 # it. Temporarily expose the transport wrapper, then restore the public 

98 # FastAPI app after startup has completed. The request transport already 

99 # captured the wrapper in TestClient.__init__. 

100 self.app = self._restly_transport_app 

101 try: 

102 super().__enter__() 

103 finally: 

104 self.app = self._restly_public_app 

105 return self 

106 

107 def get( 

108 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs: Any 

109 ) -> httpx.Response: 

110 """Make a GET request. Asserts the response status code matches `assert_status_code` (default: 200). 

111 Pass `assert_status_code=None` to accept any status below 400 instead; 

112 it does not skip the check.""" 

113 __tracebackhide__ = True 

114 response = super().get(url, **kwargs) 

115 self.assert_status(response, assert_status_code) 

116 return response 

117 

118 def post( 

119 self, url: URLTypes, *, assert_status_code: int | None = 201, **kwargs: Any 

120 ) -> httpx.Response: 

121 """Make a POST request. Asserts the response status code matches `assert_status_code` (default: 201). 

122 Pass `assert_status_code=None` to accept any status below 400 instead; 

123 it does not skip the check.""" 

124 __tracebackhide__ = True 

125 response = super().post(url, **kwargs) 

126 self.assert_status(response, assert_status_code) 

127 return response 

128 

129 def put( 

130 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs: Any 

131 ) -> httpx.Response: 

132 """Make a PUT request. Asserts the response status code matches `assert_status_code` (default: 200). 

133 Pass `assert_status_code=None` to accept any status below 400 instead; 

134 it does not skip the check.""" 

135 __tracebackhide__ = True 

136 response = super().put(url, **kwargs) 

137 self.assert_status(response, assert_status_code) 

138 return response 

139 

140 def patch( 

141 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs: Any 

142 ) -> httpx.Response: 

143 """Make a PATCH request. Asserts the response status code matches `assert_status_code` (default: 200). 

144 Pass `assert_status_code=None` to accept any status below 400 instead; 

145 it does not skip the check.""" 

146 __tracebackhide__ = True 

147 response = super().patch(url, **kwargs) 

148 self.assert_status(response, assert_status_code) 

149 return response 

150 

151 def delete( 

152 self, url: URLTypes, *, assert_status_code: int | None = 204, **kwargs: Any 

153 ) -> httpx.Response: 

154 """Make a DELETE request. Asserts the response status code matches `assert_status_code` (default: 204). 

155 Pass `assert_status_code=None` to accept any status below 400 instead; 

156 it does not skip the check.""" 

157 __tracebackhide__ = True 

158 response = super().delete(url, **kwargs) 

159 self.assert_status(response, assert_status_code) 

160 return response 

161 

162 

163class AsyncRestlyTestClient(_StatusAssertions, httpx.AsyncClient): 

164 """Async test client with the same response assertions as the sync client.""" 

165 

166 def __init__( 

167 self, app: Any, *args: Any, _transport_app: Any | None = None, **kwargs: Any 

168 ) -> None: 

169 self.app = app 

170 transport_app = app if _transport_app is None else _transport_app 

171 kwargs.setdefault("transport", httpx.ASGITransport(app=transport_app)) 

172 kwargs.setdefault("base_url", "http://testserver") 

173 kwargs.setdefault("follow_redirects", True) 

174 super().__init__(*args, **kwargs) 

175 

176 async def get( 

177 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs: Any 

178 ) -> httpx.Response: 

179 """Make a GET request and assert status 200 by default. 

180 

181 Pass ``assert_status_code=None`` to accept any status below 400; it does 

182 not skip the check. 

183 """ 

184 __tracebackhide__ = True 

185 response = await super().get(url, **kwargs) 

186 self.assert_status(response, assert_status_code) 

187 return response 

188 

189 async def post( 

190 self, url: URLTypes, *, assert_status_code: int | None = 201, **kwargs: Any 

191 ) -> httpx.Response: 

192 """Make a POST request and assert status 201 by default. 

193 

194 Pass ``assert_status_code=None`` to accept any status below 400; it does 

195 not skip the check. 

196 """ 

197 __tracebackhide__ = True 

198 response = await super().post(url, **kwargs) 

199 self.assert_status(response, assert_status_code) 

200 return response 

201 

202 async def put( 

203 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs: Any 

204 ) -> httpx.Response: 

205 """Make a PUT request and assert status 200 by default. 

206 

207 Pass ``assert_status_code=None`` to accept any status below 400; it does 

208 not skip the check. 

209 """ 

210 __tracebackhide__ = True 

211 response = await super().put(url, **kwargs) 

212 self.assert_status(response, assert_status_code) 

213 return response 

214 

215 async def patch( 

216 self, url: URLTypes, *, assert_status_code: int | None = 200, **kwargs: Any 

217 ) -> httpx.Response: 

218 """Make a PATCH request and assert status 200 by default. 

219 

220 Pass ``assert_status_code=None`` to accept any status below 400; it does 

221 not skip the check. 

222 """ 

223 __tracebackhide__ = True 

224 response = await super().patch(url, **kwargs) 

225 self.assert_status(response, assert_status_code) 

226 return response 

227 

228 async def delete( 

229 self, url: URLTypes, *, assert_status_code: int | None = 204, **kwargs: Any 

230 ) -> httpx.Response: 

231 """Make a DELETE request and assert status 204 by default. 

232 

233 Pass ``assert_status_code=None`` to accept any status below 400; it does 

234 not skip the check. 

235 """ 

236 __tracebackhide__ = True 

237 response = await super().delete(url, **kwargs) 

238 self.assert_status(response, assert_status_code) 

239 return response