extended tests

Co-authored-by: dgtlmoon <leigh@morresi.net>

changedetectionio/fetch_site_status.py

@@ -2,14 +2,14 @@ import hashlib
 import logging
 import os
 import re
-import time
 import urllib3
+import difflib
+
 
 from changedetectionio import content_fetcher, html_tools
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
-
 # Some common stuff here that can be moved to a base class
 # (set_proxy_from_list)
 class perform_site_check():
@@ -289,8 +289,23 @@ class perform_site_check():
                 else:
                     logging.debug("check_unique_lines: UUID {} had unique content".format(uuid))
 
-        # Always record the new checksum
+        if changed_detected:
+            if not watch.get("trigger_add", True) or not watch.get("trigger_del", True): # if we are supposed to filter any diff types
+                # get the diff types present in the watch
+                diff_types = watch.get_diff_types(text_content_before_ignored_filter)
+                print("Diff components found: " + str(diff_types))
+
+                # Only Additions (deletions are turned off)
+                if not watch["trigger_del"] and diff_types["del"] and not diff_types["add"]:
+                    changed_detected = False
+
+                # Only Deletions (additions are turned off)
+                elif not watch["trigger_add"] and  diff_types["add"] and not diff_types["del"]:
+                    changed_detected = False
+
+        # Always record the new checksum and the new text
         update_obj["previous_md5"] = fetched_md5
+        watch.save_previous_text(text_content_before_ignored_filter)
 
         # On the first run of a site, watch['previous_md5'] will be None, set it the current one.
         if not watch.get('previous_md5'):

changedetectionio/forms.py

@@ -323,6 +323,18 @@ class ValidateCSSJSONXPATHInput(object):
                 except:
                     raise ValidationError("A system-error occurred when validating your jq expression")
 
+class ValidateDiffFilters(object):
+    """
+    Validates that at least one filter checkbox is selected
+    """
+    def __init__(self, message=None):
+        self.message = message
+
+    def __call__(self, form, field):
+        if not form.trigger_add.data and not form.trigger_del.data:
+            message = field.gettext('At least one filter checkbox must be selected')
+            raise ValidationError(message)
+
 
 class quickWatchForm(Form):
     url = fields.URLField('URL', validators=[validateURL()])
@@ -365,6 +377,8 @@ class watchForm(commonSettingsForm):
     check_unique_lines = BooleanField('Only trigger when new lines appear', default=False)
     trigger_text = StringListField('Trigger/wait for text', [validators.Optional(), ValidateListRegex()])
     text_should_not_be_present = StringListField('Block change-detection if text matches', [validators.Optional(), ValidateListRegex()])
+    trigger_add = BooleanField('Additions', [ValidateDiffFilters()], default=True)
+    trigger_del = BooleanField('Deletions', [ValidateDiffFilters()], default=True)
 
     webdriver_js_execute_code = TextAreaField('Execute JavaScript before change detection', render_kw={"rows": "5"}, validators=[validators.Optional()])
 

changedetectionio/model/Watch.py

@@ -47,6 +47,8 @@ class model(dict):
             'consecutive_filter_failures': 0, # Every time the CSS/xPath filter cannot be located, reset when all is fine.
             'extract_title_as_title': False,
             'check_unique_lines': False, # On change-detected, compare against all history if its something new
+            'trigger_add': True,
+            'trigger_del': True,
             'proxy': None, # Preferred proxy connection
             # Re #110, so then if this is set to None, we know to use the default value instead
             # Requires setting to None on submit if it's the same as the default
@@ -206,6 +208,35 @@ class model(dict):
         # @todo bump static cache of the last timestamp so we dont need to examine the file to set a proper ''viewed'' status
         return snapshot_fname
 
+    # Save previous text snapshot for diffing - used for calculating additions and deletions
+    def save_previous_text(self, contents):
+        import logging
+
+        output_path = os.path.join(self.__datastore_path, self['uuid'])
+
+        # Incase the operator deleted it, check and create.
+        self.ensure_data_dir_exists()
+
+        snapshot_fname = os.path.join(self.watch_data_dir, "previous.txt")
+        logging.debug("Saving previous text {}".format(snapshot_fname))
+
+        with open(snapshot_fname, 'wb') as f:
+            f.write(contents)
+
+        return snapshot_fname
+
+    # Get previous text snapshot for diffing - used for calculating additions and deletions
+    def get_previous_text(self):
+
+        snapshot_fname = os.path.join(self.watch_data_dir, "previous.txt")
+        if self.history_n < 1:
+            return ""
+
+        with open(snapshot_fname, 'rb') as f:
+            contents = f.read()
+
+        return contents
+
     @property
     def has_empty_checktime(self):
         # using all() + dictionary comprehension
@@ -235,6 +266,31 @@ class model(dict):
         # if not, something new happened
         return not local_lines.issubset(existing_history)
 
+    # Get diff types (addition, deletion, modification) from the previous snapshot and new_text
+    # uses similar algorithm to customSequenceMatcher in diff.py
+    # Returns a dict of diff types and wether they are present in the diff
+    def get_diff_types(self, new_text):
+        import difflib
+
+        diff_types = {
+            'add': False,
+            'del': False,
+        }
+
+        # get diff types using difflib
+        cruncher = difflib.SequenceMatcher(isjunk=lambda x: x in " \\t", a=str(self.get_previous_text()), b=str(new_text))
+
+        for tag, alo, ahi, blo, bhi in cruncher.get_opcodes():
+            if tag == 'delete':
+                diff_types["del"] = True
+            elif tag == 'insert':
+                diff_types["add"] = True
+            elif tag == 'replace':
+                diff_types["del"] = True
+                diff_types["add"] = True
+
+        return diff_types
+
     def get_screenshot(self):
         fname = os.path.join(self.watch_data_dir, "last-screenshot.png")
         if os.path.isfile(fname):

changedetectionio/store.py

@@ -548,6 +548,10 @@ class ChangeDetectionStore:
     # `last_changed` not needed, we pull that information from the history.txt index
     def update_4(self):
         for uuid, watch in self.data['watching'].items():
+            # Be sure it's recalculated
+            p = watch.history
+            if watch.history_n < 2:
+                watch['last_changed'] = 0
             try:
                 # Remove it from the struct
                 del(watch['last_changed'])
@@ -583,3 +587,23 @@ class ChangeDetectionStore:
         for v in ['User-Agent', 'Accept', 'Accept-Encoding', 'Accept-Language']:
             if self.data['settings']['headers'].get(v):
                 del self.data['settings']['headers'][v]
+                
+    # Generate a previous.txt for all watches that do not have one and contain history
+    def update_8(self):
+        for uuid, watch in self.data['watching'].items():
+            # Make sure we actually have history
+            if (watch.history_n == 0):
+                continue
+            latest_file_name = watch.history[watch.newest_history_key]
+
+
+            # Check if the previous.txt exists
+            if not os.path.exists(os.path.join(watch.watch_data_dir, "previous.txt")):
+                # Generate a previous.txt
+                with open(os.path.join(watch.watch_data_dir, "previous.txt"), "wb") as f:
+                    # Fill it with the latest history
+                    latest_file_name = watch.history[watch.newest_history_key]
+                    with open(latest_file_name, "rb") as f2:
+                        f.write(f2.read())
+                        
+

changedetectionio/templates/edit.html

@@ -173,6 +173,16 @@ User-Agent: wonderbra 1.0") }}
                             <span class="pure-form-message-inline">Good for websites that just move the content around, and you want to know when NEW content is added, compares new lines against all history for this watch.</span>
                         </div>
                     </fieldset>
+                    <fieldset>
+                        <div class="pure-control-group">
+                            <label for="trigger-type">Filter and restrict change detection of content to</label>
+                            {{ render_checkbox_field(form.trigger_add, class="trigger-type") }}
+                            {{ render_checkbox_field(form.trigger_del, class="trigger-type") }}
+                            <span class="pure-form-message-inline">
+                                Filters the change-detection of this watch to only this type of content change. <strong>Replacements</strong> (neither additions nor deletions) are always included. The 'diff' will still include all changes.
+                            </span>
+                        </div>
+                    </fieldset>
                     <div class="pure-control-group">
                         {% set field = render_field(form.css_filter,
                             placeholder=".class-name or #some-id, or other CSS selector rule.",

changedetectionio/tests/test_backup.py

@@ -46,6 +46,6 @@ def test_backup(client, live_server):
     uuid4hex = re.compile('^[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]{3}-?[a-f0-9]{12}.*txt', re.I)
     newlist = list(filter(uuid4hex.match, l))  # Read Note below
 
-    # Should be two txt files in the archive (history and the snapshot)
-    assert len(newlist) == 2
+    # Should be three txt files in the archive (history and the snapshot)
+    assert len(newlist) == 3
 

changedetectionio/tests/test_diff_filter_changes_as_add_delete.py

@@ -0,0 +1,107 @@
+#!/usr/bin/python3
+# @NOTE:  THIS RELIES ON SOME MIDDLEWARE TO MAKE CHECKBOXES WORK WITH WTFORMS UNDER TEST CONDITION, see changedetectionio/tests/util.py
+import time
+from flask import url_for
+from .util import live_server_setup
+
+def set_original_response():
+    test_return_data = """
+        Here
+        is
+        some
+        text
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def set_response_with_deleted_word():
+    test_return_data = """
+        Here
+        is
+        text
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def set_response_with_changed_word():
+    test_return_data = """
+        Here
+        ix
+        some
+        text
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def test_diff_filter_changes_as_add_delete(client, live_server):
+    live_server_setup(live_server)
+
+    sleep_time_for_fetch_thread = 3
+
+    set_original_response()
+    # Give the endpoint time to spin up
+    time.sleep(1)
+
+    # Add our URL to the import page
+    test_url = url_for('test_endpoint', _external=True)
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+
+    assert b"1 Imported" in res.data
+    # Wait for it to read the original version
+    time.sleep(sleep_time_for_fetch_thread)
+
+    #  Make a change that ONLY includes deletes
+    set_response_with_deleted_word()
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"trigger_add": "y",
+              "trigger_del": "n",
+              "url": test_url,
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # We should NOT see a change because we chose to not know about any Deletions
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+    # Recheck to be sure
+    client.get(url_for("form_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+
+    # Now set the original response, which will include the word, which should trigger Added (because trigger_add ==y)
+    set_original_response()
+    client.get(url_for("form_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data
+
+    # Now check 'changes' are always going to be triggered
+    set_original_response()
+    client.post(
+        url_for("edit_page", uuid="first"),
+        # Neither trigger add nor del? then we should see changes still
+        data={"trigger_add": "n",
+              "trigger_del": "n",
+              "url": test_url,
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+    time.sleep(sleep_time_for_fetch_thread)
+    client.get(url_for("mark_all_viewed"), follow_redirects=True)
+    set_response_with_changed_word()
+    client.get(url_for("form_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data

changedetectionio/tests/test_diff_filter_only_additions.py

@@ -0,0 +1,83 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from .util import live_server_setup
+
+def set_original_response():
+    test_return_data = """
+        A few new lines
+        Where there is more lines originally
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def set_delete_response():
+    test_return_data = """
+        A few new lines
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def test_diff_filtering_no_del(client, live_server):
+    live_server_setup(live_server)
+
+    sleep_time_for_fetch_thread = 3
+
+    set_original_response()
+    # Give the endpoint time to spin up
+    time.sleep(1)
+
+    # Add our URL to the import page
+    test_url = url_for('test_endpoint', _external=True)
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+
+    assert b"1 Imported" in res.data
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Add our URL to the import page
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"trigger_add": "y",
+              "trigger_del": "n",
+              "url": test_url,
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+    assert b'unviewed' not in res.data
+
+    #  Make an delete change
+    set_delete_response()
+
+    time.sleep(sleep_time_for_fetch_thread)
+    # Trigger a check
+    client.get(url_for("form_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # We should NOT see the change
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+    #  Make an delete change
+    set_original_response()
+
+    time.sleep(sleep_time_for_fetch_thread)
+    # Trigger a check
+    client.get(url_for("form_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # We should see the change
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data
+

changedetectionio/tests/test_diff_filter_only_deletions.py

@@ -0,0 +1,72 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from .util import live_server_setup
+
+def set_original_response():
+    test_return_data = """
+        A few new lines
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def set_add_response():
+    test_return_data = """
+        A few new lines
+        Where there is more lines than before
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+def test_diff_filtering_no_add(client, live_server):
+    live_server_setup(live_server)
+
+    sleep_time_for_fetch_thread = 3
+
+    set_original_response()
+    # Give the endpoint time to spin up
+    time.sleep(1)
+
+    # Add our URL to the import page
+    test_url = url_for('test_endpoint', _external=True)
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+
+    assert b"1 Imported" in res.data
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Add our URL to the import page
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"trigger_add": "n",
+              "trigger_del": "y",
+              "url": test_url,
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+    assert b'unviewed' not in res.data
+
+    #  Make an add change
+    set_add_response()
+
+    time.sleep(sleep_time_for_fetch_thread)
+    # Trigger a check
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # We should NOT see the change
+    res = client.get(url_for("index"))
+    # save res.data to a file
+    
+        
+        
+    assert b'unviewed' not in res.data
+

changedetectionio/tests/test_history_consistency.py

@@ -81,4 +81,4 @@ def test_consistent_history(client, live_server):
 
 
 
-        assert len(files_in_watch_dir) == 2, "Should be just two files in the dir, history.txt and the snapshot"
+        assert len(files_in_watch_dir) == 3, "Should be just three files in the dir, history.txt, previous.txt, and the snapshot"

changedetectionio/tests/util.py

@@ -4,6 +4,12 @@ from flask import make_response, request
 from flask import url_for
 import logging
 import time
+from werkzeug import Request
+import io
+
+# This is a fix for macOS running tests.
+import multiprocessing
+multiprocessing.set_start_method("fork")
 
 def set_original_response():
     test_return_data = """<html>
@@ -159,6 +165,38 @@ def live_server_setup(live_server):
         ret = " ".join([auth.username, auth.password, auth.type])
         return ret
 
+    # Make sure any checkboxes that are supposed to be defaulted to true are set during the post request
+    # This is due to the fact that defaults are set in the HTML which we are not using during tests.
+    # This does not affect the server when running outside of a test
+    class DefaultCheckboxMiddleware(object):
+        def __init__(self, app):
+            self.app = app
+
+        def __call__(self, environ, start_response):
+            request = Request(environ)
+            if request.method == "POST" and "/edit" in request.path:
+                body = environ['wsgi.input'].read()
+
+                # if the checkboxes are not set, set them to true
+                if b"trigger_add" not in body:
+                    body += b'&trigger_add=y'
+
+                if b"trigger_del" not in body:
+                    body += b'&trigger_del=y'
+
+                # remove any checkboxes set to "n" so wtforms processes them correctly
+                body = body.replace(b"trigger_add=n", b"")
+                body = body.replace(b"trigger_del=n", b"")
+                body = body.replace(b"&&", b"&")
+
+                new_stream = io.BytesIO(body)
+                environ["CONTENT_LENGTH"] = len(body)
+                environ['wsgi.input'] = new_stream
+
+            return self.app(environ, start_response)
+
+    live_server.app.wsgi_app = DefaultCheckboxMiddleware(live_server.app.wsgi_app)
+
     # Just return some GET var
     @live_server.app.route('/test-return-query', methods=['GET'])
     def test_return_query():