Adding automated test for sending single test notification from a watch

changedetectionio/blueprint/ui/notification.py

@@ -2,7 +2,7 @@ from flask import Blueprint, request, make_response
 import random
 from loguru import logger
 
-from changedetectionio.notification_service import NotificationContextData
+from changedetectionio.notification_service import NotificationContextData, set_basic_notification_vars
 from changedetectionio.store import ChangeDetectionStore
 from changedetectionio.auth_decorator import login_optionally_required
 
@@ -95,7 +95,44 @@ def construct_blueprint(datastore: ChangeDetectionStore):
                 n_object['notification_body'] = "Test body"
 
             n_object['as_async'] = False
-            n_object.update(watch.extra_notification_token_values())
+
+            #  Same like in notification service, should be refactored
+            dates = []
+            trigger_text = ''
+            snapshot_contents = ''
+            if watch:
+                watch_history = watch.history
+                dates = list(watch_history.keys())
+                trigger_text = watch.get('trigger_text', [])
+                # Add text that was triggered
+                if len(dates):
+                    snapshot_contents = watch.get_history_snapshot(dates[-1])
+                else:
+                    snapshot_contents = "No snapshot/history available, the watch should fetch atleast once."
+
+                if len(trigger_text):
+                    from . import html_tools
+                    triggered_text = html_tools.get_triggered_text(content=snapshot_contents, trigger_text=trigger_text)
+                    if triggered_text:
+                        triggered_text = '\n'.join(triggered_text)
+
+            # Could be called as a 'test notification' with only 1 snapshot available
+            prev_snapshot = "Example text: example test\nExample text: change detection is cool\nExample text: some more examples\n"
+            current_snapshot = "Example text: example test\nExample text: change detection is fantastic\nExample text: even more examples\nExample text: a lot more examples"
+
+
+
+            if len(dates) > 1:
+                prev_snapshot = watch.get_history_snapshot(dates[-2])
+                current_snapshot = watch.get_history_snapshot(dates[-1])
+
+            n_object.update(set_basic_notification_vars(snapshot_contents=snapshot_contents,
+                                                        current_snapshot=current_snapshot,
+                                                        prev_snapshot=prev_snapshot,
+                                                        watch=watch,
+                                                        triggered_text=trigger_text))
+
+
             sent_obj = process_notification(n_object, datastore)
 
         except Exception as e:

changedetectionio/notification_service.py

@@ -9,11 +9,8 @@ for both sync and async workers
 from loguru import logger
 import time
 
-from changedetectionio.model import USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH
 from changedetectionio.notification import default_notification_format, valid_notification_formats
 
-# This gets modified on notification time (handler.py) depending on the required notification output
-CUSTOM_LINEBREAK_PLACEHOLDER='@BR@'
 
 
 # What is passed around as notification context, also used as the complete list of valid {{ tokens }}
@@ -71,6 +68,34 @@ class NotificationContextData(dict):
 
         super().__setitem__(key, value)
 
+
+def set_basic_notification_vars(snapshot_contents, current_snapshot, prev_snapshot, watch, triggered_text):
+    now = time.time()
+    from changedetectionio import diff
+
+    n_object = {
+        'current_snapshot': snapshot_contents,
+        'diff': diff.render_diff(prev_snapshot, current_snapshot),
+        'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False),
+        'diff_full': diff.render_diff(prev_snapshot, current_snapshot, include_equal=True),
+        'diff_patch': diff.render_diff(prev_snapshot, current_snapshot, patch_format=True),
+        'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False),
+        'screenshot': watch.get_screenshot() if watch and watch.get('notification_screenshot') else None,
+        'triggered_text': triggered_text,
+        'uuid': watch.get('uuid') if watch else None,
+        'watch_url': watch.get('url') if watch else None,
+        'watch_uuid': watch.get('uuid') if watch else None,
+        'watch_mime_type': watch.get('content-type')
+    }
+
+    # The \n's in the content from the above will get converted to <br> etc depending on the notification format
+
+    if watch:
+        n_object.update(watch.extra_notification_token_values())
+
+    logger.trace(f"Main rendered notification placeholders (diff_added etc) calculated in {time.time() - now:.3f}s")
+    return n_object
+
 class NotificationService:
     """
     Standalone notification service that handles all notification functionality
@@ -85,7 +110,6 @@ class NotificationService:
         """
         Queue a notification for a watch with full diff rendering and template variables
         """
-        from changedetectionio import diff
         from changedetectionio.notification import USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH
 
         if not isinstance(n_object, NotificationContextData):
@@ -94,8 +118,6 @@ class NotificationService:
         dates = []
         trigger_text = ''
 
-        now = time.time()
-
         if watch:
             watch_history = watch.history
             dates = list(watch_history.keys())
@@ -117,7 +139,7 @@ class NotificationService:
             from . import html_tools
             triggered_text = html_tools.get_triggered_text(content=snapshot_contents, trigger_text=trigger_text)
             if triggered_text:
-                triggered_text = CUSTOM_LINEBREAK_PLACEHOLDER.join(triggered_text)
+                triggered_text = '\n'.join(triggered_text)
 
         # Could be called as a 'test notification' with only 1 snapshot available
         prev_snapshot = "Example text: example test\nExample text: change detection is cool\nExample text: some more examples\n"
@@ -127,26 +149,13 @@ class NotificationService:
             prev_snapshot = watch.get_history_snapshot(dates[-2])
             current_snapshot = watch.get_history_snapshot(dates[-1])
 
-        n_object.update({
-            'current_snapshot': snapshot_contents,
-            'diff': diff.render_diff(prev_snapshot, current_snapshot),
-            'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False),
-            'diff_full': diff.render_diff(prev_snapshot, current_snapshot, include_equal=True),
-            'diff_patch': diff.render_diff(prev_snapshot, current_snapshot, patch_format=True),
-            'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False),
-            'screenshot': watch.get_screenshot() if watch and watch.get('notification_screenshot') else None,
-            'triggered_text': triggered_text,
-            'uuid': watch.get('uuid') if watch else None,
-            'watch_url': watch.get('url') if watch else None,
-            'watch_uuid': watch.get('uuid') if watch else None,
-            'watch_mime_type': watch.get('content-type')
-        })
-        # The \n's in the content from the above will get converted to <br> etc depending on the notification format
 
-        if watch:
-            n_object.update(watch.extra_notification_token_values())
+        n_object.update(set_basic_notification_vars(snapshot_contents=snapshot_contents,
+                                                    current_snapshot=current_snapshot,
+                                                    prev_snapshot=prev_snapshot,
+                                                    watch=watch,
+                                                    triggered_text=triggered_text))
 
-        logger.trace(f"Main rendered notification placeholders (diff_added etc) calculated in {time.time()-now:.3f}s")
         logger.debug("Queued notification for sending")
         self.notification_q.put(n_object)
 

changedetectionio/static/styles/scss/styles.scss

@@ -329,12 +329,18 @@ a.pure-button-selected {
 .notifications-wrapper {
   padding-top: 0.5rem;
   #notification-test-log {
-    padding-top: 1rem;
+    margin-top: 1rem;
+    padding: 1rem;
     white-space: pre-wrap;
     word-break: break-word;
     overflow-wrap: break-word;
     max-width: 100%;
     box-sizing: border-box;
+    max-height: 12rem;
+    overflow-y: scroll;
+    border: 1px solid var(--color-border-notification);
+    border-radius: 5px;
+
   }
 }
 

changedetectionio/tests/test_notification.py

@@ -404,15 +404,15 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
 
 
 #2510
+#@todo run it again as text, html, htmlcolor
 def test_global_send_test_notification(client, live_server, measure_memory_usage, datastore_path):
 
-    
     set_original_response(datastore_path=datastore_path)
     if os.path.isfile(os.path.join(datastore_path, "notification.txt")):
         os.unlink(os.path.join(datastore_path, "notification.txt")) \
 
     # 1995 UTF-8 content should be encoded
-    test_body = 'change detection is cool 网站监测 内容更新了'
+    test_body = 'change detection is cool 网站监测 内容更新了 - {{diff_full}}'
 
     # otherwise other settings would have already existed from previous tests in this file
     res = client.post(
@@ -452,7 +452,14 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
 
     with open(os.path.join(datastore_path, "notification.txt"), 'r') as f:
         x = f.read()
-        assert test_body in x
+        assert 'change detection is cool 网站监测 内容更新了' in x
+        if 'html' in default_notification_format:
+            # this should come from default text when in global/system mode here changedetectionio/notification_service.py
+            assert 'title="Changed into">Example text:' in x
+        else:
+            assert 'title="Changed into">Example text:' not in x
+            assert 'span' not in x
+            assert 'Example text:' in x
 
     os.unlink(os.path.join(datastore_path, "notification.txt"))
 
@@ -509,6 +516,47 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
     assert b"Error: You must have atleast one watch configured for 'test notification' to work" in res.data
 
 
+#2510
+def test_single_send_test_notification_on_watch(client, live_server, measure_memory_usage, datastore_path):
+
+    set_original_response(datastore_path=datastore_path)
+    if os.path.isfile(os.path.join(datastore_path, "notification.txt")):
+        os.unlink(os.path.join(datastore_path, "notification.txt")) \
+
+
+    test_url = url_for('test_endpoint', _external=True)
+    uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
+    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
+
+    test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}&+custom-header=123"
+    # 1995 UTF-8 content should be encoded
+    test_body = 'change detection is cool 网站监测 内容更新了 - {{diff_full}}'
+    ######### Test global/system settings
+    res = client.post(
+        url_for("ui.ui_notification.ajax_callback_send_notification_test")+f"/{uuid}",
+        data={"notification_urls": test_notification_url,
+              "notification_body": test_body,
+              "notification_format": default_notification_format,
+              "notification_title": "New ChangeDetection.io Notification - {{ watch_url }}",
+              },
+        follow_redirects=True
+    )
+
+    assert res.status_code != 400
+    assert res.status_code != 500
+
+    with open(os.path.join(datastore_path, "notification.txt"), 'r') as f:
+        x = f.read()
+        assert 'change detection is cool 网站监测 内容更新了' in x
+        if 'html' in default_notification_format:
+            # this should come from default text when in global/system mode here changedetectionio/notification_service.py
+            assert 'title="Changed into">Example text:' in x
+        else:
+            assert 'title="Changed into">Example text:' not in x
+            assert 'span' not in x
+            assert 'Example text:' in x
+
+    os.unlink(os.path.join(datastore_path, "notification.txt"))
 
 def _test_color_notifications(client, notification_body_token, datastore_path):