641 lines
24 KiB
Python
641 lines
24 KiB
Python
import json
|
|
from datetime import datetime, timedelta
|
|
from io import StringIO
|
|
from unittest.mock import patch
|
|
|
|
from django.conf import settings
|
|
from django.core.management import call_command
|
|
from django.test import TestCase, override_settings
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from pywebpush import WebPushException
|
|
|
|
from accounts.models import PushSubscription, User
|
|
from tracker.models import (
|
|
DailyNote,
|
|
DoseLog,
|
|
GlobalNote,
|
|
Medication,
|
|
PulseReading,
|
|
ReminderSent,
|
|
ScheduledDose,
|
|
TimeOfDay,
|
|
)
|
|
|
|
|
|
class TrackerTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_superuser("loic", password="hunter2hunter2")
|
|
|
|
# Medication 1: 1/4 every morning.
|
|
self.vetmedin = Medication.objects.create(name="Vetmedin")
|
|
self.morning_only = ScheduledDose.objects.create(
|
|
medication=self.vetmedin, time_of_day=TimeOfDay.MORNING, amount="1/4"
|
|
)
|
|
|
|
# Medication 2: 1/4 morning and 1/4 evening.
|
|
self.fortekor = Medication.objects.create(name="Fortekor")
|
|
self.morning = ScheduledDose.objects.create(
|
|
medication=self.fortekor, time_of_day=TimeOfDay.MORNING, amount="1/4"
|
|
)
|
|
self.evening = ScheduledDose.objects.create(
|
|
medication=self.fortekor, time_of_day=TimeOfDay.EVENING, amount="1/4"
|
|
)
|
|
|
|
def token(self) -> str:
|
|
response = self.client.post(
|
|
"/api/auth/login",
|
|
{"username": "loic", "password": "hunter2hunter2"},
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
return response.json()["token"]
|
|
|
|
def auth_header(self) -> dict:
|
|
return {"HTTP_AUTHORIZATION": f"Bearer {self.token()}"}
|
|
|
|
|
|
class AuthTests(TrackerTestCase):
|
|
def test_login_rejects_a_bad_password(self):
|
|
response = self.client.post(
|
|
"/api/auth/login",
|
|
{"username": "loic", "password": "wrong"},
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_endpoints_require_authentication(self):
|
|
self.assertEqual(self.client.get("/api/tracker/day").status_code, 401)
|
|
|
|
def test_health_is_public(self):
|
|
self.assertEqual(self.client.get("/api/health").status_code, 200)
|
|
|
|
|
|
class DoseApiTests(TrackerTestCase):
|
|
def test_day_lists_every_scheduled_dose_unrecorded(self):
|
|
response = self.client.get("/api/tracker/day", **self.auth_header())
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
doses = response.json()["doses"]
|
|
self.assertEqual(len(doses), 3)
|
|
self.assertTrue(all(dose["taken"] is None for dose in doses))
|
|
# Morning doses come before the evening one.
|
|
self.assertEqual([d["time_of_day"] for d in doses], [1, 1, 3])
|
|
|
|
def test_logging_a_dose_marks_only_that_dose(self):
|
|
response = self.client.post(
|
|
"/api/tracker/doses",
|
|
{"scheduled_dose_id": self.morning.id, "taken": True},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
doses = {d["id"]: d["taken"] for d in response.json()["doses"]}
|
|
self.assertIs(doses[self.morning.id], True)
|
|
self.assertIsNone(doses[self.evening.id])
|
|
self.assertIsNone(doses[self.morning_only.id])
|
|
|
|
def test_logging_the_same_dose_twice_updates_in_place(self):
|
|
for taken in (True, False):
|
|
self.client.post(
|
|
"/api/tracker/doses",
|
|
{"scheduled_dose_id": self.morning.id, "taken": taken},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
|
|
log = DoseLog.objects.get(scheduled_dose=self.morning)
|
|
self.assertFalse(log.taken)
|
|
self.assertIsNone(log.taken_at)
|
|
self.assertEqual(DoseLog.objects.count(), 1)
|
|
|
|
def test_a_dose_can_be_logged_for_a_past_day(self):
|
|
yesterday = timezone.localdate() - timedelta(days=1)
|
|
self.client.post(
|
|
"/api/tracker/doses",
|
|
{
|
|
"scheduled_dose_id": self.evening.id,
|
|
"date": yesterday.isoformat(),
|
|
"taken": True,
|
|
},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
|
|
self.assertTrue(DoseLog.objects.filter(date=yesterday, taken=True).exists())
|
|
self.assertFalse(DoseLog.objects.filter(date=timezone.localdate()).exists())
|
|
|
|
def test_inactive_medication_drops_out_of_the_day(self):
|
|
self.fortekor.is_active = False
|
|
self.fortekor.save()
|
|
|
|
response = self.client.get("/api/tracker/day", **self.auth_header())
|
|
self.assertEqual(len(response.json()["doses"]), 1)
|
|
|
|
|
|
class PulseApiTests(TrackerTestCase):
|
|
def test_recording_the_pulse_twice_in_a_day_overwrites(self):
|
|
for bpm in (180, 192):
|
|
response = self.client.post(
|
|
"/api/tracker/pulse",
|
|
{"bpm": bpm},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
self.assertEqual(PulseReading.objects.count(), 1)
|
|
self.assertEqual(PulseReading.objects.get().bpm, 192)
|
|
|
|
def test_an_implausible_pulse_is_rejected(self):
|
|
response = self.client.post(
|
|
"/api/tracker/pulse",
|
|
{"bpm": 5},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
self.assertEqual(response.status_code, 422)
|
|
|
|
def test_stats_report_a_rising_trend(self):
|
|
today = timezone.localdate()
|
|
for offset, bpm in enumerate([160, 170, 180, 190]):
|
|
PulseReading.objects.create(
|
|
date=today - timedelta(days=3 - offset), bpm=bpm
|
|
)
|
|
|
|
stats = self.client.get("/api/tracker/pulse/stats", **self.auth_header()).json()
|
|
|
|
self.assertEqual(stats["count"], 4)
|
|
self.assertEqual(stats["minimum"], 160)
|
|
self.assertEqual(stats["maximum"], 190)
|
|
self.assertEqual(stats["average"], 175.0)
|
|
self.assertEqual(stats["trend_bpm_per_day"], 10.0)
|
|
|
|
def test_stats_are_empty_without_readings(self):
|
|
stats = self.client.get("/api/tracker/pulse/stats", **self.auth_header()).json()
|
|
|
|
self.assertEqual(stats["count"], 0)
|
|
self.assertIsNone(stats["average"])
|
|
self.assertIsNone(stats["trend_bpm_per_day"])
|
|
|
|
|
|
class NoteApiTests(TrackerTestCase):
|
|
def test_the_day_includes_its_note(self):
|
|
DailyNote.objects.create(body="Ate well.")
|
|
|
|
response = self.client.get("/api/tracker/day", **self.auth_header())
|
|
|
|
self.assertEqual(response.json()["note"]["body"], "Ate well.")
|
|
|
|
def test_saving_a_daily_note_twice_overwrites(self):
|
|
for body in ("Sleepy.", "Playful after all."):
|
|
response = self.client.post(
|
|
"/api/tracker/notes/daily",
|
|
{"body": body},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
self.assertEqual(DailyNote.objects.count(), 1)
|
|
self.assertEqual(DailyNote.objects.get().body, "Playful after all.")
|
|
|
|
def test_a_daily_note_can_target_a_past_day(self):
|
|
yesterday = timezone.localdate() - timedelta(days=1)
|
|
self.client.post(
|
|
"/api/tracker/notes/daily",
|
|
{"body": "Rough night.", "date": yesterday.isoformat()},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
|
|
self.assertTrue(DailyNote.objects.filter(date=yesterday).exists())
|
|
self.assertFalse(DailyNote.objects.filter(date=timezone.localdate()).exists())
|
|
|
|
def test_an_empty_body_removes_the_daily_note(self):
|
|
DailyNote.objects.create(body="Old news.")
|
|
|
|
self.client.post(
|
|
"/api/tracker/notes/daily",
|
|
{"body": " "},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
|
|
self.assertEqual(DailyNote.objects.count(), 0)
|
|
|
|
def test_the_global_note_exists_without_ever_being_created(self):
|
|
response = self.client.get("/api/tracker/notes/global", **self.auth_header())
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["body"], "")
|
|
|
|
def test_the_global_note_is_shared_and_unique(self):
|
|
response = self.client.put(
|
|
"/api/tracker/notes/global",
|
|
{"body": "Vet on Friday."},
|
|
content_type="application/json",
|
|
**self.auth_header(),
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
other = User.objects.create_superuser("marie", password="hunter2hunter2")
|
|
self.client.force_login(other)
|
|
response = self.client.get("/api/tracker/notes/global")
|
|
|
|
self.assertEqual(response.json()["body"], "Vet on Friday.")
|
|
self.assertEqual(GlobalNote.objects.count(), 1)
|
|
|
|
|
|
class AppShellTests(TrackerTestCase):
|
|
"""The pages are one Vue SPA; Django only serves the shell and its config."""
|
|
|
|
def test_the_pages_require_login(self):
|
|
for name in ("tracker:today", "tracker:recap"):
|
|
response = self.client.get(reverse(name))
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("/admin/login/", response["Location"])
|
|
|
|
def test_both_routes_serve_the_spa_shell(self):
|
|
self.client.force_login(self.user)
|
|
for name in ("tracker:today", "tracker:recap"):
|
|
response = self.client.get(reverse(name))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertContains(response, 'id="app"')
|
|
self.assertContains(response, "osiris-config")
|
|
|
|
def test_the_shell_sets_the_csrf_cookie_for_the_api(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("tracker:today"))
|
|
|
|
self.assertIn("csrftoken", response.cookies)
|
|
self.assertContains(response, "csrfToken")
|
|
|
|
@override_settings(VAPID_PUBLIC_KEY="BMn-test-public-key")
|
|
def test_the_config_carries_the_reminder_diagnostics(self):
|
|
self.morning.reminder_time = "08:00"
|
|
self.morning.save()
|
|
PushSubscription.objects.create(
|
|
user=self.user, endpoint="https://push.example/a", p256dh="k", auth="s"
|
|
)
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.get(reverse("tracker:today"))
|
|
|
|
self.assertContains(response, "BMn-test-public-key")
|
|
self.assertContains(response, '"activeDoses": 3')
|
|
self.assertContains(response, '"dosesWithReminders": 1')
|
|
self.assertContains(response, '"subscribedDevices": 1')
|
|
|
|
@override_settings(VAPID_PUBLIC_KEY="", VAPID_PRIVATE_KEY="")
|
|
def test_missing_vapid_keys_reach_the_frontend_as_an_empty_string(self):
|
|
# The frontend's diagnostics card explains the situation to the user.
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("tracker:today"))
|
|
self.assertContains(response, '"vapidPublicKey": ""')
|
|
|
|
|
|
class RecapTests(TrackerTestCase):
|
|
def test_a_fully_ticked_day_reads_as_complete(self):
|
|
today = timezone.localdate()
|
|
for dose in (self.morning, self.morning_only, self.evening):
|
|
DoseLog.objects.create(scheduled_dose=dose, date=today, taken=True)
|
|
|
|
recap = self.client.get("/api/tracker/recap", **self.auth_header()).json()
|
|
|
|
self.assertEqual(recap[0]["date"], today.isoformat())
|
|
self.assertTrue(recap[0]["complete"])
|
|
self.assertEqual(recap[0]["taken"], 3)
|
|
self.assertEqual(recap[0]["expected"], 3)
|
|
|
|
def test_a_missed_or_unrecorded_dose_leaves_the_day_incomplete(self):
|
|
today = timezone.localdate()
|
|
DoseLog.objects.create(scheduled_dose=self.morning, date=today, taken=True)
|
|
DoseLog.objects.create(scheduled_dose=self.evening, date=today, taken=False)
|
|
# morning_only is not recorded at all.
|
|
|
|
recap = self.client.get("/api/tracker/recap", **self.auth_header()).json()
|
|
|
|
self.assertFalse(recap[0]["complete"])
|
|
self.assertEqual(recap[0]["taken"], 1)
|
|
|
|
def test_the_recap_carries_pulse_and_note(self):
|
|
yesterday = timezone.localdate() - timedelta(days=1)
|
|
PulseReading.objects.create(date=yesterday, bpm=180)
|
|
DailyNote.objects.create(date=yesterday, body="Ate well.")
|
|
|
|
recap = self.client.get("/api/tracker/recap", **self.auth_header()).json()
|
|
|
|
self.assertEqual(recap[1]["bpm"], 180)
|
|
self.assertEqual(recap[1]["note"], "Ate well.")
|
|
self.assertIsNone(recap[0]["bpm"])
|
|
|
|
def test_the_recap_lists_every_daily_note(self):
|
|
DailyNote.objects.create(body="Today's entry.")
|
|
DailyNote.objects.create(
|
|
date=timezone.localdate() - timedelta(days=30), body="A month ago."
|
|
)
|
|
|
|
response = self.client.get(
|
|
"/api/tracker/notes/daily", **self.auth_header()
|
|
).json()
|
|
|
|
self.assertEqual(
|
|
[n["body"] for n in response], ["Today's entry.", "A month ago."]
|
|
)
|
|
|
|
|
|
class PwaTests(TrackerTestCase):
|
|
def test_the_service_worker_is_served_from_the_root(self):
|
|
# Scope: under /static/ it could not control the /admin/ pages it caches.
|
|
response = self.client.get("/sw.js")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "application/javascript")
|
|
|
|
def test_the_page_links_the_manifest_and_registers_the_worker(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.get(reverse("tracker:today"))
|
|
|
|
self.assertContains(response, 'rel="manifest"')
|
|
self.assertContains(response, "serviceWorker.register")
|
|
|
|
def test_the_manifest_is_valid_json_pointing_at_real_icons(self):
|
|
response = self.client.get(reverse("manifest"))
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response["Content-Type"], "application/manifest+json")
|
|
|
|
manifest = json.loads(response.content)
|
|
self.assertEqual(manifest["start_url"], reverse("tracker:today"))
|
|
self.assertEqual(len(manifest["icons"]), 3)
|
|
for icon in manifest["icons"]:
|
|
self.assertTrue(icon["src"].startswith(settings.STATIC_URL))
|
|
|
|
def test_the_login_page_is_installable_too(self):
|
|
# The app opens on the login screen when the session lapses.
|
|
response = self.client.get("/admin/login/")
|
|
|
|
self.assertContains(response, 'rel="manifest"')
|
|
|
|
|
|
@override_settings(
|
|
VAPID_PUBLIC_KEY="BMn-test-public-key",
|
|
VAPID_PRIVATE_KEY="DiXsYTkxHVb68y2-9CGUZheAe7Iah9cLZ_vXc9DbJ5w",
|
|
)
|
|
class PushTests(TrackerTestCase):
|
|
def subscribe(self, endpoint="https://push.example/abc") -> PushSubscription:
|
|
return PushSubscription.objects.create(
|
|
user=self.user,
|
|
endpoint=endpoint,
|
|
p256dh="p256dh-key",
|
|
auth="auth-secret",
|
|
)
|
|
|
|
def test_a_device_can_subscribe_and_unsubscribe(self):
|
|
self.client.force_login(self.user)
|
|
body = {
|
|
"endpoint": "https://push.example/xyz",
|
|
"keys": {"p256dh": "key", "auth": "secret"},
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/push/subscribe", body, content_type="application/json"
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(self.user.push_subscriptions.count(), 1)
|
|
|
|
# Subscribing the same endpoint twice must not duplicate the device.
|
|
self.client.post("/api/push/subscribe", body, content_type="application/json")
|
|
self.assertEqual(self.user.push_subscriptions.count(), 1)
|
|
|
|
self.client.post(
|
|
"/api/push/unsubscribe",
|
|
{"endpoint": body["endpoint"]},
|
|
content_type="application/json",
|
|
)
|
|
self.assertEqual(self.user.push_subscriptions.count(), 0)
|
|
|
|
def test_the_test_button_needs_a_subscribed_device(self):
|
|
self.client.force_login(self.user)
|
|
response = self.client.post("/api/push/test")
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_the_test_button_pushes_to_every_device(self, webpush):
|
|
self.subscribe("https://push.example/phone")
|
|
self.subscribe("https://push.example/tablet")
|
|
self.client.force_login(self.user)
|
|
|
|
response = self.client.post("/api/push/test")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(webpush.call_count, 2)
|
|
payload = json.loads(webpush.call_args.kwargs["data"])
|
|
self.assertIn("test", payload["title"])
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_a_dead_subscription_is_dropped_rather_than_retried(self, webpush):
|
|
subscription = self.subscribe()
|
|
response = type("R", (), {"status_code": 410})()
|
|
webpush.side_effect = WebPushException("gone", response=response)
|
|
|
|
self.client.force_login(self.user)
|
|
self.client.post("/api/push/test")
|
|
|
|
self.assertFalse(
|
|
PushSubscription.objects.filter(pk=subscription.pk).exists(),
|
|
"A 410 means the browser discarded the subscription; the row is dead.",
|
|
)
|
|
|
|
|
|
@override_settings(
|
|
VAPID_PUBLIC_KEY="BMn-test-public-key",
|
|
VAPID_PRIVATE_KEY="DiXsYTkxHVb68y2-9CGUZheAe7Iah9cLZ_vXc9DbJ5w",
|
|
)
|
|
class SendRemindersTests(TrackerTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
PushSubscription.objects.create(
|
|
user=self.user,
|
|
endpoint="https://push.example/phone",
|
|
p256dh="key",
|
|
auth="secret",
|
|
)
|
|
|
|
def run_command(self, at: str, **kwargs) -> str:
|
|
"""Run send_reminders as if the clock read `at` (local time, today)."""
|
|
hour, minute = (int(part) for part in at.split(":"))
|
|
now = timezone.make_aware(
|
|
datetime.combine(timezone.localdate(), datetime.min.time()).replace(
|
|
hour=hour, minute=minute
|
|
)
|
|
)
|
|
out = StringIO()
|
|
with patch("django.utils.timezone.localtime", return_value=now):
|
|
call_command("send_reminders", stdout=out, stderr=out, **kwargs)
|
|
return out.getvalue()
|
|
|
|
def set_reminder(self, dose: ScheduledDose, at: str):
|
|
hour, minute = (int(part) for part in at.split(":"))
|
|
dose.reminder_time = datetime.min.time().replace(hour=hour, minute=minute)
|
|
dose.save()
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_a_due_dose_is_reminded_once(self, webpush):
|
|
self.set_reminder(self.morning, "08:00")
|
|
|
|
self.run_command("08:05")
|
|
|
|
self.assertEqual(webpush.call_count, 1)
|
|
payload = json.loads(webpush.call_args.kwargs["data"])
|
|
self.assertIn("Fortekor 1/4", payload["body"])
|
|
self.assertEqual(ReminderSent.objects.count(), 1)
|
|
|
|
# A second run the same day must stay quiet.
|
|
self.run_command("08:30")
|
|
self.assertEqual(webpush.call_count, 1)
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_doses_due_at_the_same_time_share_one_notification(self, webpush):
|
|
self.set_reminder(self.morning, "08:00")
|
|
self.set_reminder(self.morning_only, "08:00")
|
|
|
|
self.run_command("08:01")
|
|
|
|
self.assertEqual(webpush.call_count, 1, "One push, not one per pill.")
|
|
payload = json.loads(webpush.call_args.kwargs["data"])
|
|
self.assertIn("Fortekor", payload["body"])
|
|
self.assertIn("Vetmedin", payload["body"])
|
|
self.assertEqual(ReminderSent.objects.count(), 2)
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_a_dose_already_given_is_not_reminded(self, webpush):
|
|
self.set_reminder(self.morning, "08:00")
|
|
DoseLog.objects.create(scheduled_dose=self.morning, taken=True)
|
|
|
|
self.run_command("08:05")
|
|
|
|
webpush.assert_not_called()
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_nothing_fires_before_the_reminder_time(self, webpush):
|
|
self.set_reminder(self.evening, "19:00")
|
|
|
|
self.run_command("18:59")
|
|
|
|
webpush.assert_not_called()
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_a_long_missed_reminder_is_not_fired_late(self, webpush):
|
|
# The server was down all morning; do not buzz at 23:00 about the 08:00 dose.
|
|
self.set_reminder(self.morning, "08:00")
|
|
|
|
self.run_command("23:00")
|
|
|
|
webpush.assert_not_called()
|
|
self.assertEqual(ReminderSent.objects.count(), 0)
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_dry_run_sends_and_records_nothing(self, webpush):
|
|
self.set_reminder(self.morning, "08:00")
|
|
|
|
output = self.run_command("08:05", dry_run=True)
|
|
|
|
webpush.assert_not_called()
|
|
self.assertEqual(ReminderSent.objects.count(), 0)
|
|
self.assertIn("dry-run", output)
|
|
|
|
@patch("tracker.push.webpush")
|
|
def test_a_failed_delivery_is_retried_on_the_next_run(self, webpush):
|
|
self.set_reminder(self.morning, "08:00")
|
|
webpush.side_effect = WebPushException("push service down")
|
|
|
|
self.run_command("08:05")
|
|
self.assertEqual(
|
|
ReminderSent.objects.count(), 0, "Nothing recorded, so it can retry."
|
|
)
|
|
|
|
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)))
|