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)
python类WSGIRequest()的实例源码
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 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 make_request(*args, **kwargs):
return WSGIRequest(TRequest.blank(*args, **kwargs).environ)
def make_request(*args, **kwargs):
return WSGIRequest(TRequest.blank(*args, **kwargs).environ)
def make_request(*args, **kwargs):
return WSGIRequest(TRequest.blank(*args, **kwargs).environ)
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 request(self, **request):
"Construct a generic request object."
return WSGIRequest(self._base_environ(**request))
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 request(self, **request):
"Construct a generic request object."
return WSGIRequest(self._base_environ(**request))
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 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 __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
