Add endpoints to set user notification preference

CHANGELOG.rst

@@ -11,6 +11,9 @@ preferences. Access to the REST API is restricted by use of the
 X-Edx-Api-Key HTTP header (which must match settings.EDX_API_KEY; if
 the setting is not present, the API is disabled).
 
+LMS: Added endpoints for AJAX requests to enable/disable notifications
+(which are not yet implemented) and a one-click unsubscribe page.
+
 Common: Added *experimental* support for jsinput type.
 
 Common: Added setting to specify Celery Broker vhost

common/djangoapps/terrain/steps.py

@@ -129,6 +129,13 @@ def should_have_link_with_id_and_text(step, link_id, text):
     assert_equals(link.text, text)
 
 
+@step(r'should see a link to "([^"]*)" with the text "([^"]*)"$')
+def should_have_link_with_path_and_text(step, path, text):
+    link = world.browser.find_link_by_text(text)
+    assert len(link) > 0
+    assert_equals(link.first["href"], django_url(path))
+
+
 @step(r'should( not)? see "(.*)" (?:somewhere|anywhere) (?:in|on) (?:the|this) page')
 def should_see_in_the_page(step, doesnt_appear, text):
     if doesnt_appear:

lms/djangoapps/notification_prefs/__init__.py

@@ -0,0 +1 @@
+NOTIFICATION_PREF_KEY = "notification_pref"

lms/djangoapps/notification_prefs/features/unsubscribe.feature

@@ -0,0 +1,22 @@
+Feature: One-click unsubscribe
+  As a user with notifications enabled
+  I want to be able to unsubscribe from notifications
+
+    Scenario: Unsubscribe when not logged in
+      Given I am an edX user
+      And I am not logged in
+      And I have notifications enabled
+      When I access my unsubscribe url
+      Then my notifications should be disabled
+      And I should see "Unsubscribe Successful!" somewhere on the page
+      And I should see "Click here to return to your dashboard" somewhere on the page
+      And I should see a link to "/dashboard" with the text "here"
+
+    Scenario: Unsubscribe when logged in
+      Given I am a logged in user
+      And I have notifications enabled
+      When I access my unsubscribe url
+      Then my notifications should be disabled
+      And I should see "Unsubscribe Successful!" somewhere on the page
+      And I should see "Click here to return to your dashboard" somewhere on the page
+      And I should see a link to "/dashboard" with the text "here"

lms/djangoapps/notification_prefs/features/unsubscribe.py

@@ -0,0 +1,25 @@
+from django.contrib.auth.models import User
+from lettuce import step, world
+from notification_prefs import NOTIFICATION_PREF_KEY
+from user_api.models import UserPreference
+
+
+USERNAME = "robot"
+UNSUB_TOKEN = "av9E-14sAP1bVBRCPbrTHQ=="
+
+
+@step(u"I have notifications enabled")
+def enable_notifications(step_):
+    user = User.objects.get(username=USERNAME)
+    UserPreference.objects.create(user=user, key=NOTIFICATION_PREF_KEY, value=UNSUB_TOKEN)
+
+
+@step(u"I access my unsubscribe url")
+def access_unsubscribe_url(step_):
+    world.visit("/notification_prefs/unsubscribe/{0}/".format(UNSUB_TOKEN))
+
+
+@step(u"my notifications should be disabled")
+def notifications_should_be_disabled(step_):
+    user = User.objects.get(username=USERNAME)
+    assert not UserPreference.objects.filter(user=user, key=NOTIFICATION_PREF_KEY).exists()

lms/djangoapps/notification_prefs/tests.py

@@ -0,0 +1,202 @@
+from django.contrib.auth.models import AnonymousUser
+from django.core.exceptions import PermissionDenied
+from django.http import Http404
+from django.test import TestCase
+from django.test.client import Client, RequestFactory
+from django.test.utils import override_settings
+from mock import Mock, patch
+
+from notification_prefs import NOTIFICATION_PREF_KEY
+from notification_prefs.views import ajax_enable, ajax_disable, unsubscribe
+from student.tests.factories import UserFactory
+from user_api.models import UserPreference
+
+
+@override_settings(SECRET_KEY="test secret key")
+class NotificationPrefViewTest(TestCase):
+    INITIALIZATION_VECTOR = "\x00" * 16
+
+    @classmethod
+    def setUpClass(cls):
+        # Make sure global state is set up appropriately
+        Client().get("/")
+
+    def setUp(self):
+        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"""
+        self.assertEqual(
+            UserPreference.objects.get(user=user, key=NOTIFICATION_PREF_KEY).value,
+            self.tokens[user]
+        )
+
+    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()
+        )
+
+    # 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")
+        response = unsubscribe(request, "dummy")
+        self.assertEqual(response.status_code, 405)
+
+    def test_unsubscribe_invalid_token(self):
+        def test_invalid_token(token, message):
+            request = self.request_factory.get("dummy")
+            self.assertRaisesRegexp(Http404, "^{}$".format(message), unsubscribe, request, token)
+
+        # 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()
+            response = unsubscribe(request, self.tokens[user])
+            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()
+        unsubscribe(request, self.tokens[self.user])
+        response = unsubscribe(request, self.tokens[self.user])
+        self.assertEqual(response.status_code, 200)
+        self.assertNotPrefExists(self.user)