Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/rest_framework/test.py: 56%
237 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
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
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
17from rest_framework.compat import coreapi, requests
18from rest_framework.settings import api_settings
21def force_authenticate(request, user=None, token=None):
22 request._force_auth_user = user
23 request._force_auth_token = token
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)
31 class MockOriginalResponse:
32 def __init__(self, headers):
33 self.msg = HeaderDict(headers)
34 self.closed = False
36 def isclosed(self):
37 return self.closed
39 def close(self):
40 self.closed = True
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()
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 = {}
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']
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
75 return self.factory.generic(method, url, **kwargs).environ
77 def send(self, request, *args, **kwargs):
78 """
79 Make an outgoing request to the Django WSGI application.
80 """
81 raw_kwargs = {}
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)
92 # Make the outgoing request via WSGI.
93 environ = self.get_environ(request)
94 wsgi_response = self.app(environ, start_response)
96 # Build the underlying urllib3.HTTPResponse
97 raw_kwargs['body'] = io.BytesIO(b''.join(wsgi_response))
98 raw = requests.packages.urllib3.HTTPResponse(**raw_kwargs)
100 # Build the requests.Response
101 return self.build_response(request, raw)
103 def close(self):
104 pass
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)
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)
118else:
119 def RequestsClient(*args, **kwargs):
120 raise ImproperlyConfigured('requests must be installed in order to use RequestsClient.')
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)
130 @property
131 def session(self):
132 return self._session
134else:
135 def CoreAPIClient(*args, **kwargs):
136 raise ImproperlyConfigured('coreapi must be installed in order to use CoreAPIClient.')
139class APIRequestFactory(DjangoRequestFactory):
140 renderer_classes_list = api_settings.TEST_REQUEST_RENDERER_CLASSES
141 default_format = api_settings.TEST_REQUEST_DEFAULT_FORMAT
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)
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 """
155 if data is None:
156 return ('', content_type)
158 assert format is None or content_type is None, (
159 'You may not set both `format` and `content_type`.'
160 )
162 if content_type:
163 # Content type specified explicitly, treat data as a raw bytestring
164 ret = force_bytes(data, settings.DEFAULT_CHARSET)
166 else:
167 format = format or self.default_format
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 )
178 # Use format and render the data into a bytestring
179 renderer = self.renderer_classes[format]()
180 ret = renderer.render(data)
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 )
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)
193 return ret, content_type
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)
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)
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)
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)
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)
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)
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)
234 return super().generic(
235 method, path, data, content_type, secure, **extra)
237 def request(self, **kwargs):
238 request = super().request(**kwargs)
239 request._dont_enforce_csrf_checks = not self.enforce_csrf_checks
240 return request
243class ForceAuthClientHandler(ClientHandler):
244 """
245 A patched version of ClientHandler that can enforce authentication
246 on the outgoing requests.
247 """
249 def __init__(self, *args, **kwargs):
250 self._force_user = None
251 self._force_token = None
252 super().__init__(*args, **kwargs)
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)
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 = {}
267 def credentials(self, **kwargs):
268 """
269 Sets headers that will be used on every outgoing request.
270 """
271 self._credentials = kwargs
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
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)
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
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
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
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
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
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
334 def logout(self):
335 self._credentials = {}
337 # Also clear any `force_authenticate`
338 self.handler._force_user = None
339 self.handler._force_token = None
341 if self.session:
342 super().logout()
345class APITransactionTestCase(testcases.TransactionTestCase):
346 client_class = APIClient
349class APITestCase(testcases.TestCase):
350 client_class = APIClient
353class APISimpleTestCase(testcases.SimpleTestCase):
354 client_class = APIClient
357class APILiveServerTestCase(testcases.LiveServerTestCase):
358 client_class = APIClient
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
368class URLPatternsTestCase(testcases.SimpleTestCase):
369 """
370 Isolate URL patterns on a per-TestCase basis. For example,
372 class ATestCase(URLPatternsTestCase):
373 urlpatterns = [...]
375 def test_something(self):
376 ...
378 class AnotherTestCase(URLPatternsTestCase):
379 urlpatterns = [...]
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__)
390 if hasattr(cls._module, 'urlpatterns'):
391 cls._module_urlpatterns = cls._module.urlpatterns
393 cls._module.urlpatterns = cls.urlpatterns
395 cls._override.enable()
397 if django.VERSION > (4, 0):
398 cls.addClassCleanup(cls._override.disable)
399 cls.addClassCleanup(cleanup_url_patterns, cls)
401 super().setUpClass()
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()
409 if hasattr(cls, '_module_urlpatterns'):
410 cls._module.urlpatterns = cls._module_urlpatterns
411 else:
412 del cls._module.urlpatterns