Improving alignment with apprise

changedetectionio/notification/handler.py

@@ -1,18 +1,19 @@
 
 import time
 import apprise
+from apprise import NotifyFormat
 from loguru import logger
 from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
 from ..notification_service import NotificationContextData
 
 
-def markup_notification_message(body):
+def markup_text_links_to_html(body):
     """
     Convert plaintext to HTML with clickable links.
     Uses Jinja2's escape and Markup for XSS safety.
     """
     from linkify_it import LinkifyIt
-    from jinja2 import Markup, escape
+    from markupsafe import Markup, escape
 
     linkify = LinkifyIt()
 
@@ -44,7 +45,30 @@ def markup_notification_message(body):
         result.append(escape(body[last_index:]))
 
     # Join all parts
-    return Markup(''.join(str(part) for part in result))
+    return str(Markup(''.join(str(part) for part in result)))
+
+def notification_format_align_with_apprise(n_format : str):
+    """
+    Correctly align changedetection's formats with apprise's formats
+    Probably these are the same - but good to be sure.
+    :param n_format:
+    :return:
+    """
+
+    if n_format.lower().startswith('html'):
+        # Apprise only knows 'html' not 'htmlcolor' etc, which shouldnt matter here
+        n_format = NotifyFormat.HTML
+    elif n_format.lower().startswith('markdown'):
+        # probably the same but just to be safe
+        n_format = NotifyFormat.MARKDOWN
+    elif n_format.lower().startswith('text'):
+        # probably the same but just to be safe
+        n_format = NotifyFormat.TEXT
+    else:
+        n_format = NotifyFormat.TEXT
+
+    # Must be str for apprise notify body_format
+    return str(n_format)
 
 def process_notification(n_object: NotificationContextData, datastore):
     from changedetectionio.jinja2_custom import render as jinja_render
@@ -70,7 +94,9 @@ def process_notification(n_object: NotificationContextData, datastore):
     # If we arrived with 'System default' then look it up
     if n_format == default_notification_format_for_watch and datastore.data['settings']['application'].get('notification_format') != default_notification_format_for_watch:
         # Initially text or whatever
-        n_format = datastore.data['settings']['application'].get('notification_format', valid_notification_formats[default_notification_format])
+        n_format = datastore.data['settings']['application'].get('notification_format', valid_notification_formats[default_notification_format]).lower()
+
+    n_format = notification_format_align_with_apprise(n_format=n_format)
 
     logger.trace(f"Complete notification body including Jinja and placeholders calculated in  {time.time() - now:.2f}s")
 
@@ -95,9 +121,9 @@ def process_notification(n_object: NotificationContextData, datastore):
             n_body = jinja_render(template_str=n_object.get('notification_body', ''), **notification_parameters)
 
             if n_object.get('markup_text_to_html'):
-                n_body = markup_notification_message(body=n_body)
+                n_body = markup_text_links_to_html(body=n_body)
 
-            if n_object.get('notification_format', '').startswith('HTML'):
+            if n_format == NotifyFormat.HTML:
                 n_body = n_body.replace("\n", '<br>')
 
             n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)

changedetectionio/tests/test_filter_failure_notification.py

@@ -1,10 +1,8 @@
 import os
 import time
-from loguru import logger
 from flask import url_for
-from .util import set_original_response, live_server_setup, extract_UUID_from_client, wait_for_all_checks, \
+from .util import set_original_response,  wait_for_all_checks, wait_for_notification_endpoint_output
-    wait_for_notification_endpoint_output
+from ..notification import valid_notification_formats
-from changedetectionio.model import App
 
 
 def set_response_with_filter():
@@ -23,13 +21,14 @@ def set_response_with_filter():
         f.write(test_return_data)
     return None
 
-def run_filter_test(client, live_server, content_filter):
+def run_filter_test(client, live_server, content_filter, app_notification_format):
 
     # Response WITHOUT the filter ID element
     set_original_response()
+    live_server.app.config['DATASTORE'].data['settings']['application']['notification_format'] = app_notification_format
 
     # Goto the edit page, add our ignore text
-    notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'json')
+    notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'post')
 
     # Add our URL to the import page
     test_url = url_for('test_endpoint', _external=True)
@@ -127,8 +126,17 @@ def run_filter_test(client, live_server, content_filter):
     with open("test-datastore/notification.txt", 'r') as f:
         notification = f.read()
 
-    assert 'CSS/xPath filter was not present in the page' in notification
+    assert 'Your configured CSS/xPath filters' in notification
-    assert content_filter.replace('"', '\\"') in notification
+
+
+    # Text (or HTML conversion) markup to make the notifications a little nicer should have worked
+    if app_notification_format.startswith('html'):
+        assert 'a href' in notification
+        arrived_filter = content_filter.replace('"', '\\"')
+        assert arrived_filter in notification
+    else:
+        assert 'a href' not in notification
+        assert content_filter in notification
 
     # Remove it and prove that it doesn't trigger when not expected
     # It should register a change, but no 'filter not found'
@@ -159,14 +167,20 @@ def run_filter_test(client, live_server, content_filter):
     os.unlink("test-datastore/notification.txt")
 
 
-
-
 def test_check_include_filters_failure_notification(client, live_server, measure_memory_usage):
-#   #  live_server_setup(live_server) # Setup on conftest per function
+    #   #  live_server_setup(live_server) # Setup on conftest per function
-    run_filter_test(client, live_server,'#nope-doesnt-exist')
+    run_filter_test(client=client, live_server=live_server, content_filter='#nope-doesnt-exist', app_notification_format=valid_notification_formats.get('HTML Color'))
+    # Check markup send conversion didnt affect plaintext preference
+    run_filter_test(client=client, live_server=live_server, content_filter='#nope-doesnt-exist', app_notification_format=valid_notification_formats.get('Text'))
 
 def test_check_xpath_filter_failure_notification(client, live_server, measure_memory_usage):
-#   #  live_server_setup(live_server) # Setup on conftest per function
+    #   #  live_server_setup(live_server) # Setup on conftest per function
-    run_filter_test(client, live_server, '//*[@id="nope-doesnt-exist"]')
+    run_filter_test(client=client, live_server=live_server, content_filter='//*[@id="nope-doesnt-exist"]', app_notification_format=valid_notification_formats.get('HTML Color'))
 
 # Test that notification is never sent
+
+def test_basic_markup_from_text(client, live_server, measure_memory_usage):
+    # Test the notification error templates convert to HTML if needed (link activate)
+    from ..notification.handler import markup_text_links_to_html
+    x = markup_text_links_to_html("hello https://google.com")
+    assert 'a href' in x