| 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() |
