Refactor tests for notification error log handler (#404)

changedetectionio/__init__.py

@@ -909,7 +909,6 @@ def changedetection_app(config=None, datastore_o=None):
     @app.route("/api/delete", methods=['GET'])
     @login_required
     def api_delete():
-
         uuid = request.args.get('uuid')
         datastore.delete(uuid)
         flash('Deleted.')
@@ -1026,7 +1025,8 @@ def notification_runner():
 
                 # UUID wont be present when we submit a 'test' from the global settings
                 if 'uuid' in n_object:
-                    datastore.update_watch(uuid=n_object['uuid'], update_obj={'last_error': "Notification error detected, please see logs."})
+                    datastore.update_watch(uuid=n_object['uuid'],
+                                           update_obj={'last_notification_error': "Notification error detected, please see logs."})
 
                 log_lines = str(e).splitlines()
                 notification_debug_log += log_lines
@@ -1036,6 +1036,7 @@ def notification_runner():
 
 
 
+
 # Thread runner to check every minute, look for new watches to feed into the Queue.
 def ticker_thread_check_time_launch_checks():
     from changedetectionio import update_worker

changedetectionio/fetch_site_status.py

@@ -57,8 +57,9 @@ class perform_site_check():
         stripped_text_from_html = ""
 
         watch = self.datastore.data['watching'][uuid]
+        # Unset any existing notification error
 
-        update_obj = {}
+        update_obj = {'last_notification_error': False, 'last_error': False}
 
         extra_headers = self.datastore.get_val(uuid, 'headers')
 
@@ -141,7 +142,6 @@ class perform_site_check():
             # in the future we'll implement other mechanisms.
 
             update_obj["last_check_status"] = fetcher.get_last_status_code()
-            update_obj["last_error"] = False
 
             # If there's text to skip
             # @todo we could abstract out the get_text() to handle this cleaner

changedetectionio/templates/watch-overview.html

@@ -42,6 +42,7 @@
             <tr id="{{ watch.uuid }}"
                 class="{{ loop.cycle('pure-table-odd', 'pure-table-even') }}
                 {% if watch.last_error is defined and watch.last_error != False %}error{% endif %}
+                {% if watch.last_notification_error is defined and watch.last_notification_error != False %}error{% endif %}
                 {% if watch.paused is defined and watch.paused != False %}paused{% endif %}
                 {% if watch.newest_history_key| int > watch.last_viewed| int %}unviewed{% endif %}">
                 <td class="inline">{{ loop.index }}</td>
@@ -54,6 +55,9 @@
                     {% if watch.last_error is defined and watch.last_error != False %}
                     <div class="fetch-error">{{ watch.last_error }}</div>
                     {% endif %}
+                    {% if watch.last_notification_error is defined and watch.last_notification_error != False %}
+                    <div class="fetch-error notification-error">{{ watch.last_notification_error }}</div>
+                    {% endif %}
                     {% if not active_tag %}
                     <span class="watch-tag-list">{{ watch.tag}}</span>
                     {% endif %}

changedetectionio/tests/test_notification.py

@@ -225,33 +225,3 @@ def test_check_notification(client, live_server):
         follow_redirects=True
     )
     assert b"Notification Body and Title is required when a Notification URL is used" in res.data
-
-
-    # Check we capture the failure, we can just use trigger_check = y here
-    res = client.post(
-        url_for("edit_page", uuid="first"),
-        data={"notification_urls": "jsons://broken-url.changedetection.io/test",
-              "notification_title": "xxx",
-              "notification_body": "xxxxx",
-              "notification_format": "Text",
-              "url": test_url,
-              "tag": "my tag",
-              "title": "my title",
-              "headers": "",
-              "fetch_backend": "html_requests",
-              "trigger_check": "y"},
-        follow_redirects=True
-    )
-
-    time.sleep(3)
-
-    # The error should show in the notification logs
-    res = client.get(
-        url_for("notification_logs"))
-    assert bytes("Name or service not known".encode('utf-8')) in res.data
-
-    # And it should be listed on the watch overview
-    res = client.get(
-        url_for("index"))
-    assert bytes("Notification error detected".encode('utf-8')) in res.data
-

changedetectionio/tests/test_notification_errors.py

@@ -0,0 +1,66 @@
+import os
+import time
+import re
+from flask import url_for
+from . util import set_original_response, set_modified_response, live_server_setup
+import logging
+
+def test_check_notification_error_handling(client, live_server):
+
+    live_server_setup(live_server)
+    set_original_response()
+
+    # Give the endpoint time to spin up
+    time.sleep(3)
+
+    # use a different URL so that it doesnt interfere with the actual check until we are ready
+    test_url = url_for('test_endpoint', _external=True)
+    res = client.post(
+        url_for("api_watch_add"),
+        data={"url": "https://changedetection.io/CHANGELOG.txt", "tag": ''},
+        follow_redirects=True
+    )
+    assert b"Watch added" in res.data
+
+    time.sleep(10)
+
+    # Check we capture the failure, we can just use trigger_check = y here
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"notification_urls": "jsons://broken-url.changedetection.io/test",
+              "notification_title": "xxx",
+              "notification_body": "xxxxx",
+              "notification_format": "Text",
+              "url": test_url,
+              "tag": "",
+              "title": "",
+              "headers": "",
+              "minutes_between_check": "180",
+              "fetch_backend": "html_requests",
+              "trigger_check": "y"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+    found=False
+    for i in range(1, 10):
+        time.sleep(1)
+        logging.debug("Fetching watch overview....")
+        res = client.get(
+            url_for("index"))
+
+        if bytes("Notification error detected".encode('utf-8')) in res.data:
+            found=True
+            break
+
+
+    assert found
+
+
+    # The error should show in the notification logs
+    res = client.get(
+        url_for("notification_logs"))
+    assert bytes("Name or service not known".encode('utf-8')) in res.data
+
+
+    # And it should be listed on the watch overview