Push notifications: period soon/late reminders, test button

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_app.py

@@ -1,5 +1,7 @@
+import base64
 import json
 import sqlite3
+import sys
 import tempfile
 import threading
 import unittest
@@ -138,6 +140,21 @@ class ServerTests(unittest.TestCase):
         }})
         self.assertEqual((status, result), (200, {"added": 2}))
 
+    def test_push_is_off_without_a_helper(self):
+        _, state = self.request("GET", "/api/state", user="kayla")
+        self.assertFalse(state["push"]["available"])
+        self.assertEqual(self.request("POST", "/api/push/test", user="kayla")[0], 503)
+        _, state = self.request("GET", "/api/state", user="fred")
+        self.assertIsNone(state["push"])
+
+    def test_remind_days_setting(self):
+        self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"remindDays": 3})[1], {"shareDetails": False, "remindDays": 3})
+        self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"remindDays": 9})[0], 400)
+        self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"remindDays": True})[0], 400)
+        self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"other": 1})[0], 400)
+        _, state = self.request("GET", "/api/state", user="kayla")
+        self.assertEqual(state["push"]["remindDays"], 3)
+
     def test_static_files(self):
         status, body = self.request("GET", "/")
         self.assertEqual(status, 200)
@@ -146,5 +163,62 @@ class ServerTests(unittest.TestCase):
         self.assertEqual(self.request("GET", "/manifest.webmanifest")[0], 200)
 
 
+FAKE_NOTIFY = """#!{python}
+import json, sqlite3, sys
+command, database = sys.argv[1], sys.argv[3]
+if command == "init":
+    with sqlite3.connect(database) as db:
+        db.execute("INSERT OR IGNORE INTO settings VALUES ('vapid_public', 'PUBLIC')")
+    print(json.dumps({{"publicKey": "PUBLIC"}}))
+else:
+    print(json.dumps({{"sent": 1, "failed": 0, "removed": 0, "errors": [], "contact": sys.argv[5]}}))
+"""
+
+APPLE_SUBSCRIPTION = {
+    "endpoint": "https://web.push.apple.com/QGuQyavXutnMH",
+    "keys": {"p256dh": base64.urlsafe_b64encode(b"\x04" + b"k" * 64).decode().rstrip("="), "auth": base64.urlsafe_b64encode(b"a" * 16).decode().rstrip("=")},
+}
+
+
+class PushServerTests(ServerTests):
+    def setUp(self):
+        super().setUp()
+        script = Path(self.temp.name) / "notify"
+        script.write_text(FAKE_NOTIFY.format(python=sys.executable))
+        script.chmod(0o755)
+        self.server.notify = str(script)
+        self.server.push_contact = "https://flowtrack.example.org"
+        self.server.run_notify("init")
+
+    # Every other server test also runs again with push switched on.
+    test_push_is_off_without_a_helper = None
+
+    def test_push_state(self):
+        _, state = self.request("GET", "/api/state", user="kayla")
+        self.assertEqual(state["push"], {"available": True, "publicKey": "PUBLIC", "remindDays": 2, "devices": 0})
+
+    def test_subscribe_test_and_unsubscribe(self):
+        self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=APPLE_SUBSCRIPTION), (200, {"devices": 1}))
+        self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=APPLE_SUBSCRIPTION), (200, {"devices": 1}))
+        status, result = self.request("POST", "/api/push/test", user="kayla")
+        self.assertEqual((status, result["sent"], result["contact"]), (200, 1, "https://flowtrack.example.org"))
+        self.assertEqual(self.request("POST", "/api/push/test", user="fred")[0], 403)
+        self.assertEqual(self.request("POST", "/api/push/unsubscribe", user="kayla", body={"endpoint": APPLE_SUBSCRIPTION["endpoint"]}), (200, {"devices": 0}))
+
+    def test_only_real_push_services(self):
+        for endpoint in (
+            "https://127.0.0.1/push",
+            "http://web.push.apple.com/x",
+            "https://evilpush.apple.com.example.org/x",
+            "https://router.nordhammer.it/api",
+        ):
+            body = {**APPLE_SUBSCRIPTION, "endpoint": endpoint}
+            self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=body)[0], 400, endpoint)
+        bad_keys = {**APPLE_SUBSCRIPTION, "keys": {"p256dh": "short", "auth": APPLE_SUBSCRIPTION["keys"]["auth"]}}
+        self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=bad_keys)[0], 400)
+        self.assertEqual(self.request("POST", "/api/push/subscribe", user="fred", body=APPLE_SUBSCRIPTION)[0], 403)
+        self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body={**APPLE_SUBSCRIPTION, "endpoint": "https://fcm.googleapis.com/fcm/send/abc"})[0], 200)
+
+
 if __name__ == "__main__":
     unittest.main()