Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/rest_framework/test.py: 56%

1# Note that we import as `DjangoRequestFactory` and `DjangoClient` in order 

2# to make it harder for the user to import the wrong thing without realizing. 

3import io 

4from importlib import import_module 

5 

6import django 

7from django.conf import settings 

8from django.core.exceptions import ImproperlyConfigured 

9from django.core.handlers.wsgi import WSGIHandler 

10from django.test import override_settings, testcases 

11from django.test.client import Client as DjangoClient 

12from django.test.client import ClientHandler 

13from django.test.client import RequestFactory as DjangoRequestFactory 

14from django.utils.encoding import force_bytes 

15from django.utils.http import urlencode 

16 

17from rest_framework.compat import coreapi, requests 

18from rest_framework.settings import api_settings 

19 

20 

21def force_authenticate(request, user=None, token=None): 

22 request._force_auth_user = user 

23 request._force_auth_token = token 

24 

25 

26if requests is not None: 26 ↛ 119line 26 didn't jump to line 119, because the condition on line 26 was never false

27 class HeaderDict(requests.packages.urllib3._collections.HTTPHeaderDict): 

28 def get_all(self, key, default): 

29 return self.getheaders(key) 

30 

31 class MockOriginalResponse: 

32 def __init__(self, headers): 

33 self.msg = HeaderDict(headers) 

34 self.closed = False 

35 

36 def isclosed(self): 

37 return self.closed 

38 

39 def close(self): 

40 self.closed = True 

41 

42 class DjangoTestAdapter(requests.adapters.HTTPAdapter): 

43 """ 

44 A transport adapter for `requests`, that makes requests via the 

45 Django WSGI app, rather than making actual HTTP requests over the network. 

46 """ 

47 def __init__(self): 

48 self.app = WSGIHandler() 

49 self.factory = DjangoRequestFactory() 

50 

51 def get_environ(self, request): 

52 """ 

53 Given a `requests.PreparedRequest` instance, return a WSGI environ dict. 

54 """ 

55 method = request.method 

56 url = request.url 

57 kwargs = {} 

58 

59 # Set request content, if any exists. 

60 if request.body is not None: 

61 if hasattr(request.body, 'read'): 

62 kwargs['data'] = request.body.read() 

63 else: 

64 kwargs['data'] = request.body 

65 if 'content-type' in request.headers: 

66 kwargs['content_type'] = request.headers['content-type'] 

67 

68 # Set request headers. 

69 for key, value in request.headers.items(): 

70 key = key.upper() 

71 if key in ('CONNECTION', 'CONTENT-LENGTH', 'CONTENT-TYPE'): 

72 continue 

73 kwargs['HTTP_%s' % key.replace('-', '_')] = value 

74 

75 return self.factory.generic(method, url, **kwargs).environ 

76 

77 def send(self, request, *args, **kwargs): 

78 """ 

79 Make an outgoing request to the Django WSGI application. 

80 """ 

81 raw_kwargs = {} 

82 

83 def start_response(wsgi_status, wsgi_headers, exc_info=None): 

84 status, _, reason = wsgi_status.partition(' ') 

85 raw_kwargs['status'] = int(status) 

86 raw_kwargs['reason'] = reason 

87 raw_kwargs['headers'] = wsgi_headers 

88 raw_kwargs['version'] = 11 

89 raw_kwargs['preload_content'] = False 

90 raw_kwargs['original_response'] = MockOriginalResponse(wsgi_headers) 

91 

92 # Make the outgoing request via WSGI. 

93 environ = self.get_environ(request) 

94 wsgi_response = self.app(environ, start_response) 

95 

96 # Build the underlying urllib3.HTTPResponse 

97 raw_kwargs['body'] = io.BytesIO(b''.join(wsgi_response)) 

98 raw = requests.packages.urllib3.HTTPResponse(**raw_kwargs) 

99 

100 # Build the requests.Response 

101 return self.build_response(request, raw) 

102 

103 def close(self): 

104 pass 

105 

106 class RequestsClient(requests.Session): 

107 def __init__(self, *args, **kwargs): 

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

109 adapter = DjangoTestAdapter() 

110 self.mount('http://', adapter) 

111 self.mount('https://', adapter) 

112 

113 def request(self, method, url, *args, **kwargs): 

114 if not url.startswith('http'): 

115 raise ValueError('Missing "http:" or "https:". Use a fully qualified URL, eg "http://testserver%s"' % url) 

116 return super().request(method, url, *args, **kwargs) 

117 

118else: 

119 def RequestsClient(*args, **kwargs): 

120 raise ImproperlyConfigured('requests must be installed in order to use RequestsClient.') 

121 

122 

123if coreapi is not None: 123 ↛ 135line 123 didn't jump to line 135, because the condition on line 123 was never false

124 class CoreAPIClient(coreapi.Client): 

125 def __init__(self, *args, **kwargs): 

126 self._session = RequestsClient() 

127 kwargs['transports'] = [coreapi.transports.HTTPTransport(session=self.session)] 

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

129 

130 @property 

131 def session(self): 

132 return self._session 

133 

134else: 

135 def CoreAPIClient(*args, **kwargs): 

136 raise ImproperlyConfigured('coreapi must be installed in order to use CoreAPIClient.') 

137 

138 

139class APIRequestFactory(DjangoRequestFactory): 

140 renderer_classes_list = api_settings.TEST_REQUEST_RENDERER_CLASSES 

141 default_format = api_settings.TEST_REQUEST_DEFAULT_FORMAT 

142 

143 def __init__(self, enforce_csrf_checks=False, **defaults): 

144 self.enforce_csrf_checks = enforce_csrf_checks 

145 self.renderer_classes = {} 

146 for cls in self.renderer_classes_list: 

147 self.renderer_classes[cls.format] = cls 

148 super().__init__(**defaults) 

149 

150 def _encode_data(self, data, format=None, content_type=None): 

151 """ 

152 Encode the data returning a two tuple of (bytes, content_type) 

153 """ 

154 

155 if data is None: 

156 return ('', content_type) 

157 

158 assert format is None or content_type is None, ( 

159 'You may not set both `format` and `content_type`.' 

160 ) 

161 

162 if content_type: 

163 # Content type specified explicitly, treat data as a raw bytestring 

164 ret = force_bytes(data, settings.DEFAULT_CHARSET) 

165 

166 else: 

167 format = format or self.default_format 

168 

169 assert format in self.renderer_classes, ( 169 ↛ exitline 169 didn't jump to the function exit

170 "Invalid format '{}'. Available formats are {}. " 

171 "Set TEST_REQUEST_RENDERER_CLASSES to enable " 

172 "extra request formats.".format( 

173 format, 

174 ', '.join(["'" + fmt + "'" for fmt in self.renderer_classes]) 

175 ) 

176 ) 

177 

178 # Use format and render the data into a bytestring 

179 renderer = self.renderer_classes[format]() 

180 ret = renderer.render(data) 

181 

182 # Determine the content-type header from the renderer 

183 content_type = renderer.media_type 

184 if renderer.charset: 184 ↛ 190line 184 didn't jump to line 190, because the condition on line 184 was never false

185 content_type = "{}; charset={}".format( 

186 content_type, renderer.charset 

187 ) 

188 

189 # Coerce text to bytes if required. 

190 if isinstance(ret, str): 190 ↛ 191line 190 didn't jump to line 191, because the condition on line 190 was never true

191 ret = ret.encode(renderer.charset) 

192 

193 return ret, content_type 

194 

195 def get(self, path, data=None, **extra): 

196 r = { 

197 'QUERY_STRING': urlencode(data or {}, doseq=True), 

198 } 

199 if not data and '?' in path: 199 ↛ 202line 199 didn't jump to line 202, because the condition on line 199 was never true

200 # Fix to support old behavior where you have the arguments in the 

201 # url. See #1461. 

202 query_string = force_bytes(path.split('?')[1]) 

203 query_string = query_string.decode('iso-8859-1') 

204 r['QUERY_STRING'] = query_string 

205 r.update(extra) 

206 return self.generic('GET', path, **r) 

207 

208 def post(self, path, data=None, format=None, content_type=None, **extra): 

209 data, content_type = self._encode_data(data, format, content_type) 

210 return self.generic('POST', path, data, content_type, **extra) 

211 

212 def put(self, path, data=None, format=None, content_type=None, **extra): 

213 data, content_type = self._encode_data(data, format, content_type) 

214 return self.generic('PUT', path, data, content_type, **extra) 

215 

216 def patch(self, path, data=None, format=None, content_type=None, **extra): 

217 data, content_type = self._encode_data(data, format, content_type) 

218 return self.generic('PATCH', path, data, content_type, **extra) 

219 

220 def delete(self, path, data=None, format=None, content_type=None, **extra): 

221 data, content_type = self._encode_data(data, format, content_type) 

222 return self.generic('DELETE', path, data, content_type, **extra) 

223 

224 def options(self, path, data=None, format=None, content_type=None, **extra): 

225 data, content_type = self._encode_data(data, format, content_type) 

226 return self.generic('OPTIONS', path, data, content_type, **extra) 

227 

228 def generic(self, method, path, data='', 

229 content_type='application/octet-stream', secure=False, **extra): 

230 # Include the CONTENT_TYPE, regardless of whether or not data is empty. 

231 if content_type is not None: 

232 extra['CONTENT_TYPE'] = str(content_type) 

233 

234 return super().generic( 

235 method, path, data, content_type, secure, **extra) 

236 

237 def request(self, **kwargs): 

238 request = super().request(**kwargs) 

239 request._dont_enforce_csrf_checks = not self.enforce_csrf_checks 

240 return request 

241 

242 

243class ForceAuthClientHandler(ClientHandler): 

244 """ 

245 A patched version of ClientHandler that can enforce authentication 

246 on the outgoing requests. 

247 """ 

248 

249 def __init__(self, *args, **kwargs): 

250 self._force_user = None 

251 self._force_token = None 

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

253 

254 def get_response(self, request): 

255 # This is the simplest place we can hook into to patch the 

256 # request object. 

257 force_authenticate(request, self._force_user, self._force_token) 

258 return super().get_response(request) 

259 

260 

261class APIClient(APIRequestFactory, DjangoClient): 

262 def __init__(self, enforce_csrf_checks=False, **defaults): 

263 super().__init__(**defaults) 

264 self.handler = ForceAuthClientHandler(enforce_csrf_checks) 

265 self._credentials = {} 

266 

267 def credentials(self, **kwargs): 

268 """ 

269 Sets headers that will be used on every outgoing request. 

270 """ 

271 self._credentials = kwargs 

272 

273 def force_authenticate(self, user=None, token=None): 

274 """ 

275 Forcibly authenticates outgoing requests with the given 

276 user and/or token. 

277 """ 

278 self.handler._force_user = user 

279 self.handler._force_token = token 

280 if user is None: 

281 self.logout() # Also clear any possible session info if required 

282 

283 def request(self, **kwargs): 

284 # Ensure that any credentials set get added to every request. 

285 kwargs.update(self._credentials) 

286 return super().request(**kwargs) 

287 

288 def get(self, path, data=None, follow=False, **extra): 

289 response = super().get(path, data=data, **extra) 

290 if follow: 290 ↛ 291line 290 didn't jump to line 291, because the condition on line 290 was never true

291 response = self._handle_redirects(response, **extra) 

292 return response 

293 

294 def post(self, path, data=None, format=None, content_type=None, 

295 follow=False, **extra): 

296 response = super().post( 

297 path, data=data, format=format, content_type=content_type, **extra) 

298 if follow: 298 ↛ 299line 298 didn't jump to line 299, because the condition on line 298 was never true

299 response = self._handle_redirects(response, **extra) 

300 return response 

301 

302 def put(self, path, data=None, format=None, content_type=None, 

303 follow=False, **extra): 

304 response = super().put( 

305 path, data=data, format=format, content_type=content_type, **extra) 

306 if follow: 

307 response = self._handle_redirects(response, **extra) 

308 return response 

309 

310 def patch(self, path, data=None, format=None, content_type=None, 

311 follow=False, **extra): 

312 response = super().patch( 

313 path, data=data, format=format, content_type=content_type, **extra) 

314 if follow: 314 ↛ 315line 314 didn't jump to line 315, because the condition on line 314 was never true

315 response = self._handle_redirects(response, **extra) 

316 return response 

317 

318 def delete(self, path, data=None, format=None, content_type=None, 

319 follow=False, **extra): 

320 response = super().delete( 

321 path, data=data, format=format, content_type=content_type, **extra) 

322 if follow: 322 ↛ 323line 322 didn't jump to line 323, because the condition on line 322 was never true

323 response = self._handle_redirects(response, **extra) 

324 return response 

325 

326 def options(self, path, data=None, format=None, content_type=None, 

327 follow=False, **extra): 

328 response = super().options( 

329 path, data=data, format=format, content_type=content_type, **extra) 

330 if follow: 

331 response = self._handle_redirects(response, **extra) 

332 return response 

333 

334 def logout(self): 

335 self._credentials = {} 

336 

337 # Also clear any `force_authenticate` 

338 self.handler._force_user = None 

339 self.handler._force_token = None 

340 

341 if self.session: 

342 super().logout() 

343 

344 

345class APITransactionTestCase(testcases.TransactionTestCase): 

346 client_class = APIClient 

347 

348 

349class APITestCase(testcases.TestCase): 

350 client_class = APIClient 

351 

352 

353class APISimpleTestCase(testcases.SimpleTestCase): 

354 client_class = APIClient 

355 

356 

357class APILiveServerTestCase(testcases.LiveServerTestCase): 

358 client_class = APIClient 

359 

360 

361def cleanup_url_patterns(cls): 

362 if hasattr(cls, '_module_urlpatterns'): 

363 cls._module.urlpatterns = cls._module_urlpatterns 

364 else: 

365 del cls._module.urlpatterns 

366 

367 

368class URLPatternsTestCase(testcases.SimpleTestCase): 

369 """ 

370 Isolate URL patterns on a per-TestCase basis. For example, 

371 

372 class ATestCase(URLPatternsTestCase): 

373 urlpatterns = [...] 

374 

375 def test_something(self): 

376 ... 

377 

378 class AnotherTestCase(URLPatternsTestCase): 

379 urlpatterns = [...] 

380 

381 def test_something_else(self): 

382 ... 

383 """ 

384 @classmethod 

385 def setUpClass(cls): 

386 # Get the module of the TestCase subclass 

387 cls._module = import_module(cls.__module__) 

388 cls._override = override_settings(ROOT_URLCONF=cls.__module__) 

389 

390 if hasattr(cls._module, 'urlpatterns'): 

391 cls._module_urlpatterns = cls._module.urlpatterns 

392 

393 cls._module.urlpatterns = cls.urlpatterns 

394 

395 cls._override.enable() 

396 

397 if django.VERSION > (4, 0): 

398 cls.addClassCleanup(cls._override.disable) 

399 cls.addClassCleanup(cleanup_url_patterns, cls) 

400 

401 super().setUpClass() 

402 

403 if django.VERSION < (4, 0): 403 ↛ 404line 403 didn't jump to line 404, because the condition on line 403 was never true

404 @classmethod 

405 def tearDownClass(cls): 

406 super().tearDownClass() 

407 cls._override.disable() 

408 

409 if hasattr(cls, '_module_urlpatterns'): 

410 cls._module.urlpatterns = cls._module_urlpatterns 

411 else: 

412 del cls._module.urlpatterns