tests.py 10 KB
Newer Older
1 2
import json

3 4 5 6
from django.contrib.auth.models import AnonymousUser
from django.core.exceptions import PermissionDenied
from django.http import Http404
from django.test import TestCase
Ned Batchelder committed
7
from django.test.client import RequestFactory
8 9 10 11
from django.test.utils import override_settings
from mock import Mock, patch

from notification_prefs import NOTIFICATION_PREF_KEY
12
from notification_prefs.views import ajax_enable, ajax_disable, ajax_status, set_subscription, UsernameCipher
13
from student.tests.factories import UserFactory
14
from edxmako.tests import mako_middleware_process_request
15
from openedx.core.djangoapps.user_api.models import UserPreference
16
from util.testing import UrlResetMixin
17 18 19


@override_settings(SECRET_KEY="test secret key")
20
class NotificationPrefViewTest(UrlResetMixin, TestCase):
21 22
    INITIALIZATION_VECTOR = "\x00" * 16

23
    @patch.dict("django.conf.settings.FEATURES", {"ENABLE_DISCUSSION_SERVICE": True})
24
    def setUp(self):
25
        super(NotificationPrefViewTest, self).setUp()
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49
        self.user = UserFactory.create(username="testuser")
        # Tokens are intentionally hard-coded instead of computed to help us
        # avoid breaking existing links.
        self.tokens = {
            self.user: "AAAAAAAAAAAAAAAAAAAAAA8mMQo96FZfb1YKv1R5X6s=",
            # Username with length equal to AES block length to test padding
            UserFactory.create(username="sixteencharsuser"):
            "AAAAAAAAAAAAAAAAAAAAAPxPWCuI2Ay9TATBVnfw7eIj-hUh6erQ_-VkbDqHqm8D",
            # Even longer username
            UserFactory.create(username="thisusernameissoveryverylong"):
            "AAAAAAAAAAAAAAAAAAAAAPECbYqPI7_W4mRF8LbTaHuHt3tNXPggZ1Bke-zDyEiZ",
            # Non-ASCII username
            UserFactory.create(username=u"\u4e2d\u56fd"):
            "AAAAAAAAAAAAAAAAAAAAAMjfGAhZKIZsI3L-Z7nflTA="
        }
        self.request_factory = RequestFactory()

    def create_prefs(self):
        """Create all test preferences in the database"""
        for (user, token) in self.tokens.items():
            UserPreference.objects.create(user=user, key=NOTIFICATION_PREF_KEY, value=token)

    def assertPrefValid(self, user):
        """Ensure that the correct preference for the user is persisted"""
50 51 52 53 54 55
        pref = UserPreference.objects.get(user=user, key=NOTIFICATION_PREF_KEY)
        self.assertTrue(pref)  # check exists and only 1 (.get)
        # now coerce username to utf-8 encoded str, since we test with non-ascii unicdoe above and
        # the unittest framework has hard time coercing to unicode.
        # decrypt also can't take a unicode input, so coerce its input to str
        self.assertEqual(str(user.username.encode('utf-8')), UsernameCipher().decrypt(str(pref.value)))
56 57 58 59 60 61 62

    def assertNotPrefExists(self, user):
        """Ensure that the user does not have a persisted preference"""
        self.assertFalse(
            UserPreference.objects.filter(user=user, key=NOTIFICATION_PREF_KEY).exists()
        )

63 64 65 66 67 68 69
    # AJAX status view

    def test_ajax_status_get_0(self):
        request = self.request_factory.get("dummy")
        request.user = self.user
        response = ajax_status(request)
        self.assertEqual(response.status_code, 200)
70
        self.assertEqual(json.loads(response.content), {"status": 0})
71 72 73 74 75 76 77

    def test_ajax_status_get_1(self):
        self.create_prefs()
        request = self.request_factory.get("dummy")
        request.user = self.user
        response = ajax_status(request)
        self.assertEqual(response.status_code, 200)
78
        self.assertEqual(json.loads(response.content), {"status": 1})
79 80 81 82 83 84 85 86 87 88 89 90

    def test_ajax_status_post(self):
        request = self.request_factory.post("dummy")
        request.user = self.user
        response = ajax_status(request)
        self.assertEqual(response.status_code, 405)

    def test_ajax_status_anon_user(self):
        request = self.request_factory.get("dummy")
        request.user = AnonymousUser()
        self.assertRaises(PermissionDenied, ajax_status, request)

91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177
    # AJAX enable view

    def test_ajax_enable_get(self):
        request = self.request_factory.get("dummy")
        request.user = self.user
        response = ajax_enable(request)
        self.assertEqual(response.status_code, 405)
        self.assertNotPrefExists(self.user)

    def test_ajax_enable_anon_user(self):
        request = self.request_factory.post("dummy")
        request.user = AnonymousUser()
        self.assertRaises(PermissionDenied, ajax_enable, request)
        self.assertNotPrefExists(self.user)

    @patch("Crypto.Random.new")
    def test_ajax_enable_success(self, mock_random_new):
        mock_stream = Mock()
        mock_stream.read.return_value = self.INITIALIZATION_VECTOR
        mock_random_new.return_value = mock_stream

        def test_user(user):
            request = self.request_factory.post("dummy")
            request.user = user
            response = ajax_enable(request)
            self.assertEqual(response.status_code, 204)
            self.assertPrefValid(user)

        for user in self.tokens.keys():
            test_user(user)

    def test_ajax_enable_already_enabled(self):
        self.create_prefs()
        request = self.request_factory.post("dummy")
        request.user = self.user
        response = ajax_enable(request)
        self.assertEqual(response.status_code, 204)
        self.assertPrefValid(self.user)

    def test_ajax_enable_distinct_values(self):
        request = self.request_factory.post("dummy")
        request.user = self.user
        ajax_enable(request)
        other_user = UserFactory.create()
        request.user = other_user
        ajax_enable(request)
        self.assertNotEqual(
            UserPreference.objects.get(user=self.user, key=NOTIFICATION_PREF_KEY).value,
            UserPreference.objects.get(user=other_user, key=NOTIFICATION_PREF_KEY).value
        )

    # AJAX disable view

    def test_ajax_disable_get(self):
        self.create_prefs()
        request = self.request_factory.get("dummy")
        request.user = self.user
        response = ajax_disable(request)
        self.assertEqual(response.status_code, 405)
        self.assertPrefValid(self.user)

    def test_ajax_disable_anon_user(self):
        self.create_prefs()
        request = self.request_factory.post("dummy")
        request.user = AnonymousUser()
        self.assertRaises(PermissionDenied, ajax_disable, request)
        self.assertPrefValid(self.user)

    def test_ajax_disable_success(self):
        self.create_prefs()
        request = self.request_factory.post("dummy")
        request.user = self.user
        response = ajax_disable(request)
        self.assertEqual(response.status_code, 204)
        self.assertNotPrefExists(self.user)

    def test_ajax_disable_already_disabled(self):
        request = self.request_factory.post("dummy")
        request.user = self.user
        response = ajax_disable(request)
        self.assertEqual(response.status_code, 204)
        self.assertNotPrefExists(self.user)

    # Unsubscribe view

    def test_unsubscribe_post(self):
        request = self.request_factory.post("dummy")
178
        response = set_subscription(request, "dummy", subscribe=False)
179 180 181 182 183
        self.assertEqual(response.status_code, 405)

    def test_unsubscribe_invalid_token(self):
        def test_invalid_token(token, message):
            request = self.request_factory.get("dummy")
184
            self.assertRaisesRegexp(Http404, "^{}$".format(message), set_subscription, request, token, False)
185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218

        # Invalid base64 encoding
        test_invalid_token("ZOMG INVALID BASE64 CHARS!!!", "base64url")
        test_invalid_token("Non-ASCII\xff", "base64url")
        test_invalid_token(self.tokens[self.user][:-1], "base64url")

        # Token not long enough to contain initialization vector
        test_invalid_token("AAAAAAAAAAA=", "initialization_vector")

        # Token length not a multiple of AES block length
        test_invalid_token(self.tokens[self.user][:-4], "aes")

        # Invalid padding (ends in 0 byte)
        # Encrypted value: "testuser" + "\x00" * 8
        test_invalid_token("AAAAAAAAAAAAAAAAAAAAAMoazRI7ePLjEWXN1N7keLw=", "padding")

        # Invalid padding (ends in byte > 16)
        # Encrypted value: "testusertestuser"
        test_invalid_token("AAAAAAAAAAAAAAAAAAAAAC6iLXGhjkFytJoJSBJZzJ4=", "padding")

        # Invalid padding (entire string is padding)
        # Encrypted value: "\x10" * 16
        test_invalid_token("AAAAAAAAAAAAAAAAAAAAANRGw8HDEmlcLVFawgY9wI8=", "padding")

        # Nonexistent user
        # Encrypted value: "nonexistentuser\x01"
        test_invalid_token("AAAAAAAAAAAAAAAAAAAAACpyUxTGIrUjnpuUsNi7mAY=", "username")

    def test_unsubscribe_success(self):
        self.create_prefs()

        def test_user(user):
            request = self.request_factory.get("dummy")
            request.user = AnonymousUser()
219 220

            mako_middleware_process_request(request)
221
            response = set_subscription(request, self.tokens[user], subscribe=False)
222 223 224 225 226 227 228 229 230 231
            self.assertEqual(response.status_code, 200)
            self.assertNotPrefExists(user)

        for user in self.tokens.keys():
            test_user(user)

    def test_unsubscribe_twice(self):
        self.create_prefs()
        request = self.request_factory.get("dummy")
        request.user = AnonymousUser()
232 233

        mako_middleware_process_request(request)
234 235
        set_subscription(request, self.tokens[self.user], False)
        response = set_subscription(request, self.tokens[self.user], subscribe=False)
236 237
        self.assertEqual(response.status_code, 200)
        self.assertNotPrefExists(self.user)
238 239 240 241 242 243 244

    def test_resubscribe_success(self):
        def test_user(user):
            # start without a pref key
            self.assertFalse(UserPreference.objects.filter(user=user, key=NOTIFICATION_PREF_KEY))
            request = self.request_factory.get("dummy")
            request.user = AnonymousUser()
245 246

            mako_middleware_process_request(request)
247
            response = set_subscription(request, self.tokens[user], subscribe=True)
248
            self.assertEqual(response.status_code, 200)
249
            self.assertPrefValid(user)
250 251 252

        for user in self.tokens.keys():
            test_user(user)