From c17eec085fce24894fece5cbe23237cbbb5e6b49 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Thu, 28 Aug 2025 23:18:35 +0200 Subject: [PATCH] juggling tests --- .../static/js/browser-notifications.js | 3 + .../tests/unit/test_browser_notifications.py | 436 ++++++++++++++++++ 2 files changed, 439 insertions(+) create mode 100644 changedetectionio/tests/unit/test_browser_notifications.py diff --git a/changedetectionio/static/js/browser-notifications.js b/changedetectionio/static/js/browser-notifications.js index ee4deb806..02f9fd780 100644 --- a/changedetectionio/static/js/browser-notifications.js +++ b/changedetectionio/static/js/browser-notifications.js @@ -24,6 +24,9 @@ class BrowserNotifications { // Register service worker await this.registerServiceWorker(); + // Check existing subscription state + await this.checkExistingSubscription(); + // Initialize UI elements this.initializeUI(); diff --git a/changedetectionio/tests/unit/test_browser_notifications.py b/changedetectionio/tests/unit/test_browser_notifications.py new file mode 100644 index 000000000..549736547 --- /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