我们从Python开源项目中，提取了以下34个代码示例，用于说明如何使用django.core.handlers.wsgi.WSGIRequest()。
def request(self, **request): """ Similar to parent class, but returns the request object as soon as it has created it. """ environ = { 'HTTP_COOKIE': self.cookies, 'PATH_INFO': '/', 'QUERY_STRING': '', 'REQUEST_METHOD': 'GET', 'SCRIPT_NAME': '', 'SERVER_NAME': 'testserver', 'SERVER_PORT': 80, 'SERVER_PROTOCOL': 'HTTP/1.1', 'wsgi.version': (1,0), 'wsgi.url_scheme': 'http', 'wsgi.input': FakePayload(''), 'wsgi.errors': self.errors, 'wsgi.multiprocess': True, 'wsgi.multithread': False, 'wsgi.run_once': False, } environ.update(self.defaults) environ.update(request) return WSGIRequest(environ)
def test_raw_post_data_partial_read(django_elasticapm_client): v = compat.b('{"foo": "bar"}') request = WSGIRequest(environ={ 'wsgi.input': compat.BytesIO(v + compat.b('\r\n\r\n')), 'REQUEST_METHOD': 'POST', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'CONTENT_TYPE': 'application/json', 'CONTENT_LENGTH': len(v), 'ACCEPT': 'application/json', }) request.read(1) django_elasticapm_client.capture('Message', message='foo', request=request) assert len(django_elasticapm_client.events) == 1 event = django_elasticapm_client.events.pop(0)['errors'][0] assert 'request' in event['context'] request = event['context']['request'] assert request['method'] == 'POST' assert request['body'] == '<unavailable>'
def test_post_data(django_elasticapm_client): request = WSGIRequest(environ={ 'wsgi.input': compat.BytesIO(), 'REQUEST_METHOD': 'POST', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'CONTENT_TYPE': 'application/json', 'ACCEPT': 'application/json', }) request.POST = QueryDict("x=1&y=2") django_elasticapm_client.capture('Message', message='foo', request=request) assert len(django_elasticapm_client.events) == 1 event = django_elasticapm_client.events.pop(0)['errors'][0] assert 'request' in event['context'] request = event['context']['request'] assert request['method'] == 'POST' assert request['body'] == {'x': '1', 'y': '2'}
def test_post_raw_data(django_elasticapm_client): request = WSGIRequest(environ={ 'wsgi.input': compat.BytesIO(compat.b('foobar')), 'wsgi.url_scheme': 'http', 'REQUEST_METHOD': 'POST', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'CONTENT_TYPE': 'application/json', 'ACCEPT': 'application/json', 'CONTENT_LENGTH': '6', }) django_elasticapm_client.capture('Message', message='foo', request=request) assert len(django_elasticapm_client.events) == 1 event = django_elasticapm_client.events.pop(0)['errors'][0] assert 'request' in event['context'] request = event['context']['request'] assert request['method'] == 'POST' assert request['body'] == compat.b('foobar')
def test_django_logging_request_kwarg(django_elasticapm_client): handler = LoggingHandler() logger = logging.getLogger(__name__) logger.handlers = [] logger.addHandler(handler) logger.error('This is a test error', extra={ 'request': WSGIRequest(environ={ 'wsgi.input': compat.StringIO(), 'REQUEST_METHOD': 'POST', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'CONTENT_TYPE': 'application/json', 'ACCEPT': 'application/json', }) }) assert len(django_elasticapm_client.events) == 1 event = django_elasticapm_client.events.pop(0)['errors'][0] assert 'request' in event['context'] request = event['context']['request'] assert request['method'] == 'POST'
def get_menu(context, request): if not isinstance(request, WSGIRequest): return None # Try to get app list if hasattr(request, 'current_app'): # Django 1.8 uses request.current_app instead of context.current_app template_response = get_admin_site(request.current_app).index(request) else: template_response = get_admin_site(context.current_app).index(request) try: app_list = template_response.context_data['app_list'] except Exception: return return Menu(context, request, app_list).get_app_list()
def request(self, **request): """ Similar to parent class, but returns the request object as soon as it has created it. """ environ = { 'HTTP_COOKIE': self.cookies, 'PATH_INFO': '/', 'QUERY_STRING': '', 'REQUEST_METHOD': 'GET', 'SCRIPT_NAME': '', 'SERVER_NAME': 'testserver', 'SERVER_PORT': 80, 'SERVER_PROTOCOL': 'HTTP/1.1', } environ.update(self.defaults) environ.update(request) return WSGIRequest(environ)
def handle_http_event(app): environ = wsgi(json.loads(sys.stdin.read())) app.load_middleware() resp = app.get_response(WSGIRequest(environ)) body = '' if resp.streaming: for content in resp.streaming_content: body += content else: body = resp.content.decode('utf-8') headers = {} for header in resp.items(): headers[header[0]] = header[1] resp.close() sys.stdout.write(json.dumps({ 'body': body, 'status_code': resp.status_code, 'headers': headers, }))
def parse_headers_and_body_with_mimer(cls, headers, body): """Use piston's Mimer functionality to handle the content. :return: The value of 'request.data' after using Piston's translate_mime on the input. """ # JAM 2012-10-09 Importing emitters has a side effect of registering # mime type handlers with utils.translate_mime. from piston3 import emitters emitters # Imported for side-effects. from piston3.utils import translate_mime environ = {'wsgi.input': BytesIO(body.encode("utf8"))} for name, value in headers.items(): environ[name.upper().replace('-', '_')] = value environ['REQUEST_METHOD'] = 'POST' environ['SCRIPT_NAME'] = '' environ['PATH_INFO'] = '' from django.core.handlers.wsgi import WSGIRequest request = WSGIRequest(environ) translate_mime(request) return request.data
def __call__(self, environ): # Set up middleware if needed. We couldn't do this earlier, because # settings weren't available. if self._request_middleware is None: self.load_middleware() request_started.disconnect(close_old_connections) request_started.send(sender=self.__class__, environ=environ) request_started.connect(close_old_connections) request = WSGIRequest(environ) # sneaky little hack so that we can easily get round # CsrfViewMiddleware. This makes life easier, and is probably # required for backwards compatibility with external tests against # admin views. request._dont_enforce_csrf_checks = not self.enforce_csrf_checks # Request goes through middleware. response = self.get_response(request) # Attach the originating request to the response so that it could be # later retrieved. response.wsgi_request = request # We're emulating a WSGI server; we must call the close method # on completion. if response.streaming: response.streaming_content = closing_iterator_wrapper( response.streaming_content, response.close) else: request_finished.disconnect(close_old_connections) response.close() # will fire request_finished request_finished.connect(close_old_connections) return response
def request(self, **request): "Construct a generic request object." return WSGIRequest(self._base_environ(**request))
def validate_error_on_unauthenticated(self, url_shortcut: str, request: Callable[[str], WSGIRequest], url_shortcut_args=[]): url = reverse(url_shortcut, args=url_shortcut_args) response = request(url) self.assertEqual(response.status_code, HTTP_403_FORBIDDEN) self.assertEqual(response.json(), {'detail': 'Authentication credentials were not provided.'})
def test_disallowed_hosts_error_django_19(django_elasticapm_client): request = WSGIRequest(environ={ 'wsgi.input': compat.BytesIO(), 'wsgi.url_scheme': 'http', 'REQUEST_METHOD': 'POST', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'CONTENT_TYPE': 'application/json', 'ACCEPT': 'application/json', }) with override_settings(ALLOWED_HOSTS=['example.com']): # this should not raise a DisallowedHost exception django_elasticapm_client.capture('Message', message='foo', request=request) event = django_elasticapm_client.events.pop(0)['errors'][0] assert event['context']['request']['url']['raw'] == 'http://testserver/'
def test_disallowed_hosts_error_django_18(django_elasticapm_client): request = WSGIRequest(environ={ 'wsgi.input': compat.BytesIO(), 'wsgi.url_scheme': 'http', 'REQUEST_METHOD': 'POST', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'CONTENT_TYPE': 'application/json', 'ACCEPT': 'application/json', }) with override_settings(ALLOWED_HOSTS=['example.com']): # this should not raise a DisallowedHost exception django_elasticapm_client.capture('Message', message='foo', request=request) event = django_elasticapm_client.events.pop(0)['errors'][0] assert event['context']['request']['url'] == {'raw': 'DisallowedHost'}
def make_request(*args, **kwargs): return WSGIRequest(TRequest.blank(*args, **kwargs).environ)
def parse_headers_and_body_with_mimer(headers, body): """Use piston's Mimer functionality to handle the content. :return: The value of 'request.data' after using Piston's translate_mime on the input. """ # JAM 2012-10-09 Importing emitters has a side effect of registering mime # type handlers with utils.translate_mime. So we must import it, even # though we don't use it. However, piston loads Django's QuerySet code # which fails if you don't have a settings.py available. Which we don't # during 'test.pserv'. So we import this late. from piston import emitters ignore_unused(emitters) from piston.utils import translate_mime environ = {'wsgi.input': BytesIO(body)} for name, value in headers.items(): environ[name.upper().replace('-', '_')] = value environ['REQUEST_METHOD'] = 'POST' environ['SCRIPT_NAME'] = '' environ['PATH_INFO'] = '' # Django 1.6 needs DJANGO_SETTINGS_MODULE to be defined # when importing WSGIRequest. os.environ['DJANGO_SETTINGS_MODULE'] = 'maas.development' from django.core.handlers.wsgi import WSGIRequest request = WSGIRequest(environ) translate_mime(request) return request.data
def __call__(self, environ): # Set up middleware if needed. We couldn't do this earlier, because # settings weren't available. if self._middleware_chain is None: self.load_middleware() request_started.disconnect(close_old_connections) request_started.send(sender=self.__class__, environ=environ) request_started.connect(close_old_connections) request = WSGIRequest(environ) # sneaky little hack so that we can easily get round # CsrfViewMiddleware. This makes life easier, and is probably # required for backwards compatibility with external tests against # admin views. request._dont_enforce_csrf_checks = not self.enforce_csrf_checks # Request goes through middleware. response = self.get_response(request) # Simulate behaviors of most Web servers. conditional_content_removal(request, response) # Attach the originating request to the response so that it could be # later retrieved. response.wsgi_request = request # We're emulating a WSGI server; we must call the close method # on completion. if response.streaming: response.streaming_content = closing_iterator_wrapper( response.streaming_content, response.close) else: request_finished.disconnect(close_old_connections) response.close() # will fire request_finished request_finished.connect(close_old_connections) return response
def build_dummy_request(newsitem): """ Construct a HttpRequest object that is, as far as possible, representative of ones that would receive this page as a response. Used for previewing / moderation and any other place where we want to display a view of this page in the admin interface without going through the regular page routing logic. """ url = newsitem.full_url if url: url_info = urlparse(url) hostname = url_info.hostname path = url_info.path port = url_info.port or 80 else: # Cannot determine a URL to this page - cobble one together based on # whatever we find in ALLOWED_HOSTS try: hostname = settings.ALLOWED_HOSTS[0] except IndexError: hostname = 'localhost' path = '/' port = 80 request = WSGIRequest({ 'REQUEST_METHOD': 'GET', 'PATH_INFO': path, 'SERVER_NAME': hostname, 'SERVER_PORT': port, 'HTTP_HOST': hostname, 'wsgi.input': StringIO(), }) # Apply middleware to the request - see http://www.mellowmorning.com/2011/04/18/mock-django-request-for-testing/ handler = BaseHandler() handler.load_middleware() # call each middleware in turn and throw away any responses that they might return for middleware_method in handler._request_middleware: middleware_method(request) return request
def request(self, **request): """ Rather than issuing a request and returning the response, this simply constructs an ``HttpRequest`` object and returns it. """ environ = { 'HTTP_COOKIE': self.cookies, 'PATH_INFO': '/', 'QUERY_STRING': '', 'REMOTE_ADDR': '127.0.0.1', 'REQUEST_METHOD': 'GET', 'SCRIPT_NAME': '', 'SERVER_NAME': 'testserver', 'SERVER_PORT': '80', 'SERVER_PROTOCOL': 'HTTP/1.1', 'wsgi.version': (1,0), 'wsgi.url_scheme': 'http', 'wsgi.errors': self.errors, 'wsgi.multiprocess':True, 'wsgi.multithread': False, 'wsgi.run_once': False, 'wsgi.input': None, } environ.update(self.defaults) environ.update(request) request = WSGIRequest(environ) # We have to manually add a session since we'll be bypassing # the middleware chain. session_middleware = SessionMiddleware() session_middleware.process_request(request) return request
def setUp(self): super(APIClientTestCase, self).setUp() # Django 1.8 and 1.11 need to be configured before we can use # WSGIRequest and MultiPartParser. if not django.conf.settings.configured: django.conf.settings.configure(DEBUG=True)
