[feat] Add unsubscribe link to email notifications

.github/workflows/build.yml

@@ -11,6 +11,7 @@ on:
       - master
       - dev
       - gsoc24-rebased
+      - notification-preferences
 
 jobs:
 

openwisp_notifications/static/openwisp-notifications/css/unsubscribe.css

@@ -0,0 +1,40 @@
+#content, .content {
+  padding: 0 !important;
+}
+.unsubscribe-container {
+  display: flex;
+  justify-content: center;
+  align-items: center;
+  flex-direction: column;
+  height: 95vh;
+}
+.unsubscribe-content {
+  padding: 40px;
+  border-radius: 12px;
+  text-align: center;
+  box-shadow: 0 6px 12px rgba(0, 0, 0, 0.1);
+  max-width: 500px;
+  width: 100%;
+}
+.unsubscribe-content h1 {
+  padding-top: 10px;
+}
+.logo {
+  width: 200px;
+  margin-bottom: 80px;
+}
+.email-icon {
+  background-image: url('../../openwisp-notifications/images/icons/icon-email.svg');
+  width: 50px;
+  height: 50px;
+  margin: 0 auto;
+  transform: scale(2) translate(25%, 25%);
+}
+.footer {
+  margin-top: 20px;
+}
+#confirmation-msg {
+  color: green;
+  margin-top: 20px;
+  font-weight: bold;
+}

openwisp_notifications/static/openwisp-notifications/js/unsubscribe.js

@@ -0,0 +1,49 @@
+'use strict';
+if (typeof gettext === 'undefined') {
+    var gettext = function(word) { return word; };
+}
+
+function updateSubscription(subscribe) {
+    fetch(window.location.href, {
+        method: 'POST',
+        headers: {
+            'Content-Type': 'application/json',
+        },
+        body: JSON.stringify({ subscribe: subscribe })
+    })
+    .then(response => response.json())
+    .then(data => {
+        if (data.success) {
+            const toggleBtn = document.getElementById('toggle-btn');
+            const statusMessage = document.getElementById('status-message');
+            const confirmationMsg = document.getElementById('confirmation-msg');
+
+            if (subscribe) {
+                statusMessage.textContent = gettext('You are currently subscribed to notifications.');
+                toggleBtn.textContent = gettext('Unsubscribe');
+                toggleBtn.setAttribute('data-hasSubscribe', 'true');
+            } else {
+                statusMessage.textContent = gettext('You are currently unsubscribed from notifications.');
+                toggleBtn.textContent = gettext('Subscribe');
+                toggleBtn.setAttribute('data-hasSubscribe', 'false');
+            }
+
+            confirmationMsg.textContent = data.message;
+            confirmationMsg.style.display = 'block';
+        } else {
+            window.alert(data.message);
+        }
+    })
+    .catch(error => {
+        window.console.error('Error:', error);
+    });
+}
+
+document.addEventListener('DOMContentLoaded', function() {
+    const toggleBtn = document.getElementById('toggle-btn');
+    toggleBtn.addEventListener('click', function() {
+        const isSubscribed = toggleBtn.getAttribute('data-hasSubscribe') === 'true';
+        const subscribe = !isSubscribed;
+        updateSubscription(subscribe);
+    });
+});

openwisp_notifications/tasks.py

@@ -9,12 +9,16 @@
 from django.db.utils import OperationalError
 from django.template.loader import render_to_string
 from django.utils import timezone
+from django.utils.safestring import mark_safe
 from django.utils.translation import gettext as _
 
 from openwisp_notifications import settings as app_settings
 from openwisp_notifications import types
 from openwisp_notifications.swapper import load_model, swapper_load_model
-from openwisp_notifications.utils import send_notification_email
+from openwisp_notifications.utils import (
+    generate_unsubscribe_link,
+    send_notification_email,
+)
 from openwisp_utils.admin_theme.email import send_email
 from openwisp_utils.tasks import OpenwispCeleryTask
 
@@ -273,17 +277,26 @@ def send_batched_email_notifications(instance_id):
             .replace('pm', 'p.m.')
         ) + ' UTC'
 
+        user = User.objects.get(id=instance_id)
+
         context = {
             'notifications': unsent_notifications[:display_limit],
             'notifications_count': notifications_count,
             'site_name': current_site.name,
             'start_time': starting_time,
         }
 
-        extra_context = {}
+        unsubscribe_link = generate_unsubscribe_link(user)
+
+        extra_context = {
+            'footer': mark_safe(
+                'To unsubscribe from these notifications, '
+                f'<a href="{unsubscribe_link}">click here</a>.'
+            ),
+        }
         if notifications_count > display_limit:
             extra_context = {
-                'call_to_action_url': f"https://{current_site.domain}/admin/#notifications",
+                'call_to_action_url': f'https://{current_site.domain}/admin/#notifications',
                 'call_to_action_text': _('View all Notifications'),
             }
         context.update(extra_context)
@@ -298,6 +311,10 @@ def send_batched_email_notifications(instance_id):
             body_html=html_content,
             recipients=[email_id],
             extra_context=extra_context,
+            headers={
+                'List-Unsubscribe-Post': 'List-Unsubscribe=One-Click',
+                'List-Unsubscribe': f'<{unsubscribe_link}>',
+            },
         )
 
     unsent_notifications_query.update(emailed=True)

openwisp_notifications/templates/openwisp_notifications/unsubscribe.html

@@ -0,0 +1,55 @@
+{% extends 'account/base_entrance.html' %}
+{% load i18n %}
+{% load static %}
+
+{% block head_title %}
+  <title>{% trans 'Manage Subscription Preferences' %}</title>
+{% endblock %}
+
+{% block extrastyle %}
+  {{ block.super }}
+  <link rel="stylesheet" type="text/css" href="{% static 'openwisp-notifications/css/unsubscribe.css' %}">
+  <script src="{% static 'openwisp-notifications/js/unsubscribe.js' %}"></script>
+{% endblock %}
+
+{% block menu-bar %}
+{% endblock %}
+
+{% block content %}
+  <div class="unsubscribe-container">
+    <img
+      src="{% static 'ui/openwisp/images/openwisp-logo-black.svg' %}"
+      alt="{% trans 'OpenWISP Logo' %}"
+      class="logo"
+    />
+    <div class="unsubscribe-content">
+      <div class="icon email-icon"></div>
+      <h1>{% trans 'Manage Notification Preferences' %}</h1>
+      {% if valid %}
+        <p id="status-message">
+          {% if is_subscribed %}
+            {% trans 'You are currently subscribed to notifications.' %}
+          {% else %}
+            {% trans 'You are currently unsubscribed from notifications.' %}
+          {% endif %}
+        </p>
+        <button id="toggle-btn" class="button default" data-hasSubscribe="{% if is_subscribed %}true{% else %}false{% endif %}">
+          {% if is_subscribed %}
+            {% trans 'Unsubscribe' %}
+          {% else %}
+            {% trans 'Subscribe' %}
+          {% endif %}
+        </button>
+        <p id="confirmation-msg" style="display:none;"></p>
+      {% else %}
+        <h1>{% trans 'Invalid or Expired Link' %}</h1>
+        <p>{% trans 'The link you used is invalid or expired.' %}</p>
+      {% endif %}
+      <div class="footer">
+        <p>
+          {% trans 'Manage Other Preferences?' %} <a href="{% url 'notifications:notification_preference' %}">{% trans 'Click here' %}</a>
+        </p>
+      </div>
+    </div>
+  </div>
+{% endblock %}

openwisp_notifications/tests/test_notifications.py

@@ -1,3 +1,4 @@
+import json
 from datetime import datetime, timedelta
 from unittest.mock import patch
 from uuid import uuid4
@@ -31,11 +32,12 @@
     register_notification_type,
     unregister_notification_type,
 )
+from openwisp_notifications.tokens import email_token_generator
 from openwisp_notifications.types import (
     _unregister_notification_choice,
     get_notification_configuration,
 )
-from openwisp_notifications.utils import _get_absolute_url
+from openwisp_notifications.utils import _get_absolute_url, generate_unsubscribe_link
 from openwisp_users.tests.utils import TestOrganizationMixin
 from openwisp_utils.tests import capture_any_output
 
@@ -1053,6 +1055,113 @@ def test_that_the_notification_is_only_sent_once_to_the_user(self):
         self._create_notification()
         self.assertEqual(notification_queryset.count(), 1)
 
+    def test_email_unsubscribe_token(self):
+        token = email_token_generator.make_token(self.admin)
+
+        with self.subTest('Valid token for the user'):
+            is_valid = email_token_generator.check_token(self.admin, token)
+            self.assertTrue(is_valid)
+
+        with self.subTest('Token used with a different user'):
+            test_user = self._create_user(username='test')
+            is_valid = email_token_generator.check_token(test_user, token)
+            self.assertFalse(is_valid)
+
+        with self.subTest('Token invalidated after password change'):
+            self.admin.set_password('new_password')
+            self.admin.save()
+            is_valid = email_token_generator.check_token(self.admin, token)
+            self.assertFalse(is_valid)
+
+    def test_email_unsubscribe_view(self):
+        unsubscribe_link_generated = generate_unsubscribe_link(self.admin, False)
+        token = unsubscribe_link_generated.split('?token=')[1]
+        local_unsubscribe_url = reverse('notifications:unsubscribe')
+        unsubscribe_url = f"{local_unsubscribe_url}?token={token}"
+
+        with self.subTest('Test GET request with valid token'):
+            response = self.client.get(unsubscribe_url)
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test POST request with valid token'):
+            response = self.client.post(unsubscribe_url)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json()['message'], 'Successfully unsubscribed')
+
+        with self.subTest('Test GET request with invalid token'):
+            response = self.client.get(f"{local_unsubscribe_url}?token=invalid_token")
+            self.assertContains(response, 'Invalid or Expired Link')
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test POST request with invalid token'):
+            response = self.client.post(f"{local_unsubscribe_url}?token=invalid_token")
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest('Test GET request with invalid user'):
+            tester = self._create_user(username='tester')
+            tester_link_generated = generate_unsubscribe_link(tester)
+            token = tester_link_generated.split('?token=')[1]
+            tester_unsubscribe_url = f"{local_unsubscribe_url}?token={token}"
+            tester.delete()
+            response = self.client.get(tester_unsubscribe_url)
+            self.assertContains(response, 'Invalid or Expired Link')
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test POST request with invalid user'):
+            tester = self._create_user(username='tester')
+            tester_link_generated = generate_unsubscribe_link(tester)
+            token = tester_link_generated.split('?token=')[1]
+            tester_unsubscribe_url = f"{local_unsubscribe_url}?token={token}"
+            tester.delete()
+            response = self.client.post(tester_unsubscribe_url)
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest('Test GET request with no token'):
+            response = self.client.get(local_unsubscribe_url)
+            self.assertEqual(response.status_code, 200)
+            self.assertContains(response, 'Invalid or Expired Link')
+            self.assertFalse(response.context['valid'])
+
+        with self.subTest('Test POST request with no token'):
+            response = self.client.post(local_unsubscribe_url)
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(response.json()['message'], 'No token provided')
+
+        with self.subTest('Test POST request with empty JSON body'):
+            response = self.client.post(
+                unsubscribe_url, content_type='application/json'
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json()['message'], 'Successfully unsubscribed')
+
+        with self.subTest('Test POST request with subscribe=True in JSON'):
+            response = self.client.post(
+                unsubscribe_url,
+                data=json.dumps({'subscribe': True}),
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json()['message'], 'Successfully subscribed')
+
+        with self.subTest('Test POST request with subscribe=False in JSON'):
+            response = self.client.post(
+                unsubscribe_url,
+                data=json.dumps({'subscribe': False}),
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.json()['message'], 'Successfully unsubscribed')
+
+        with self.subTest('Test POST request with invalid JSON'):
+            invalid_json = "{'data: invalid}"
+            response = self.client.post(
+                unsubscribe_url,
+                data=invalid_json,
+                content_type='application/json',
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(response.json()['message'], 'Invalid JSON data')
+
     def test_notification_preference_page(self):
         preference_page = 'notifications:user_notification_preference'
         tester = self._create_user(username='tester')