{{ render_field(form.notification_title, class="m-d notification-title", placeholder=settings_application['notification_title']) }}
diff --git a/changedetectionio/templates/base.html b/changedetectionio/templates/base.html
index 365a4c6dbfe..21b1456c9f1 100644
--- a/changedetectionio/templates/base.html
+++ b/changedetectionio/templates/base.html
@@ -35,6 +35,7 @@
+
{% if socket_io_enabled %}
diff --git a/changedetectionio/tests/unit/test_browser_notifications.py b/changedetectionio/tests/unit/test_browser_notifications.py
new file mode 100644
index 00000000000..5497365474a
--- /dev/null
+++ b/changedetectionio/tests/unit/test_browser_notifications.py
@@ -0,0 +1,436 @@
+"""
+Tests for browser notification functionality
+Tests VAPID key handling, subscription management, and notification sending
+"""
+
+import json
+import sys
+import tempfile
+import os
+import unittest
+from unittest.mock import patch, Mock, MagicMock
+from py_vapid import Vapid
+
+from changedetectionio.notification.apprise_plugin.browser_notification_helpers import (
+ convert_pem_private_key_for_pywebpush,
+ convert_pem_public_key_for_browser,
+ send_push_notifications,
+ create_notification_payload,
+ get_vapid_config_from_datastore,
+ get_browser_subscriptions,
+ save_browser_subscriptions
+)
+
+
+class TestVAPIDKeyHandling(unittest.TestCase):
+ """Test VAPID key generation, conversion, and validation"""
+
+ def test_create_notification_payload(self):
+ """Test notification payload creation"""
+ payload = create_notification_payload("Test Title", "Test Body", "/test-icon.png")
+
+ self.assertEqual(payload['title'], "Test Title")
+ self.assertEqual(payload['body'], "Test Body")
+ self.assertEqual(payload['icon'], "/test-icon.png")
+ self.assertEqual(payload['badge'], "/static/favicons/favicon-32x32.png")
+ self.assertIn('timestamp', payload)
+ self.assertIsInstance(payload['timestamp'], int)
+
+ def test_create_notification_payload_defaults(self):
+ """Test notification payload with default values"""
+ payload = create_notification_payload("Title", "Body")
+
+ self.assertEqual(payload['icon'], "/static/favicons/favicon-32x32.png")
+ self.assertEqual(payload['badge'], "/static/favicons/favicon-32x32.png")
+
+ def test_convert_pem_private_key_for_pywebpush_with_valid_pem(self):
+ """Test conversion of valid PEM private key to Vapid instance"""
+ # Generate a real VAPID key
+ vapid = Vapid()
+ vapid.generate_keys()
+ private_pem = vapid.private_pem().decode()
+
+ # Convert using our function
+ converted_key = convert_pem_private_key_for_pywebpush(private_pem)
+
+ # Should return a Vapid instance
+ self.assertIsInstance(converted_key, Vapid)
+
+ def test_convert_pem_private_key_invalid_input(self):
+ """Test conversion with invalid input returns original"""
+ invalid_key = "not-a-pem-key"
+ result = convert_pem_private_key_for_pywebpush(invalid_key)
+ self.assertEqual(result, invalid_key)
+
+ none_key = None
+ result = convert_pem_private_key_for_pywebpush(none_key)
+ self.assertEqual(result, none_key)
+
+ def test_convert_pem_public_key_for_browser(self):
+ """Test conversion of PEM public key to browser format"""
+ # Generate a real VAPID key pair
+ vapid = Vapid()
+ vapid.generate_keys()
+ public_pem = vapid.public_pem().decode()
+
+ # Convert to browser format
+ browser_key = convert_pem_public_key_for_browser(public_pem)
+
+ # Should return URL-safe base64 string
+ self.assertIsInstance(browser_key, str)
+ self.assertGreater(len(browser_key), 0)
+ # Should not contain padding
+ self.assertFalse(browser_key.endswith('='))
+
+ def test_convert_pem_public_key_invalid(self):
+ """Test public key conversion with invalid input"""
+ result = convert_pem_public_key_for_browser("invalid-pem")
+ self.assertIsNone(result)
+
+
+class TestDatastoreIntegration(unittest.TestCase):
+ """Test datastore operations for VAPID and subscriptions"""
+
+ def test_get_vapid_config_from_datastore(self):
+ """Test retrieving VAPID config from datastore"""
+ mock_datastore = Mock()
+ mock_datastore.data = {
+ 'settings': {
+ 'application': {
+ 'vapid': {
+ 'private_key': 'test-private-key',
+ 'public_key': 'test-public-key',
+ 'contact_email': 'test@example.com'
+ }
+ }
+ }
+ }
+
+ private_key, public_key, contact_email = get_vapid_config_from_datastore(mock_datastore)
+
+ self.assertEqual(private_key, 'test-private-key')
+ self.assertEqual(public_key, 'test-public-key')
+ self.assertEqual(contact_email, 'test@example.com')
+
+ def test_get_vapid_config_missing_email(self):
+ """Test VAPID config with missing contact email uses default"""
+ mock_datastore = Mock()
+ mock_datastore.data = {
+ 'settings': {
+ 'application': {
+ 'vapid': {
+ 'private_key': 'test-private-key',
+ 'public_key': 'test-public-key'
+ }
+ }
+ }
+ }
+
+ private_key, public_key, contact_email = get_vapid_config_from_datastore(mock_datastore)
+
+ self.assertEqual(contact_email, 'citizen@example.com')
+
+ def test_get_vapid_config_empty_datastore(self):
+ """Test VAPID config with empty datastore returns None values"""
+ mock_datastore = Mock()
+ mock_datastore.data = {}
+
+ private_key, public_key, contact_email = get_vapid_config_from_datastore(mock_datastore)
+
+ self.assertIsNone(private_key)
+ self.assertIsNone(public_key)
+ self.assertEqual(contact_email, 'citizen@example.com')
+
+ def test_get_browser_subscriptions(self):
+ """Test retrieving browser subscriptions from datastore"""
+ mock_datastore = Mock()
+ test_subscriptions = [
+ {
+ 'endpoint': 'https://fcm.googleapis.com/fcm/send/test1',
+ 'keys': {'p256dh': 'key1', 'auth': 'auth1'}
+ },
+ {
+ 'endpoint': 'https://fcm.googleapis.com/fcm/send/test2',
+ 'keys': {'p256dh': 'key2', 'auth': 'auth2'}
+ }
+ ]
+ mock_datastore.data = {
+ 'settings': {
+ 'application': {
+ 'browser_subscriptions': test_subscriptions
+ }
+ }
+ }
+
+ subscriptions = get_browser_subscriptions(mock_datastore)
+
+ self.assertEqual(len(subscriptions), 2)
+ self.assertEqual(subscriptions, test_subscriptions)
+
+ def test_get_browser_subscriptions_empty(self):
+ """Test getting subscriptions from empty datastore returns empty list"""
+ mock_datastore = Mock()
+ mock_datastore.data = {}
+
+ subscriptions = get_browser_subscriptions(mock_datastore)
+
+ self.assertEqual(subscriptions, [])
+
+ def test_save_browser_subscriptions(self):
+ """Test saving browser subscriptions to datastore"""
+ mock_datastore = Mock()
+ mock_datastore.data = {'settings': {'application': {}}}
+
+ test_subscriptions = [
+ {'endpoint': 'test1', 'keys': {'p256dh': 'key1', 'auth': 'auth1'}}
+ ]
+
+ save_browser_subscriptions(mock_datastore, test_subscriptions)
+
+ self.assertEqual(mock_datastore.data['settings']['application']['browser_subscriptions'], test_subscriptions)
+ self.assertTrue(mock_datastore.needs_write)
+
+
+class TestNotificationSending(unittest.TestCase):
+ """Test notification sending with mocked pywebpush"""
+
+ @patch('pywebpush.webpush')
+ def test_send_push_notifications_success(self, mock_webpush):
+ """Test successful notification sending"""
+ mock_webpush.return_value = True
+
+ mock_datastore = Mock()
+ mock_datastore.needs_write = False
+
+ subscriptions = [
+ {
+ 'endpoint': 'https://fcm.googleapis.com/fcm/send/test1',
+ 'keys': {'p256dh': 'key1', 'auth': 'auth1'}
+ }
+ ]
+
+ # Generate a real VAPID key for testing
+ vapid = Vapid()
+ vapid.generate_keys()
+ private_key = vapid.private_pem().decode()
+
+ notification_payload = {
+ 'title': 'Test Title',
+ 'body': 'Test Body'
+ }
+
+ success_count, total_count = send_push_notifications(
+ subscriptions=subscriptions,
+ notification_payload=notification_payload,
+ private_key=private_key,
+ contact_email='test@example.com',
+ datastore=mock_datastore
+ )
+
+ self.assertEqual(success_count, 1)
+ self.assertEqual(total_count, 1)
+ self.assertTrue(mock_webpush.called)
+
+ # Verify webpush was called with correct parameters
+ call_args = mock_webpush.call_args
+ self.assertEqual(call_args[1]['subscription_info'], subscriptions[0])
+ self.assertEqual(json.loads(call_args[1]['data']), notification_payload)
+ self.assertIn('vapid_private_key', call_args[1])
+ self.assertEqual(call_args[1]['vapid_claims']['sub'], 'mailto:test@example.com')
+
+ @patch('pywebpush.webpush')
+ def test_send_push_notifications_webpush_exception(self, mock_webpush):
+ """Test handling of WebPushException with invalid subscription removal"""
+ from pywebpush import WebPushException
+
+ # Mock a 410 response (subscription gone)
+ mock_response = Mock()
+ mock_response.status_code = 410
+
+ mock_webpush.side_effect = WebPushException("Subscription expired", response=mock_response)
+
+ mock_datastore = Mock()
+ mock_datastore.needs_write = False
+
+ subscriptions = [
+ {
+ 'endpoint': 'https://fcm.googleapis.com/fcm/send/test1',
+ 'keys': {'p256dh': 'key1', 'auth': 'auth1'}
+ }
+ ]
+
+ vapid = Vapid()
+ vapid.generate_keys()
+ private_key = vapid.private_pem().decode()
+
+ success_count, total_count = send_push_notifications(
+ subscriptions=subscriptions,
+ notification_payload={'title': 'Test', 'body': 'Test'},
+ private_key=private_key,
+ contact_email='test@example.com',
+ datastore=mock_datastore
+ )
+
+ self.assertEqual(success_count, 0)
+ self.assertEqual(total_count, 1)
+ self.assertTrue(mock_datastore.needs_write) # Should mark for subscription cleanup
+
+ def test_send_push_notifications_no_pywebpush(self):
+ """Test graceful handling when pywebpush is not available"""
+ with patch.dict('sys.modules', {'pywebpush': None}):
+ subscriptions = [{'endpoint': 'test', 'keys': {}}]
+
+ success_count, total_count = send_push_notifications(
+ subscriptions=subscriptions,
+ notification_payload={'title': 'Test', 'body': 'Test'},
+ private_key='test-key',
+ contact_email='test@example.com',
+ datastore=Mock()
+ )
+
+ self.assertEqual(success_count, 0)
+ self.assertEqual(total_count, 1)
+
+
+class TestBrowserIntegration(unittest.TestCase):
+ """Test browser integration aspects (file existence)"""
+
+ def test_javascript_browser_notifications_class_exists(self):
+ """Test that browser notifications JavaScript file exists and has expected structure"""
+ js_file = "/var/www/changedetection.io/changedetectionio/static/js/browser-notifications.js"
+
+ self.assertTrue(os.path.exists(js_file))
+
+ with open(js_file, 'r') as f:
+ content = f.read()
+
+ # Check for key class and methods
+ self.assertIn('class BrowserNotifications', content)
+ self.assertIn('async init()', content)
+ self.assertIn('async subscribe()', content)
+ self.assertIn('async sendTestNotification()', content)
+ self.assertIn('setupNotificationUrlMonitoring()', content)
+
+ def test_service_worker_exists(self):
+ """Test that service worker file exists"""
+ sw_file = "/var/www/changedetection.io/changedetectionio/static/js/service-worker.js"
+
+ self.assertTrue(os.path.exists(sw_file))
+
+ with open(sw_file, 'r') as f:
+ content = f.read()
+
+ # Check for key service worker functionality
+ self.assertIn('push', content)
+ self.assertIn('notificationclick', content)
+
+
+class TestAPIEndpoints(unittest.TestCase):
+ """Test browser notification API endpoints"""
+
+ def test_browser_notifications_module_exists(self):
+ """Test that BrowserNotifications API module exists"""
+ api_file = "/var/www/changedetection.io/changedetectionio/notification/BrowserNotifications.py"
+
+ self.assertTrue(os.path.exists(api_file))
+
+ with open(api_file, 'r') as f:
+ content = f.read()
+
+ # Check for key API classes
+ self.assertIn('BrowserNotificationsVapidPublicKey', content)
+ self.assertIn('BrowserNotificationsSubscribe', content)
+ self.assertIn('BrowserNotificationsUnsubscribe', content)
+
+ def test_vapid_public_key_conversion(self):
+ """Test VAPID public key conversion for browser use"""
+ # Generate a real key pair
+ vapid = Vapid()
+ vapid.generate_keys()
+ public_pem = vapid.public_pem().decode()
+
+ # Convert to browser format
+ browser_key = convert_pem_public_key_for_browser(public_pem)
+
+ # Verify it's a valid URL-safe base64 string
+ self.assertIsInstance(browser_key, str)
+ self.assertGreater(len(browser_key), 80) # P-256 uncompressed point should be ~88 chars
+
+ # Should not have padding
+ self.assertFalse(browser_key.endswith('='))
+
+ # Should only contain URL-safe base64 characters
+ import re
+ self.assertRegex(browser_key, r'^[A-Za-z0-9_-]+$')
+
+
+class TestIntegrationFlow(unittest.TestCase):
+ """Test complete integration flow"""
+
+ @patch('pywebpush.webpush')
+ def test_complete_notification_flow(self, mock_webpush):
+ """Test complete flow from subscription to notification"""
+ mock_webpush.return_value = True
+
+ # Create mock datastore with VAPID keys
+ mock_datastore = Mock()
+ vapid = Vapid()
+ vapid.generate_keys()
+
+ mock_datastore.data = {
+ 'settings': {
+ 'application': {
+ 'vapid': {
+ 'private_key': vapid.private_pem().decode(),
+ 'public_key': vapid.public_pem().decode(),
+ 'contact_email': 'test@example.com'
+ },
+ 'browser_subscriptions': [
+ {
+ 'endpoint': 'https://fcm.googleapis.com/fcm/send/test123',
+ 'keys': {
+ 'p256dh': 'test-p256dh-key',
+ 'auth': 'test-auth-key'
+ }
+ }
+ ]
+ }
+ }
+ }
+ mock_datastore.needs_write = False
+
+ # Get configuration
+ private_key, public_key, contact_email = get_vapid_config_from_datastore(mock_datastore)
+ subscriptions = get_browser_subscriptions(mock_datastore)
+
+ # Create notification
+ payload = create_notification_payload("Test Title", "Test Message")
+
+ # Send notification
+ success_count, total_count = send_push_notifications(
+ subscriptions=subscriptions,
+ notification_payload=payload,
+ private_key=private_key,
+ contact_email=contact_email,
+ datastore=mock_datastore
+ )
+
+ # Verify success
+ self.assertEqual(success_count, 1)
+ self.assertEqual(total_count, 1)
+ self.assertTrue(mock_webpush.called)
+
+ # Verify webpush call parameters
+ call_args = mock_webpush.call_args
+ self.assertIn('subscription_info', call_args[1])
+ self.assertIn('vapid_private_key', call_args[1])
+ self.assertIn('vapid_claims', call_args[1])
+
+ # Verify vapid_claims format
+ vapid_claims = call_args[1]['vapid_claims']
+ self.assertEqual(vapid_claims['sub'], 'mailto:test@example.com')
+ self.assertEqual(vapid_claims['aud'], 'https://fcm.googleapis.com')
+
+
+if __name__ == '__main__':
+ unittest.main()
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index cb220097f0a..aeaccb1313a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -142,3 +142,6 @@ pre_commit >= 4.2.0
# For events between checking and socketio updates
blinker
+
+# For Web Push notifications (browser notifications)
+pywebpush