Refs #26601 -- Deprecated passing None as get_response arg to middleware classes.

This is the new contract since middleware refactoring in Django 1.10.

Co-authored-by: Carlton Gibson <carlton.gibson@noumenal.es>
Co-authored-by: Mariusz Felisiak <felisiak.mariusz@gmail.com>
This commit is contained in:
Claude Paroz 2019-09-26 19:06:35 +02:00 committed by Carlton Gibson
parent a34cb5a6d4
commit 4d973f5939
23 changed files with 668 additions and 502 deletions

View File

@ -10,7 +10,10 @@ from django.utils.http import http_date
class SessionMiddleware(MiddlewareMixin):
# RemovedInDjango40Warning: when the deprecation ends, replace with:
# def __init__(self, get_response):
def __init__(self, get_response=None):
self._get_response_none_deprecation(get_response)
self.get_response = get_response
engine = import_module(settings.SESSION_ENGINE)
self.SessionStore = engine.SessionStore

View File

@ -61,7 +61,10 @@ class UpdateCacheMiddleware(MiddlewareMixin):
UpdateCacheMiddleware must be the first piece of middleware in MIDDLEWARE
so that it'll get called last during the response phase.
"""
# RemovedInDjango40Warning: when the deprecation ends, replace with:
# def __init__(self, get_response):
def __init__(self, get_response=None):
self._get_response_none_deprecation(get_response)
self.cache_timeout = settings.CACHE_MIDDLEWARE_SECONDS
self.page_timeout = None
self.key_prefix = settings.CACHE_MIDDLEWARE_KEY_PREFIX
@ -122,7 +125,10 @@ class FetchFromCacheMiddleware(MiddlewareMixin):
FetchFromCacheMiddleware must be the last piece of middleware in MIDDLEWARE
so that it'll get called last during the request phase.
"""
# RemovedInDjango40Warning: when the deprecation ends, replace with:
# def __init__(self, get_response):
def __init__(self, get_response=None):
self._get_response_none_deprecation(get_response)
self.key_prefix = settings.CACHE_MIDDLEWARE_KEY_PREFIX
self.cache_alias = settings.CACHE_MIDDLEWARE_ALIAS
self.cache = caches[self.cache_alias]
@ -164,7 +170,10 @@ class CacheMiddleware(UpdateCacheMiddleware, FetchFromCacheMiddleware):
Also used as the hook point for the cache decorator, which is generated
using the decorator-from-middleware utility.
"""
# RemovedInDjango40Warning: when the deprecation ends, replace with:
# def __init__(self, get_response, cache_timeout=None, page_timeout=None, **kwargs):
def __init__(self, get_response=None, cache_timeout=None, page_timeout=None, **kwargs):
self._get_response_none_deprecation(get_response)
self.get_response = get_response
# We need to differentiate between "provided, but using default value",
# and "not provided". If the value is provided using a default, then

View File

@ -6,7 +6,10 @@ from django.utils.deprecation import MiddlewareMixin
class SecurityMiddleware(MiddlewareMixin):
# RemovedInDjango40Warning: when the deprecation ends, replace with:
# def __init__(self, get_response):
def __init__(self, get_response=None):
self._get_response_none_deprecation(get_response)
self.sts_seconds = settings.SECURE_HSTS_SECONDS
self.sts_include_subdomains = settings.SECURE_HSTS_INCLUDE_SUBDOMAINS
self.sts_preload = settings.SECURE_HSTS_PRELOAD

View File

@ -113,9 +113,9 @@ def decorator_from_middleware(middleware_class):
def make_middleware_decorator(middleware_class):
def _make_decorator(*m_args, **m_kwargs):
middleware = middleware_class(*m_args, **m_kwargs)
def _decorator(view_func):
middleware = middleware_class(view_func, *m_args, **m_kwargs)
@wraps(view_func)
def _wrapped_view(request, *args, **kwargs):
if hasattr(middleware, 'process_request'):

View File

@ -80,7 +80,10 @@ class DeprecationInstanceCheck(type):
class MiddlewareMixin:
# RemovedInDjango40Warning: when the deprecation ends, replace with:
# def __init__(self, get_response):
def __init__(self, get_response=None):
self._get_response_none_deprecation(get_response)
self.get_response = get_response
super().__init__()
@ -92,3 +95,11 @@ class MiddlewareMixin:
if hasattr(self, 'process_response'):
response = self.process_response(request, response)
return response
def _get_response_none_deprecation(self, get_response):
if get_response is None:
warnings.warn(
'Passing None for the middleware get_response argument is '
'deprecated.',
RemovedInDjango40Warning, stacklevel=3,
)

View File

@ -52,6 +52,10 @@ details on these changes.
* Support for the pre-Django 3.1 password reset tokens in the admin site (that
use the SHA-1 hashing algorithm) will be removed.
* The ``get_request`` argument for
``django.utils.deprecation.MiddlewareMixin.__init__()`` will be required and
won't accept ``None``.
See the :ref:`Django 3.1 release notes <deprecated-features-3.1>` for more
details on these changes.

View File

@ -525,6 +525,9 @@ Miscellaneous
be reproduced exactly as
``request.headers.get('x-requested-with') == 'XMLHttpRequest'``.
* Passing ``None`` as the first argument to
``django.utils.deprecation.MiddlewareMixin.__init__()`` is deprecated.
* The encoding format of cookies values used by
:class:`~django.contrib.messages.storage.cookie.CookieStorage` is different
from the format generated by older versions of Django. Support for the old

View File

@ -295,8 +295,8 @@ middleware classes that are compatible with both :setting:`MIDDLEWARE` and the
old ``MIDDLEWARE_CLASSES``. All middleware classes included with Django
are compatible with both settings.
The mixin provides an ``__init__()`` method that accepts an optional
``get_response`` argument and stores it in ``self.get_response``.
The mixin provides an ``__init__()`` method that requires a ``get_response``
argument and stores it in ``self.get_response``.
The ``__call__()`` method:

View File

@ -1,20 +1,20 @@
from django.contrib.auth.middleware import AuthenticationMiddleware
from django.contrib.auth.models import User
from django.http import HttpRequest
from django.http import HttpRequest, HttpResponse
from django.test import TestCase
class TestAuthenticationMiddleware(TestCase):
def setUp(self):
self.user = User.objects.create_user('test_user', 'test@example.com', 'test_password')
self.middleware = AuthenticationMiddleware()
self.middleware = AuthenticationMiddleware(lambda req: HttpResponse())
self.client.force_login(self.user)
self.request = HttpRequest()
self.request.session = self.client.session
def test_no_password_change_doesnt_invalidate_session(self):
self.request.session = self.client.session
self.middleware.process_request(self.request)
self.middleware(self.request)
self.assertIsNotNone(self.request.user)
self.assertFalse(self.request.user.is_anonymous)
@ -22,7 +22,7 @@ class TestAuthenticationMiddleware(TestCase):
# After password change, user should be anonymous
self.user.set_password('new_password')
self.user.save()
self.middleware.process_request(self.request)
self.middleware(self.request)
self.assertIsNotNone(self.request.user)
self.assertTrue(self.request.user.is_anonymous)
# session should be flushed

View File

@ -25,7 +25,7 @@ from django.contrib.sessions.middleware import SessionMiddleware
from django.contrib.sites.requests import RequestSite
from django.core import mail
from django.db import connection
from django.http import HttpRequest
from django.http import HttpRequest, HttpResponse
from django.middleware.csrf import CsrfViewMiddleware, get_token
from django.test import Client, TestCase, override_settings
from django.test.client import RedirectCycleError
@ -650,15 +650,17 @@ class LoginTest(AuthViewsTestCase):
"""
Makes sure that a login rotates the currently-used CSRF token.
"""
def get_response(request):
return HttpResponse()
# Do a GET to establish a CSRF token
# The test client isn't used here as it's a test for middleware.
req = HttpRequest()
CsrfViewMiddleware().process_view(req, LoginView.as_view(), (), {})
CsrfViewMiddleware(get_response).process_view(req, LoginView.as_view(), (), {})
# get_token() triggers CSRF token inclusion in the response
get_token(req)
resp = LoginView.as_view()(req)
resp2 = CsrfViewMiddleware().process_response(req, resp)
csrf_cookie = resp2.cookies.get(settings.CSRF_COOKIE_NAME, None)
resp = CsrfViewMiddleware(LoginView.as_view())(req)
csrf_cookie = resp.cookies.get(settings.CSRF_COOKIE_NAME, None)
token1 = csrf_cookie.coded_value
# Prepare the POST request
@ -668,13 +670,12 @@ class LoginTest(AuthViewsTestCase):
req.POST = {'username': 'testclient', 'password': 'password', 'csrfmiddlewaretoken': token1}
# Use POST request to log in
SessionMiddleware().process_request(req)
CsrfViewMiddleware().process_view(req, LoginView.as_view(), (), {})
SessionMiddleware(get_response).process_request(req)
CsrfViewMiddleware(get_response).process_view(req, LoginView.as_view(), (), {})
req.META["SERVER_NAME"] = "testserver" # Required to have redirect work in login view
req.META["SERVER_PORT"] = 80
resp = LoginView.as_view()(req)
resp2 = CsrfViewMiddleware().process_response(req, resp)
csrf_cookie = resp2.cookies.get(settings.CSRF_COOKIE_NAME, None)
resp = CsrfViewMiddleware(LoginView.as_view())(req)
csrf_cookie = resp.cookies.get(settings.CSRF_COOKIE_NAME, None)
token2 = csrf_cookie.coded_value
# Check the CSRF token switched

105
tests/cache/tests.py vendored
View File

@ -59,6 +59,10 @@ class Unpicklable:
raise pickle.PickleError()
def empty_response(request):
return HttpResponse()
KEY_ERRORS_WITH_MEMCACHED_MSG = (
'Cache key contains characters that will cause errors if used with '
'memcached: %r'
@ -908,30 +912,31 @@ class BaseCacheTests:
self.assertEqual(caches['custom_key2'].get('answer2'), 42)
def test_cache_write_unpicklable_object(self):
update_middleware = UpdateCacheMiddleware()
update_middleware.cache = cache
fetch_middleware = FetchFromCacheMiddleware()
fetch_middleware = FetchFromCacheMiddleware(empty_response)
fetch_middleware.cache = cache
request = self.factory.get('/cache/test')
request._cache_update_cache = True
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertIsNone(get_cache_data)
response = HttpResponse()
content = 'Testing cookie serialization.'
response.content = content
response.set_cookie('foo', 'bar')
update_middleware.process_response(request, response)
def get_response(req):
response = HttpResponse(content)
response.set_cookie('foo', 'bar')
return response
update_middleware = UpdateCacheMiddleware(get_response)
update_middleware.cache = cache
response = update_middleware(request)
get_cache_data = fetch_middleware.process_request(request)
self.assertIsNotNone(get_cache_data)
self.assertEqual(get_cache_data.content, content.encode())
self.assertEqual(get_cache_data.cookies, response.cookies)
update_middleware.process_response(request, get_cache_data)
UpdateCacheMiddleware(lambda req: get_cache_data)(request)
get_cache_data = fetch_middleware.process_request(request)
self.assertIsNotNone(get_cache_data)
self.assertEqual(get_cache_data.content, content.encode())
@ -1769,9 +1774,7 @@ class CacheHEADTest(SimpleTestCase):
cache.clear()
def _set_cache(self, request, msg):
response = HttpResponse()
response.content = msg
return UpdateCacheMiddleware().process_response(request, response)
return UpdateCacheMiddleware(lambda req: HttpResponse(msg))(request)
def test_head_caches_correctly(self):
test_content = 'test content'
@ -1782,7 +1785,7 @@ class CacheHEADTest(SimpleTestCase):
request = self.factory.head(self.path)
request._cache_update_cache = True
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertIsNotNone(get_cache_data)
self.assertEqual(test_content.encode(), get_cache_data.content)
@ -1794,7 +1797,7 @@ class CacheHEADTest(SimpleTestCase):
self._set_cache(request, test_content)
request = self.factory.head(self.path)
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertIsNotNone(get_cache_data)
self.assertEqual(test_content.encode(), get_cache_data.content)
@ -1932,30 +1935,33 @@ class CacheI18nTest(SimpleTestCase):
)
def test_middleware(self):
def set_cache(request, lang, msg):
def get_response(req):
return HttpResponse(msg)
translation.activate(lang)
response = HttpResponse()
response.content = msg
return UpdateCacheMiddleware().process_response(request, response)
return UpdateCacheMiddleware(get_response)(request)
# cache with non empty request.GET
request = self.factory.get(self.path, {'foo': 'bar', 'other': 'true'})
request._cache_update_cache = True
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
# first access, cache must return None
self.assertIsNone(get_cache_data)
response = HttpResponse()
content = 'Check for cache with QUERY_STRING'
response.content = content
UpdateCacheMiddleware().process_response(request, response)
get_cache_data = FetchFromCacheMiddleware().process_request(request)
def get_response(req):
return HttpResponse(content)
UpdateCacheMiddleware(get_response)(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
# cache must return content
self.assertIsNotNone(get_cache_data)
self.assertEqual(get_cache_data.content, content.encode())
# different QUERY_STRING, cache must be empty
request = self.factory.get(self.path, {'foo': 'bar', 'somethingelse': 'true'})
request._cache_update_cache = True
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertIsNone(get_cache_data)
# i18n tests
@ -1965,7 +1971,7 @@ class CacheI18nTest(SimpleTestCase):
request = self.factory.get(self.path)
request._cache_update_cache = True
set_cache(request, 'en', en_message)
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
# The cache can be recovered
self.assertIsNotNone(get_cache_data)
self.assertEqual(get_cache_data.content, en_message.encode())
@ -1976,11 +1982,11 @@ class CacheI18nTest(SimpleTestCase):
# change again the language
translation.activate('en')
# retrieve the content from cache
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertEqual(get_cache_data.content, en_message.encode())
# change again the language
translation.activate('es')
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertEqual(get_cache_data.content, es_message.encode())
# reset the language
translation.deactivate()
@ -1991,14 +1997,15 @@ class CacheI18nTest(SimpleTestCase):
)
def test_middleware_doesnt_cache_streaming_response(self):
request = self.factory.get(self.path)
get_cache_data = FetchFromCacheMiddleware().process_request(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertIsNone(get_cache_data)
content = ['Check for cache with streaming content.']
response = StreamingHttpResponse(content)
UpdateCacheMiddleware().process_response(request, response)
def get_stream_response(req):
return StreamingHttpResponse(['Check for cache with streaming content.'])
get_cache_data = FetchFromCacheMiddleware().process_request(request)
UpdateCacheMiddleware(get_stream_response)(request)
get_cache_data = FetchFromCacheMiddleware(empty_response).process_request(request)
self.assertIsNone(get_cache_data)
@ -2055,17 +2062,18 @@ class CacheMiddlewareTest(SimpleTestCase):
Middleware vs. usage of CacheMiddleware as view decorator and setting attributes
appropriately.
"""
# If no arguments are passed in construction, it's being used as middleware.
middleware = CacheMiddleware()
# If only one argument is passed in construction, it's being used as
# middleware.
middleware = CacheMiddleware(empty_response)
# Now test object attributes against values defined in setUp above
self.assertEqual(middleware.cache_timeout, 30)
self.assertEqual(middleware.key_prefix, 'middlewareprefix')
self.assertEqual(middleware.cache_alias, 'other')
# If arguments are being passed in construction, it's being used as a decorator.
# First, test with "defaults":
as_view_decorator = CacheMiddleware(cache_alias=None, key_prefix=None)
# If more arguments are being passed in construction, it's being used
# as a decorator. First, test with "defaults":
as_view_decorator = CacheMiddleware(empty_response, cache_alias=None, key_prefix=None)
self.assertEqual(as_view_decorator.cache_timeout, 30) # Timeout value for 'default' cache, i.e. 30
self.assertEqual(as_view_decorator.key_prefix, '')
@ -2073,16 +2081,18 @@ class CacheMiddlewareTest(SimpleTestCase):
self.assertEqual(as_view_decorator.cache_alias, 'default')
# Next, test with custom values:
as_view_decorator_with_custom = CacheMiddleware(cache_timeout=60, cache_alias='other', key_prefix='foo')
as_view_decorator_with_custom = CacheMiddleware(
hello_world_view, cache_timeout=60, cache_alias='other', key_prefix='foo'
)
self.assertEqual(as_view_decorator_with_custom.cache_timeout, 60)
self.assertEqual(as_view_decorator_with_custom.key_prefix, 'foo')
self.assertEqual(as_view_decorator_with_custom.cache_alias, 'other')
def test_middleware(self):
middleware = CacheMiddleware()
prefix_middleware = CacheMiddleware(key_prefix='prefix1')
timeout_middleware = CacheMiddleware(cache_timeout=1)
middleware = CacheMiddleware(hello_world_view)
prefix_middleware = CacheMiddleware(hello_world_view, key_prefix='prefix1')
timeout_middleware = CacheMiddleware(hello_world_view, cache_timeout=1)
request = self.factory.get('/view/')
@ -2223,18 +2233,13 @@ class CacheMiddlewareTest(SimpleTestCase):
Django must prevent caching of responses that set a user-specific (and
maybe security sensitive) cookie in response to a cookie-less request.
"""
csrf_middleware = CsrfViewMiddleware()
cache_middleware = CacheMiddleware()
request = self.factory.get('/view/')
self.assertIsNone(cache_middleware.process_request(request))
csrf_middleware = CsrfViewMiddleware(csrf_view)
csrf_middleware.process_view(request, csrf_view, (), {})
cache_middleware = CacheMiddleware(csrf_middleware)
response = csrf_view(request)
response = csrf_middleware.process_response(request, response)
response = cache_middleware.process_response(request, response)
self.assertIsNone(cache_middleware.process_request(request))
cache_middleware(request)
# Inserting a CSRF cookie in a cookie-less request prevented caching.
self.assertIsNone(cache_middleware.process_request(request))

View File

@ -3,7 +3,7 @@ import re
from django.conf import settings
from django.contrib.sessions.backends.cache import SessionStore
from django.core.exceptions import ImproperlyConfigured
from django.http import HttpRequest
from django.http import HttpRequest, HttpResponse
from django.middleware.csrf import (
CSRF_SESSION_KEY, CSRF_TOKEN_LENGTH, REASON_BAD_TOKEN,
REASON_NO_CSRF_COOKIE, CsrfViewMiddleware,
@ -37,7 +37,6 @@ class CsrfViewMiddlewareTestMixin:
"""
_csrf_id = _csrf_id_cookie = '1bcdefghij2bcdefghij3bcdefghij4bcdefghij5bcdefghij6bcdefghijABCD'
mw = CsrfViewMiddleware()
def _get_GET_no_csrf_cookie_request(self):
return TestingHttpRequest()
@ -82,12 +81,12 @@ class CsrfViewMiddlewareTestMixin:
# does use the csrf request processor. By using this, we are testing
# that the view processor is properly lazy and doesn't call get_token()
# until needed.
self.mw.process_request(req)
self.mw.process_view(req, non_token_view_using_request_processor, (), {})
resp = non_token_view_using_request_processor(req)
resp2 = self.mw.process_response(req, resp)
mw = CsrfViewMiddleware(non_token_view_using_request_processor)
mw.process_request(req)
mw.process_view(req, non_token_view_using_request_processor, (), {})
resp = mw(req)
csrf_cookie = resp2.cookies.get(settings.CSRF_COOKIE_NAME, False)
csrf_cookie = resp.cookies.get(settings.CSRF_COOKIE_NAME, False)
self.assertIs(csrf_cookie, False)
# Check the request processing
@ -97,10 +96,11 @@ class CsrfViewMiddlewareTestMixin:
request. This will stop login CSRF.
"""
req = self._get_POST_no_csrf_cookie_request()
self.mw.process_request(req)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
with self.assertLogs('django.security.csrf', 'WARNING') as cm:
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, req2.status_code)
resp = mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, resp.status_code)
self.assertEqual(cm.records[0].getMessage(), 'Forbidden (%s): ' % REASON_NO_CSRF_COOKIE)
def test_process_request_csrf_cookie_no_token(self):
@ -109,10 +109,11 @@ class CsrfViewMiddlewareTestMixin:
the incoming request.
"""
req = self._get_POST_csrf_cookie_request()
self.mw.process_request(req)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
with self.assertLogs('django.security.csrf', 'WARNING') as cm:
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, req2.status_code)
resp = mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, resp.status_code)
self.assertEqual(cm.records[0].getMessage(), 'Forbidden (%s): ' % REASON_BAD_TOKEN)
def test_process_request_csrf_cookie_and_token(self):
@ -120,9 +121,10 @@ class CsrfViewMiddlewareTestMixin:
If both a cookie and a token is present, the middleware lets it through.
"""
req = self._get_POST_request_with_token()
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
def test_process_request_csrf_cookie_no_token_exempt_view(self):
"""
@ -130,9 +132,10 @@ class CsrfViewMiddlewareTestMixin:
has been applied to the view, the middleware lets it through
"""
req = self._get_POST_csrf_cookie_request()
self.mw.process_request(req)
req2 = self.mw.process_view(req, csrf_exempt(post_form_view), (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, csrf_exempt(post_form_view), (), {})
self.assertIsNone(resp)
def test_csrf_token_in_header(self):
"""
@ -140,9 +143,10 @@ class CsrfViewMiddlewareTestMixin:
"""
req = self._get_POST_csrf_cookie_request()
req.META['HTTP_X_CSRFTOKEN'] = self._csrf_id
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
@override_settings(CSRF_HEADER_NAME='HTTP_X_CSRFTOKEN_CUSTOMIZED')
def test_csrf_token_in_header_with_customized_name(self):
@ -151,9 +155,10 @@ class CsrfViewMiddlewareTestMixin:
"""
req = self._get_POST_csrf_cookie_request()
req.META['HTTP_X_CSRFTOKEN_CUSTOMIZED'] = self._csrf_id
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
def test_put_and_delete_rejected(self):
"""
@ -161,16 +166,17 @@ class CsrfViewMiddlewareTestMixin:
"""
req = TestingHttpRequest()
req.method = 'PUT'
mw = CsrfViewMiddleware(post_form_view)
with self.assertLogs('django.security.csrf', 'WARNING') as cm:
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, req2.status_code)
resp = mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, resp.status_code)
self.assertEqual(cm.records[0].getMessage(), 'Forbidden (%s): ' % REASON_NO_CSRF_COOKIE)
req = TestingHttpRequest()
req.method = 'DELETE'
with self.assertLogs('django.security.csrf', 'WARNING') as cm:
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, req2.status_code)
resp = mw.process_view(req, post_form_view, (), {})
self.assertEqual(403, resp.status_code)
self.assertEqual(cm.records[0].getMessage(), 'Forbidden (%s): ' % REASON_NO_CSRF_COOKIE)
def test_put_and_delete_allowed(self):
@ -180,16 +186,17 @@ class CsrfViewMiddlewareTestMixin:
req = self._get_GET_csrf_cookie_request()
req.method = 'PUT'
req.META['HTTP_X_CSRFTOKEN'] = self._csrf_id
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
req = self._get_GET_csrf_cookie_request()
req.method = 'DELETE'
req.META['HTTP_X_CSRFTOKEN'] = self._csrf_id
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
# Tests for the template tag method
def test_token_node_no_csrf_cookie(self):
@ -209,7 +216,8 @@ class CsrfViewMiddlewareTestMixin:
"""
req = self._get_GET_no_csrf_cookie_request()
req.COOKIES[settings.CSRF_COOKIE_NAME] = ""
self.mw.process_view(req, token_view, (), {})
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = token_view(req)
token = get_token(req)
@ -221,8 +229,9 @@ class CsrfViewMiddlewareTestMixin:
CsrfTokenNode works when a CSRF cookie is set.
"""
req = self._get_GET_csrf_cookie_request()
self.mw.process_request(req)
self.mw.process_view(req, token_view, (), {})
mw = CsrfViewMiddleware(token_view)
mw.process_request(req)
mw.process_view(req, token_view, (), {})
resp = token_view(req)
self._check_token_present(resp)
@ -231,8 +240,9 @@ class CsrfViewMiddlewareTestMixin:
get_token still works for a view decorated with 'csrf_exempt'.
"""
req = self._get_GET_csrf_cookie_request()
self.mw.process_request(req)
self.mw.process_view(req, csrf_exempt(token_view), (), {})
mw = CsrfViewMiddleware(token_view)
mw.process_request(req)
mw.process_view(req, csrf_exempt(token_view), (), {})
resp = token_view(req)
self._check_token_present(resp)
@ -250,10 +260,10 @@ class CsrfViewMiddlewareTestMixin:
the middleware (when one was not already present)
"""
req = self._get_GET_no_csrf_cookie_request()
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp2 = self.mw.process_response(req, resp)
csrf_cookie = resp2.cookies[settings.CSRF_COOKIE_NAME]
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
csrf_cookie = resp.cookies[settings.CSRF_COOKIE_NAME]
self._check_token_present(resp, csrf_id=csrf_cookie.value)
def test_cookie_not_reset_on_accepted_request(self):
@ -263,10 +273,10 @@ class CsrfViewMiddlewareTestMixin:
requests. If it appears in the response, it should keep its value.
"""
req = self._get_POST_request_with_token()
self.mw.process_request(req)
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp = self.mw.process_response(req, resp)
mw = CsrfViewMiddleware(token_view)
mw.process_request(req)
mw.process_view(req, token_view, (), {})
resp = mw(req)
csrf_cookie = resp.cookies.get(settings.CSRF_COOKIE_NAME, None)
if csrf_cookie:
self.assertEqual(
@ -284,7 +294,8 @@ class CsrfViewMiddlewareTestMixin:
req.META['HTTP_HOST'] = 'www.example.com'
req.META['HTTP_REFERER'] = 'https://www.evil.org/somepage'
req.META['SERVER_PORT'] = '443'
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(
response,
'Referer checking failed - https://www.evil.org/somepage does not '
@ -302,7 +313,8 @@ class CsrfViewMiddlewareTestMixin:
req.META['HTTP_HOST'] = '@malformed'
req.META['HTTP_REFERER'] = 'https://www.evil.org/somepage'
req.META['SERVER_PORT'] = '443'
response = self.mw.process_view(req, token_view, (), {})
mw = CsrfViewMiddleware(token_view)
response = mw.process_view(req, token_view, (), {})
self.assertEqual(response.status_code, 403)
@override_settings(DEBUG=True)
@ -314,7 +326,8 @@ class CsrfViewMiddlewareTestMixin:
req = self._get_POST_request_with_token()
req._is_secure_override = True
req.META['HTTP_REFERER'] = 'http://http://www.example.com/'
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(
response,
'Referer checking failed - Referer is insecure while host is secure.',
@ -322,23 +335,23 @@ class CsrfViewMiddlewareTestMixin:
)
# Empty
req.META['HTTP_REFERER'] = ''
response = self.mw.process_view(req, post_form_view, (), {})
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(response, malformed_referer_msg, status_code=403)
# Non-ASCII
req.META['HTTP_REFERER'] = 'ØBöIß'
response = self.mw.process_view(req, post_form_view, (), {})
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(response, malformed_referer_msg, status_code=403)
# missing scheme
# >>> urlparse('//example.com/')
# ParseResult(scheme='', netloc='example.com', path='/', params='', query='', fragment='')
req.META['HTTP_REFERER'] = '//example.com/'
response = self.mw.process_view(req, post_form_view, (), {})
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(response, malformed_referer_msg, status_code=403)
# missing netloc
# >>> urlparse('https://')
# ParseResult(scheme='https', netloc='', path='', params='', query='', fragment='')
req.META['HTTP_REFERER'] = 'https://'
response = self.mw.process_view(req, post_form_view, (), {})
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(response, malformed_referer_msg, status_code=403)
@override_settings(ALLOWED_HOSTS=['www.example.com'])
@ -350,9 +363,10 @@ class CsrfViewMiddlewareTestMixin:
req._is_secure_override = True
req.META['HTTP_HOST'] = 'www.example.com'
req.META['HTTP_REFERER'] = 'https://www.example.com/somepage'
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
@override_settings(ALLOWED_HOSTS=['www.example.com'])
def test_https_good_referer_2(self):
@ -365,9 +379,10 @@ class CsrfViewMiddlewareTestMixin:
req._is_secure_override = True
req.META['HTTP_HOST'] = 'www.example.com'
req.META['HTTP_REFERER'] = 'https://www.example.com'
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
def _test_https_good_referer_behind_proxy(self):
req = self._get_POST_request_with_token()
@ -379,9 +394,10 @@ class CsrfViewMiddlewareTestMixin:
'HTTP_X_FORWARDED_HOST': 'www.example.com',
'HTTP_X_FORWARDED_PORT': '443',
})
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
@override_settings(ALLOWED_HOSTS=['www.example.com'], CSRF_TRUSTED_ORIGINS=['dashboard.example.com'])
def test_https_csrf_trusted_origin_allowed(self):
@ -393,9 +409,10 @@ class CsrfViewMiddlewareTestMixin:
req._is_secure_override = True
req.META['HTTP_HOST'] = 'www.example.com'
req.META['HTTP_REFERER'] = 'https://dashboard.example.com'
self.mw.process_request(req)
req2 = self.mw.process_view(req, post_form_view, (), {})
self.assertIsNone(req2)
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
@override_settings(ALLOWED_HOSTS=['www.example.com'], CSRF_TRUSTED_ORIGINS=['.example.com'])
def test_https_csrf_wildcard_trusted_origin_allowed(self):
@ -407,8 +424,9 @@ class CsrfViewMiddlewareTestMixin:
req._is_secure_override = True
req.META['HTTP_HOST'] = 'www.example.com'
req.META['HTTP_REFERER'] = 'https://dashboard.example.com'
self.mw.process_request(req)
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
response = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(response)
def _test_https_good_referer_matches_cookie_domain(self):
@ -416,8 +434,9 @@ class CsrfViewMiddlewareTestMixin:
req._is_secure_override = True
req.META['HTTP_REFERER'] = 'https://foo.example.com/'
req.META['SERVER_PORT'] = '443'
self.mw.process_request(req)
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
response = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(response)
def _test_https_good_referer_matches_cookie_domain_with_different_port(self):
@ -426,8 +445,9 @@ class CsrfViewMiddlewareTestMixin:
req.META['HTTP_HOST'] = 'www.example.com'
req.META['HTTP_REFERER'] = 'https://foo.example.com:4443/'
req.META['SERVER_PORT'] = '4443'
self.mw.process_request(req)
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
response = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(response)
def test_ensures_csrf_cookie_no_logging(self):
@ -479,14 +499,15 @@ class CsrfViewMiddlewareTestMixin:
token = ('ABC' + self._csrf_id)[:CSRF_TOKEN_LENGTH]
req = CsrfPostRequest(token, raise_error=False)
self.mw.process_request(req)
resp = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
mw.process_request(req)
resp = mw.process_view(req, post_form_view, (), {})
self.assertIsNone(resp)
req = CsrfPostRequest(token, raise_error=True)
self.mw.process_request(req)
mw.process_request(req)
with self.assertLogs('django.security.csrf', 'WARNING') as cm:
resp = self.mw.process_view(req, post_form_view, (), {})
resp = mw.process_view(req, post_form_view, (), {})
self.assertEqual(resp.status_code, 403)
self.assertEqual(cm.records[0].getMessage(), 'Forbidden (%s): ' % REASON_BAD_TOKEN)
@ -523,11 +544,11 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
enabled.
"""
req = self._get_GET_no_csrf_cookie_request()
self.mw.process_view(req, ensure_csrf_cookie_view, (), {})
resp = ensure_csrf_cookie_view(req)
resp2 = self.mw.process_response(req, resp)
self.assertTrue(resp2.cookies.get(settings.CSRF_COOKIE_NAME, False))
self.assertIn('Cookie', resp2.get('Vary', ''))
mw = CsrfViewMiddleware(ensure_csrf_cookie_view)
mw.process_view(req, ensure_csrf_cookie_view, (), {})
resp = mw(req)
self.assertTrue(resp.cookies.get(settings.CSRF_COOKIE_NAME, False))
self.assertIn('Cookie', resp.get('Vary', ''))
def test_csrf_cookie_age(self):
"""
@ -543,11 +564,10 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
CSRF_COOKIE_SECURE=True,
CSRF_COOKIE_HTTPONLY=True):
# token_view calls get_token() indirectly
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp2 = self.mw.process_response(req, resp)
max_age = resp2.cookies.get('csrfcookie').get('max-age')
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
max_age = resp.cookies.get('csrfcookie').get('max-age')
self.assertEqual(max_age, MAX_AGE)
def test_csrf_cookie_age_none(self):
@ -565,20 +585,19 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
CSRF_COOKIE_SECURE=True,
CSRF_COOKIE_HTTPONLY=True):
# token_view calls get_token() indirectly
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp2 = self.mw.process_response(req, resp)
max_age = resp2.cookies.get('csrfcookie').get('max-age')
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
max_age = resp.cookies.get('csrfcookie').get('max-age')
self.assertEqual(max_age, '')
def test_csrf_cookie_samesite(self):
req = self._get_GET_no_csrf_cookie_request()
with self.settings(CSRF_COOKIE_NAME='csrfcookie', CSRF_COOKIE_SAMESITE='Strict'):
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp2 = self.mw.process_response(req, resp)
self.assertEqual(resp2.cookies['csrfcookie']['samesite'], 'Strict')
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
self.assertEqual(resp.cookies['csrfcookie']['samesite'], 'Strict')
def test_process_view_token_too_long(self):
"""
@ -587,10 +606,10 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
"""
req = self._get_GET_no_csrf_cookie_request()
req.COOKIES[settings.CSRF_COOKIE_NAME] = 'x' * 100000
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp2 = self.mw.process_response(req, resp)
csrf_cookie = resp2.cookies.get(settings.CSRF_COOKIE_NAME, False)
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
csrf_cookie = resp.cookies.get(settings.CSRF_COOKIE_NAME, False)
self.assertEqual(len(csrf_cookie.value), CSRF_TOKEN_LENGTH)
def test_process_view_token_invalid_chars(self):
@ -601,10 +620,10 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
token = ('!@#' + self._csrf_id)[:CSRF_TOKEN_LENGTH]
req = self._get_GET_no_csrf_cookie_request()
req.COOKIES[settings.CSRF_COOKIE_NAME] = token
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
resp2 = self.mw.process_response(req, resp)
csrf_cookie = resp2.cookies.get(settings.CSRF_COOKIE_NAME, False)
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
csrf_cookie = resp.cookies.get(settings.CSRF_COOKIE_NAME, False)
self.assertEqual(len(csrf_cookie.value), CSRF_TOKEN_LENGTH)
self.assertNotEqual(csrf_cookie.value, token)
@ -613,11 +632,11 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
The csrf token is reset from a bare secret.
"""
req = self._get_POST_bare_secret_csrf_cookie_request_with_token()
self.mw.process_request(req)
req2 = self.mw.process_view(req, token_view, (), {})
self.assertIsNone(req2)
resp = token_view(req)
resp = self.mw.process_response(req, resp)
mw = CsrfViewMiddleware(token_view)
mw.process_request(req)
resp = mw.process_view(req, token_view, (), {})
self.assertIsNone(resp)
resp = mw(req)
self.assertIn(settings.CSRF_COOKIE_NAME, resp.cookies, "Cookie was not reset from bare secret")
csrf_cookie = resp.cookies[settings.CSRF_COOKIE_NAME]
self.assertEqual(len(csrf_cookie.value), CSRF_TOKEN_LENGTH)
@ -655,7 +674,8 @@ class CsrfViewMiddlewareTests(CsrfViewMiddlewareTestMixin, SimpleTestCase):
req._is_secure_override = True
req.META['HTTP_REFERER'] = 'http://example.com/'
req.META['SERVER_PORT'] = '443'
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(
response,
'Referer checking failed - Referer is insecure while host is secure.',
@ -685,7 +705,8 @@ class CsrfViewMiddlewareUseSessionsTests(CsrfViewMiddlewareTestMixin, SimpleTest
'SessionMiddleware must appear before CsrfViewMiddleware in MIDDLEWARE.'
)
with self.assertRaisesMessage(ImproperlyConfigured, msg):
self.mw.process_request(HttpRequest())
mw = CsrfViewMiddleware(lambda req: HttpResponse())
mw.process_request(HttpRequest())
def test_process_response_get_token_used(self):
"""The ensure_csrf_cookie() decorator works without middleware."""
@ -696,14 +717,13 @@ class CsrfViewMiddlewareUseSessionsTests(CsrfViewMiddlewareTestMixin, SimpleTest
def test_session_modify(self):
"""The session isn't saved if the CSRF cookie is unchanged."""
req = self._get_GET_no_csrf_cookie_request()
self.mw.process_view(req, ensure_csrf_cookie_view, (), {})
resp = ensure_csrf_cookie_view(req)
self.mw.process_response(req, resp)
mw = CsrfViewMiddleware(ensure_csrf_cookie_view)
mw.process_view(req, ensure_csrf_cookie_view, (), {})
mw(req)
self.assertIsNotNone(req.session.get(CSRF_SESSION_KEY))
req.session.modified = False
self.mw.process_view(req, ensure_csrf_cookie_view, (), {})
resp = ensure_csrf_cookie_view(req)
self.mw.process_response(req, resp)
mw.process_view(req, ensure_csrf_cookie_view, (), {})
mw(req)
self.assertFalse(req.session.modified)
def test_ensures_csrf_cookie_with_middleware(self):
@ -712,9 +732,9 @@ class CsrfViewMiddlewareUseSessionsTests(CsrfViewMiddlewareTestMixin, SimpleTest
enabled.
"""
req = self._get_GET_no_csrf_cookie_request()
self.mw.process_view(req, ensure_csrf_cookie_view, (), {})
resp = ensure_csrf_cookie_view(req)
self.mw.process_response(req, resp)
mw = CsrfViewMiddleware(ensure_csrf_cookie_view)
mw.process_view(req, ensure_csrf_cookie_view, (), {})
mw(req)
self.assertTrue(req.session.get(CSRF_SESSION_KEY, False))
def test_token_node_with_new_csrf_cookie(self):
@ -723,9 +743,9 @@ class CsrfViewMiddlewareUseSessionsTests(CsrfViewMiddlewareTestMixin, SimpleTest
(when one was not already present).
"""
req = self._get_GET_no_csrf_cookie_request()
self.mw.process_view(req, token_view, (), {})
resp = token_view(req)
self.mw.process_response(req, resp)
mw = CsrfViewMiddleware(token_view)
mw.process_view(req, token_view, (), {})
resp = mw(req)
csrf_cookie = req.session[CSRF_SESSION_KEY]
self._check_token_present(resp, csrf_id=csrf_cookie)
@ -766,7 +786,8 @@ class CsrfViewMiddlewareUseSessionsTests(CsrfViewMiddlewareTestMixin, SimpleTest
req._is_secure_override = True
req.META['HTTP_REFERER'] = 'http://example.com/'
req.META['SERVER_PORT'] = '443'
response = self.mw.process_view(req, post_form_view, (), {})
mw = CsrfViewMiddleware(post_form_view)
response = mw.process_view(req, post_form_view, (), {})
self.assertContains(
response,
'Referer checking failed - Referer is insecure while host is secure.',

View File

@ -466,7 +466,7 @@ class XFrameOptionsDecoratorsTests(TestCase):
# Since the real purpose of the exempt decorator is to suppress
# the middleware's functionality, let's make sure it actually works...
r = XFrameOptionsMiddleware().process_response(req, resp)
r = XFrameOptionsMiddleware(a_view)(req)
self.assertIsNone(r.get('X-Frame-Options', None))

View File

@ -0,0 +1,39 @@
from django.contrib.sessions.middleware import SessionMiddleware
from django.middleware.cache import (
CacheMiddleware, FetchFromCacheMiddleware, UpdateCacheMiddleware,
)
from django.middleware.common import CommonMiddleware
from django.middleware.security import SecurityMiddleware
from django.test import SimpleTestCase
from django.utils.deprecation import RemovedInDjango40Warning
class MiddlewareMixinTests(SimpleTestCase):
"""
Deprecation warning is raised when using get_response=None.
"""
msg = 'Passing None for the middleware get_response argument is deprecated.'
def test_deprecation(self):
with self.assertRaisesMessage(RemovedInDjango40Warning, self.msg):
CommonMiddleware()
def test_passing_explicit_none(self):
with self.assertRaisesMessage(RemovedInDjango40Warning, self.msg):
CommonMiddleware(None)
def test_subclass_deprecation(self):
"""
Deprecation warning is raised in subclasses overriding __init__()
without calling super().
"""
for middleware in [
SessionMiddleware,
CacheMiddleware,
FetchFromCacheMiddleware,
UpdateCacheMiddleware,
SecurityMiddleware,
]:
with self.subTest(middleware=middleware):
with self.assertRaisesMessage(RemovedInDjango40Warning, self.msg):
middleware()

View File

@ -2,7 +2,7 @@ import os
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.http import HttpResponsePermanentRedirect
from django.http import HttpResponse, HttpResponsePermanentRedirect
from django.middleware.locale import LocaleMiddleware
from django.template import Context, Template
from django.test import SimpleTestCase, override_settings
@ -100,7 +100,7 @@ class RequestURLConfTests(SimpleTestCase):
def test_request_urlconf_considered(self):
request = RequestFactory().get('/nl/')
request.urlconf = 'i18n.patterns.urls.default'
middleware = LocaleMiddleware()
middleware = LocaleMiddleware(lambda req: HttpResponse())
with translation.override('nl'):
middleware.process_request(request)
self.assertEqual(request.LANGUAGE_CODE, 'nl')

View File

@ -6,13 +6,10 @@ from django.http import HttpRequest, HttpResponse
class MiddlewareTests(unittest.TestCase):
def setUp(self):
self.middleware = MessageMiddleware()
def test_response_without_messages(self):
"""
MessageMiddleware is tolerant of messages not existing on request.
"""
request = HttpRequest()
response = HttpResponse()
self.middleware.process_response(request, response)
MessageMiddleware(lambda req: HttpResponse()).process_response(request, response)

View File

@ -4,21 +4,22 @@ from django.test.utils import override_settings
class SecurityMiddlewareTest(SimpleTestCase):
@property
def middleware(self):
def middleware(self, *args, **kwargs):
from django.middleware.security import SecurityMiddleware
return SecurityMiddleware()
return SecurityMiddleware(self.response(*args, **kwargs))
@property
def secure_request_kwargs(self):
return {"wsgi.url_scheme": "https"}
def response(self, *args, headers=None, **kwargs):
def get_response(req):
response = HttpResponse(*args, **kwargs)
if headers:
for k, v in headers.items():
response[k] = v
return response
return get_response
def process_response(self, *args, secure=False, request=None, **kwargs):
request_kwargs = {}
@ -26,11 +27,10 @@ class SecurityMiddlewareTest(SimpleTestCase):
request_kwargs.update(self.secure_request_kwargs)
if request is None:
request = self.request.get("/some/url", **request_kwargs)
ret = self.middleware.process_request(request)
ret = self.middleware(*args, **kwargs).process_request(request)
if ret:
return ret
return self.middleware.process_response(
request, self.response(*args, **kwargs))
return self.middleware(*args, **kwargs)(request)
request = RequestFactory()
@ -38,7 +38,7 @@ class SecurityMiddlewareTest(SimpleTestCase):
if secure:
kwargs.update(self.secure_request_kwargs)
req = getattr(self.request, method.lower())(*args, **kwargs)
return self.middleware.process_request(req)
return self.middleware().process_request(req)
@override_settings(SECURE_HSTS_SECONDS=3600)
def test_sts_on(self):

View File

@ -23,6 +23,14 @@ from django.test import RequestFactory, SimpleTestCase, override_settings
int2byte = struct.Struct(">B").pack
def get_response_empty(request):
return HttpResponse()
def get_response_404(request):
return HttpResponseNotFound()
@override_settings(ROOT_URLCONF='middleware.urls')
class CommonMiddlewareTest(SimpleTestCase):
@ -34,19 +42,23 @@ class CommonMiddlewareTest(SimpleTestCase):
URLs with slashes should go unmolested.
"""
request = self.rf.get('/slash/')
self.assertIsNone(CommonMiddleware().process_request(request))
response = HttpResponseNotFound()
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertIsNone(CommonMiddleware(get_response_404).process_request(request))
self.assertEqual(CommonMiddleware(get_response_404)(request).status_code, 404)
@override_settings(APPEND_SLASH=True)
def test_append_slash_slashless_resource(self):
"""
Matches to explicit slashless URLs should go unmolested.
"""
def get_response(req):
return HttpResponse("Here's the text of the Web page.")
request = self.rf.get('/noslash')
self.assertIsNone(CommonMiddleware().process_request(request))
response = HttpResponse("Here's the text of the Web page.")
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertIsNone(CommonMiddleware(get_response).process_request(request))
self.assertEqual(
CommonMiddleware(get_response)(request).content,
b"Here's the text of the Web page.",
)
@override_settings(APPEND_SLASH=True)
def test_append_slash_slashless_unknown(self):
@ -54,8 +66,8 @@ class CommonMiddlewareTest(SimpleTestCase):
APPEND_SLASH should not redirect to unknown resources.
"""
request = self.rf.get('/unknown')
response = HttpResponseNotFound()
self.assertEqual(CommonMiddleware().process_response(request, response), response)
response = CommonMiddleware(get_response_404)(request)
self.assertEqual(response.status_code, 404)
@override_settings(APPEND_SLASH=True)
def test_append_slash_redirect(self):
@ -63,7 +75,7 @@ class CommonMiddlewareTest(SimpleTestCase):
APPEND_SLASH should redirect slashless URLs to a valid pattern.
"""
request = self.rf.get('/slash')
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
@override_settings(APPEND_SLASH=True)
@ -72,9 +84,8 @@ class CommonMiddlewareTest(SimpleTestCase):
APPEND_SLASH should preserve querystrings when redirecting.
"""
request = self.rf.get('/slash?test=1')
response = HttpResponseNotFound()
r = CommonMiddleware().process_response(request, response)
self.assertEqual(r.url, '/slash/?test=1')
resp = CommonMiddleware(get_response_404)(request)
self.assertEqual(resp.url, '/slash/?test=1')
@override_settings(APPEND_SLASH=True)
def test_append_slash_redirect_querystring_have_slash(self):
@ -83,10 +94,9 @@ class CommonMiddlewareTest(SimpleTestCase):
with a querystring ending with slash.
"""
request = self.rf.get('/slash?test=slash/')
response = HttpResponseNotFound()
r = CommonMiddleware().process_response(request, response)
self.assertIsInstance(r, HttpResponsePermanentRedirect)
self.assertEqual(r.url, '/slash/?test=slash/')
resp = CommonMiddleware(get_response_404)(request)
self.assertIsInstance(resp, HttpResponsePermanentRedirect)
self.assertEqual(resp.url, '/slash/?test=slash/')
@override_settings(APPEND_SLASH=True, DEBUG=True)
def test_append_slash_no_redirect_on_POST_in_DEBUG(self):
@ -98,17 +108,16 @@ class CommonMiddlewareTest(SimpleTestCase):
msg = "maintaining %s data. Change your form to point to testserver/slash/"
request = self.rf.get('/slash')
request.method = 'POST'
response = HttpResponseNotFound()
with self.assertRaisesMessage(RuntimeError, msg % request.method):
CommonMiddleware().process_response(request, response)
CommonMiddleware(get_response_404)(request)
request = self.rf.get('/slash')
request.method = 'PUT'
with self.assertRaisesMessage(RuntimeError, msg % request.method):
CommonMiddleware().process_response(request, response)
CommonMiddleware(get_response_404)(request)
request = self.rf.get('/slash')
request.method = 'PATCH'
with self.assertRaisesMessage(RuntimeError, msg % request.method):
CommonMiddleware().process_response(request, response)
CommonMiddleware(get_response_404)(request)
@override_settings(APPEND_SLASH=False)
def test_append_slash_disabled(self):
@ -116,8 +125,7 @@ class CommonMiddlewareTest(SimpleTestCase):
Disabling append slash functionality should leave slashless URLs alone.
"""
request = self.rf.get('/slash')
response = HttpResponseNotFound()
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertEqual(CommonMiddleware(get_response_404)(request).status_code, 404)
@override_settings(APPEND_SLASH=True)
def test_append_slash_quoted(self):
@ -125,8 +133,7 @@ class CommonMiddlewareTest(SimpleTestCase):
URLs which require quoting should be redirected to their slash version.
"""
request = self.rf.get(quote('/needsquoting#'))
response = HttpResponseNotFound()
r = CommonMiddleware().process_response(request, response)
r = CommonMiddleware(get_response_404)(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, '/needsquoting%23/')
@ -141,32 +148,31 @@ class CommonMiddlewareTest(SimpleTestCase):
"""
# Use 4 slashes because of RequestFactory behavior.
request = self.rf.get('////evil.com/security')
response = HttpResponseNotFound()
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_404).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, '/%2Fevil.com/security/')
r = CommonMiddleware().process_response(request, response)
r = CommonMiddleware(get_response_404)(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, '/%2Fevil.com/security/')
@override_settings(APPEND_SLASH=False, PREPEND_WWW=True)
def test_prepend_www(self):
request = self.rf.get('/path/')
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, 'http://www.testserver/path/')
@override_settings(APPEND_SLASH=True, PREPEND_WWW=True)
def test_prepend_www_append_slash_have_slash(self):
request = self.rf.get('/slash/')
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, 'http://www.testserver/slash/')
@override_settings(APPEND_SLASH=True, PREPEND_WWW=True)
def test_prepend_www_append_slash_slashless(self):
request = self.rf.get('/slash')
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, 'http://www.testserver/slash/')
@ -180,20 +186,21 @@ class CommonMiddlewareTest(SimpleTestCase):
"""
request = self.rf.get('/customurlconf/slash/')
request.urlconf = 'middleware.extra_urls'
self.assertIsNone(CommonMiddleware().process_request(request))
response = HttpResponseNotFound()
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertIsNone(CommonMiddleware(get_response_404).process_request(request))
self.assertEqual(CommonMiddleware(get_response_404)(request).status_code, 404)
@override_settings(APPEND_SLASH=True)
def test_append_slash_slashless_resource_custom_urlconf(self):
"""
Matches to explicit slashless URLs should go unmolested.
"""
def get_response(req):
return HttpResponse("Web content")
request = self.rf.get('/customurlconf/noslash')
request.urlconf = 'middleware.extra_urls'
self.assertIsNone(CommonMiddleware().process_request(request))
response = HttpResponse("Here's the text of the Web page.")
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertIsNone(CommonMiddleware(get_response).process_request(request))
self.assertEqual(CommonMiddleware(get_response)(request).content, b'Web content')
@override_settings(APPEND_SLASH=True)
def test_append_slash_slashless_unknown_custom_urlconf(self):
@ -202,9 +209,8 @@ class CommonMiddlewareTest(SimpleTestCase):
"""
request = self.rf.get('/customurlconf/unknown')
request.urlconf = 'middleware.extra_urls'
self.assertIsNone(CommonMiddleware().process_request(request))
response = HttpResponseNotFound()
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertIsNone(CommonMiddleware(get_response_404).process_request(request))
self.assertEqual(CommonMiddleware(get_response_404)(request).status_code, 404)
@override_settings(APPEND_SLASH=True)
def test_append_slash_redirect_custom_urlconf(self):
@ -213,8 +219,7 @@ class CommonMiddlewareTest(SimpleTestCase):
"""
request = self.rf.get('/customurlconf/slash')
request.urlconf = 'middleware.extra_urls'
response = HttpResponseNotFound()
r = CommonMiddleware().process_response(request, response)
r = CommonMiddleware(get_response_404)(request)
self.assertIsNotNone(r, "CommonMiddleware failed to return APPEND_SLASH redirect using request.urlconf")
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, '/customurlconf/slash/')
@ -229,9 +234,8 @@ class CommonMiddlewareTest(SimpleTestCase):
request = self.rf.get('/customurlconf/slash')
request.urlconf = 'middleware.extra_urls'
request.method = 'POST'
response = HttpResponseNotFound()
with self.assertRaisesMessage(RuntimeError, 'end in a slash'):
CommonMiddleware().process_response(request, response)
CommonMiddleware(get_response_404)(request)
@override_settings(APPEND_SLASH=False)
def test_append_slash_disabled_custom_urlconf(self):
@ -240,9 +244,8 @@ class CommonMiddlewareTest(SimpleTestCase):
"""
request = self.rf.get('/customurlconf/slash')
request.urlconf = 'middleware.extra_urls'
self.assertIsNone(CommonMiddleware().process_request(request))
response = HttpResponseNotFound()
self.assertEqual(CommonMiddleware().process_response(request, response), response)
self.assertIsNone(CommonMiddleware(get_response_404).process_request(request))
self.assertEqual(CommonMiddleware(get_response_404)(request).status_code, 404)
@override_settings(APPEND_SLASH=True)
def test_append_slash_quoted_custom_urlconf(self):
@ -251,8 +254,7 @@ class CommonMiddlewareTest(SimpleTestCase):
"""
request = self.rf.get(quote('/customurlconf/needsquoting#'))
request.urlconf = 'middleware.extra_urls'
response = HttpResponseNotFound()
r = CommonMiddleware().process_response(request, response)
r = CommonMiddleware(get_response_404)(request)
self.assertIsNotNone(r, "CommonMiddleware failed to return APPEND_SLASH redirect using request.urlconf")
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, '/customurlconf/needsquoting%23/')
@ -261,7 +263,7 @@ class CommonMiddlewareTest(SimpleTestCase):
def test_prepend_www_custom_urlconf(self):
request = self.rf.get('/customurlconf/path/')
request.urlconf = 'middleware.extra_urls'
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, 'http://www.testserver/customurlconf/path/')
@ -269,7 +271,7 @@ class CommonMiddlewareTest(SimpleTestCase):
def test_prepend_www_append_slash_have_slash_custom_urlconf(self):
request = self.rf.get('/customurlconf/slash/')
request.urlconf = 'middleware.extra_urls'
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, 'http://www.testserver/customurlconf/slash/')
@ -277,29 +279,39 @@ class CommonMiddlewareTest(SimpleTestCase):
def test_prepend_www_append_slash_slashless_custom_urlconf(self):
request = self.rf.get('/customurlconf/slash')
request.urlconf = 'middleware.extra_urls'
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, 'http://www.testserver/customurlconf/slash/')
# Tests for the Content-Length header
def test_content_length_header_added(self):
def get_response(req):
response = HttpResponse('content')
self.assertNotIn('Content-Length', response)
response = CommonMiddleware().process_response(HttpRequest(), response)
return response
response = CommonMiddleware(get_response)(self.rf.get('/'))
self.assertEqual(int(response['Content-Length']), len(response.content))
def test_content_length_header_not_added_for_streaming_response(self):
def get_response(req):
response = StreamingHttpResponse('content')
self.assertNotIn('Content-Length', response)
response = CommonMiddleware().process_response(HttpRequest(), response)
return response
response = CommonMiddleware(get_response)(self.rf.get('/'))
self.assertNotIn('Content-Length', response)
def test_content_length_header_not_changed(self):
bad_content_length = 500
def get_response(req):
response = HttpResponse()
bad_content_length = len(response.content) + 10
response['Content-Length'] = bad_content_length
response = CommonMiddleware().process_response(HttpRequest(), response)
return response
response = CommonMiddleware(get_response)(self.rf.get('/'))
self.assertEqual(int(response['Content-Length']), bad_content_length)
# Other tests
@ -309,19 +321,18 @@ class CommonMiddlewareTest(SimpleTestCase):
request = self.rf.get('/slash')
request.META['HTTP_USER_AGENT'] = 'foo'
with self.assertRaisesMessage(PermissionDenied, 'Forbidden user agent'):
CommonMiddleware().process_request(request)
CommonMiddleware(get_response_empty).process_request(request)
def test_non_ascii_query_string_does_not_crash(self):
"""Regression test for #15152"""
request = self.rf.get('/slash')
request.META['QUERY_STRING'] = 'drink=café'
r = CommonMiddleware().process_request(request)
r = CommonMiddleware(get_response_empty).process_request(request)
self.assertEqual(r.status_code, 301)
def test_response_redirect_class(self):
request = self.rf.get('/slash')
response = HttpResponseNotFound()
r = CommonMiddleware().process_response(request, response)
r = CommonMiddleware(get_response_404)(request)
self.assertEqual(r.status_code, 301)
self.assertEqual(r.url, '/slash/')
self.assertIsInstance(r, HttpResponsePermanentRedirect)
@ -331,8 +342,7 @@ class CommonMiddlewareTest(SimpleTestCase):
response_redirect_class = HttpResponseRedirect
request = self.rf.get('/slash')
response = HttpResponseNotFound()
r = MyCommonMiddleware().process_response(request, response)
r = MyCommonMiddleware(get_response_404)(request)
self.assertEqual(r.status_code, 302)
self.assertEqual(r.url, '/slash/')
self.assertIsInstance(r, HttpResponseRedirect)
@ -348,21 +358,23 @@ class BrokenLinkEmailsMiddlewareTest(SimpleTestCase):
def setUp(self):
self.req = self.rf.get('/regular_url/that/does/not/exist')
self.resp = self.client.get(self.req.path)
def get_response(self, req):
return self.client.get(req.path)
def test_404_error_reporting(self):
self.req.META['HTTP_REFERER'] = '/another/url/'
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 1)
self.assertIn('Broken', mail.outbox[0].subject)
def test_404_error_reporting_no_referer(self):
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
def test_404_error_reporting_ignored_url(self):
self.req.path = self.req.path_info = 'foo_url/that/does/not/exist'
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
def test_custom_request_checker(self):
@ -378,10 +390,10 @@ class BrokenLinkEmailsMiddlewareTest(SimpleTestCase):
self.req.META['HTTP_REFERER'] = '/another/url/'
self.req.META['HTTP_USER_AGENT'] = 'Spider machine 3.4'
SubclassedMiddleware().process_response(self.req, self.resp)
SubclassedMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
self.req.META['HTTP_USER_AGENT'] = 'My user agent'
SubclassedMiddleware().process_response(self.req, self.resp)
SubclassedMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 1)
def test_referer_equal_to_requested_url(self):
@ -390,12 +402,12 @@ class BrokenLinkEmailsMiddlewareTest(SimpleTestCase):
an referer check (#25302).
"""
self.req.META['HTTP_REFERER'] = self.req.path
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
# URL with scheme and domain should also be ignored
self.req.META['HTTP_REFERER'] = 'http://testserver%s' % self.req.path
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
# URL with a different scheme should be ignored as well because bots
@ -403,26 +415,26 @@ class BrokenLinkEmailsMiddlewareTest(SimpleTestCase):
self.req.META['HTTP_X_PROTO'] = 'https'
self.req.META['SERVER_PORT'] = 443
with self.settings(SECURE_PROXY_SSL_HEADER=('HTTP_X_PROTO', 'https')):
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
def test_referer_equal_to_requested_url_on_another_domain(self):
self.req.META['HTTP_REFERER'] = 'http://anotherserver%s' % self.req.path
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 1)
@override_settings(APPEND_SLASH=True)
def test_referer_equal_to_requested_url_without_trailing_slash_when_append_slash_is_set(self):
self.req.path = self.req.path_info = '/regular_url/that/does/not/exist/'
self.req.META['HTTP_REFERER'] = self.req.path_info[:-1]
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 0)
@override_settings(APPEND_SLASH=False)
def test_referer_equal_to_requested_url_without_trailing_slash_when_append_slash_is_unset(self):
self.req.path = self.req.path_info = '/regular_url/that/does/not/exist/'
self.req.META['HTTP_REFERER'] = self.req.path_info[:-1]
BrokenLinkEmailsMiddleware().process_response(self.req, self.resp)
BrokenLinkEmailsMiddleware(self.get_response)(self.req)
self.assertEqual(len(mail.outbox), 1)
@ -432,139 +444,171 @@ class ConditionalGetMiddlewareTest(SimpleTestCase):
def setUp(self):
self.req = self.request_factory.get('/')
self.resp = self.client.get(self.req.path_info)
self.resp_headers = {}
def get_response(self, req):
resp = self.client.get(req.path_info)
for key, value in self.resp_headers.items():
resp[key] = value
return resp
# Tests for the ETag header
def test_middleware_calculates_etag(self):
self.assertNotIn('ETag', self.resp)
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
self.assertNotEqual('', self.resp['ETag'])
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
self.assertNotEqual('', resp['ETag'])
def test_middleware_wont_overwrite_etag(self):
self.resp['ETag'] = 'eggs'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
self.assertEqual('eggs', self.resp['ETag'])
self.resp_headers['ETag'] = 'eggs'
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
self.assertEqual('eggs', resp['ETag'])
def test_no_etag_streaming_response(self):
res = StreamingHttpResponse(['content'])
self.assertFalse(ConditionalGetMiddleware().process_response(self.req, res).has_header('ETag'))
def get_response(req):
return StreamingHttpResponse(['content'])
self.assertFalse(ConditionalGetMiddleware(get_response)(self.req).has_header('ETag'))
def test_no_etag_response_empty_content(self):
res = HttpResponse()
self.assertFalse(
ConditionalGetMiddleware().process_response(self.req, res).has_header('ETag')
)
def get_response(req):
return HttpResponse()
self.assertFalse(ConditionalGetMiddleware(get_response)(self.req).has_header('ETag'))
def test_no_etag_no_store_cache(self):
self.resp['Cache-Control'] = 'No-Cache, No-Store, Max-age=0'
self.assertFalse(ConditionalGetMiddleware().process_response(self.req, self.resp).has_header('ETag'))
self.resp_headers['Cache-Control'] = 'No-Cache, No-Store, Max-age=0'
self.assertFalse(ConditionalGetMiddleware(self.get_response)(self.req).has_header('ETag'))
def test_etag_extended_cache_control(self):
self.resp['Cache-Control'] = 'my-directive="my-no-store"'
self.assertTrue(ConditionalGetMiddleware().process_response(self.req, self.resp).has_header('ETag'))
self.resp_headers['Cache-Control'] = 'my-directive="my-no-store"'
self.assertTrue(ConditionalGetMiddleware(self.get_response)(self.req).has_header('ETag'))
def test_if_none_match_and_no_etag(self):
self.req.META['HTTP_IF_NONE_MATCH'] = 'spam'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
def test_no_if_none_match_and_etag(self):
self.resp['ETag'] = 'eggs'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
self.resp_headers['ETag'] = 'eggs'
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
def test_if_none_match_and_same_etag(self):
self.req.META['HTTP_IF_NONE_MATCH'] = self.resp['ETag'] = '"spam"'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 304)
self.req.META['HTTP_IF_NONE_MATCH'] = '"spam"'
self.resp_headers['ETag'] = '"spam"'
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 304)
def test_if_none_match_and_different_etag(self):
self.req.META['HTTP_IF_NONE_MATCH'] = 'spam'
self.resp['ETag'] = 'eggs'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
self.resp_headers['ETag'] = 'eggs'
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
def test_if_none_match_and_redirect(self):
self.req.META['HTTP_IF_NONE_MATCH'] = self.resp['ETag'] = 'spam'
self.resp['Location'] = '/'
self.resp.status_code = 301
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 301)
def get_response(req):
resp = self.client.get(req.path_info)
resp['ETag'] = 'spam'
resp['Location'] = '/'
resp.status_code = 301
return resp
self.req.META['HTTP_IF_NONE_MATCH'] = 'spam'
resp = ConditionalGetMiddleware(get_response)(self.req)
self.assertEqual(resp.status_code, 301)
def test_if_none_match_and_client_error(self):
self.req.META['HTTP_IF_NONE_MATCH'] = self.resp['ETag'] = 'spam'
self.resp.status_code = 400
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 400)
def get_response(req):
resp = self.client.get(req.path_info)
resp['ETag'] = 'spam'
resp.status_code = 400
return resp
self.req.META['HTTP_IF_NONE_MATCH'] = 'spam'
resp = ConditionalGetMiddleware(get_response)(self.req)
self.assertEqual(resp.status_code, 400)
# Tests for the Last-Modified header
def test_if_modified_since_and_no_last_modified(self):
self.req.META['HTTP_IF_MODIFIED_SINCE'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
def test_no_if_modified_since_and_last_modified(self):
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 200)
self.resp_headers['Last-Modified'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 200)
def test_if_modified_since_and_same_last_modified(self):
self.req.META['HTTP_IF_MODIFIED_SINCE'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.resp_headers['Last-Modified'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(self.resp.status_code, 304)
def test_if_modified_since_and_last_modified_in_the_past(self):
self.req.META['HTTP_IF_MODIFIED_SINCE'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 304)
self.resp_headers['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(resp.status_code, 304)
def test_if_modified_since_and_last_modified_in_the_future(self):
self.req.META['HTTP_IF_MODIFIED_SINCE'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:41:44 GMT'
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.resp_headers['Last-Modified'] = 'Sat, 12 Feb 2011 17:41:44 GMT'
self.resp = ConditionalGetMiddleware(self.get_response)(self.req)
self.assertEqual(self.resp.status_code, 200)
def test_if_modified_since_and_redirect(self):
def get_response(req):
resp = self.client.get(req.path_info)
resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
resp['Location'] = '/'
resp.status_code = 301
return resp
self.req.META['HTTP_IF_MODIFIED_SINCE'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
self.resp['Location'] = '/'
self.resp.status_code = 301
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 301)
resp = ConditionalGetMiddleware(get_response)(self.req)
self.assertEqual(resp.status_code, 301)
def test_if_modified_since_and_client_error(self):
def get_response(req):
resp = self.client.get(req.path_info)
resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
resp.status_code = 400
return resp
self.req.META['HTTP_IF_MODIFIED_SINCE'] = 'Sat, 12 Feb 2011 17:38:44 GMT'
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
self.resp.status_code = 400
self.resp = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.assertEqual(self.resp.status_code, 400)
resp = ConditionalGetMiddleware(get_response)(self.req)
self.assertEqual(resp.status_code, 400)
def test_not_modified_headers(self):
"""
The 304 Not Modified response should include only the headers required
by section 4.1 of RFC 7232, Last-Modified, and the cookies.
"""
self.req.META['HTTP_IF_NONE_MATCH'] = self.resp['ETag'] = '"spam"'
self.resp['Date'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
self.resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
self.resp['Expires'] = 'Sun, 13 Feb 2011 17:35:44 GMT'
self.resp['Vary'] = 'Cookie'
self.resp['Cache-Control'] = 'public'
self.resp['Content-Location'] = '/alt'
self.resp['Content-Language'] = 'en' # shouldn't be preserved
self.resp.set_cookie('key', 'value')
def get_response(req):
resp = self.client.get(req.path_info)
resp['Date'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
resp['Last-Modified'] = 'Sat, 12 Feb 2011 17:35:44 GMT'
resp['Expires'] = 'Sun, 13 Feb 2011 17:35:44 GMT'
resp['Vary'] = 'Cookie'
resp['Cache-Control'] = 'public'
resp['Content-Location'] = '/alt'
resp['Content-Language'] = 'en' # shouldn't be preserved
resp['ETag'] = '"spam"'
resp.set_cookie('key', 'value')
return resp
new_response = ConditionalGetMiddleware().process_response(self.req, self.resp)
self.req.META['HTTP_IF_NONE_MATCH'] = '"spam"'
new_response = ConditionalGetMiddleware(get_response)(self.req)
self.assertEqual(new_response.status_code, 304)
base_response = get_response(self.req)
for header in ('Cache-Control', 'Content-Location', 'Date', 'ETag', 'Expires', 'Last-Modified', 'Vary'):
self.assertEqual(new_response[header], self.resp[header])
self.assertEqual(new_response.cookies, self.resp.cookies)
self.assertEqual(new_response[header], base_response[header])
self.assertEqual(new_response.cookies, base_response.cookies)
self.assertNotIn('Content-Language', new_response)
def test_no_unsafe(self):
@ -574,11 +618,13 @@ class ConditionalGetMiddlewareTest(SimpleTestCase):
ConditionalGetMiddleware is called, so it's too late to return a 412
Precondition Failed.
"""
get_response = ConditionalGetMiddleware().process_response(self.req, self.resp)
etag = get_response['ETag']
def get_200_response(req):
return HttpResponse(status=200)
response = ConditionalGetMiddleware(self.get_response)(self.req)
etag = response['ETag']
put_request = self.request_factory.put('/', HTTP_IF_MATCH=etag)
put_response = HttpResponse(status=200)
conditional_get_response = ConditionalGetMiddleware().process_response(put_request, put_response)
conditional_get_response = ConditionalGetMiddleware(get_200_response)(put_request)
self.assertEqual(conditional_get_response.status_code, 200) # should never be a 412
def test_no_head(self):
@ -587,9 +633,11 @@ class ConditionalGetMiddlewareTest(SimpleTestCase):
HEAD request since it can't do so accurately without access to the
response body of the corresponding GET.
"""
def get_200_response(req):
return HttpResponse(status=200)
request = self.request_factory.head('/')
response = HttpResponse(status=200)
conditional_get_response = ConditionalGetMiddleware().process_response(request, response)
conditional_get_response = ConditionalGetMiddleware(get_200_response)(request)
self.assertNotIn('ETag', conditional_get_response)
@ -604,11 +652,11 @@ class XFrameOptionsMiddlewareTest(SimpleTestCase):
middleware use that value for the HTTP header.
"""
with override_settings(X_FRAME_OPTIONS='SAMEORIGIN'):
r = XFrameOptionsMiddleware().process_response(HttpRequest(), HttpResponse())
r = XFrameOptionsMiddleware(get_response_empty)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
with override_settings(X_FRAME_OPTIONS='sameorigin'):
r = XFrameOptionsMiddleware().process_response(HttpRequest(), HttpResponse())
r = XFrameOptionsMiddleware(get_response_empty)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
def test_deny(self):
@ -617,11 +665,11 @@ class XFrameOptionsMiddlewareTest(SimpleTestCase):
use that value for the HTTP header.
"""
with override_settings(X_FRAME_OPTIONS='DENY'):
r = XFrameOptionsMiddleware().process_response(HttpRequest(), HttpResponse())
r = XFrameOptionsMiddleware(get_response_empty)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'DENY')
with override_settings(X_FRAME_OPTIONS='deny'):
r = XFrameOptionsMiddleware().process_response(HttpRequest(), HttpResponse())
r = XFrameOptionsMiddleware(get_response_empty)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'DENY')
def test_defaults_sameorigin(self):
@ -631,7 +679,7 @@ class XFrameOptionsMiddlewareTest(SimpleTestCase):
"""
with override_settings(X_FRAME_OPTIONS=None):
del settings.X_FRAME_OPTIONS # restored by override_settings
r = XFrameOptionsMiddleware().process_response(HttpRequest(), HttpResponse())
r = XFrameOptionsMiddleware(get_response_empty)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'DENY')
def test_dont_set_if_set(self):
@ -639,16 +687,22 @@ class XFrameOptionsMiddlewareTest(SimpleTestCase):
If the X-Frame-Options header is already set then the middleware does
not attempt to override it.
"""
with override_settings(X_FRAME_OPTIONS='DENY'):
def same_origin_response(request):
response = HttpResponse()
response['X-Frame-Options'] = 'SAMEORIGIN'
r = XFrameOptionsMiddleware().process_response(HttpRequest(), response)
return response
def deny_response(request):
response = HttpResponse()
response['X-Frame-Options'] = 'DENY'
return response
with override_settings(X_FRAME_OPTIONS='DENY'):
r = XFrameOptionsMiddleware(same_origin_response)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
with override_settings(X_FRAME_OPTIONS='SAMEORIGIN'):
response = HttpResponse()
response['X-Frame-Options'] = 'DENY'
r = XFrameOptionsMiddleware().process_response(HttpRequest(), response)
r = XFrameOptionsMiddleware(deny_response)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'DENY')
def test_response_exempt(self):
@ -656,15 +710,21 @@ class XFrameOptionsMiddlewareTest(SimpleTestCase):
If the response has an xframe_options_exempt attribute set to False
then it still sets the header, but if it's set to True then it doesn't.
"""
with override_settings(X_FRAME_OPTIONS='SAMEORIGIN'):
response = HttpResponse()
response.xframe_options_exempt = False
r = XFrameOptionsMiddleware().process_response(HttpRequest(), response)
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
def xframe_exempt_response(request):
response = HttpResponse()
response.xframe_options_exempt = True
r = XFrameOptionsMiddleware().process_response(HttpRequest(), response)
return response
def xframe_not_exempt_response(request):
response = HttpResponse()
response.xframe_options_exempt = False
return response
with override_settings(X_FRAME_OPTIONS='SAMEORIGIN'):
r = XFrameOptionsMiddleware(xframe_not_exempt_response)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
r = XFrameOptionsMiddleware(xframe_exempt_response)(HttpRequest())
self.assertIsNone(r.get('X-Frame-Options'))
def test_is_extendable(self):
@ -682,19 +742,22 @@ class XFrameOptionsMiddlewareTest(SimpleTestCase):
return 'SAMEORIGIN'
return 'DENY'
with override_settings(X_FRAME_OPTIONS='DENY'):
def same_origin_response(request):
response = HttpResponse()
response.sameorigin = True
r = OtherXFrameOptionsMiddleware().process_response(HttpRequest(), response)
return response
with override_settings(X_FRAME_OPTIONS='DENY'):
r = OtherXFrameOptionsMiddleware(same_origin_response)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
request = HttpRequest()
request.sameorigin = True
r = OtherXFrameOptionsMiddleware().process_response(request, HttpResponse())
r = OtherXFrameOptionsMiddleware(get_response_empty)(request)
self.assertEqual(r['X-Frame-Options'], 'SAMEORIGIN')
with override_settings(X_FRAME_OPTIONS='SAMEORIGIN'):
r = OtherXFrameOptionsMiddleware().process_response(HttpRequest(), HttpResponse())
r = OtherXFrameOptionsMiddleware(get_response_empty)(HttpRequest())
self.assertEqual(r['X-Frame-Options'], 'DENY')
@ -717,10 +780,9 @@ class GZipMiddlewareTest(SimpleTestCase):
self.resp.status_code = 200
self.resp.content = self.compressible_string
self.resp['Content-Type'] = 'text/html; charset=UTF-8'
self.stream_resp = StreamingHttpResponse(self.sequence)
self.stream_resp['Content-Type'] = 'text/html; charset=UTF-8'
self.stream_resp_unicode = StreamingHttpResponse(self.sequence_unicode)
self.stream_resp_unicode['Content-Type'] = 'text/html; charset=UTF-8'
def get_response(self, request):
return self.resp
@staticmethod
def decompress(gzipped_string):
@ -737,7 +799,7 @@ class GZipMiddlewareTest(SimpleTestCase):
"""
Compression is performed on responses with compressible content.
"""
r = GZipMiddleware().process_response(self.req, self.resp)
r = GZipMiddleware(self.get_response)(self.req)
self.assertEqual(self.decompress(r.content), self.compressible_string)
self.assertEqual(r.get('Content-Encoding'), 'gzip')
self.assertEqual(r.get('Content-Length'), str(len(r.content)))
@ -746,7 +808,12 @@ class GZipMiddlewareTest(SimpleTestCase):
"""
Compression is performed on responses with streaming content.
"""
r = GZipMiddleware().process_response(self.req, self.stream_resp)
def get_stream_response(request):
resp = StreamingHttpResponse(self.sequence)
resp['Content-Type'] = 'text/html; charset=UTF-8'
return resp
r = GZipMiddleware(get_stream_response)(self.req)
self.assertEqual(self.decompress(b''.join(r)), b''.join(self.sequence))
self.assertEqual(r.get('Content-Encoding'), 'gzip')
self.assertFalse(r.has_header('Content-Length'))
@ -755,7 +822,12 @@ class GZipMiddlewareTest(SimpleTestCase):
"""
Compression is performed on responses with streaming Unicode content.
"""
r = GZipMiddleware().process_response(self.req, self.stream_resp_unicode)
def get_stream_response_unicode(request):
resp = StreamingHttpResponse(self.sequence_unicode)
resp['Content-Type'] = 'text/html; charset=UTF-8'
return resp
r = GZipMiddleware(get_stream_response_unicode)(self.req)
self.assertEqual(
self.decompress(b''.join(r)),
b''.join(x.encode() for x in self.sequence_unicode)
@ -768,9 +840,12 @@ class GZipMiddlewareTest(SimpleTestCase):
Compression is performed on FileResponse.
"""
with open(__file__, 'rb') as file1:
def get_response(req):
file_resp = FileResponse(file1)
file_resp['Content-Type'] = 'text/html; charset=UTF-8'
r = GZipMiddleware().process_response(self.req, file_resp)
return file_resp
r = GZipMiddleware(get_response)(self.req)
with open(__file__, 'rb') as file2:
self.assertEqual(self.decompress(b''.join(r)), file2.read())
self.assertEqual(r.get('Content-Encoding'), 'gzip')
@ -782,7 +857,7 @@ class GZipMiddlewareTest(SimpleTestCase):
(#10762).
"""
self.resp.status_code = 404
r = GZipMiddleware().process_response(self.req, self.resp)
r = GZipMiddleware(self.get_response)(self.req)
self.assertEqual(self.decompress(r.content), self.compressible_string)
self.assertEqual(r.get('Content-Encoding'), 'gzip')
@ -791,7 +866,7 @@ class GZipMiddlewareTest(SimpleTestCase):
Compression isn't performed on responses with short content.
"""
self.resp.content = self.short_string
r = GZipMiddleware().process_response(self.req, self.resp)
r = GZipMiddleware(self.get_response)(self.req)
self.assertEqual(r.content, self.short_string)
self.assertIsNone(r.get('Content-Encoding'))
@ -800,7 +875,7 @@ class GZipMiddlewareTest(SimpleTestCase):
Compression isn't performed on responses that are already compressed.
"""
self.resp['Content-Encoding'] = 'deflate'
r = GZipMiddleware().process_response(self.req, self.resp)
r = GZipMiddleware(self.get_response)(self.req)
self.assertEqual(r.content, self.compressible_string)
self.assertEqual(r.get('Content-Encoding'), 'deflate')
@ -809,7 +884,7 @@ class GZipMiddlewareTest(SimpleTestCase):
Compression isn't performed on responses with incompressible content.
"""
self.resp.content = self.incompressible_string
r = GZipMiddleware().process_response(self.req, self.resp)
r = GZipMiddleware(self.get_response)(self.req)
self.assertEqual(r.content, self.incompressible_string)
self.assertIsNone(r.get('Content-Encoding'))
@ -821,8 +896,8 @@ class GZipMiddlewareTest(SimpleTestCase):
ConditionalGetMiddleware from recognizing conditional matches
on gzipped content).
"""
r1 = GZipMiddleware().process_response(self.req, self.resp)
r2 = GZipMiddleware().process_response(self.req, self.resp)
r1 = GZipMiddleware(self.get_response)(self.req)
r2 = GZipMiddleware(self.get_response)(self.req)
self.assertEqual(r1.content, r2.content)
self.assertEqual(self.get_mtime(r1.content), 0)
self.assertEqual(self.get_mtime(r2.content), 0)
@ -839,35 +914,42 @@ class ETagGZipMiddlewareTest(SimpleTestCase):
"""
GZipMiddleware makes a strong ETag weak.
"""
request = self.rf.get('/', HTTP_ACCEPT_ENCODING='gzip, deflate')
def get_response(req):
response = HttpResponse(self.compressible_string)
response['ETag'] = '"eggs"'
gzip_response = GZipMiddleware().process_response(request, response)
return response
request = self.rf.get('/', HTTP_ACCEPT_ENCODING='gzip, deflate')
gzip_response = GZipMiddleware(get_response)(request)
self.assertEqual(gzip_response['ETag'], 'W/"eggs"')
def test_weak_etag_not_modified(self):
"""
GZipMiddleware doesn't modify a weak ETag.
"""
request = self.rf.get('/', HTTP_ACCEPT_ENCODING='gzip, deflate')
def get_response(req):
response = HttpResponse(self.compressible_string)
response['ETag'] = 'W/"eggs"'
gzip_response = GZipMiddleware().process_response(request, response)
return response
request = self.rf.get('/', HTTP_ACCEPT_ENCODING='gzip, deflate')
gzip_response = GZipMiddleware(get_response)(request)
self.assertEqual(gzip_response['ETag'], 'W/"eggs"')
def test_etag_match(self):
"""
GZipMiddleware allows 304 Not Modified responses.
"""
def get_response(req):
response = HttpResponse(self.compressible_string)
return response
def get_cond_response(req):
return ConditionalGetMiddleware(get_response)(req)
request = self.rf.get('/', HTTP_ACCEPT_ENCODING='gzip, deflate')
response = GZipMiddleware().process_response(
request,
ConditionalGetMiddleware().process_response(request, HttpResponse(self.compressible_string))
)
response = GZipMiddleware(get_cond_response)(request)
gzip_etag = response['ETag']
next_request = self.rf.get('/', HTTP_ACCEPT_ENCODING='gzip, deflate', HTTP_IF_NONE_MATCH=gzip_etag)
next_response = ConditionalGetMiddleware().process_response(
next_request,
HttpResponse(self.compressible_string)
)
next_response = ConditionalGetMiddleware(get_response)(next_request)
self.assertEqual(next_response.status_code, 304)

View File

@ -1,5 +1,6 @@
from django.conf import settings
from django.core.exceptions import MiddlewareNotUsed
from django.http import HttpResponse
from django.test import RequestFactory, SimpleTestCase, override_settings
from . import middleware as mw
@ -114,7 +115,7 @@ class RootUrlconfTests(SimpleTestCase):
class MyMiddleware:
def __init__(self, get_response=None):
def __init__(self, get_response):
raise MiddlewareNotUsed
def process_request(self, request):
@ -123,7 +124,7 @@ class MyMiddleware:
class MyMiddlewareWithExceptionMessage:
def __init__(self, get_response=None):
def __init__(self, get_response):
raise MiddlewareNotUsed('spam eggs')
def process_request(self, request):
@ -142,7 +143,7 @@ class MiddlewareNotUsedTests(SimpleTestCase):
def test_raise_exception(self):
request = self.rf.get('middleware_exceptions/view/')
with self.assertRaises(MiddlewareNotUsed):
MyMiddleware().process_request(request)
MyMiddleware(lambda req: HttpResponse()).process_request(request)
@override_settings(MIDDLEWARE=['middleware_exceptions.tests.MyMiddleware'])
def test_log(self):

View File

@ -649,32 +649,27 @@ class CacheSessionTests(SessionTestsMixin, unittest.TestCase):
class SessionMiddlewareTests(TestCase):
request_factory = RequestFactory()
@staticmethod
def get_response_touching_session(request):
request.session['hello'] = 'world'
return HttpResponse('Session test')
@override_settings(SESSION_COOKIE_SECURE=True)
def test_secure_session_cookie(self):
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
# Simulate a request the modifies the session
middleware.process_request(request)
request.session['hello'] = 'world'
middleware = SessionMiddleware(self.get_response_touching_session)
# Handle the response through the middleware
response = middleware.process_response(request, response)
response = middleware(request)
self.assertIs(response.cookies[settings.SESSION_COOKIE_NAME]['secure'], True)
@override_settings(SESSION_COOKIE_HTTPONLY=True)
def test_httponly_session_cookie(self):
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
# Simulate a request the modifies the session
middleware.process_request(request)
request.session['hello'] = 'world'
middleware = SessionMiddleware(self.get_response_touching_session)
# Handle the response through the middleware
response = middleware.process_response(request, response)
response = middleware(request)
self.assertIs(response.cookies[settings.SESSION_COOKIE_NAME]['httponly'], True)
self.assertIn(
cookies.Morsel._reserved['httponly'],
@ -684,25 +679,15 @@ class SessionMiddlewareTests(TestCase):
@override_settings(SESSION_COOKIE_SAMESITE='Strict')
def test_samesite_session_cookie(self):
request = self.request_factory.get('/')
response = HttpResponse()
middleware = SessionMiddleware()
middleware.process_request(request)
request.session['hello'] = 'world'
response = middleware.process_response(request, response)
middleware = SessionMiddleware(self.get_response_touching_session)
response = middleware(request)
self.assertEqual(response.cookies[settings.SESSION_COOKIE_NAME]['samesite'], 'Strict')
@override_settings(SESSION_COOKIE_HTTPONLY=False)
def test_no_httponly_session_cookie(self):
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
# Simulate a request the modifies the session
middleware.process_request(request)
request.session['hello'] = 'world'
# Handle the response through the middleware
response = middleware.process_response(request, response)
middleware = SessionMiddleware(self.get_response_touching_session)
response = middleware(request)
self.assertEqual(response.cookies[settings.SESSION_COOKIE_NAME]['httponly'], '')
self.assertNotIn(
cookies.Morsel._reserved['httponly'],
@ -710,30 +695,27 @@ class SessionMiddlewareTests(TestCase):
)
def test_session_save_on_500(self):
request = self.request_factory.get('/')
def response_500(requset):
response = HttpResponse('Horrible error')
response.status_code = 500
middleware = SessionMiddleware()
# Simulate a request the modifies the session
middleware.process_request(request)
request.session['hello'] = 'world'
return response
# Handle the response through the middleware
response = middleware.process_response(request, response)
request = self.request_factory.get('/')
SessionMiddleware(response_500)(request)
# The value wasn't saved above.
self.assertNotIn('hello', request.session.load())
def test_session_update_error_redirect(self):
path = '/foo/'
request = self.request_factory.get(path)
response = HttpResponse()
middleware = SessionMiddleware()
def response_delete_session(request):
request.session = DatabaseSession()
request.session.save(must_create=True)
request.session.delete()
return HttpResponse()
request = self.request_factory.get('/foo/')
middleware = SessionMiddleware(response_delete_session)
msg = (
"The request's session was deleted before the request completed. "
@ -743,22 +725,21 @@ class SessionMiddlewareTests(TestCase):
# Handle the response through the middleware. It will try to save
# the deleted session which will cause an UpdateError that's caught
# and raised as a SuspiciousOperation.
middleware.process_response(request, response)
middleware(request)
def test_session_delete_on_end(self):
def response_ending_session(request):
request.session.flush()
return HttpResponse('Session test')
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
middleware = SessionMiddleware(response_ending_session)
# Before deleting, there has to be an existing cookie
request.COOKIES[settings.SESSION_COOKIE_NAME] = 'abc'
# Simulate a request that ends the session
middleware.process_request(request)
request.session.flush()
# Handle the response through the middleware
response = middleware.process_response(request, response)
response = middleware(request)
# The cookie was deleted, not recreated.
# A deleted cookie header looks like:
@ -776,19 +757,18 @@ class SessionMiddlewareTests(TestCase):
@override_settings(SESSION_COOKIE_DOMAIN='.example.local', SESSION_COOKIE_PATH='/example/')
def test_session_delete_on_end_with_custom_domain_and_path(self):
def response_ending_session(request):
request.session.flush()
return HttpResponse('Session test')
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
middleware = SessionMiddleware(response_ending_session)
# Before deleting, there has to be an existing cookie
request.COOKIES[settings.SESSION_COOKIE_NAME] = 'abc'
# Simulate a request that ends the session
middleware.process_request(request)
request.session.flush()
# Handle the response through the middleware
response = middleware.process_response(request, response)
response = middleware(request)
# The cookie was deleted, not recreated.
# A deleted cookie header with a custom domain and path looks like:
@ -804,16 +784,15 @@ class SessionMiddlewareTests(TestCase):
)
def test_flush_empty_without_session_cookie_doesnt_set_cookie(self):
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
# Simulate a request that ends the session
middleware.process_request(request)
def response_ending_session(request):
request.session.flush()
return HttpResponse('Session test')
request = self.request_factory.get('/')
middleware = SessionMiddleware(response_ending_session)
# Handle the response through the middleware
response = middleware.process_response(request, response)
response = middleware(request)
# A cookie should not be set.
self.assertEqual(response.cookies, {})
@ -825,15 +804,16 @@ class SessionMiddlewareTests(TestCase):
If a session is emptied of data but still has a key, it should still
be updated.
"""
request = self.request_factory.get('/')
response = HttpResponse('Session test')
middleware = SessionMiddleware()
def response_set_session(request):
# Set a session key and some data.
middleware.process_request(request)
request.session['foo'] = 'bar'
return HttpResponse('Session test')
request = self.request_factory.get('/')
middleware = SessionMiddleware(response_set_session)
# Handle the response through the middleware.
response = middleware.process_response(request, response)
response = middleware(request)
self.assertEqual(tuple(request.session.items()), (('foo', 'bar'),))
# A cookie should be set, along with Vary: Cookie.
self.assertIn(

View File

@ -1,7 +1,7 @@
import re
from django.forms import CharField, Form, Media
from django.http import HttpRequest
from django.http import HttpRequest, HttpResponse
from django.middleware.csrf import (
CsrfViewMiddleware, _compare_salted_tokens as equivalent_tokens, get_token,
)
@ -76,7 +76,7 @@ class TemplateStringsTests(SimpleTestCase):
def test_csrf_token(self):
request = HttpRequest()
CsrfViewMiddleware().process_view(request, lambda r: None, (), {})
CsrfViewMiddleware(lambda req: HttpResponse()).process_view(request, lambda r: None, (), {})
template = self.engine.get_template('template_backends/csrf.html')
content = template.render(request=request)

View File

@ -10,7 +10,6 @@ from django.test import (
RequestFactory, SimpleTestCase, modify_settings, override_settings,
)
from django.test.utils import require_jinja2
from django.utils.deprecation import MiddlewareMixin
from .utils import TEMPLATE_DIR
@ -23,9 +22,11 @@ test_processor_name = 'template_tests.test_response.test_processor'
# A test middleware that installs a temporary URLConf
class CustomURLConfMiddleware(MiddlewareMixin):
def process_request(self, request):
def custom_urlconf_middleware(get_response):
def middleware(request):
request.urlconf = 'template_tests.alternate_urls'
return get_response(request)
return middleware
class SimpleTemplateResponseTest(SimpleTestCase):
@ -319,7 +320,7 @@ class TemplateResponseTest(SimpleTestCase):
pickle.dumps(unpickled_response)
@modify_settings(MIDDLEWARE={'append': ['template_tests.test_response.CustomURLConfMiddleware']})
@modify_settings(MIDDLEWARE={'append': ['template_tests.test_response.custom_urlconf_middleware']})
@override_settings(ROOT_URLCONF='template_tests.urls')
class CustomURLConfTest(SimpleTestCase):

View File

@ -6,6 +6,9 @@ from django.utils.decorators import decorator_from_middleware
class ProcessViewMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def process_view(self, request, view_func, view_args, view_kwargs):
pass
@ -27,6 +30,9 @@ class_process_view = process_view_dec(ClassProcessView())
class FullMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def process_request(self, request):
request.process_request_reached = True