437 lines
17 KiB
Python
437 lines
17 KiB
Python
import warnings
|
|
|
|
from django import http
|
|
from django.test import TestCase
|
|
from django.conf import settings
|
|
from django.utils.translation import ugettext_lazy
|
|
from django.utils.unittest import skipIf
|
|
from django.contrib.messages import constants, utils, get_level, set_level
|
|
from django.contrib.messages.api import MessageFailure
|
|
from django.contrib.messages.storage import default_storage, base
|
|
from django.contrib.messages.storage.base import Message
|
|
from django.core.urlresolvers import reverse
|
|
from django.contrib.auth.models import User
|
|
|
|
|
|
def skipUnlessAuthIsInstalled(func):
|
|
return skipIf(
|
|
'django.contrib.auth' not in settings.INSTALLED_APPS,
|
|
"django.contrib.auth isn't installed")(func)
|
|
|
|
|
|
def add_level_messages(storage):
|
|
"""
|
|
Adds 6 messages from different levels (including a custom one) to a storage
|
|
instance.
|
|
"""
|
|
storage.add(constants.INFO, 'A generic info message')
|
|
storage.add(29, 'Some custom level')
|
|
storage.add(constants.DEBUG, 'A debugging message', extra_tags='extra-tag')
|
|
storage.add(constants.WARNING, 'A warning')
|
|
storage.add(constants.ERROR, 'An error')
|
|
storage.add(constants.SUCCESS, 'This was a triumph.')
|
|
|
|
|
|
class BaseTest(TestCase):
|
|
storage_class = default_storage
|
|
restore_settings = ['MESSAGE_LEVEL', 'MESSAGE_TAGS']
|
|
urls = 'django.contrib.messages.tests.urls'
|
|
levels = {
|
|
'debug': constants.DEBUG,
|
|
'info': constants.INFO,
|
|
'success': constants.SUCCESS,
|
|
'warning': constants.WARNING,
|
|
'error': constants.ERROR,
|
|
}
|
|
|
|
def setUp(self):
|
|
self._remembered_settings = {}
|
|
for setting in self.restore_settings:
|
|
if hasattr(settings, setting):
|
|
self._remembered_settings[setting] = getattr(settings, setting)
|
|
delattr(settings._wrapped, setting)
|
|
# Backup these manually because we do not want them deleted.
|
|
self._middleware_classes = settings.MIDDLEWARE_CLASSES
|
|
self._template_context_processors = \
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS
|
|
self._installed_apps = settings.INSTALLED_APPS
|
|
self._message_storage = settings.MESSAGE_STORAGE
|
|
settings.MESSAGE_STORAGE = '%s.%s' % (self.storage_class.__module__,
|
|
self.storage_class.__name__)
|
|
self.save_warnings_state()
|
|
warnings.filterwarnings('ignore', category=DeprecationWarning,
|
|
module='django.contrib.auth.models')
|
|
|
|
def tearDown(self):
|
|
for setting in self.restore_settings:
|
|
self.restore_setting(setting)
|
|
# Restore these manually (see above).
|
|
settings.MIDDLEWARE_CLASSES = self._middleware_classes
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS = \
|
|
self._template_context_processors
|
|
settings.INSTALLED_APPS = self._installed_apps
|
|
settings.MESSAGE_STORAGE = self._message_storage
|
|
self.restore_warnings_state()
|
|
|
|
def restore_setting(self, setting):
|
|
if setting in self._remembered_settings:
|
|
value = self._remembered_settings.pop(setting)
|
|
setattr(settings, setting, value)
|
|
elif hasattr(settings, setting):
|
|
delattr(settings._wrapped, setting)
|
|
|
|
def get_request(self):
|
|
return http.HttpRequest()
|
|
|
|
def get_response(self):
|
|
return http.HttpResponse()
|
|
|
|
def get_storage(self, data=None):
|
|
"""
|
|
Returns the storage backend, setting its loaded data to the ``data``
|
|
argument.
|
|
|
|
This method avoids the storage ``_get`` method from getting called so
|
|
that other parts of the storage backend can be tested independent of
|
|
the message retrieval logic.
|
|
"""
|
|
storage = self.storage_class(self.get_request())
|
|
storage._loaded_data = data or []
|
|
return storage
|
|
|
|
def test_add(self):
|
|
storage = self.get_storage()
|
|
self.assertFalse(storage.added_new)
|
|
storage.add(constants.INFO, 'Test message 1')
|
|
self.assertTrue(storage.added_new)
|
|
storage.add(constants.INFO, 'Test message 2', extra_tags='tag')
|
|
self.assertEqual(len(storage), 2)
|
|
|
|
def test_add_lazy_translation(self):
|
|
storage = self.get_storage()
|
|
response = self.get_response()
|
|
|
|
storage.add(constants.INFO, ugettext_lazy('lazy message'))
|
|
storage.update(response)
|
|
|
|
storing = self.stored_messages_count(storage, response)
|
|
self.assertEqual(storing, 1)
|
|
|
|
def test_no_update(self):
|
|
storage = self.get_storage()
|
|
response = self.get_response()
|
|
storage.update(response)
|
|
storing = self.stored_messages_count(storage, response)
|
|
self.assertEqual(storing, 0)
|
|
|
|
def test_add_update(self):
|
|
storage = self.get_storage()
|
|
response = self.get_response()
|
|
|
|
storage.add(constants.INFO, 'Test message 1')
|
|
storage.add(constants.INFO, 'Test message 1', extra_tags='tag')
|
|
storage.update(response)
|
|
|
|
storing = self.stored_messages_count(storage, response)
|
|
self.assertEqual(storing, 2)
|
|
|
|
def test_existing_add_read_update(self):
|
|
storage = self.get_existing_storage()
|
|
response = self.get_response()
|
|
|
|
storage.add(constants.INFO, 'Test message 3')
|
|
list(storage) # Simulates a read
|
|
storage.update(response)
|
|
|
|
storing = self.stored_messages_count(storage, response)
|
|
self.assertEqual(storing, 0)
|
|
|
|
def test_existing_read_add_update(self):
|
|
storage = self.get_existing_storage()
|
|
response = self.get_response()
|
|
|
|
list(storage) # Simulates a read
|
|
storage.add(constants.INFO, 'Test message 3')
|
|
storage.update(response)
|
|
|
|
storing = self.stored_messages_count(storage, response)
|
|
self.assertEqual(storing, 1)
|
|
|
|
def test_full_request_response_cycle(self):
|
|
"""
|
|
With the message middleware enabled, tests that messages are properly
|
|
stored and then retrieved across the full request/redirect/response
|
|
cycle.
|
|
"""
|
|
settings.MESSAGE_LEVEL = constants.DEBUG
|
|
data = {
|
|
'messages': ['Test message %d' % x for x in xrange(10)],
|
|
}
|
|
show_url = reverse('django.contrib.messages.tests.urls.show')
|
|
for level in ('debug', 'info', 'success', 'warning', 'error'):
|
|
add_url = reverse('django.contrib.messages.tests.urls.add',
|
|
args=(level,))
|
|
response = self.client.post(add_url, data, follow=True)
|
|
self.assertRedirects(response, show_url)
|
|
self.assertTrue('messages' in response.context)
|
|
messages = [Message(self.levels[level], msg) for msg in
|
|
data['messages']]
|
|
self.assertEqual(list(response.context['messages']), messages)
|
|
for msg in data['messages']:
|
|
self.assertContains(response, msg)
|
|
|
|
def test_with_template_response(self):
|
|
settings.MESSAGE_LEVEL = constants.DEBUG
|
|
data = {
|
|
'messages': ['Test message %d' % x for x in xrange(10)],
|
|
}
|
|
show_url = reverse('django.contrib.messages.tests.urls.show_template_response')
|
|
for level in self.levels.keys():
|
|
add_url = reverse('django.contrib.messages.tests.urls.add_template_response',
|
|
args=(level,))
|
|
response = self.client.post(add_url, data, follow=True)
|
|
self.assertRedirects(response, show_url)
|
|
self.assertTrue('messages' in response.context)
|
|
for msg in data['messages']:
|
|
self.assertContains(response, msg)
|
|
|
|
# there shouldn't be any messages on second GET request
|
|
response = self.client.get(show_url)
|
|
for msg in data['messages']:
|
|
self.assertNotContains(response, msg)
|
|
|
|
def test_multiple_posts(self):
|
|
"""
|
|
Tests that messages persist properly when multiple POSTs are made
|
|
before a GET.
|
|
"""
|
|
settings.MESSAGE_LEVEL = constants.DEBUG
|
|
data = {
|
|
'messages': ['Test message %d' % x for x in xrange(10)],
|
|
}
|
|
show_url = reverse('django.contrib.messages.tests.urls.show')
|
|
messages = []
|
|
for level in ('debug', 'info', 'success', 'warning', 'error'):
|
|
messages.extend([Message(self.levels[level], msg) for msg in
|
|
data['messages']])
|
|
add_url = reverse('django.contrib.messages.tests.urls.add',
|
|
args=(level,))
|
|
self.client.post(add_url, data)
|
|
response = self.client.get(show_url)
|
|
self.assertTrue('messages' in response.context)
|
|
self.assertEqual(list(response.context['messages']), messages)
|
|
for msg in data['messages']:
|
|
self.assertContains(response, msg)
|
|
|
|
@skipUnlessAuthIsInstalled
|
|
def test_middleware_disabled_auth_user(self):
|
|
"""
|
|
Tests that the messages API successfully falls back to using
|
|
user.message_set to store messages directly when the middleware is
|
|
disabled.
|
|
"""
|
|
settings.MESSAGE_LEVEL = constants.DEBUG
|
|
user = User.objects.create_user('test', 'test@example.com', 'test')
|
|
self.client.login(username='test', password='test')
|
|
settings.INSTALLED_APPS = list(settings.INSTALLED_APPS)
|
|
settings.INSTALLED_APPS.remove(
|
|
'django.contrib.messages',
|
|
)
|
|
settings.MIDDLEWARE_CLASSES = list(settings.MIDDLEWARE_CLASSES)
|
|
settings.MIDDLEWARE_CLASSES.remove(
|
|
'django.contrib.messages.middleware.MessageMiddleware',
|
|
)
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS = \
|
|
list(settings.TEMPLATE_CONTEXT_PROCESSORS)
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS.remove(
|
|
'django.contrib.messages.context_processors.messages',
|
|
)
|
|
data = {
|
|
'messages': ['Test message %d' % x for x in xrange(10)],
|
|
}
|
|
show_url = reverse('django.contrib.messages.tests.urls.show')
|
|
for level in ('debug', 'info', 'success', 'warning', 'error'):
|
|
add_url = reverse('django.contrib.messages.tests.urls.add',
|
|
args=(level,))
|
|
response = self.client.post(add_url, data, follow=True)
|
|
self.assertRedirects(response, show_url)
|
|
self.assertTrue('messages' in response.context)
|
|
context_messages = list(response.context['messages'])
|
|
for msg in data['messages']:
|
|
self.assertTrue(msg in context_messages)
|
|
self.assertContains(response, msg)
|
|
|
|
def test_middleware_disabled_anon_user(self):
|
|
"""
|
|
Tests that, when the middleware is disabled and a user is not logged
|
|
in, an exception is raised when one attempts to store a message.
|
|
"""
|
|
settings.MESSAGE_LEVEL = constants.DEBUG
|
|
settings.INSTALLED_APPS = list(settings.INSTALLED_APPS)
|
|
settings.INSTALLED_APPS.remove(
|
|
'django.contrib.messages',
|
|
)
|
|
settings.MIDDLEWARE_CLASSES = list(settings.MIDDLEWARE_CLASSES)
|
|
settings.MIDDLEWARE_CLASSES.remove(
|
|
'django.contrib.messages.middleware.MessageMiddleware',
|
|
)
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS = \
|
|
list(settings.TEMPLATE_CONTEXT_PROCESSORS)
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS.remove(
|
|
'django.contrib.messages.context_processors.messages',
|
|
)
|
|
data = {
|
|
'messages': ['Test message %d' % x for x in xrange(10)],
|
|
}
|
|
show_url = reverse('django.contrib.messages.tests.urls.show')
|
|
for level in ('debug', 'info', 'success', 'warning', 'error'):
|
|
add_url = reverse('django.contrib.messages.tests.urls.add',
|
|
args=(level,))
|
|
self.assertRaises(MessageFailure, self.client.post, add_url,
|
|
data, follow=True)
|
|
|
|
def test_middleware_disabled_anon_user_fail_silently(self):
|
|
"""
|
|
Tests that, when the middleware is disabled and a user is not logged
|
|
in, an exception is not raised if 'fail_silently' = True
|
|
"""
|
|
settings.MESSAGE_LEVEL = constants.DEBUG
|
|
settings.INSTALLED_APPS = list(settings.INSTALLED_APPS)
|
|
settings.INSTALLED_APPS.remove(
|
|
'django.contrib.messages',
|
|
)
|
|
settings.MIDDLEWARE_CLASSES = list(settings.MIDDLEWARE_CLASSES)
|
|
settings.MIDDLEWARE_CLASSES.remove(
|
|
'django.contrib.messages.middleware.MessageMiddleware',
|
|
)
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS = \
|
|
list(settings.TEMPLATE_CONTEXT_PROCESSORS)
|
|
settings.TEMPLATE_CONTEXT_PROCESSORS.remove(
|
|
'django.contrib.messages.context_processors.messages',
|
|
)
|
|
data = {
|
|
'messages': ['Test message %d' % x for x in xrange(10)],
|
|
'fail_silently': True,
|
|
}
|
|
show_url = reverse('django.contrib.messages.tests.urls.show')
|
|
for level in ('debug', 'info', 'success', 'warning', 'error'):
|
|
add_url = reverse('django.contrib.messages.tests.urls.add',
|
|
args=(level,))
|
|
response = self.client.post(add_url, data, follow=True)
|
|
self.assertRedirects(response, show_url)
|
|
self.assertTrue('messages' in response.context)
|
|
self.assertEqual(list(response.context['messages']), [])
|
|
|
|
def stored_messages_count(self, storage, response):
|
|
"""
|
|
Returns the number of messages being stored after a
|
|
``storage.update()`` call.
|
|
"""
|
|
raise NotImplementedError('This method must be set by a subclass.')
|
|
|
|
def test_get(self):
|
|
raise NotImplementedError('This method must be set by a subclass.')
|
|
|
|
def get_existing_storage(self):
|
|
return self.get_storage([Message(constants.INFO, 'Test message 1'),
|
|
Message(constants.INFO, 'Test message 2',
|
|
extra_tags='tag')])
|
|
|
|
def test_existing_read(self):
|
|
"""
|
|
Tests that reading the existing storage doesn't cause the data to be
|
|
lost.
|
|
"""
|
|
storage = self.get_existing_storage()
|
|
self.assertFalse(storage.used)
|
|
# After iterating the storage engine directly, the used flag is set.
|
|
data = list(storage)
|
|
self.assert_(storage.used)
|
|
# The data does not disappear because it has been iterated.
|
|
self.assertEqual(data, list(storage))
|
|
|
|
def test_existing_add(self):
|
|
storage = self.get_existing_storage()
|
|
self.assertFalse(storage.added_new)
|
|
storage.add(constants.INFO, 'Test message 3')
|
|
self.assert_(storage.added_new)
|
|
|
|
def test_default_level(self):
|
|
# get_level works even with no storage on the request.
|
|
request = self.get_request()
|
|
self.assertEqual(get_level(request), constants.INFO)
|
|
|
|
# get_level returns the default level if it hasn't been set.
|
|
storage = self.get_storage()
|
|
request._messages = storage
|
|
self.assertEqual(get_level(request), constants.INFO)
|
|
|
|
# Only messages of sufficient level get recorded.
|
|
add_level_messages(storage)
|
|
self.assertEqual(len(storage), 5)
|
|
|
|
def test_low_level(self):
|
|
request = self.get_request()
|
|
storage = self.storage_class(request)
|
|
request._messages = storage
|
|
|
|
self.assert_(set_level(request, 5))
|
|
self.assertEqual(get_level(request), 5)
|
|
|
|
add_level_messages(storage)
|
|
self.assertEqual(len(storage), 6)
|
|
|
|
def test_high_level(self):
|
|
request = self.get_request()
|
|
storage = self.storage_class(request)
|
|
request._messages = storage
|
|
|
|
self.assert_(set_level(request, 30))
|
|
self.assertEqual(get_level(request), 30)
|
|
|
|
add_level_messages(storage)
|
|
self.assertEqual(len(storage), 2)
|
|
|
|
def test_settings_level(self):
|
|
request = self.get_request()
|
|
storage = self.storage_class(request)
|
|
|
|
settings.MESSAGE_LEVEL = 29
|
|
self.assertEqual(get_level(request), 29)
|
|
|
|
add_level_messages(storage)
|
|
self.assertEqual(len(storage), 3)
|
|
|
|
def test_tags(self):
|
|
storage = self.get_storage()
|
|
storage.level = 0
|
|
add_level_messages(storage)
|
|
tags = [msg.tags for msg in storage]
|
|
self.assertEqual(tags,
|
|
['info', '', 'extra-tag debug', 'warning', 'error',
|
|
'success'])
|
|
|
|
def test_custom_tags(self):
|
|
settings.MESSAGE_TAGS = {
|
|
constants.INFO: 'info',
|
|
constants.DEBUG: '',
|
|
constants.WARNING: '',
|
|
constants.ERROR: 'bad',
|
|
29: 'custom',
|
|
}
|
|
# LEVEL_TAGS is a constant defined in the
|
|
# django.contrib.messages.storage.base module, so after changing
|
|
# settings.MESSAGE_TAGS, we need to update that constant too.
|
|
base.LEVEL_TAGS = utils.get_level_tags()
|
|
try:
|
|
storage = self.get_storage()
|
|
storage.level = 0
|
|
add_level_messages(storage)
|
|
tags = [msg.tags for msg in storage]
|
|
self.assertEqual(tags,
|
|
['info', 'custom', 'extra-tag', '', 'bad', 'success'])
|
|
finally:
|
|
# Ensure the level tags constant is put back like we found it.
|
|
self.restore_setting('MESSAGE_TAGS')
|
|
base.LEVEL_TAGS = utils.get_level_tags()
|