Merge pull request #10701 from uditijmehta/feature/duplicate-notifica…

…tions-fix

[ENG-5075] Add Admin Screen to Manage Duplicate Notifications

admin/base/urls.py

@@ -36,6 +36,7 @@
             re_path(r'^schema_responses/', include('admin.schema_responses.urls', namespace='schema_responses')),
             re_path(r'^registration_schemas/', include('admin.registration_schemas.urls', namespace='registration_schemas')),
             re_path(r'^cedar_metadata_templates/', include('admin.cedar.urls', namespace='cedar_metadata_templates')),
+            re_path(r'^notifications/', include('admin.notifications.urls', namespace='notifications')),
         ]),
     ),
 ]

admin/notifications/urls.py

@@ -0,0 +1,8 @@
+from django.urls import re_path
+from admin.notifications import views
+
+app_name = 'notifications'
+
+urlpatterns = [
+    re_path(r'^$', views.handle_duplicate_notifications, name='handle_duplicate_notifications'),
+]

admin/notifications/views.py

@@ -0,0 +1,54 @@
+from django.contrib.auth.decorators import user_passes_test
+from django.shortcuts import render, redirect
+from admin.base.utils import osf_staff_check
+from osf.models.notifications import NotificationSubscription
+from django.db.models import Count
+
+def delete_selected_notifications(selected_ids):
+    NotificationSubscription.objects.filter(id__in=selected_ids).delete()
+
+def detect_duplicate_notifications():
+    duplicates = (
+        NotificationSubscription.objects.values('user', 'node', 'event_name')
+        .annotate(count=Count('id'))
+        .filter(count__gt=1)
+    )
+
+    detailed_duplicates = []
+    for dup in duplicates:
+        notifications = NotificationSubscription.objects.filter(
+            user=dup['user'], node=dup['node'], event_name=dup['event_name']
+        ).order_by('created')
+
+        for notification in notifications:
+            detailed_duplicates.append({
+                'id': notification.id,
+                'user': notification.user,
+                'node': notification.node,
+                'event_name': notification.event_name,
+                'created': notification.created,
+                'count': dup['count']
+            })
+
+    return detailed_duplicates
+
+def process_duplicate_notifications(request):
+    detailed_duplicates = detect_duplicate_notifications()
+
+    if request.method == 'POST':
+        selected_ids = request.POST.getlist('selected_notifications')
+        delete_selected_notifications(selected_ids)
+        return detailed_duplicates, 'Selected duplicate notifications have been deleted.', True
+
+    return detailed_duplicates, '', False
+
+@user_passes_test(osf_staff_check)
+def handle_duplicate_notifications(request):
+    detailed_duplicates, message, is_post = process_duplicate_notifications(request)
+
+    context = {'duplicates': detailed_duplicates}
+    if is_post:
+        context['message'] = message
+        return redirect('notifications:handle_duplicate_notifications')
+
+    return render(request, 'notifications/handle_duplicate_notifications.html', context)

admin/templates/base.html

@@ -311,6 +311,9 @@
           {% if perms.osf.change_maintenancestate %}
             <li><a href="{% url 'maintenance:display' %}"><i class='fa fa-link'></i> <span>Maintenance Alerts</span></a></li>
           {% endif %}
+          {% if perms.osf.view_notification %}
+            <li><a href="{% url 'notifications:handle_duplicate_notifications' %}"><i class='fa fa-link'></i><span>Duplicate Notifications</span> </a></li>
+          {% endif %}
           </ul><!-- /.sidebar-menu -->
         </section>
         <!-- /.sidebar -->

admin/templates/notifications/handle_duplicate_notifications.html

@@ -0,0 +1,54 @@
+{% extends "base.html" %}
+{% load render_bundle from webpack_loader %}
+{% load static %}
+
+{% block title %}
+    <title>Duplicate Notifications</title>
+{% endblock title %}
+
+{% block content %}
+    <h2>Duplicate Notifications</h2>
+
+    {% if message %}
+        <div class="alert alert-success">
+            {{ message }}
+        </div>
+    {% endif %}
+
+    {% if duplicates %}
+        <form method="post">
+            {% csrf_token %}
+            <table class="table table-striped table-hover table-responsive">
+                <thead>
+                    <tr>
+                        <th>Select</th>
+                        <th>User</th>
+                        <th>Node</th>
+                        <th>Event Name</th>
+                        <th>Created</th>
+                        <th>Count</th>
+                    </tr>
+                </thead>
+                <tbody>
+                {% for notification in duplicates %}
+                    <tr>
+                        <td><input type="checkbox" name="selected_notifications" value="{{ notification.id }}"></td>
+                        <td>{{ notification.user }}</td>
+                        <td>{{ notification.node }}</td>
+                        <td>{{ notification.event_name }}</td>
+                        <td>{{ notification.created }}</td>
+                        <td>{{ notification.count }}</td>
+                    </tr>
+                {% empty %}
+                    <tr>
+                        <td colspan="6">No duplicate notifications found!</td>
+                    </tr>
+                {% endfor %}
+                </tbody>
+            </table>
+            <button type="submit" class="btn btn-danger">Delete Selected</button>
+        </form>
+    {% else %}
+        <p>No duplicate notifications found.</p>
+    {% endif %}
+{% endblock content %}

admin_tests/notifications/test_views.py

@@ -0,0 +1,69 @@
+import pytest
+from django.test import RequestFactory
+from osf.models import OSFUser, NotificationSubscription, Node
+from admin.notifications.views import (
+    delete_selected_notifications,
+    detect_duplicate_notifications,
+    process_duplicate_notifications
+)
+from tests.base import AdminTestCase
+
+pytestmark = pytest.mark.django_db
+
+class TestNotificationFunctions(AdminTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.user = OSFUser.objects.create(username='admin', is_staff=True)
+        self.node = Node.objects.create(creator=self.user, title='Test Node')
+        self.request_factory = RequestFactory()
+
+    def test_delete_selected_notifications(self):
+        notification1 = NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event1')
+        notification2 = NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event2')
+        notification3 = NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event3')
+
+        delete_selected_notifications([notification1.id, notification2.id])
+
+        assert not NotificationSubscription.objects.filter(id__in=[notification1.id, notification2.id]).exists()
+        assert NotificationSubscription.objects.filter(id=notification3.id).exists()
+
+    def test_detect_duplicate_notifications(self):
+        NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event1')
+        NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event1')
+        NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event2')
+
+        duplicates = detect_duplicate_notifications()
+
+        assert len(duplicates) == 2
+        assert duplicates[0]['user'] == self.user
+        assert duplicates[0]['node'] == self.node
+        assert duplicates[0]['event_name'] == 'event1'
+        assert duplicates[0]['count'] == 2
+
+        non_duplicate_event = [dup for dup in duplicates if dup['event_name'] == 'event2']
+        assert len(non_duplicate_event) == 0
+
+    def test_process_duplicate_notifications_get(self):
+        request = self.request_factory.get('/fake_path')
+        request.user = self.user
+
+        detailed_duplicates, message, is_post = process_duplicate_notifications(request)
+
+        assert detailed_duplicates == []
+        assert message == ''
+        assert not is_post
+
+    def test_process_duplicate_notifications_post(self):
+        notification1 = NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event1')
+        notification2 = NotificationSubscription.objects.create(user=self.user, node=self.node, event_name='event1')
+
+        request = self.request_factory.post('/fake_path', {'selected_notifications': [notification1.id]})
+        request.user = self.user
+
+        detailed_duplicates, message, is_post = process_duplicate_notifications(request)
+
+        assert message == 'Selected duplicate notifications have been deleted.'
+        assert is_post
+        assert not NotificationSubscription.objects.filter(id=notification1.id).exists()
+        assert NotificationSubscription.objects.filter(id=notification2.id).exists()

osf/management/commands/create_test_notifications.py

@@ -0,0 +1,27 @@
+from django.core.management.base import BaseCommand
+from osf.models.notifications import NotificationSubscription
+from osf.models import OSFUser, Node
+from django.utils.crypto import get_random_string
+from django.utils import timezone
+
+class Command(BaseCommand):
+    help = 'Create duplicate notifications for testing'
+
+    def handle(self, *args, **kwargs):
+        user = OSFUser.objects.first()
+        node = Node.objects.first()
+        event_name = 'file_added'
+
+        for _ in range(3):
+            unique_id = get_random_string(length=32)
+            notification = NotificationSubscription.objects.create(
+                user=user,
+                node=node,
+                event_name=event_name,
+                _id=unique_id,
+                created=timezone.now()
+            )
+            notification.email_transactional.add(user)
+            notification.save()
+
+        self.stdout.write(self.style.SUCCESS('Successfully created duplicate notifications'))