Add endpoints to set user notification preference

CHANGELOG.rst

@@ -5,6 +5,9 @@ These are notable changes in edx-platform.  This is a rolling list of changes,
 in roughly chronological order, most recent first.  Add your entries at or near
 the top.  Include a label indicating the component affected.
 
+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,185 @@
+from django.contrib.auth.models import AnonymousUser
+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 mitxmako.middleware import MakoMiddleware
+from student.tests.factories import UserFactory
+from user_api.models import UserPreference
+from notification_prefs import NOTIFICATION_PREF_KEY
+from notification_prefs.views import ajax_enable, ajax_disable, unsubscribe
+
+
+@override_settings(SECRET_KEY="test secret key")
+class NotificationPrefViewTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        # Make sure global state is set up appropriately
+        Client().get("/")
+
+    def setUp(self):
+        self.user = UserFactory.create(username="testuser")
+        # Username with length equal to AES block length to test padding
+        self.aes_block_length_user = UserFactory.create(username="sixteencharsuser")
+        # Tokens are intentionally hard-coded instead of computed to help us
+        # avoid breaking existing links.
+        self.tokens = {
+            # Encrypted value: "testuser" + "\x08" * 8
+            self.user: "DyYxCj3oVl9vVgq_VHlfqw==",
+            # Encrypted value: "sixteencharsuser" + "\x10" * 16
+            self.aes_block_length_user: "_E9YK4jYDL1MBMFWd_Dt4tRGw8HDEmlcLVFawgY9wI8=",
+        }
+        self.request_factory = RequestFactory()
+
+    def create_prefs(self):
+        for (user, token) in self.tokens.items():
+            UserPreference.objects.create(user=user, key=NOTIFICATION_PREF_KEY, value=token)
+
+    def assertPrefValid(self, user):
+        self.assertEqual(
+            UserPreference.objects.get(user=user, key=NOTIFICATION_PREF_KEY).value,
+            self.tokens[user]
+        )
+
+    def assertNotPrefExists(self, user):
+        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()
+        response = ajax_enable(request)
+        self.assertEqual(response.status_code, 403)
+        self.assertNotPrefExists(self.user)
+
+    def test_ajax_enable_success(self):
+        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)
+
+        test_user(self.user)
+        test_user(self.aes_block_length_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()
+        response = ajax_disable(request)
+        self.assertEqual(response.status_code, 403)
+        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):
+            request = self.request_factory.get("dummy")
+            self.assertRaises(Http404, unsubscribe, request, token)
+
+        # Invalid base64 encoding
+        test_invalid_token("Non-ASCII\xff")
+        test_invalid_token("ZOMG INVALID BASE64 CHARS!!!")
+        test_invalid_token(self.tokens[self.user][:-1])
+
+        # Token of wrong length
+        test_invalid_token(self.tokens[self.user][:-4])
+
+        # Invalid padding (ends in 0 byte)
+        # Encrypted value: "testuser" + "\x00" * 8
+        test_invalid_token("yhrNEjt48uMRZc3U3uR4vA==")
+
+        # Invalid padding (ends in byte > 16)
+        # Encrypted value: "testusertestuser"
+        test_invalid_token("LqItcaGOQXK0mglIElnMng==")
+
+        # Invalid padding (entire string is padding)
+        # Encrypted value: "\x10" * 16
+        test_invalid_token("1EbDwcMSaVwtUVrCBj3Ajw==")
+
+        # Nonexistent user
+        # Encrypted value: "nonexistentuser\x01"
+        test_invalid_token("KnJTFMYitSOem5Sw2LuYBg==")
+
+    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)
+
+        test_user(self.user)
+        test_user(self.aes_block_length_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)

lms/djangoapps/notification_prefs/views.py

@@ -0,0 +1,119 @@
+from base64 import urlsafe_b64encode, urlsafe_b64decode
+from Crypto.Cipher import AES
+from django.conf import settings
+from django.contrib.auth.models import User
+from django.http import Http404, HttpResponse, HttpResponseForbidden, HttpResponseNotAllowed
+from hashlib import sha256
+from mitxmako.shortcuts import render_to_response
+from notification_prefs import NOTIFICATION_PREF_KEY
+from user_api.models import UserPreference
+
+
+class UsernameCodec(object):
+    AES_BLOCK_LEN = 16
+
+    def __init__(self):
+        hash_ = sha256()
+        hash_.update(settings.SECRET_KEY)
+        self.cipher = AES.new(hash_.digest())
+
+    def _add_padding(self, str):
+        """Return str with PKCS#7 padding added"""
+        padding_len = self.AES_BLOCK_LEN - (len(str) % self.AES_BLOCK_LEN)
+        return str + (padding_len * chr(padding_len))
+
+    def _remove_padding(self, str):
+        """Return str with PKCS#7 padding trimmed"""
+        num_pad_bytes = ord(str[-1])
+        if num_pad_bytes < 1 or num_pad_bytes > self.AES_BLOCK_LEN or num_pad_bytes >= len(str):
+            return None
+        return str[:-num_pad_bytes]
+
+    def encode(self, username):
+        return urlsafe_b64encode(self.cipher.encrypt(self._add_padding(username)))
+
+    def decode(self, encoded):
+        return self._remove_padding(self.cipher.decrypt(urlsafe_b64decode(encoded)))
+
+
+def _validate_ajax(request):
+    """
+    Ensure that `request` is valid
+
+    If the request is invalid, an appropriate response is returned. Otherwise,
+    None is returned.
+    """
+    if request.method != "POST":
+        return HttpResponseNotAllowed(["POST"])
+
+    if not request.user.is_authenticated():
+        return HttpResponseForbidden()
+
+
+def ajax_enable(request):
+    """
+    A view that enables notifications for the authenticated user
+
+    This view should be invoked by an AJAX POST call. It returns status 204
+    (no content) or an error. If notifications were already enabled for this
+    user, this has no effect. Otherwise, a preference is created with the
+    unsubscribe token (an ecnryption of the username) as the value.unsernam
+    """
+    validation_response = _validate_ajax(request)
+    if validation_response is not None:
+        return validation_response
+
+    UserPreference.objects.get_or_create(
+        user=request.user,
+        key=NOTIFICATION_PREF_KEY,
+        defaults={
+            "value": UsernameCodec().encode(request.user.username)
+        }
+    )
+
+    return HttpResponse(status=204)
+
+
+def ajax_disable(request):
+    """
+    A view that disables notifications for the authenticated user
+
+    This view should be invoked by an AJAX POST call. It returns status 204
+    (no content) or an error.
+    """
+    validation_response = _validate_ajax(request)
+    if validation_response is not None:
+        return validation_response
+
+    UserPreference.objects.filter(
+        user=request.user,
+        key=NOTIFICATION_PREF_KEY
+    ).delete()
+
+    return HttpResponse(status=204)
+
+
+def unsubscribe(request, token):
+    """
+    A view that disables notifications for a user who may not be authenticated
+
+    This view is meant to be the target of an unsubscribe link. The request
+    must be a GET, and the `token` parameter must decrypt to a valid username.
+
+    A 405 will be returned if the request method is not GET. A 404 will be
+    returned if the token parameter is missing or if the given token does not
+    decrypt to a valid username. On success, the response will contain a page
+    indicating success.
+    """
+    if request.method != "GET":
+        return HttpResponseNotAllowed(["GET"])
+
+    try:
+        username = UsernameCodec().decode(token.encode())
+        user = User.objects.get(username=username)
+    except Exception as e:
+        raise Http404(e.message)
+
+    UserPreference.objects.filter(user=user, key=NOTIFICATION_PREF_KEY).delete()
+
+    return render_to_response("unsubscribe.html", {})