changedetection.io/changedetectionio/tests/test_notification.py

import json
import os
import time
import re
from flask import url_for
from loguru import logger

from .util import set_original_response, set_modified_response, set_more_modified_response, live_server_setup, wait_for_all_checks, wait_for_notification_endpoint_output
from . util import  extract_UUID_from_client
import logging
import base64

from changedetectionio.notification import (
    default_notification_body,
    default_notification_format,
    default_notification_title, valid_notification_formats
)
from ..diff import HTML_CHANGED_STYLE
from ..model import USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH


# 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, measure_memory_usage, datastore_path):
    
    set_original_response(datastore_path=datastore_path)

    # Re 360 - new install should have defaults set
    res = client.get(url_for("settings.settings_page"))
    notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'json')+"?status_code=204"

    assert default_notification_body.encode() in res.data
    assert default_notification_title.encode() in res.data

    #####################
    # Set this up for when we remove the notification from the watch, it should fallback with these details
    res = client.post(
        url_for("settings.settings_page"),
        data={"application-notification_urls": notification_url,
              "application-notification_title": "fallback-title "+default_notification_title,
              "application-notification_body": "fallback-body "+default_notification_body,
              "application-notification_format": default_notification_format,
              "requests-time_between_check-minutes": 180,
              'application-fetch_backend': "html_requests"},
        follow_redirects=True
    )

    assert b"Settings updated." in res.data

    res = client.get(url_for("settings.settings_page"))
    for k,v in valid_notification_formats.items():
        if k == USE_SYSTEM_DEFAULT_NOTIFICATION_FORMAT_FOR_WATCH:
            continue
        assert f'value="{k}"'.encode() in res.data # Should be by key NOT value
        assert f'value="{v}"'.encode() not in res.data # Should be by key NOT value


    # When test mode is in BASE_URL env mode, we should see this already configured
    env_base_url = os.getenv('BASE_URL', '').strip()
    if len(env_base_url):
        logging.debug(">>> BASE_URL enabled, looking for %s", env_base_url)
        res = client.get(url_for("settings.settings_page"))
        assert bytes(env_base_url.encode('utf-8')) in res.data
    else:
        logging.debug(">>> SKIPPING BASE_URL check")

    # re #242 - when you edited an existing new entry, it would not correctly show the notification settings
    # Add our URL to the import page
    test_url = url_for('test_endpoint', _external=True)
    res = client.post(
        url_for("ui.ui_views.form_quick_watch_add"),
        data={"url": test_url, "tags": ''},
        follow_redirects=True
    )
    assert b"Watch added" in res.data

    # Give the thread time to pick up the first version
    wait_for_all_checks(client)

    # We write the PNG to disk, but a JPEG should appear in the notification
    # Write the last screenshot png
    testimage_png = 'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII='


    uuid = next(iter(live_server.app.config['DATASTORE'].data['watching']))
    screenshot_dir = os.path.join(datastore_path, str(uuid))
    os.makedirs(screenshot_dir, exist_ok=True)
    with open(os.path.join(screenshot_dir, 'last-screenshot.png'), 'wb') as f:
        f.write(base64.b64decode(testimage_png))

    # Goto the edit page, add our ignore text
    # Add our URL to the import page

    print (">>>> Notification URL: "+notification_url)

    notification_form_data = {"notification_urls": notification_url,
                              "notification_title": "New ChangeDetection.io Notification - {{watch_url}}",
                              "notification_body": "BASE URL: {{base_url}}\n"
                                                   "Watch URL: {{watch_url}}\n"
                                                   "Watch UUID: {{watch_uuid}}\n"
                                                   "Watch title: {{watch_title}}\n"
                                                   "Watch tag: {{watch_tag}}\n"
                                                   "Preview: {{preview_url}}\n"
                                                   "Diff URL: {{diff_url}}\n"
                                                   "Snapshot: {{current_snapshot}}\n"
                                                   "Diff: {{diff}}\n"
                                                   "Diff Added: {{diff_added}}\n"
                                                   "Diff Removed: {{diff_removed}}\n"
                                                   "Diff Full: {{diff_full}}\n"
                                                   "Diff as Patch: {{diff_patch}}\n"
                                                   ":-)",
                              "notification_screenshot": True,
                              "notification_format": 'text'}

    notification_form_data.update({
        "url": test_url,
        "tags": "my tag, my second tag",
        "title": "my title",
        "headers": "",
        "fetch_backend": "html_requests",
        "time_between_check_use_default": "y"})

    res = client.post(
        url_for("ui.ui_edit.edit_page", uuid="first"),
        data=notification_form_data,
        follow_redirects=True
    )
    assert b"Updated watch." in res.data


    # Hit the edit page, be sure that we saved it
    # Re #242 - wasnt saving?
    res = client.get(
        url_for("ui.ui_edit.edit_page", uuid="first"))
    assert bytes(notification_url.encode('utf-8')) in res.data
    assert bytes("New ChangeDetection.io Notification".encode('utf-8')) in res.data



    ## Now recheck, and it should have sent the notification
    wait_for_all_checks(client)
    set_modified_response(datastore_path=datastore_path)

    # Trigger a check
    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    wait_for_all_checks(client)
    wait_for_notification_endpoint_output(datastore_path=datastore_path)

    # Check no errors were recorded
    res = client.get(url_for("watchlist.index"))
    assert b'notification-error' not in res.data


    # Verify what was sent as a notification, this file should exist
    with open(os.path.join(datastore_path, "notification.txt"), "r") as f:
        notification_submission = f.read()
    os.unlink(os.path.join(datastore_path, "notification.txt"))

    # Did we see the URL that had a change, in the notification?
    # Diff was correctly executed

    assert "Diff Full: Some initial text" in notification_submission
    assert "Diff: (changed) Which is across multiple lines" in notification_submission
    assert "(into) which has this one new line" in notification_submission
    # Re #342 - check for accidental python byte encoding of non-utf8/string
    assert "b'" not in notification_submission
    assert re.search('Watch UUID: [0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}', notification_submission, re.IGNORECASE)
    assert "Watch title: my title" in notification_submission
    assert "Watch tag: my tag, my second tag" in notification_submission
    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
    assert test_url in notification_submission
    assert ':-)' in notification_submission
    # Check the attachment was added, and that it is a JPEG from the original PNG
    notification_submission_object = json.loads(notification_submission)
    assert notification_submission_object

    # We keep PNG screenshots for now
    # IF THIS FAILS YOU SHOULD BE TESTING WITH ENV VAR REMOVE_REQUESTS_OLD_SCREENSHOTS=False
    assert notification_submission_object['attachments'][0]['filename'] == 'last-screenshot.png'
    assert len(notification_submission_object['attachments'][0]['base64'])
    assert notification_submission_object['attachments'][0]['mimetype'] == 'image/png'
    jpeg_in_attachment = base64.b64decode(notification_submission_object['attachments'][0]['base64'])

    # Assert that the JPEG is readable (didn't get chewed up somewhere)
    from PIL import Image
    import io
    assert Image.open(io.BytesIO(jpeg_in_attachment))

    if env_base_url:
        # Re #65 - did we see our BASE_URl ?
        logging.debug (">>> BASE_URL checking in notification: %s", env_base_url)
        assert env_base_url in notification_submission
    else:
        logging.debug(">>> Skipping BASE_URL check")


    # This should insert the {current_snapshot}
    set_more_modified_response(datastore_path=datastore_path)
    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    wait_for_all_checks(client)
    wait_for_notification_endpoint_output(datastore_path=datastore_path)
    # Verify what was sent as a notification, this file should exist
    with open(os.path.join(datastore_path, "notification.txt"), "r") as f:
        notification_submission = f.read()
    assert "Ohh yeah awesome" in notification_submission


    # Prove that "content constantly being marked as Changed with no Updating causes notification" is not a thing
    # https://github.com/dgtlmoon/changedetection.io/discussions/192
    os.unlink(os.path.join(datastore_path, "notification.txt"))

    # Trigger a check
    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    wait_for_all_checks(client)
    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    wait_for_all_checks(client)
    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    wait_for_all_checks(client)
    assert os.path.exists(os.path.join(datastore_path, "notification.txt")) == False

    res = client.get(url_for("settings.notification_logs"))
    # be sure we see it in the output log
    assert b'New ChangeDetection.io Notification - ' + test_url.encode('utf-8') in res.data

    set_original_response(datastore_path=datastore_path)
    res = client.post(
        url_for("ui.ui_edit.edit_page", uuid="first"),
        data={
        "url": test_url,
        "tags": "my tag",
        "title": "my title",
        "notification_urls": '',
        "notification_title": '',
        "notification_body": '',
        "notification_format": default_notification_format,
        "fetch_backend": "html_requests",
        "time_between_check_use_default": "y"},
        follow_redirects=True
    )
    assert b"Updated watch." in res.data

    wait_for_all_checks(client)
    wait_for_notification_endpoint_output(datastore_path=datastore_path)

    # Verify what was sent as a notification, this file should exist
    with open(os.path.join(datastore_path, "notification.txt"), "r") as f:
        notification_submission = f.read()
    assert "fallback-title" in notification_submission
    assert "fallback-body" in notification_submission

    # cleanup for the next
    client.get(
        url_for("ui.form_delete", uuid="all"),
        follow_redirects=True
    )


def test_notification_urls_jinja2_apprise_integration(client, live_server, measure_memory_usage, datastore_path):

    #
    # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#header-manipulation
    test_notification_url = "hassio://127.0.0.1/longaccesstoken?verify=no&nid={{watch_uuid}}"

    res = client.post(
        url_for("settings.settings_page"),
        data={
              "application-fetch_backend": "html_requests",
              "application-minutes_between_check": 180,
              "application-notification_body": '{ "url" : "{{ watch_url }}", "secret": 444, "somebug": "网站监测 内容更新了", "another": "{{diff|truncate(1500)}}" }',
              "application-notification_format": default_notification_format,
              "application-notification_urls": test_notification_url,
              # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
              "application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }}  {{diff|truncate(200)}} ",
              },
        follow_redirects=True
    )
    assert b'Settings updated' in res.data
    assert '网站监测'.encode() in res.data
    assert b'{{diff|truncate(1500)}}' in res.data
    assert b'{{diff|truncate(200)}}' in res.data




def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_memory_usage, datastore_path):
    

    # test_endpoint - that sends the contents of a file
    # test_notification_endpoint - that takes a POST and writes it to file (test-datastore/notification.txt)

    # CUSTOM JSON BODY CHECK for POST://
    set_original_response(datastore_path=datastore_path)
    # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#header-manipulation
    test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?status_code=204&watch_uuid={{ watch_uuid }}&xxx={{ watch_url }}&now={% now 'Europe/London', '%Y-%m-%d' %}&+custom-header=123&+second=hello+world%20%22space%22"

    res = client.post(
        url_for("settings.settings_page"),
        data={
              "application-fetch_backend": "html_requests",
              "application-minutes_between_check": 180,
              "application-notification_body": '{ "url" : "{{ watch_url }}", "secret": 444, "somebug": "网站监测 内容更新了" }',
              "application-notification_format": default_notification_format,
              "application-notification_urls": test_notification_url,
              # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
              "application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }} ",
              },
        follow_redirects=True
    )
    assert b'Settings updated' in res.data

    # Add a watch and trigger a HTTP POST
    test_url = url_for('test_endpoint', _external=True)
    res = client.post(
        url_for("ui.ui_views.form_quick_watch_add"),
        data={"url": test_url, "tags": 'nice one'},
        follow_redirects=True
    )

    assert b"Watch added" in res.data
    watch_uuid = next(iter(live_server.app.config['DATASTORE'].data['watching']))

    wait_for_all_checks(client)
    set_modified_response(datastore_path=datastore_path)

    client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    wait_for_all_checks(client)

    wait_for_notification_endpoint_output(datastore_path=datastore_path)


    # Check no errors were recorded, because we asked for 204 which is slightly uncommon but is still OK
    res = client.get(url_for("watchlist.index"))
    assert b'notification-error' not in res.data

    with open(os.path.join(datastore_path, "notification.txt"), 'r') as f:
        x = f.read()
        j = json.loads(x)
        assert j['url'].startswith('http://localhost')
        assert j['secret'] == 444
        assert j['somebug'] == '网站监测 内容更新了'


    # URL check, this will always be converted to lowercase
    assert os.path.isfile(os.path.join(datastore_path, "notification-url.txt"))
    with open(os.path.join(datastore_path, "notification-url.txt"), 'r') as f:
        notification_url = f.read()
        assert 'xxx=http' in notification_url
        # apprise style headers should be stripped
        assert 'custom-header' not in notification_url
        # Check jinja2 custom arrow/jinja2-time replace worked
        assert 'now=2' in notification_url
        # Check our watch_uuid appeared
        assert f'watch_uuid={watch_uuid}' in notification_url


    with open(os.path.join(datastore_path, "notification-headers.txt"), 'r') as f:
        notification_headers = f.read()
        assert 'custom-header: 123' in notification_headers.lower()
        assert 'second: hello world "space"' in notification_headers.lower()


    # Should always be automatically detected as JSON content type even when we set it as 'Plain Text' (default)
    assert os.path.isfile(os.path.join(datastore_path, "notification-content-type.txt"))
    with open(os.path.join(datastore_path, "notification-content-type.txt"), 'r') as f:
        assert 'application/json' in f.read()

    os.unlink(os.path.join(datastore_path, "notification-url.txt"))

    client.get(
        url_for("ui.form_delete", uuid="all"),
        follow_redirects=True
    )


#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 网站监测 内容更新了 - {{diff_full}}'

    # otherwise other settings would have already existed from previous tests in this file
    res = client.post(
        url_for("settings.settings_page"),
        data={
            "application-fetch_backend": "html_requests",
            "application-minutes_between_check": 180,
            "application-notification_body": test_body,
            "application-notification_format": default_notification_format,
            "application-notification_urls": "",
            "application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }}",
        },
        follow_redirects=True
    )
    assert b'Settings updated' in res.data

    test_url = url_for('test_endpoint', _external=True)
    res = client.post(
        url_for("ui.ui_views.form_quick_watch_add"),
        data={"url": test_url, "tags": 'nice one'},
        follow_redirects=True
    )

    assert b"Watch added" in res.data

    test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}&+custom-header=123"

    ######### Test global/system settings
    res = client.post(
        url_for("ui.ui_notification.ajax_callback_send_notification_test")+"?mode=global-settings",
        data={"notification_urls": test_notification_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"))

    ######### Test group/tag settings
    res = client.post(
        url_for("ui.ui_notification.ajax_callback_send_notification_test")+"?mode=group-settings",
        data={"notification_urls": test_notification_url},
        follow_redirects=True
    )

    assert res.status_code != 400
    assert res.status_code != 500

    # Give apprise time to fire
    wait_for_notification_endpoint_output(datastore_path=datastore_path)

    with open(os.path.join(datastore_path, "notification.txt"), 'r') as f:
        x = f.read()
        # Should come from notification.py default handler when there is no notification body to pull from
        assert 'change detection is cool 网站监测 内容更新了' in x

    ## Check that 'test' catches errors
    test_notification_url = 'post://akjsdfkjasdkfjasdkfjasdkjfas232323/should-error'

    ######### Test global/system settings
    res = client.post(
        url_for("ui.ui_notification.ajax_callback_send_notification_test")+"?mode=global-settings",
        data={"notification_urls": test_notification_url},
        follow_redirects=True
    )
    assert res.status_code == 400
    assert (
        b"No address found" in res.data or
        b"Name or service not known" in res.data or
        b"nodename nor servname provided" in res.data or
        b"Temporary failure in name resolution" in res.data or
        b"Failed to establish a new connection" in res.data or
        b"Connection error occurred" in res.data
    )
    
    client.get(
        url_for("ui.form_delete", uuid="all"),
        follow_redirects=True
    )

    ######### Test global/system settings - When everything is deleted it should give a helpful error
    # See #2727
    res = client.post(
        url_for("ui.ui_notification.ajax_callback_send_notification_test")+"?mode=global-settings",
        data={"notification_urls": test_notification_url},
        follow_redirects=True
    )
    assert res.status_code == 400
    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)
    wait_for_all_checks(client)

    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}}\n\nCurrent snapshot: {{current_snapshot}}'
    ######### 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
        #3720 current_snapshot check, was working but lets test it exactly.
        assert 'Current snapshot: Example text: example test' in x
    os.unlink(os.path.join(datastore_path, "notification.txt"))

def _test_color_notifications(client, notification_body_token, 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_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}&+custom-header=123"


    # otherwise other settings would have already existed from previous tests in this file
    res = client.post(
        url_for("settings.settings_page"),
        data={
            "application-fetch_backend": "html_requests",
            "application-minutes_between_check": 180,
            "application-notification_body": notification_body_token,
            "application-notification_format": "htmlcolor",
            "application-notification_urls": test_notification_url,
            "application-notification_title": "New ChangeDetection.io Notification - {{ watch_url }}",
        },
        follow_redirects=True
    )
    assert b'Settings updated' in res.data

    test_url = url_for('test_endpoint', _external=True)
    res = client.post(
        url_for("ui.ui_views.form_quick_watch_add"),
        data={"url": test_url, "tags": 'nice one'},
        follow_redirects=True
    )

    assert b"Watch added" in res.data

    wait_for_all_checks(client)

    set_modified_response(datastore_path=datastore_path)


    res = client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
    assert b'Queued 1 watch for rechecking.' in res.data

    wait_for_all_checks(client)
    wait_for_notification_endpoint_output(datastore_path=datastore_path)

    with open(os.path.join(datastore_path, "notification.txt"), 'r') as f:
        x = f.read()
        s = f'<span style="{HTML_CHANGED_STYLE}" role="note" aria-label="Changed text" title="Changed text">Which is across multiple lines</span><br>'
        assert s in x

    client.get(
        url_for("ui.form_delete", uuid="all"),
        follow_redirects=True
    )

# Just checks the format of the colour notifications was correct
def test_html_color_notifications(client, live_server, measure_memory_usage, datastore_path):
    _test_color_notifications(client, '{{diff}}',datastore_path=datastore_path)
    _test_color_notifications(client, '{{diff_full}}',datastore_path=datastore_path)