kopia lustrzana https://github.com/dgtlmoon/changedetection.io
download-latest-html
rodzic
ed00f67a80
commit
8aae7dfae0
|
@ -532,12 +532,21 @@ def changedetection_app(config=None, datastore_o=None):
|
|||
@login_optionally_required
|
||||
def ajax_callback_send_notification_test(watch_uuid=None):
|
||||
|
||||
# Watch_uuid could be unsuet in the case its used in tag editor, global setings
|
||||
# Watch_uuid could be unset in the case its used in tag editor, global setings
|
||||
import apprise
|
||||
import random
|
||||
from .apprise_asset import asset
|
||||
apobj = apprise.Apprise(asset=asset)
|
||||
|
||||
watch = datastore.data['watching'].get(watch_uuid) if watch_uuid else None
|
||||
is_global_settings_form = request.args.get('mode', '') == 'global-settings'
|
||||
is_group_settings_form = request.args.get('mode', '') == 'group-settings'
|
||||
|
||||
# Use an existing random one on the global/main settings form
|
||||
if not watch_uuid and (is_global_settings_form or is_group_settings_form):
|
||||
logger.debug(f"Send test notification - Choosing random Watch {watch_uuid}")
|
||||
watch_uuid = random.choice(list(datastore.data['watching'].keys()))
|
||||
|
||||
watch = datastore.data['watching'].get(watch_uuid)
|
||||
|
||||
notification_urls = request.form['notification_urls'].strip().splitlines()
|
||||
|
||||
|
@ -549,8 +558,6 @@ def changedetection_app(config=None, datastore_o=None):
|
|||
tag = datastore.tag_exists_by_name(k.strip())
|
||||
notification_urls = tag.get('notifications_urls') if tag and tag.get('notifications_urls') else None
|
||||
|
||||
is_global_settings_form = request.args.get('mode', '') == 'global-settings'
|
||||
is_group_settings_form = request.args.get('mode', '') == 'group-settings'
|
||||
if not notification_urls and not is_global_settings_form and not is_group_settings_form:
|
||||
# In the global settings, use only what is typed currently in the text box
|
||||
logger.debug("Test notification - Trying by global system settings notifications")
|
||||
|
@ -569,7 +576,7 @@ def changedetection_app(config=None, datastore_o=None):
|
|||
try:
|
||||
# use the same as when it is triggered, but then override it with the form test values
|
||||
n_object = {
|
||||
'watch_url': request.form['window_url'],
|
||||
'watch_url': request.form.get('window_url', "https://changedetection.io"),
|
||||
'notification_urls': notification_urls
|
||||
}
|
||||
|
||||
|
|
|
@ -157,7 +157,7 @@ def process_notification(n_object, datastore):
|
|||
logger.warning(f"Process Notification: skipping empty notification URL.")
|
||||
continue
|
||||
|
||||
logger.info(">> Process Notification: AppRise notifying {}".format(url))
|
||||
logger.info(f">> Process Notification: AppRise notifying {url}")
|
||||
url = jinja_render(template_str=url, **notification_parameters)
|
||||
|
||||
# Re 323 - Limit discord length to their 2000 char limit total or it wont send.
|
||||
|
@ -230,6 +230,7 @@ def process_notification(n_object, datastore):
|
|||
log_value = logs.getvalue()
|
||||
|
||||
if log_value and 'WARNING' in log_value or 'ERROR' in log_value:
|
||||
logger.critical(log_value)
|
||||
raise Exception(log_value)
|
||||
|
||||
# Return what was sent for better logging - after the for loop
|
||||
|
|
|
@ -3,6 +3,8 @@ 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, \
|
||||
set_longer_modified_response
|
||||
from . util import extract_UUID_from_client
|
||||
|
@ -347,3 +349,81 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
|
|||
url_for("form_delete", uuid="all"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
|
||||
#2510
|
||||
def test_global_send_test_notification(client, live_server, measure_memory_usage):
|
||||
|
||||
|
||||
#live_server_setup(live_server)
|
||||
set_original_response()
|
||||
|
||||
# otherwise other settings would have already existed from previous tests in this file
|
||||
res = client.post(
|
||||
url_for("settings_page"),
|
||||
data={
|
||||
"application-fetch_backend": "html_requests",
|
||||
"application-minutes_between_check": 180,
|
||||
"application-notification_body": 'change detection is cool',
|
||||
"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("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("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
|
||||
|
||||
# Give apprise time to fire
|
||||
time.sleep(4)
|
||||
|
||||
with open("test-datastore/notification.txt", 'r') as f:
|
||||
x = f.read()
|
||||
assert 'change detection is coo' in x
|
||||
|
||||
|
||||
os.unlink("test-datastore/notification.txt")
|
||||
|
||||
######### Test group/tag settings
|
||||
res = client.post(
|
||||
url_for("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
|
||||
time.sleep(4)
|
||||
|
||||
with open("test-datastore/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 coo' in x
|
||||
|
||||
client.get(
|
||||
url_for("form_delete", uuid="all"),
|
||||
follow_redirects=True
|
||||
)
|
||||
|
||||
|
|
Ładowanie…
Reference in New Issue