Apprise notifications (#43)
* issue #4 Adding settings screen for apprise URLS * Adding test notification mechanism * Move Worker module to own class file * Adding basic notification URL runner * Tests for notifications * Tweak readme with notification info * Move notification test to main test_backend.py * Fix spacing * Adding notifications screenshot * Cleanup more files from test * Offer send notification test on individual edits and main/default * Process global notifications * All branches test * Wrap worker notification process in try/catch, use global if nothing set * Fix syntax * Handle exception, increase wait time for liveserver to come up * Fixing test setup * remove debug * Split tests into their own totally isolated setups, if you know a better way to make live_server() work, MR :) * Tidying up lint/importspull/59/head
dgtlmoon
3 years ago
committed by
GitHub
parent
b752690f89
commit
f877af75b9
@ -0,0 +1,16 @@
|
||||
#!/bin/bash
|
||||
|
||||
|
||||
# live_server will throw errors even with live_server_scope=function if I have the live_server setup in different functions
|
||||
# and I like to restart the server for each test (and have the test cleanup after each test)
|
||||
# merge request welcome :)
|
||||
|
||||
|
||||
# exit when any command fails
|
||||
set -e
|
||||
|
||||
find tests/test_*py -type f|while read test_name
|
||||
do
|
||||
echo "TEST RUNNING $test_name"
|
||||
pytest $test_name
|
||||
done
|
||||
@ -0,0 +1,58 @@
|
||||
from flask import url_for
|
||||
|
||||
def test_check_access_control(app, client):
|
||||
# Still doesnt work, but this is closer.
|
||||
return
|
||||
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"},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Password protection enabled." in res.data
|
||||
assert b"LOG OUT" not in res.data
|
||||
print ("SESSION:", res.session)
|
||||
# Check we hit the login
|
||||
|
||||
res = c.get(url_for("settings_page"), follow_redirects=True)
|
||||
res = c.get(url_for("login"), follow_redirects=True)
|
||||
|
||||
assert b"Login" in res.data
|
||||
|
||||
print ("DEBUG >>>>>",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", "email": "defaultuser@changedetection.io"},
|
||||
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
|
||||
|
||||
c.get(url_for("settings_page", removepassword="true"))
|
||||
c.get(url_for("import_page"))
|
||||
assert b"LOG OUT" not in res.data
|
||||
|
||||
@ -0,0 +1,66 @@
|
||||
|
||||
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 it up
|
||||
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": ""},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Updated watch." 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
|
||||
|
||||
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
|
||||
|
||||
|
||||
# Check it triggered
|
||||
res = client.get(
|
||||
url_for("test_notification_counter"),
|
||||
)
|
||||
print (res.data)
|
||||
|
||||
assert bytes("we hit it".encode('utf-8')) in res.data
|
||||
@ -0,0 +1,60 @@
|
||||
#!/usr/bin/python3
|
||||
|
||||
|
||||
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>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
with open("test-datastore/output.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>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
with open("test-datastore/output.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/output.txt", "r") as f:
|
||||
return f.read()
|
||||
|
||||
@live_server.app.route('/test_notification_endpoint', methods=['POST'])
|
||||
def test_notification_endpoint():
|
||||
with open("test-datastore/count.txt", "w") as f:
|
||||
f.write("we hit it")
|
||||
print("\n>> Test notification endpoint was hit.\n")
|
||||
return "Text was set"
|
||||
|
||||
# And this should return not zero.
|
||||
@live_server.app.route('/test_notification_counter')
|
||||
def test_notification_counter():
|
||||
try:
|
||||
with open("test-datastore/count.txt", "r") as f:
|
||||
return f.read()
|
||||
except FileNotFoundError:
|
||||
return "nope :("
|
||||
|
||||
live_server.start()
|
||||
@ -0,0 +1,67 @@
|
||||
import threading
|
||||
import queue
|
||||
|
||||
# Requests for checking on the site use a pool of thread Workers managed by a Queue.
|
||||
class update_worker(threading.Thread):
|
||||
current_uuid = None
|
||||
|
||||
def __init__(self, q, notification_q, app, datastore, *args, **kwargs):
|
||||
self.q = q
|
||||
self.app = app
|
||||
self.notification_q = notification_q
|
||||
self.datastore = datastore
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def run(self):
|
||||
from backend import fetch_site_status
|
||||
|
||||
update_handler = fetch_site_status.perform_site_check(datastore=self.datastore)
|
||||
|
||||
while not self.app.config.exit.is_set():
|
||||
|
||||
try:
|
||||
uuid = self.q.get(block=False)
|
||||
except queue.Empty:
|
||||
pass
|
||||
|
||||
else:
|
||||
self.current_uuid = uuid
|
||||
|
||||
if uuid in list(self.datastore.data['watching'].keys()):
|
||||
try:
|
||||
changed_detected, result, contents = update_handler.run(uuid)
|
||||
|
||||
except PermissionError as s:
|
||||
self.app.logger.error("File permission error updating", uuid, str(s))
|
||||
else:
|
||||
if result:
|
||||
try:
|
||||
self.datastore.update_watch(uuid=uuid, update_obj=result)
|
||||
if changed_detected:
|
||||
# A change was detected
|
||||
self.datastore.save_history_text(uuid=uuid, contents=contents, result_obj=result)
|
||||
|
||||
watch = self.datastore.data['watching'][uuid]
|
||||
|
||||
# Did it have any notification alerts to hit?
|
||||
if len(watch['notification_urls']):
|
||||
print("Processing notifications for UUID: {}".format(uuid))
|
||||
n_object = {'watch_url': self.datastore.data['watching'][uuid]['url'],
|
||||
'notification_urls': watch['notification_urls']}
|
||||
self.notification_q.put(n_object)
|
||||
|
||||
|
||||
# No? maybe theres a global setting, queue them all
|
||||
elif len(self.datastore.data['settings']['application']['notification_urls']):
|
||||
print("Processing GLOBAL notifications for UUID: {}".format(uuid))
|
||||
n_object = {'watch_url': self.datastore.data['watching'][uuid]['url'],
|
||||
'notification_urls': self.datastore.data['settings']['application'][
|
||||
'notification_urls']}
|
||||
self.notification_q.put(n_object)
|
||||
except Exception as e:
|
||||
print("!!!! Exception in update_worker !!!\n", e)
|
||||
|
||||
self.current_uuid = None # Done
|
||||
self.q.task_done()
|
||||
|
||||
self.app.config.exit.wait(1)
|
||||
|
After Width: | Height: | Size: 27 KiB |
Loading…
Reference in new issue