feat: add test notif button on admin

accounts/admin.py

@@ -1,8 +1,50 @@
-from django.contrib import admin
+from django.contrib import admin, messages
 from django.contrib.auth.admin import UserAdmin
+from django.http import HttpResponseRedirect
+from django.urls import reverse
 
 from accounts.models import ApiToken, PushSubscription, User
 
+TEST_PAYLOAD = {
+    "title": "Osiris — test",
+    "body": "Reminders are working. This is what a dose reminder looks like.",
+    "url": "/admin/today/",
+    "tag": "test",
+}
+
+
+def send_test_push(request, subscription: PushSubscription) -> bool:
+    """Push a test notification and report the outcome as an admin message."""
+    # Imported here rather than at module scope: tracker.push imports this module's
+    # models, so a top-level import would be circular.
+    from tracker.push import push_is_configured, send_to_subscription
+
+    if not push_is_configured():
+        messages.error(
+            request,
+            "VAPID keys are not set on the server — run manage.py generate_vapid_keys.",
+        )
+        return False
+
+    pk = subscription.pk
+    if send_to_subscription(subscription, TEST_PAYLOAD):
+        messages.success(request, f"Test notification sent to {subscription}.")
+        return True
+
+    # send_to_subscription deletes the row on a 404/410, which is the single most
+    # useful thing to know here: the device is gone, not merely unreachable.
+    if not PushSubscription.objects.filter(pk=pk).exists():
+        messages.warning(
+            request,
+            "That device had discarded its subscription (the push service returned "
+            "404/410), so it has been removed. Re-enable reminders on the device.",
+        )
+    else:
+        messages.error(
+            request, "The push service rejected the message. See the server log."
+        )
+    return False
+
 
 @admin.register(User)
 class OsirisUserAdmin(UserAdmin):
@@ -15,6 +57,38 @@ class PushSubscriptionAdmin(admin.ModelAdmin):
 
     list_display = ["user", "user_agent", "created_at", "last_success_at"]
     readonly_fields = ["endpoint", "p256dh", "auth", "created_at", "last_success_at"]
+    actions = ["send_test_notification"]
+    change_form_template = "admin/accounts/pushsubscription/change_form.html"
+
+    @admin.action(description="Send a test notification to the selected devices")
+    def send_test_notification(self, request, queryset):
+        for subscription in queryset:
+            send_test_push(request, subscription)
+
+    def change_view(self, request, object_id, form_url="", extra_context=None):
+        """Handle the 'Send test notification' button.
+
+        Intercepted before the form is validated or saved: sending a test is not an
+        edit, and it should not be blocked by an unrelated validation error.
+        """
+        if request.method == "POST" and "_send_test" in request.POST:
+            subscription = self.get_object(request, object_id)
+            if subscription is None:
+                return self._get_obj_does_not_exist_redirect(
+                    request, self.opts, object_id
+                )
+
+            send_test_push(request, subscription)
+
+            # Back to this object so the message is read in context — unless the push
+            # came back 404/410, in which case the row was deleted as dead.
+            if PushSubscription.objects.filter(pk=subscription.pk).exists():
+                return HttpResponseRedirect(request.get_full_path())
+            return HttpResponseRedirect(
+                reverse("admin:accounts_pushsubscription_changelist")
+            )
+
+        return super().change_view(request, object_id, form_url, extra_context)
 
 
 @admin.register(ApiToken)

templates/admin/accounts/pushsubscription/change_form.html

@@ -0,0 +1,9 @@
+{% extends "admin/change_form.html" %}
+
+{% block submit_buttons_bottom %}
+  {{ block.super }}
+  <div class="submit-row" style="display:flex; justify-content:flex-end">
+    {# Posts the change form, which admin.response_change() intercepts. #}
+    <input type="submit" name="_send_test" value="Send test notification">
+  </div>
+{% endblock %}

tracker/tests.py

@@ -486,3 +486,81 @@ class SendRemindersTests(TrackerTestCase):
         webpush.side_effect = None
         self.run_command("08:10")
         self.assertEqual(ReminderSent.objects.count(), 1)
+
+
+@override_settings(
+    VAPID_PUBLIC_KEY="BMn-test-public-key",
+    VAPID_PRIVATE_KEY="DiXsYTkxHVb68y2-9CGUZheAe7Iah9cLZ_vXc9DbJ5w",
+)
+class PushSubscriptionAdminTests(TrackerTestCase):
+    def setUp(self):
+        super().setUp()
+        self.client.force_login(self.user)
+        self.subscription = PushSubscription.objects.create(
+            user=self.user,
+            endpoint="https://push.example/phone",
+            p256dh="key",
+            auth="secret",
+        )
+        self.url = reverse(
+            "admin:accounts_pushsubscription_change", args=[self.subscription.pk]
+        )
+
+    def messages_from(self, response) -> list[str]:
+        return [str(m) for m in response.context["messages"]]
+
+    def test_the_change_page_offers_the_button(self):
+        response = self.client.get(self.url)
+        self.assertContains(response, "_send_test")
+        self.assertContains(response, "Send test notification")
+
+    @patch("tracker.push.webpush")
+    def test_the_button_pushes_to_that_device(self, webpush):
+        response = self.client.post(self.url, {"_send_test": ""}, follow=True)
+
+        webpush.assert_called_once()
+        sent_to = webpush.call_args.kwargs["subscription_info"]
+        self.assertEqual(sent_to["endpoint"], self.subscription.endpoint)
+        self.assertIn("Test notification sent", " ".join(self.messages_from(response)))
+
+    @patch("tracker.push.webpush")
+    def test_a_dead_device_is_reported_and_removed(self, webpush):
+        webpush.side_effect = WebPushException(
+            "gone", response=type("R", (), {"status_code": 410})()
+        )
+
+        response = self.client.post(self.url, {"_send_test": ""}, follow=True)
+
+        self.assertFalse(
+            PushSubscription.objects.filter(pk=self.subscription.pk).exists()
+        )
+        self.assertIn("has been removed", " ".join(self.messages_from(response)))
+
+    @patch("tracker.push.webpush")
+    def test_the_bulk_action_pushes_to_each_selected_device(self, webpush):
+        other = PushSubscription.objects.create(
+            user=self.user,
+            endpoint="https://push.example/tablet",
+            p256dh="key",
+            auth="secret",
+        )
+
+        response = self.client.post(
+            reverse("admin:accounts_pushsubscription_changelist"),
+            {
+                "action": "send_test_notification",
+                "_selected_action": [self.subscription.pk, other.pk],
+            },
+            follow=True,
+        )
+
+        self.assertEqual(webpush.call_count, 2)
+        self.assertEqual(len(self.messages_from(response)), 2)
+
+    @override_settings(VAPID_PUBLIC_KEY="", VAPID_PRIVATE_KEY="")
+    @patch("tracker.push.webpush")
+    def test_it_says_so_when_the_server_has_no_keys(self, webpush):
+        response = self.client.post(self.url, {"_send_test": ""}, follow=True)
+
+        webpush.assert_not_called()
+        self.assertIn("VAPID keys are not set", " ".join(self.messages_from(response)))