Installation via pip (#186)

Builder for https://pypi.org/project/changedetection.io/

changedetectionio/tests/__init__.py

@@ -0,0 +1,2 @@
+"""Tests for the app."""
+

changedetectionio/tests/conftest.py

@@ -0,0 +1,57 @@
+#!/usr/bin/python3
+
+import pytest
+from changedetectionio import changedetection_app
+from changedetectionio import store
+import os
+
+# https://github.com/pallets/flask/blob/1.1.2/examples/tutorial/tests/test_auth.py
+# Much better boilerplate than the docs
+# https://www.python-boilerplate.com/py3+flask+pytest/
+
+global app
+
+
+def cleanup(datastore_path):
+    # Unlink test output files
+    files = ['output.txt',
+             'url-watches.json',
+             'notification.txt',
+             'count.txt',
+             'endpoint-content.txt']
+    for file in files:
+        try:
+            os.unlink("{}/{}".format(datastore_path, file))
+            x = 1
+        except FileNotFoundError:
+            pass
+
+@pytest.fixture(scope='session')
+def app(request):
+    """Create application for the tests."""
+    datastore_path = "./test-datastore"
+
+    try:
+        os.mkdir(datastore_path)
+    except FileExistsError:
+        pass
+
+    # Enable a BASE_URL for notifications to work (so we can look for diff/ etc URLs)
+    os.environ["BASE_URL"] = "http://mysite.com/"
+    cleanup(datastore_path)
+
+
+    app_config = {'datastore_path': datastore_path}
+    cleanup(app_config['datastore_path'])
+    datastore = store.ChangeDetectionStore(datastore_path=app_config['datastore_path'], include_default_watches=False)
+    app = changedetection_app(app_config, datastore)
+    app.config['STOP_THREADS'] = True
+
+    def teardown():
+        datastore.stop_thread = True
+        app.config.exit.set()
+        cleanup(app_config['datastore_path'])
+
+       
+    request.addfinalizer(teardown)
+    yield app

changedetectionio/tests/test_access_control.py

@@ -0,0 +1,108 @@
+from flask import url_for
+
+
+def test_check_access_control(app, client):
+    # Still doesnt work, but this is closer.
+
+    with app.test_client() as c:
+        # Check we dont have any password protection enabled yet.
+        res = c.get(url_for("settings_page"))
+        assert b"Remove password" not in res.data
+
+        # Enable password check.
+        res = c.post(
+            url_for("settings_page"),
+            data={"password": "foobar",
+                  "minutes_between_check": 180,
+                  'fetch_backend': "html_requests"},
+            follow_redirects=True
+        )
+
+        assert b"Password protection enabled." in res.data
+        assert b"LOG OUT" not in res.data
+
+        # Check we hit the login
+        res = c.get(url_for("index"), follow_redirects=True)
+
+        assert b"Login" in res.data
+
+        # Menu should not be available yet
+        #        assert b"SETTINGS" not in res.data
+        #        assert b"BACKUP" not in res.data
+        #        assert b"IMPORT" not in res.data
+
+        # defaultuser@changedetection.io is actually hardcoded for now, we only use a single password
+        res = c.post(
+            url_for("login"),
+            data={"password": "foobar"},
+            follow_redirects=True
+        )
+
+        assert b"LOG OUT" in res.data
+        res = c.get(url_for("settings_page"))
+
+        # Menu should be available now
+        assert b"SETTINGS" in res.data
+        assert b"BACKUP" in res.data
+        assert b"IMPORT" in res.data
+        assert b"LOG OUT" in res.data
+
+        # Now remove the password so other tests function, @todo this should happen before each test automatically
+        res = c.get(url_for("settings_page", removepassword="yes"),
+              follow_redirects=True)
+        assert b"Password protection removed." in res.data
+
+        res = c.get(url_for("index"))
+        assert b"LOG OUT" not in res.data
+
+
+# There was a bug where saving the settings form would submit a blank password
+def test_check_access_control_no_blank_password(app, client):
+    # Still doesnt work, but this is closer.
+
+    with app.test_client() as c:
+        # Check we dont have any password protection enabled yet.
+        res = c.get(url_for("settings_page"))
+        assert b"Remove password" not in res.data
+
+        # Enable password check.
+        res = c.post(
+            url_for("settings_page"),
+            data={"password": "",
+                  "minutes_between_check": 180,
+                  'fetch_backend': "html_requests"},
+
+        follow_redirects=True
+        )
+
+        assert b"Password protection enabled." not in res.data
+        assert b"Login" not in res.data
+
+
+# There was a bug where saving the settings form would submit a blank password
+def test_check_access_no_remote_access_to_remove_password(app, client):
+    # Still doesnt work, but this is closer.
+
+    with app.test_client() as c:
+        # Check we dont have any password protection enabled yet.
+        res = c.get(url_for("settings_page"))
+        assert b"Remove password" not in res.data
+
+        # Enable password check.
+        res = c.post(
+            url_for("settings_page"),
+            data={"password": "password", "minutes_between_check": 180,
+                  'fetch_backend': "html_requests"},
+            follow_redirects=True
+        )
+
+        assert b"Password protection enabled." in res.data
+        assert b"Login" in res.data
+
+        res = c.get(url_for("settings_page", removepassword="yes"),
+              follow_redirects=True)
+        assert b"Password protection removed." not in res.data
+
+        res = c.get(url_for("index"),
+              follow_redirects=True)
+        assert b"watch-table-wrapper" not in res.data

changedetectionio/tests/test_backend.py

@@ -0,0 +1,107 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from urllib.request import urlopen
+from . util import set_original_response, set_modified_response, live_server_setup
+
+sleep_time_for_fetch_thread = 3
+
+
+def test_check_basic_change_detection_functionality(client, live_server):
+    set_original_response()
+    live_server_setup(live_server)
+
+    # Add our URL to the import page
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": url_for('test_endpoint', _external=True)},
+        follow_redirects=True
+    )
+    assert b"1 Imported" in res.data
+
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Do this a few times.. ensures we dont accidently set the status
+    for n in range(3):
+        client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+        # Give the thread time to pick it up
+        time.sleep(sleep_time_for_fetch_thread)
+
+        # It should report nothing found (no new 'unviewed' class)
+        res = client.get(url_for("index"))
+        assert b'unviewed' not in res.data
+        assert b'test-endpoint' in res.data
+
+        # Default no password set, this stuff should be always available.
+
+        assert b"SETTINGS" in res.data
+        assert b"BACKUP" in res.data
+        assert b"IMPORT" in res.data
+
+    #####################
+
+    # Make a change
+    set_modified_response()
+
+    res = urlopen(url_for('test_endpoint', _external=True))
+    assert b'which has this one new line' in res.read()
+
+    # Force recheck
+    res = client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    assert b'1 watches are rechecking.' in res.data
+
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Now something should be ready, indicated by having a 'unviewed' class
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data
+
+    # #75, and it should be in the RSS feed
+    res = client.get(url_for("index", rss="true"))
+    expected_url = url_for('test_endpoint', _external=True)
+    assert b'<rss' in res.data
+    assert expected_url.encode('utf-8') in res.data
+
+    # Following the 'diff' link, it should no longer display as 'unviewed' even after we recheck it a few times
+    res = client.get(url_for("diff_history_page", uuid="first"))
+    assert b'Compare newest' in res.data
+
+    time.sleep(2)
+
+    # Do this a few times.. ensures we dont accidently set the status
+    for n in range(2):
+        client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+        # Give the thread time to pick it up
+        time.sleep(sleep_time_for_fetch_thread)
+
+        # It should report nothing found (no new 'unviewed' class)
+        res = client.get(url_for("index"))
+        assert b'unviewed' not in res.data
+        assert b'head title' not in res.data # Should not be present because this is off by default
+        assert b'test-endpoint' in res.data
+
+    set_original_response()
+
+    # Enable auto pickup of <title> in settings
+    res = client.post(
+        url_for("settings_page"),
+        data={"extract_title_as_title": "1", "minutes_between_check": 180, 'fetch_backend': "html_requests"},
+        follow_redirects=True
+    )
+
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data
+    # It should have picked up the <title>
+    assert b'head title' in res.data
+
+    #
+    # Cleanup everything
+    res = client.get(url_for("api_delete", uuid="all"), follow_redirects=True)
+    assert b'Deleted' in res.data
+

changedetectionio/tests/test_css_selector.py

@@ -0,0 +1,128 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+
+from ..html_tools import *
+
+def test_setup(live_server):
+    live_server_setup(live_server)
+
+def set_original_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     <div id="sametext">Some text thats the same</div>
+     <div id="changetext">Some text that will change</div>
+     </body>
+     </html>
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+    return None
+
+def set_modified_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>which has this one new line</p>
+     </br>
+     So let's see what happens.  </br>
+     <div id="sametext">Some text thats the same</div>
+     <div id="changetext">Some text that changes</div>
+     </body>
+     </html>
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+    return None
+
+
+# Test that the CSS extraction works how we expect, important here is the right placing of new lines \n's
+def test_css_filter_output():
+    from changedetectionio import fetch_site_status
+    from inscriptis import get_text
+
+    # Check text with sub-parts renders correctly
+    content = """<html> <body><div id="thingthing" >  Some really <b>bold</b> text  </div> </body> </html>"""
+    html_blob = css_filter(css_filter="#thingthing", html_content=content)
+    text = get_text(html_blob)
+    assert text == "  Some really bold text"
+
+    content = """<html> <body>
+    <p>foo bar blah</p>
+    <div class="parts">Block A</div> <div class="parts">Block B</div></body> 
+    </html>
+"""
+    html_blob = css_filter(css_filter=".parts", html_content=content)
+    text = get_text(html_blob)
+
+    # Divs are converted to 4 whitespaces by inscriptis
+    assert text == "    Block A\n    Block B"
+
+
+# Tests the whole stack works with the CSS Filter
+def test_check_markup_css_filter_restriction(client, live_server):
+    sleep_time_for_fetch_thread = 3
+
+    css_filter = "#sametext"
+
+    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
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Goto the edit page, add our ignore text
+    # Add our URL to the import page
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"css_filter": css_filter, "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+    # Check it saved
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+    )
+    assert bytes(css_filter.encode('utf-8')) in res.data
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+    #  Make a change
+    set_modified_response()
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should have 'unviewed' still
+    # Because it should be looking at only that 'sametext' id
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data

changedetectionio/tests/test_headers.py

@@ -0,0 +1,80 @@
+import json
+import time
+from flask import url_for
+from . util import set_original_response, set_modified_response, live_server_setup
+
+# Hard to just add more live server URLs when one test is already running (I think)
+# So we add our test here (was in a different file)
+def test_headers_in_request(client, live_server):
+    live_server_setup(live_server)
+
+    # Add our URL to the import page
+    test_url = url_for('test_headers', _external=True)
+
+    # Add the test URL twice, we will check
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+    assert b"1 Imported" in res.data
+
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+    assert b"1 Imported" in res.data
+
+    cookie_header = '_ga=GA1.2.1022228332; cookie-preferences=analytics:accepted;'
+
+
+    # Add some headers to a request
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={
+              "url": test_url,
+              "tag": "",
+              "fetch_backend": "html_requests",
+              "headers": "xxx:ooo\ncool:yeah\r\ncookie:"+cookie_header},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+
+    # Give the thread time to pick up the first version
+    time.sleep(5)
+
+    # The service should echo back the request headers
+    res = client.get(
+        url_for("preview_page", uuid="first"),
+        follow_redirects=True
+    )
+
+    # Flask will convert the header key to uppercase
+    assert b"Xxx:ooo" in res.data
+    assert b"Cool:yeah" in res.data
+
+    # The test call service will return the headers as the body
+    from html import escape
+    assert escape(cookie_header).encode('utf-8') in res.data
+
+    time.sleep(5)
+
+    # Re #137 -  Examine the JSON index file, it should have only one set of headers entered
+    watches_with_headers = 0
+    with open('test-datastore/url-watches.json') as f:
+        app_struct = json.load(f)
+        for uuid in app_struct['watching']:
+            if (len(app_struct['watching'][uuid]['headers'])):
+                watches_with_headers += 1
+
+    # Should be only one with headers set
+    assert watches_with_headers==1
+
+
+
+
+
+
+

changedetectionio/tests/test_ignore_regex_text.py

@@ -0,0 +1,31 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+
+def test_setup(live_server):
+    live_server_setup(live_server)
+
+# Unit test of the stripper
+# Always we are dealing in utf-8
+def test_strip_regex_text_func():
+    from changedetectionio import fetch_site_status
+
+    test_content = """
+    but sometimes we want to remove the lines.
+    
+    but 1 lines
+    but including 1234 lines
+    igNORe-cAse text we dont want to keep    
+    but not always."""
+
+    ignore_lines = ["sometimes", "/\s\d{2,3}\s/", "/ignore-case text/"]
+
+    fetcher = fetch_site_status.perform_site_check(datastore=False)
+    stripped_content = fetcher.strip_ignore_text(test_content, ignore_lines)
+
+    assert b"but 1 lines" in stripped_content
+    assert b"igNORe-cAse text" not in stripped_content
+    assert b"but 1234 lines" not in stripped_content
+

changedetectionio/tests/test_ignore_text.py

@@ -0,0 +1,153 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+
+def test_setup(live_server):
+    live_server_setup(live_server)
+
+# Unit test of the stripper
+# Always we are dealing in utf-8
+def test_strip_text_func():
+    from changedetectionio import fetch_site_status
+
+    test_content = """
+    Some content
+    is listed here
+
+    but sometimes we want to remove the lines.
+
+    but not always."""
+
+    ignore_lines = ["sometimes"]
+
+    fetcher = fetch_site_status.perform_site_check(datastore=False)
+    stripped_content = fetcher.strip_ignore_text(test_content, ignore_lines)
+
+    assert b"sometimes" not in stripped_content
+    assert b"Some content" in stripped_content
+
+
+def set_original_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+def set_modified_original_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some NEW nice initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+# Is the same but includes ZZZZZ, 'ZZZZZ' is the last line in ignore_text
+def set_modified_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     <P>ZZZZZ</P>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+def test_check_ignore_text_functionality(client, live_server):
+    sleep_time_for_fetch_thread = 3
+
+    ignore_text = "XXXXX\r\nYYYYY\r\nZZZZZ"
+    set_original_ignore_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
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Goto the edit page, add our ignore text
+    # Add our URL to the import page
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"ignore_text": ignore_text, "url": test_url, 'fetch_backend': "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+    # Check it saved
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+    )
+    assert bytes(ignore_text.encode('utf-8')) in res.data
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (no new 'unviewed' class)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+    assert b'/test-endpoint' in res.data
+
+    #  Make a change
+    set_modified_ignore_response()
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (no new 'unviewed' class)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+    assert b'/test-endpoint' in res.data
+
+    # Just to be sure.. set a regular modified change..
+    set_modified_original_ignore_response()
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data
+
+    res = client.get(url_for("api_delete", uuid="all"), follow_redirects=True)
+    assert b'Deleted' in res.data

changedetectionio/tests/test_jsonpath_selector.py

@@ -0,0 +1,163 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+import pytest
+
+def test_unittest_inline_html_extract():
+    # So lets pretend that the JSON we want is inside some HTML
+    content="""
+    <html>
+    
+    food and stuff and more
+    <script>
+    alert('nothing really good here');
+    </script>
+    
+    <script type="application/ld+json">
+  xx {"@context":"http://schema.org","@type":"Product","name":"Nan Optipro Stage 1 Baby Formula  800g","description":"During the first year of life, nutrition is critical for your baby. NAN OPTIPRO 1 is tailored to ensure your formula fed infant receives balanced, high quality nutrition.<br />Starter infant formula. The age optimised protein source (whey dominant) is from cow’s milk.<br />Backed by more than 150 years of Nestlé expertise.<br />For hygiene and convenience, it is available in an innovative packaging format with a separate storage area for the scoop, and a semi-transparent window which allows you to see how much powder is left in the can without having to open it.","image":"https://cdn0.woolworths.media/content/wowproductimages/large/155536.jpg","brand":{"@context":"http://schema.org","@type":"Organization","name":"Nan"},"gtin13":"7613287517388","offers":{"@context":"http://schema.org","@type":"Offer","potentialAction":{"@context":"http://schema.org","@type":"BuyAction"},"availability":"http://schema.org/InStock","itemCondition":"http://schema.org/NewCondition","price":23.5,"priceCurrency":"AUD"},"review":[],"sku":"155536"}
+</script>
+<body>
+and it can also be repeated
+<script type="application/ld+json">
+  {"@context":"http://schema.org","@type":"Product","name":"Nan Optipro Stage 1 Baby Formula  800g","description":"During the first year of life, nutrition is critical for your baby. NAN OPTIPRO 1 is tailored to ensure your formula fed infant receives balanced, high quality nutrition.<br />Starter infant formula. The age optimised protein source (whey dominant) is from cow’s milk.<br />Backed by more than 150 years of Nestlé expertise.<br />For hygiene and convenience, it is available in an innovative packaging format with a separate storage area for the scoop, and a semi-transparent window which allows you to see how much powder is left in the can without having to open it.","image":"https://cdn0.woolworths.media/content/wowproductimages/large/155536.jpg","brand":{"@context":"http://schema.org","@type":"Organization","name":"Nan"},"gtin13":"7613287517388","offers":{"@context":"http://schema.org","@type":"Offer","potentialAction":{"@context":"http://schema.org","@type":"BuyAction"},"availability":"http://schema.org/InStock","itemCondition":"http://schema.org/NewCondition","price":23.5,"priceCurrency":"AUD"},"review":[],"sku":"155536"}
+</script>
+<h4>ok</h4>
+</body>
+</html>
+
+    """
+    from .. import html_tools
+
+    # See that we can find the second <script> one, which is not broken, and matches our filter
+    text = html_tools.extract_json_as_string(content, "$.offers.price")
+    assert text == "23.5"
+
+    text = html_tools.extract_json_as_string('{"id":5}', "$.id")
+    assert text == "5"
+
+    # When nothing at all is found, it should throw JSONNOTFound
+    # Which is caught and shown to the user in the watch-overview table
+    with pytest.raises(html_tools.JSONNotFound) as e_info:
+        html_tools.extract_json_as_string('COMPLETE GIBBERISH, NO JSON!', "$.id")
+
+
+def set_original_response():
+    test_return_data = """
+    {
+      "employees": [
+        {
+          "id": 1,
+          "name": "Pankaj",
+          "salary": "10000"
+        },
+        {
+          "name": "David",
+          "salary": "5000",
+          "id": 2
+        }
+      ],
+      "boss": {
+        "name": "Fat guy"
+      }
+    }
+    """
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+    return None
+
+def set_modified_response():
+    test_return_data = """
+    {
+      "employees": [
+        {
+          "id": 1,
+          "name": "Pankaj",
+          "salary": "10000"
+        },
+        {
+          "name": "David",
+          "salary": "5000",
+          "id": 2
+        }
+      ],
+      "boss": {
+        "name": "Foobar"
+      }
+    }
+        """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+    return None
+
+
+
+def test_check_json_filter(client, live_server):
+    live_server_setup(live_server)
+
+    json_filter = 'json:boss.name'
+
+    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
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(3)
+
+    # Goto the edit page, add our ignore text
+    # Add our URL to the import page
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"css_filter": json_filter,
+              "url": test_url,
+              "tag": "",
+              "headers": "",
+              "fetch_backend": "html_requests"
+              },
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+    # Check it saved
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+    )
+    assert bytes(json_filter.encode('utf-8')) in res.data
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(3)
+    #  Make a change
+    set_modified_response()
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    # Give the thread time to pick it up
+    time.sleep(4)
+
+    # It should have 'unviewed' still
+    res = client.get(url_for("index"))
+    assert b'unviewed' in res.data
+
+    # Should not see this, because its not in the JSONPath we entered
+    res = client.get(url_for("diff_history_page", uuid="first"))
+    # But the change should be there, tho its hard to test the change was detected because it will show old and new versions
+    assert b'Foobar' in res.data

changedetectionio/tests/test_notification.py

@@ -0,0 +1,134 @@
+import os
+import time
+from flask import url_for
+from . util import set_original_response, set_modified_response, live_server_setup
+
+# Hard to just add more live server URLs when one test is already running (I think)
+# So we add our test here (was in a different file)
+def test_check_notification(client, live_server):
+
+    live_server_setup(live_server)
+    set_original_response()
+
+    # Give the endpoint time to spin up
+    time.sleep(3)
+
+    # 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
+
+    # Give the thread time to pick up the first version
+    time.sleep(3)
+
+    # Goto the edit page, add our ignore text
+    # Add our URL to the import page
+    url = url_for('test_notification_endpoint', _external=True)
+    notification_url = url.replace('http', 'json')
+
+    print (">>>> Notification URL: "+notification_url)
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"notification_urls": notification_url,
+              "url": test_url,
+              "tag": "",
+              "headers": "",
+              "fetch_backend": "html_requests",
+              "trigger_check": "y"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+    assert b"Notifications queued" in res.data
+
+    # Hit the edit page, be sure that we saved it
+    res = client.get(
+        url_for("edit_page", uuid="first"))
+    assert bytes(notification_url.encode('utf-8')) in res.data
+
+
+    # Because we hit 'send test notification on save'
+    time.sleep(3)
+
+    notification_submission = None
+
+    # Verify what was sent as a notification, this file should exist
+    with open("test-datastore/notification.txt", "r") as f:
+        notification_submission = f.read()
+        # Did we see the URL that had a change, in the notification?
+
+    assert test_url in notification_submission
+
+    os.unlink("test-datastore/notification.txt")
+
+
+    set_modified_response()
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(3)
+
+    # Did the front end see it?
+    res = client.get(
+        url_for("index"))
+
+    assert bytes("just now".encode('utf-8')) in res.data
+
+    notification_submission=None
+    # Verify what was sent as a notification
+    with open("test-datastore/notification.txt", "r") as f:
+        notification_submission = f.read()
+        # Did we see the URL that had a change, in the notification?
+
+    assert test_url in notification_submission
+
+        # Re #65 - did we see our foobar.com BASE_URL ?
+        #assert bytes("https://foobar.com".encode('utf-8')) in notification_submission
+
+
+    ##  Now configure something clever, we go into custom config (non-default) mode, this is returned by the endpoint
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(";jasdhflkjadshf kjhsdfkjl ahslkjf haslkjd hfaklsj hf\njl;asdhfkasj stuff we will detect\n")
+
+    res = client.post(
+        url_for("settings_page"),
+        data={"notification_title": "New ChangeDetection.io Notification - {watch_url}",
+              "notification_body": "{base_url}\n{watch_url}\n{preview_url}\n{diff_url}\n{current_snapshot}\n:-)",
+              "notification_urls": "json://foobar.com", #Re #143 should not see that it sent without [test checkbox]
+              "minutes_between_check": 180,
+              "fetch_backend": "html_requests",
+              },
+        follow_redirects=True
+    )
+    assert b"Settings updated." in res.data
+    # Re #143 - should not see this if we didnt hit the test box
+    assert b"Notifications queued" not in res.data
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(3)
+
+    # Did the front end see it?
+    res = client.get(
+        url_for("index"))
+
+    assert bytes("just now".encode('utf-8')) in res.data
+
+    with open("test-datastore/notification.txt", "r") as f:
+        notification_submission = f.read()
+
+        # @todo regex that diff/uuid-31123-123-etc
+
+        assert "diff/" in notification_submission
+        assert "preview/" in notification_submission
+        assert ":-)" in notification_submission
+        assert "New ChangeDetection.io Notification - {}".format(test_url) in notification_submission
+        # This should insert the {current_snapshot}
+        assert "stuff we will detect" in notification_submission

changedetectionio/tests/test_trigger.py

@@ -0,0 +1,131 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+
+
+def set_original_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+def set_modified_original_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some NEW nice initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+def set_modified_with_trigger_text_response():
+    test_return_data = """<html>
+       <body>
+     Some NEW nice initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     foobar123
+     <br/>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+def test_trigger_functionality(client, live_server):
+
+    live_server_setup(live_server)
+
+    sleep_time_for_fetch_thread = 3
+    trigger_text = "foobar123"
+    set_original_ignore_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
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # Goto the edit page, add our ignore text
+    # Add our URL to the import page
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"trigger_text": trigger_text,
+              "url": test_url,
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+    # Check it saved
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+    )
+    assert bytes(trigger_text.encode('utf-8')) in res.data
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (no new 'unviewed' class)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+    assert b'/test-endpoint' in res.data
+
+    #  Make a change
+    set_modified_original_ignore_response()
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (no new 'unviewed' class)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+    # Just to be sure.. set a regular modified change..
+    time.sleep(sleep_time_for_fetch_thread)
+    set_modified_with_trigger_text_response()
+
+    client.get(url_for("api_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_trigger_regex.py

@@ -0,0 +1,81 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+
+
+def set_original_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+
+def test_trigger_regex_functionality(client, live_server):
+
+    live_server_setup(live_server)
+
+    sleep_time_for_fetch_thread = 3
+
+    set_original_ignore_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
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (just a new one shouldnt have anything)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+    ### test regex
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"trigger_text": '/something \d{3}/',
+              "url": test_url,
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("some new noise")
+
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (nothing should match the regex)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("regex test123<br/>\nsomething 123")
+
+    client.get(url_for("api_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_trigger_regex_with_filter.py

@@ -0,0 +1,84 @@
+#!/usr/bin/python3
+
+import time
+from flask import url_for
+from . util import live_server_setup
+
+
+def set_original_ignore_response():
+    test_return_data = """<html>
+       <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+
+def test_trigger_regex_functionality(client, live_server):
+
+    live_server_setup(live_server)
+
+    sleep_time_for_fetch_thread = 3
+
+    set_original_ignore_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
+
+    # Trigger a check
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+
+    # Give the thread time to pick it up
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (just a new one shouldnt have anything)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+    ### test regex with filter
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"trigger_text": "/cool.stuff\d/",
+              "url": test_url,
+              "css_filter": '#in-here',
+              "fetch_backend": "html_requests"},
+        follow_redirects=True
+    )
+
+    # Check that we have the expected text.. but it's not in the css filter we want
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("<html>some new noise with cool stuff2 ok</html>")
+
+    client.get(url_for("api_watch_checknow"), follow_redirects=True)
+    time.sleep(sleep_time_for_fetch_thread)
+
+    # It should report nothing found (nothing should match the regex and filter)
+    res = client.get(url_for("index"))
+    assert b'unviewed' not in res.data
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("<html>some new noise with <span id=in-here>cool stuff6</span> ok</html>")
+
+    client.get(url_for("api_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_watch_fields_storage.py

@@ -0,0 +1,150 @@
+import time
+from flask import url_for
+from urllib.request import urlopen
+from . util import set_original_response, set_modified_response, live_server_setup
+
+
+def test_check_watch_field_storage(client, live_server):
+    set_original_response()
+    live_server_setup(live_server)
+
+    test_url = "http://somerandomsitewewatch.com"
+
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+    assert b"1 Imported" in res.data
+
+
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={ "notification_urls": "http://myapi.com",
+               "minutes_between_check": 126,
+               "css_filter" : ".fooclass",
+               "title" : "My title",
+               "ignore_text" : "ignore this",
+               "url": test_url,
+               "tag": "woohoo",
+               "headers": "curl:foo",
+               'fetch_backend': "html_requests"
+               },
+        follow_redirects=True
+    )
+    assert b"Updated watch." in res.data
+
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+        follow_redirects=True
+    )
+
+    assert b"http://myapi.com" in res.data
+    assert b"126" in res.data
+    assert b".fooclass" in res.data
+    assert b"My title" in res.data
+    assert b"ignore this" in res.data
+    assert b"http://somerandomsitewewatch.com" in res.data
+    assert b"woohoo" in res.data
+    assert b"curl: foo" in res.data
+
+
+
+# Re https://github.com/dgtlmoon/changedetection.io/issues/110
+def test_check_recheck_global_setting(client, live_server):
+
+    res = client.post(
+        url_for("settings_page"),
+        data={
+               "minutes_between_check": 1566,
+               'fetch_backend': "html_requests"
+               },
+        follow_redirects=True
+    )
+    assert b"Settings updated." in res.data
+
+    # Now add a record
+
+    test_url = "http://somerandomsitewewatch.com"
+
+    res = client.post(
+        url_for("import_page"),
+        data={"urls": test_url},
+        follow_redirects=True
+    )
+    assert b"1 Imported" in res.data
+
+    # Now visit the edit page, it should have the default minutes
+
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+        follow_redirects=True
+    )
+
+    # Should show the default minutes
+    assert b"change to another value if you want to be specific" in res.data
+    assert b"1566" in res.data
+
+    res = client.post(
+        url_for("settings_page"),
+        data={
+               "minutes_between_check": 222,
+                'fetch_backend': "html_requests"
+               },
+        follow_redirects=True
+    )
+    assert b"Settings updated." in res.data
+
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+        follow_redirects=True
+    )
+
+    # Should show the default minutes
+    assert b"change to another value if you want to be specific" in res.data
+    assert b"222" in res.data
+
+    # Now change it specifically, it should show the new minutes
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"url": test_url,
+              "minutes_between_check": 55,
+              'fetch_backend': "html_requests"
+              },
+        follow_redirects=True
+    )
+
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+        follow_redirects=True
+    )
+    assert b"55" in res.data
+
+    # Now submit an empty field, it should give back the default global minutes
+    res = client.post(
+        url_for("settings_page"),
+        data={
+               "minutes_between_check": 666,
+                'fetch_backend': "html_requests"
+               },
+        follow_redirects=True
+    )
+    assert b"Settings updated." in res.data
+
+    res = client.post(
+        url_for("edit_page", uuid="first"),
+        data={"url": test_url,
+              "minutes_between_check": "",
+              'fetch_backend': "html_requests"
+              },
+        follow_redirects=True
+    )
+
+    assert b"Updated watch." in res.data
+
+    res = client.get(
+        url_for("edit_page", uuid="first"),
+        follow_redirects=True
+    )
+    assert b"666" in res.data
+

changedetectionio/tests/util.py

@@ -0,0 +1,74 @@
+#!/usr/bin/python3
+
+
+def set_original_response():
+    test_return_data = """<html>
+    <head><title>head title</title></head>
+    <body>
+     Some initial text</br>
+     <p>Which is across multiple lines</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+    return None
+
+def set_modified_response():
+    test_return_data = """<html>
+    <head><title>modified head title</title></head>
+    <body>
+     Some initial text</br>
+     <p>which has this one new line</p>
+     </br>
+     So let's see what happens.  </br>
+     </body>
+     </html>
+    """
+
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+    return None
+
+
+def live_server_setup(live_server):
+
+
+    @live_server.app.route('/test-endpoint')
+    def test_endpoint():
+        # Tried using a global var here but didn't seem to work, so reading from a file instead.
+        with open("test-datastore/endpoint-content.txt", "r") as f:
+            return f.read()
+
+    # Just return the headers in the request
+    @live_server.app.route('/test-headers')
+    def test_headers():
+
+        from flask import request
+        output= []
+
+        for header in request.headers:
+             output.append("{}:{}".format(str(header[0]),str(header[1])   ))
+
+        return "\n".join(output)
+
+
+    # Where we POST to as a notification
+    @live_server.app.route('/test_notification_endpoint', methods=['POST'])
+    def test_notification_endpoint():
+        from flask import request
+
+        with open("test-datastore/notification.txt", "wb") as f:
+            # Debug method, dump all POST to file also, used to prove #65
+            data = request.stream.read()
+            if data != None:
+                f.write(data)
+
+        print("\n>> Test notification endpoint was hit.\n")
+        return "Text was set"
+
+    live_server.start()