1 from django import http

     2 from django.test import TestCase

     3 from django.conf import settings

     4 from django.utils.translation import ugettext_lazy

     5 from django.contrib.messages import constants, utils, get_level, set_level

     6 from django.contrib.messages.api import MessageFailure

     7 from django.contrib.messages.storage import default_storage, base

     8 from django.contrib.messages.storage.base import Message

     9 from django.core.urlresolvers import reverse

    10 from django.contrib.auth.models import User

    11 

    12 

    13 def add_level_messages(storage):

    14     """

    15     Adds 6 messages from different levels (including a custom one) to a storage

    16     instance.

    17     """

    18     storage.add(constants.INFO, 'A generic info message')

    19     storage.add(29, 'Some custom level')

    20     storage.add(constants.DEBUG, 'A debugging message', extra_tags='extra-tag')

    21     storage.add(constants.WARNING, 'A warning')

    22     storage.add(constants.ERROR, 'An error')

    23     storage.add(constants.SUCCESS, 'This was a triumph.')

    24 

    25 

    26 class BaseTest(TestCase):

    27     storage_class = default_storage

    28     restore_settings = ['MESSAGE_LEVEL', 'MESSAGE_TAGS']

    29     urls = 'django.contrib.messages.tests.urls'

    30     levels = {

    31         'debug': constants.DEBUG,

    32         'info': constants.INFO,

    33         'success': constants.SUCCESS,

    34         'warning': constants.WARNING,

    35         'error': constants.ERROR,

    36     }

    37 

    38     def setUp(self):

    39         self._remembered_settings = {}

    40         for setting in self.restore_settings:

    41             if hasattr(settings, setting):

    42                 self._remembered_settings[setting] = getattr(settings, setting)

    43                 delattr(settings._wrapped, setting)

    44         # Backup these manually because we do not want them deleted.

    45         self._middleware_classes = settings.MIDDLEWARE_CLASSES

    46         self._template_context_processors = \

    47            settings.TEMPLATE_CONTEXT_PROCESSORS

    48         self._installed_apps = settings.INSTALLED_APPS

    49         self._message_storage = settings.MESSAGE_STORAGE

    50         settings.MESSAGE_STORAGE = '%s.%s' % (self.storage_class.__module__,

    51                                               self.storage_class.__name__)

    52 

    53     def tearDown(self):

    54         for setting in self.restore_settings:

    55             self.restore_setting(setting)

    56         # Restore these manually (see above).

    57         settings.MIDDLEWARE_CLASSES = self._middleware_classes

    58         settings.TEMPLATE_CONTEXT_PROCESSORS = \

    59            self._template_context_processors

    60         settings.INSTALLED_APPS = self._installed_apps

    61         settings.MESSAGE_STORAGE = self._message_storage

    62 

    63     def restore_setting(self, setting):

    64         if setting in self._remembered_settings:

    65             value = self._remembered_settings.pop(setting)

    66             setattr(settings, setting, value)

    67         elif hasattr(settings, setting):

    68             delattr(settings._wrapped, setting)

    69 

    70     def get_request(self):

    71         return http.HttpRequest()

    72 

    73     def get_response(self):

    74         return http.HttpResponse()

    75 

    76     def get_storage(self, data=None):

    77         """

    78         Returns the storage backend, setting its loaded data to the ``data``

    79         argument.

    80 

    81         This method avoids the storage ``_get`` method from getting called so

    82         that other parts of the storage backend can be tested independent of

    83         the message retrieval logic.

    84         """

    85         storage = self.storage_class(self.get_request())

    86         storage._loaded_data = data or []

    87         return storage

    88 

    89     def test_add(self):

    90         storage = self.get_storage()

    91         self.assertFalse(storage.added_new)

    92         storage.add(constants.INFO, 'Test message 1')

    93         self.assert_(storage.added_new)

    94         storage.add(constants.INFO, 'Test message 2', extra_tags='tag')

    95         self.assertEqual(len(storage), 2)

    96 

    97     def test_add_lazy_translation(self):

    98         storage = self.get_storage()

    99         response = self.get_response()

   100 

   101         storage.add(constants.INFO, ugettext_lazy('lazy message'))

   102         storage.update(response)

   103 

   104         storing = self.stored_messages_count(storage, response)

   105         self.assertEqual(storing, 1)

   106 

   107     def test_no_update(self):

   108         storage = self.get_storage()

   109         response = self.get_response()

   110         storage.update(response)

   111         storing = self.stored_messages_count(storage, response)

   112         self.assertEqual(storing, 0)

   113 

   114     def test_add_update(self):

   115         storage = self.get_storage()

   116         response = self.get_response()

   117 

   118         storage.add(constants.INFO, 'Test message 1')

   119         storage.add(constants.INFO, 'Test message 1', extra_tags='tag')

   120         storage.update(response)

   121 

   122         storing = self.stored_messages_count(storage, response)

   123         self.assertEqual(storing, 2)

   124 

   125     def test_existing_add_read_update(self):

   126         storage = self.get_existing_storage()

   127         response = self.get_response()

   128 

   129         storage.add(constants.INFO, 'Test message 3')

   130         list(storage)   # Simulates a read

   131         storage.update(response)

   132 

   133         storing = self.stored_messages_count(storage, response)

   134         self.assertEqual(storing, 0)

   135 

   136     def test_existing_read_add_update(self):

   137         storage = self.get_existing_storage()

   138         response = self.get_response()

   139 

   140         list(storage)   # Simulates a read

   141         storage.add(constants.INFO, 'Test message 3')

   142         storage.update(response)

   143 

   144         storing = self.stored_messages_count(storage, response)

   145         self.assertEqual(storing, 1)

   146 

   147     def test_full_request_response_cycle(self):

   148         """

   149         With the message middleware enabled, tests that messages are properly

   150         stored and then retrieved across the full request/redirect/response

   151         cycle.

   152         """

   153         settings.MESSAGE_LEVEL = constants.DEBUG

   154         data = {

   155             'messages': ['Test message %d' % x for x in xrange(10)],

   156         }

   157         show_url = reverse('django.contrib.messages.tests.urls.show')

   158         for level in ('debug', 'info', 'success', 'warning', 'error'):

   159             add_url = reverse('django.contrib.messages.tests.urls.add',

   160                               args=(level,))

   161             response = self.client.post(add_url, data, follow=True)

   162             self.assertRedirects(response, show_url)

   163             self.assertTrue('messages' in response.context)

   164             messages = [Message(self.levels[level], msg) for msg in

   165                                                          data['messages']]

   166             self.assertEqual(list(response.context['messages']), messages)

   167             for msg in data['messages']:

   168                 self.assertContains(response, msg)

   169 

   170     def test_multiple_posts(self):

   171         """

   172         Tests that messages persist properly when multiple POSTs are made

   173         before a GET.

   174         """

   175         settings.MESSAGE_LEVEL = constants.DEBUG

   176         data = {

   177             'messages': ['Test message %d' % x for x in xrange(10)],

   178         }

   179         show_url = reverse('django.contrib.messages.tests.urls.show')

   180         messages = []

   181         for level in ('debug', 'info', 'success', 'warning', 'error'):

   182             messages.extend([Message(self.levels[level], msg) for msg in

   183                                                              data['messages']])

   184             add_url = reverse('django.contrib.messages.tests.urls.add',

   185                               args=(level,))

   186             self.client.post(add_url, data)

   187         response = self.client.get(show_url)

   188         self.assertTrue('messages' in response.context)

   189         self.assertEqual(list(response.context['messages']), messages)

   190         for msg in data['messages']:

   191             self.assertContains(response, msg)

   192 

   193     def test_middleware_disabled_auth_user(self):

   194         """

   195         Tests that the messages API successfully falls back to using

   196         user.message_set to store messages directly when the middleware is

   197         disabled.

   198         """

   199         settings.MESSAGE_LEVEL = constants.DEBUG

   200         user = User.objects.create_user('test', 'test@example.com', 'test')

   201         self.client.login(username='test', password='test')

   202         settings.INSTALLED_APPS = list(settings.INSTALLED_APPS)

   203         settings.INSTALLED_APPS.remove(

   204             'django.contrib.messages',

   205         )

   206         settings.MIDDLEWARE_CLASSES = list(settings.MIDDLEWARE_CLASSES)

   207         settings.MIDDLEWARE_CLASSES.remove(

   208             'django.contrib.messages.middleware.MessageMiddleware',

   209         )

   210         settings.TEMPLATE_CONTEXT_PROCESSORS = \

   211           list(settings.TEMPLATE_CONTEXT_PROCESSORS)

   212         settings.TEMPLATE_CONTEXT_PROCESSORS.remove(

   213             'django.contrib.messages.context_processors.messages',

   214         )

   215         data = {

   216             'messages': ['Test message %d' % x for x in xrange(10)],

   217         }

   218         show_url = reverse('django.contrib.messages.tests.urls.show')

   219         for level in ('debug', 'info', 'success', 'warning', 'error'):

   220             add_url = reverse('django.contrib.messages.tests.urls.add',

   221                               args=(level,))

   222             response = self.client.post(add_url, data, follow=True)

   223             self.assertRedirects(response, show_url)

   224             self.assertTrue('messages' in response.context)

   225             context_messages = list(response.context['messages'])

   226             for msg in data['messages']:

   227                 self.assertTrue(msg in context_messages)

   228                 self.assertContains(response, msg)

   229 

   230     def test_middleware_disabled_anon_user(self):

   231         """

   232         Tests that, when the middleware is disabled and a user is not logged

   233         in, an exception is raised when one attempts to store a message.

   234         """

   235         settings.MESSAGE_LEVEL = constants.DEBUG

   236         settings.INSTALLED_APPS = list(settings.INSTALLED_APPS)

   237         settings.INSTALLED_APPS.remove(

   238             'django.contrib.messages',

   239         )

   240         settings.MIDDLEWARE_CLASSES = list(settings.MIDDLEWARE_CLASSES)

   241         settings.MIDDLEWARE_CLASSES.remove(

   242             'django.contrib.messages.middleware.MessageMiddleware',

   243         )

   244         settings.TEMPLATE_CONTEXT_PROCESSORS = \

   245           list(settings.TEMPLATE_CONTEXT_PROCESSORS)

   246         settings.TEMPLATE_CONTEXT_PROCESSORS.remove(

   247             'django.contrib.messages.context_processors.messages',

   248         )

   249         data = {

   250             'messages': ['Test message %d' % x for x in xrange(10)],

   251         }

   252         show_url = reverse('django.contrib.messages.tests.urls.show')

   253         for level in ('debug', 'info', 'success', 'warning', 'error'):

   254             add_url = reverse('django.contrib.messages.tests.urls.add',

   255                               args=(level,))

   256             self.assertRaises(MessageFailure, self.client.post, add_url,

   257                               data, follow=True)

   258 

   259     def test_middleware_disabled_anon_user_fail_silently(self):

   260         """

   261         Tests that, when the middleware is disabled and a user is not logged

   262         in, an exception is not raised if 'fail_silently' = True

   263         """

   264         settings.MESSAGE_LEVEL = constants.DEBUG

   265         settings.INSTALLED_APPS = list(settings.INSTALLED_APPS)

   266         settings.INSTALLED_APPS.remove(

   267             'django.contrib.messages',

   268         )

   269         settings.MIDDLEWARE_CLASSES = list(settings.MIDDLEWARE_CLASSES)

   270         settings.MIDDLEWARE_CLASSES.remove(

   271             'django.contrib.messages.middleware.MessageMiddleware',

   272         )

   273         settings.TEMPLATE_CONTEXT_PROCESSORS = \

   274           list(settings.TEMPLATE_CONTEXT_PROCESSORS)

   275         settings.TEMPLATE_CONTEXT_PROCESSORS.remove(

   276             'django.contrib.messages.context_processors.messages',

   277         )

   278         data = {

   279             'messages': ['Test message %d' % x for x in xrange(10)],

   280             'fail_silently': True,

   281         }

   282         show_url = reverse('django.contrib.messages.tests.urls.show')

   283         for level in ('debug', 'info', 'success', 'warning', 'error'):

   284             add_url = reverse('django.contrib.messages.tests.urls.add',

   285                               args=(level,))

   286             response = self.client.post(add_url, data, follow=True)

   287             self.assertRedirects(response, show_url)

   288             self.assertTrue('messages' in response.context)

   289             self.assertEqual(list(response.context['messages']), [])

   290 

   291     def stored_messages_count(self, storage, response):

   292         """

   293         Returns the number of messages being stored after a

   294         ``storage.update()`` call.

   295         """

   296         raise NotImplementedError('This method must be set by a subclass.')

   297 

   298     def test_get(self):

   299         raise NotImplementedError('This method must be set by a subclass.')

   300 

   301     def get_existing_storage(self):

   302         return self.get_storage([Message(constants.INFO, 'Test message 1'),

   303                                  Message(constants.INFO, 'Test message 2',

   304                                               extra_tags='tag')])

   305 

   306     def test_existing_read(self):

   307         """

   308         Tests that reading the existing storage doesn't cause the data to be

   309         lost.

   310         """

   311         storage = self.get_existing_storage()

   312         self.assertFalse(storage.used)

   313         # After iterating the storage engine directly, the used flag is set.

   314         data = list(storage)

   315         self.assert_(storage.used)

   316         # The data does not disappear because it has been iterated.

   317         self.assertEqual(data, list(storage))

   318 

   319     def test_existing_add(self):

   320         storage = self.get_existing_storage()

   321         self.assertFalse(storage.added_new)

   322         storage.add(constants.INFO, 'Test message 3')

   323         self.assert_(storage.added_new)

   324 

   325     def test_default_level(self):

   326         # get_level works even with no storage on the request.

   327         request = self.get_request()

   328         self.assertEqual(get_level(request), constants.INFO)

   329 

   330         # get_level returns the default level if it hasn't been set.

   331         storage = self.get_storage()

   332         request._messages = storage

   333         self.assertEqual(get_level(request), constants.INFO)

   334 

   335         # Only messages of sufficient level get recorded.

   336         add_level_messages(storage)

   337         self.assertEqual(len(storage), 5)

   338 

   339     def test_low_level(self):

   340         request = self.get_request()

   341         storage = self.storage_class(request)

   342         request._messages = storage

   343 

   344         self.assert_(set_level(request, 5))

   345         self.assertEqual(get_level(request), 5)

   346 

   347         add_level_messages(storage)

   348         self.assertEqual(len(storage), 6)

   349 

   350     def test_high_level(self):

   351         request = self.get_request()

   352         storage = self.storage_class(request)

   353         request._messages = storage

   354 

   355         self.assert_(set_level(request, 30))

   356         self.assertEqual(get_level(request), 30)

   357 

   358         add_level_messages(storage)

   359         self.assertEqual(len(storage), 2)

   360 

   361     def test_settings_level(self):

   362         request = self.get_request()

   363         storage = self.storage_class(request)

   364 

   365         settings.MESSAGE_LEVEL = 29

   366         self.assertEqual(get_level(request), 29)

   367 

   368         add_level_messages(storage)

   369         self.assertEqual(len(storage), 3)

   370 

   371     def test_tags(self):

   372         storage = self.get_storage()

   373         storage.level = 0

   374         add_level_messages(storage)

   375         tags = [msg.tags for msg in storage]

   376         self.assertEqual(tags,

   377                          ['info', '', 'extra-tag debug', 'warning', 'error',

   378                           'success'])

   379 

   380     def test_custom_tags(self):

   381         settings.MESSAGE_TAGS = {

   382             constants.INFO: 'info',

   383             constants.DEBUG: '',

   384             constants.WARNING: '',

   385             constants.ERROR: 'bad',

   386             29: 'custom',

   387         }

   388         # LEVEL_TAGS is a constant defined in the

   389         # django.contrib.messages.storage.base module, so after changing

   390         # settings.MESSAGE_TAGS, we need to update that constant too.

   391         base.LEVEL_TAGS = utils.get_level_tags()

   392         try:

   393             storage = self.get_storage()

   394             storage.level = 0

   395             add_level_messages(storage)

   396             tags = [msg.tags for msg in storage]

   397             self.assertEqual(tags,

   398                          ['info', 'custom', 'extra-tag', '', 'bad', 'success'])

   399         finally:

   400             # Ensure the level tags constant is put back like we found it.

   401             self.restore_setting('MESSAGE_TAGS')

   402             base.LEVEL_TAGS = utils.get_level_tags()