You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
changedetection.io/changedetectionio/tests/test_notification.py

337 lines
13 KiB

import json
import os
import time
import re
from flask import url_for
from .util import set_original_response, set_modified_response, set_more_modified_response, live_server_setup, wait_for_all_checks, \
set_longer_modified_response
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,
)
def test_setup(live_server):
live_server_setup(live_server)
# 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()
# Re 360 - new install should have defaults set
res = client.get(url_for("settings_page"))
notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'json')
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_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
# 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_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("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 = extract_UUID_from_client(client)
datastore = 'test-datastore'
with open(os.path.join(datastore, str(uuid), '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"
":-)",
"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"})
res = client.post(
url_for("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("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()
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(3)
# Verify what was sent as a notification, this file should exist
with open("test-datastore/notification.txt", "r") as f:
notification_submission = f.read()
os.unlink("test-datastore/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)
# We keep PNG screenshots for now
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()
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(3)
# Verify what was sent as a notification, this file should exist
with open("test-datastore/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("test-datastore/notification.txt")
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
client.get(url_for("form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
client.get(url_for("form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
assert os.path.exists("test-datastore/notification.txt") == False
res = client.get(url_for("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()
res = client.post(
url_for("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"},
follow_redirects=True
)
assert b"Updated watch." in res.data
time.sleep(2)
# Verify what was sent as a notification, this file should exist
with open("test-datastore/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("form_delete", uuid="all"),
follow_redirects=True
)
def test_notification_validation(client, live_server):
time.sleep(1)
# 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("form_quick_watch_add"),
data={"url": test_url, "tags": 'nice one'},
follow_redirects=True
)
assert b"Watch added" in res.data
# Re #360 some validation
# res = client.post(
# url_for("edit_page", uuid="first"),
# data={"notification_urls": 'json://localhost/foobar',
# "notification_title": "",
# "notification_body": "",
# "notification_format": "Text",
# "url": test_url,
# "tag": "my tag",
# "title": "my title",
# "headers": "",
# "fetch_backend": "html_requests"},
# follow_redirects=True
# )
# assert b"Notification Body and Title is required when a Notification URL is used" in res.data
# cleanup for the next
client.get(
url_for("form_delete", uuid="all"),
follow_redirects=True
)
def test_notification_custom_endpoint_and_jinja2(client, live_server):
#live_server_setup(live_server)
# 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()
test_notification_url = url_for('test_notification_endpoint', _external=True).replace('http://', 'post://')+"?xxx={{ watch_url }}"
res = client.post(
url_for("settings_page"),
data={
"application-fetch_backend": "html_requests",
"application-minutes_between_check": 180,
"application-notification_body": '{ "url" : "{{ watch_url }}", "secret": 444 }',
"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
client.get(
url_for("form_delete", uuid="all"),
follow_redirects=True
)
# Add a watch and trigger a HTTP POST
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("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()
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(2)
with open("test-datastore/notification.txt", 'r') as f:
x = f.read()
j = json.loads(x)
assert j['url'].startswith('http://localhost')
assert j['secret'] == 444
# URL check, this will always be converted to lowercase
assert os.path.isfile("test-datastore/notification-url.txt")
with open("test-datastore/notification-url.txt", 'r') as f:
notification_url = f.read()
assert 'xxx=http' in notification_url
# Should always be automatically detected as JSON content type even when we set it as 'Text' (default)
assert os.path.isfile("test-datastore/notification-content-type.txt")
with open("test-datastore/notification-content-type.txt", 'r') as f:
assert 'application/json' in f.read()
os.unlink("test-datastore/notification-url.txt")