cphipps/changedetection.io
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UI/Functionality - Ability to manage/apply filters and notifications across tags/groups

Browse Source
This commit is contained in:
dgtlmoon
2023-06-19 23:29:13 +02:00
committed by GitHub
parent 72311fb845
commit 52f2c00308
49 changed files with 1161 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
changedetectionio/tests/proxy_list/test_multiple_proxy.py
View File

@@ -28,7 +28,7 @@ def test_preferred_proxy(client, live_server):
"fetch_backend": "html_requests",
"headers": "",
"proxy": "proxy-two",
"tag": "",
"tags": "",
"url": url,
},
follow_redirects=True

2
changedetectionio/tests/restock/test_restock.py
View File

@@ -77,7 +77,7 @@ def test_restock_detection(client, live_server):
client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": '', 'processor': 'restock_diff'},
data={"url": test_url, "tags": '', 'processor': 'restock_diff'},
follow_redirects=True
)

9
changedetectionio/tests/test_access_control.py
View File

@@ -45,6 +45,15 @@ def test_check_access_control(app, client, live_server):
res = client.get(url_for("diff_history_page", uuid="first"))
assert b'Random content' in res.data
# Check wrong password does not let us in
res = c.post(
url_for("login"),
data={"password": "WRONG PASSWORD"},
follow_redirects=True
)
assert b"LOG OUT" not in res.data
assert b"Incorrect password" in res.data
# Menu should not be available yet

29
changedetectionio/tests/test_add_replace_remove_filter.py
View File

@@ -2,7 +2,7 @@
import time
from flask import url_for
from .util import live_server_setup
from .util import live_server_setup, wait_for_all_checks
from changedetectionio import html_tools
@@ -39,7 +39,6 @@ def test_setup(client, live_server):
live_server_setup(live_server)
def test_check_removed_line_contains_trigger(client, live_server):
sleep_time_for_fetch_thread = 3
# Give the endpoint time to spin up
time.sleep(1)
@@ -54,7 +53,7 @@ def test_check_removed_line_contains_trigger(client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
@@ -67,20 +66,20 @@ def test_check_removed_line_contains_trigger(client, live_server):
follow_redirects=True
)
assert b"Updated watch." in res.data
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
set_original(excluding='Something irrelevant')
# A line thats not the trigger should not trigger anything
res = client.get(url_for("form_watch_checknow"), follow_redirects=True)
assert b'1 watches queued for rechecking.' in res.data
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' not in res.data
# The trigger line is REMOVED, this should trigger
set_original(excluding='The golden line')
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data
@@ -89,14 +88,14 @@ def test_check_removed_line_contains_trigger(client, live_server):
client.get(url_for("mark_all_viewed"), follow_redirects=True)
set_original(excluding=None)
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' not in res.data
# Remove it again, and we should get a trigger
set_original(excluding='The golden line')
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data
@@ -105,8 +104,7 @@ def test_check_removed_line_contains_trigger(client, live_server):
def test_check_add_line_contains_trigger(client, live_server):
sleep_time_for_fetch_thread = 3
#live_server_setup(live_server)
# Give the endpoint time to spin up
time.sleep(1)
@@ -136,8 +134,7 @@ def test_check_add_line_contains_trigger(client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
@@ -150,23 +147,25 @@ def test_check_add_line_contains_trigger(client, live_server):
follow_redirects=True
)
assert b"Updated watch." in res.data
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
set_original(excluding='Something irrelevant')
# A line thats not the trigger should not trigger anything
res = client.get(url_for("form_watch_checknow"), follow_redirects=True)
assert b'1 watches queued for rechecking.' in res.data
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' not in res.data
# The trigger line is ADDED, this should trigger
set_original(add_line='<p>Oh yes please</p>')
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data
# Takes a moment for apprise to fire
time.sleep(3)
with open("test-datastore/notification.txt", 'r') as f:
response= f.read()
assert '-Oh yes please-' in response

14
changedetectionio/tests/test_api.py
View File

@@ -2,7 +2,7 @@
import time
from flask import url_for
from .util import live_server_setup, extract_api_key_from_UI
from .util import live_server_setup, extract_api_key_from_UI, wait_for_all_checks
import json
import uuid
@@ -57,6 +57,7 @@ def test_setup(client, live_server):
live_server_setup(live_server)
def test_api_simple(client, live_server):
#live_server_setup(live_server)
api_key = extract_api_key_from_UI(client)
@@ -86,7 +87,7 @@ def test_api_simple(client, live_server):
watch_uuid = res.json.get('uuid')
assert res.status_code == 201
time.sleep(3)
wait_for_all_checks(client)
# Verify its in the list and that recheck worked
res = client.get(
@@ -107,7 +108,7 @@ def test_api_simple(client, live_server):
)
assert len(res.json) == 0
time.sleep(2)
wait_for_all_checks(client)
set_modified_response()
# Trigger recheck of all ?recheck_all=1
@@ -115,7 +116,7 @@ def test_api_simple(client, live_server):
url_for("createwatch", recheck_all='1'),
headers={'x-api-key': api_key},
)
time.sleep(3)
wait_for_all_checks(client)
# Did the recheck fire?
res = client.get(
@@ -297,6 +298,8 @@ def test_api_watch_PUT_update(client, live_server):
url_for("edit_page", uuid=watch_uuid),
)
assert b"cookie: yum" in res.data, "'cookie: yum' found in 'headers' section"
assert b"One" in res.data, "Tag 'One' was found"
assert b"Two" in res.data, "Tag 'Two' was found"
# HTTP PUT ( UPDATE an existing watch )
res = client.put(
@@ -319,7 +322,8 @@ def test_api_watch_PUT_update(client, live_server):
)
assert b"new title" in res.data, "new title found in edit page"
assert b"552" in res.data, "552 minutes found in edit page"
assert b"One, Two" in res.data, "Tag 'One, Two' was found"
assert b"One" in res.data, "Tag 'One' was found"
assert b"Two" in res.data, "Tag 'Two' was found"
assert b"cookie: all eaten" in res.data, "'cookie: all eaten' found in 'headers' section"
######################################################

2
changedetectionio/tests/test_auth.py
View File

@@ -24,7 +24,7 @@ def test_basic_auth(client, live_server):
# Check form validation
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": "", "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": "", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data

14
changedetectionio/tests/test_block_while_text_present.py
View File

@@ -2,7 +2,7 @@
import time
from flask import url_for
from . util import live_server_setup
from .util import live_server_setup, wait_for_all_checks
from changedetectionio import html_tools
def set_original_ignore_response():
@@ -61,7 +61,7 @@ def set_modified_response_minus_block_text():
def test_check_block_changedetection_text_NOT_present(client, live_server):
sleep_time_for_fetch_thread = 3
live_server_setup(live_server)
# Use a mix of case in ZzZ to prove it works case-insensitive.
ignore_text = "out of stoCk\r\nfoobar"
@@ -81,7 +81,7 @@ def test_check_block_changedetection_text_NOT_present(client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
@@ -96,7 +96,7 @@ def test_check_block_changedetection_text_NOT_present(client, live_server):
assert b"Updated watch." in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Check it saved
res = client.get(
url_for("edit_page", uuid="first"),
@@ -107,7 +107,7 @@ def test_check_block_changedetection_text_NOT_present(client, live_server):
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
@@ -120,7 +120,7 @@ def test_check_block_changedetection_text_NOT_present(client, live_server):
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
@@ -131,7 +131,7 @@ def test_check_block_changedetection_text_NOT_present(client, live_server):
# Now we set a change where the text is gone, it should now trigger
set_modified_response_minus_block_text()
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data

4
changedetectionio/tests/test_css_selector.py
View File

@@ -96,7 +96,7 @@ def test_check_markup_include_filters_restriction(client, live_server):
# Add our URL to the import page
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": include_filters, "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": include_filters, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data
@@ -157,7 +157,7 @@ def test_check_multiple_filters(client, live_server):
url_for("edit_page", uuid="first"),
data={"include_filters": include_filters,
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
'fetch_backend': "html_requests"},
follow_redirects=True

2
changedetectionio/tests/test_element_removal.py
View File

@@ -129,7 +129,7 @@ def test_element_removal_full(client, live_server):
data={
"subtractive_selectors": subtractive_selectors_data,
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
"fetch_backend": "html_requests",
},

4
changedetectionio/tests/test_extract_regex.py
View File

@@ -91,7 +91,7 @@ def test_check_filter_multiline(client, live_server):
data={"include_filters": '',
'extract_text': '/something.+?6 billion.+?lines/si',
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
'fetch_backend': "html_requests"
},
@@ -146,7 +146,7 @@ def test_check_filter_and_regex_extract(client, live_server):
data={"include_filters": include_filters,
'extract_text': '\d+ online\r\n\d+ guests\r\n/somecase insensitive \d+/i\r\n/somecase insensitive (345\d)/i',
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
'fetch_backend': "html_requests"
},

4
changedetectionio/tests/test_filter_exist_changes.py
View File

@@ -56,7 +56,7 @@ def test_filter_doesnt_exist_then_exists_should_get_notification(client, live_se
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": 'cinema'},
data={"url": test_url, "tags": 'cinema'},
follow_redirects=True
)
assert b"Watch added" in res.data
@@ -89,7 +89,7 @@ def test_filter_doesnt_exist_then_exists_should_get_notification(client, live_se
notification_form_data.update({
"url": test_url,
"tag": "my tag",
"tags": "my tag",
"title": "my title",
"headers": "",
"include_filters": '.ticket-available',

40
changedetectionio/tests/test_filter_failure_notification.py
View File

@@ -1,7 +1,7 @@
import os
import time
from flask import url_for
from .util import set_original_response, live_server_setup, extract_UUID_from_client
from .util import set_original_response, live_server_setup, extract_UUID_from_client, wait_for_all_checks
from changedetectionio.model import App
@@ -37,14 +37,14 @@ def run_filter_test(client, content_filter):
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": ''},
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
time.sleep(3)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
@@ -71,8 +71,8 @@ def run_filter_test(client, content_filter):
notification_form_data.update({
"url": test_url,
"tag": "my tag",
"title": "my title",
"tags": "my tag",
"title": "my title 123",
"headers": "",
"filter_failure_notification_send": 'y',
"include_filters": content_filter,
@@ -85,43 +85,55 @@ def run_filter_test(client, content_filter):
)
assert b"Updated watch." in res.data
time.sleep(3)
wait_for_all_checks(client)
# Now the notification should not exist, because we didnt reach the threshold
assert not os.path.isfile("test-datastore/notification.txt")
for i in range(0, App._FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT):
# -2 because we would have checked twice above (on adding and on edit)
for i in range(0, App._FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT-2):
res = client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(3)
wait_for_all_checks(client)
assert not os.path.isfile("test-datastore/notification.txt"), f"test-datastore/notification.txt should not exist - Attempt {i}"
# We should see something in the frontend
assert b'Warning, no filters were found' in res.data
# One more check should trigger it (see -2 above)
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)
# Now it should exist and contain our "filter not found" alert
assert os.path.isfile("test-datastore/notification.txt")
notification = False
with open("test-datastore/notification.txt", 'r') as f:
notification = f.read()
assert 'CSS/xPath filter was not present in the page' in notification
assert content_filter.replace('"', '\\"') in notification
# Remove it and prove that it doesnt trigger when not expected
# Remove it and prove that it doesn't trigger when not expected
# It should register a change, but no 'filter not found'
os.unlink("test-datastore/notification.txt")
set_response_with_filter()
# Try several times, it should NOT have 'filter not found'
for i in range(0, App._FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT):
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(3)
wait_for_all_checks(client)
# It should have sent a notification, but..
assert os.path.isfile("test-datastore/notification.txt")
# but it should not contain the info about the failed filter
# but it should not contain the info about a failed filter (because there was none in this case)
with open("test-datastore/notification.txt", 'r') as f:
notification = f.read()
assert not 'CSS/xPath filter was not present in the page' in notification
# Re #1247 - All tokens got replaced
# Re #1247 - All tokens got replaced correctly in the notification
res = client.get(url_for("index"))
uuid = extract_UUID_from_client(client)
# UUID is correct, but notification contains tag uuid as UUIID wtf
assert uuid in notification
# cleanup for the next
@@ -137,7 +149,7 @@ def test_setup(live_server):
def test_check_include_filters_failure_notification(client, live_server):
set_original_response()
time.sleep(1)
wait_for_all_checks(client)
run_filter_test(client, '#nope-doesnt-exist')
def test_check_xpath_filter_failure_notification(client, live_server):

262
changedetectionio/tests/test_group.py Normal file
View File

@@ -0,0 +1,262 @@
#!/usr/bin/python3
import time
from flask import url_for
from .util import live_server_setup, wait_for_all_checks, extract_rss_token_from_UI, get_UUID_for_tag_name
import os
def test_setup(client, live_server):
live_server_setup(live_server)
def set_original_response():
test_return_data = """<html>
<body>
Some initial text<br>
<p id="only-this">Should be only this</p>
<br>
<p id="not-this">And never this</p>
</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 id="only-this">Should be REALLY only this</p>
<br>
<p id="not-this">And never this</p>
</body>
</html>
"""
with open("test-datastore/endpoint-content.txt", "w") as f:
f.write(test_return_data)
return None
def test_setup_group_tag(client, live_server):
#live_server_setup(live_server)
set_original_response()
# Add a tag with some config, import a tag and it should roughly work
res = client.post(
url_for("tags.form_tag_add"),
data={"name": "test-tag"},
follow_redirects=True
)
assert b"Tag added" in res.data
assert b"test-tag" in res.data
res = client.post(
url_for("tags.form_tag_edit_submit", uuid="first"),
data={"name": "test-tag",
"include_filters": '#only-this',
"subtractive_selectors": '#not-this'},
follow_redirects=True
)
assert b"Updated" in res.data
tag_uuid = get_UUID_for_tag_name(client, name="test-tag")
res = client.get(
url_for("tags.form_tag_edit", uuid="first")
)
assert b"#only-this" in res.data
assert b"#not-this" in res.data
# Tag should be setup and ready, now add a watch
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("import_page"),
data={"urls": test_url + "?first-imported=1 test-tag, extra-import-tag"},
follow_redirects=True
)
assert b"1 Imported" in res.data
res = client.get(url_for("index"))
assert b'import-tag' in res.data
assert b'extra-import-tag' in res.data
res = client.get(
url_for("tags.tags_overview_page"),
follow_redirects=True
)
assert b'import-tag' in res.data
assert b'extra-import-tag' in res.data
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'Warning, no filters were found' not in res.data
res = client.get(
url_for("preview_page", uuid="first"),
follow_redirects=True
)
assert b'Should be only this' in res.data
assert b'And never this' not in res.data
# RSS Group tag filter
# An extra one that should be excluded
res = client.post(
url_for("import_page"),
data={"urls": test_url + "?should-be-excluded=1 some-tag"},
follow_redirects=True
)
assert b"1 Imported" in res.data
wait_for_all_checks(client)
set_modified_response()
res = client.get(url_for("form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
rss_token = extract_rss_token_from_UI(client)
res = client.get(
url_for("rss", token=rss_token, tag="extra-import-tag", _external=True),
follow_redirects=True
)
assert b"should-be-excluded" not in res.data
assert res.status_code == 200
assert b"first-imported=1" in res.data
res = client.get(url_for("form_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data
def test_tag_import_singular(client, live_server):
#live_server_setup(live_server)
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("import_page"),
data={"urls": test_url + " test-tag, test-tag\r\n"+ test_url + "?x=1 test-tag, test-tag\r\n"},
follow_redirects=True
)
assert b"2 Imported" in res.data
res = client.get(
url_for("tags.tags_overview_page"),
follow_redirects=True
)
# Should be only 1 tag because they both had the same
assert res.data.count(b'test-tag') == 1
res = client.get(url_for("form_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data
def test_tag_add_in_ui(client, live_server):
#live_server_setup(live_server)
#
res = client.post(
url_for("tags.form_tag_add"),
data={"name": "new-test-tag"},
follow_redirects=True
)
assert b"Tag added" in res.data
assert b"new-test-tag" in res.data
res = client.get(url_for("form_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data
def test_group_tag_notification(client, live_server):
#live_server_setup(live_server)
set_original_response()
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tags": 'test-tag, other-tag'},
follow_redirects=True
)
assert b"Watch added" in res.data
notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'json')
notification_form_data = {"notification_urls": notification_url,
"notification_title": "New GROUP TAG 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",
"title": "test-tag"}
res = client.post(
url_for("tags.form_tag_edit_submit", uuid=get_UUID_for_tag_name(client, name="test-tag")),
data=notification_form_data,
follow_redirects=True
)
assert b"Updated" in res.data
wait_for_all_checks(client)
set_modified_response()
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(3)
assert os.path.isfile("test-datastore/notification.txt")
# 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 test_url in notification_submission
assert ':-)' in notification_submission
assert "Diff Full: Some initial text" in notification_submission
assert "New GROUP TAG ChangeDetection.io" in notification_submission
assert "test-tag" in notification_submission
assert "other-tag" in notification_submission
res = client.get(url_for("form_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data
#@todo Test that multiple notifications fired
#@todo Test that each of multiple notifications with different settings
def test_limit_tag_ui(client, live_server):
#live_server_setup(live_server)
test_url = url_for('test_endpoint', _external=True)
urls=[]
for i in range(20):
urls.append(test_url+"?x="+str(i)+" test-tag")
for i in range(20):
urls.append(test_url+"?non-grouped="+str(i))
res = client.post(
url_for("import_page"),
data={"urls": "\r\n".join(urls)},
follow_redirects=True
)
assert b"40 Imported" in res.data
res = client.get(url_for("index"))
assert b'test-tag' in res.data
# All should be here
assert res.data.count(b'processor-text_json_diff') == 40
tag_uuid = get_UUID_for_tag_name(client, name="test-tag")
res = client.get(url_for("index", tag=tag_uuid))
# Just a subset should be here
assert b'test-tag' in res.data
assert res.data.count(b'processor-text_json_diff') == 20
assert b"object at" not in res.data

20
changedetectionio/tests/test_ignore_text.py
View File

@@ -2,7 +2,7 @@
import time
from flask import url_for
from . util import live_server_setup
from .util import live_server_setup, wait_for_all_checks
from changedetectionio import html_tools
def test_setup(live_server):
@@ -84,7 +84,6 @@ def set_modified_ignore_response():
def test_check_ignore_text_functionality(client, live_server):
sleep_time_for_fetch_thread = 3
# Use a mix of case in ZzZ to prove it works case-insensitive.
ignore_text = "XXXXX\r\nYYYYY\r\nzZzZZ\r\nnew ignore stuff"
@@ -103,7 +102,7 @@ def test_check_ignore_text_functionality(client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
@@ -124,7 +123,7 @@ def test_check_ignore_text_functionality(client, live_server):
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
@@ -137,7 +136,7 @@ def test_check_ignore_text_functionality(client, live_server):
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
@@ -151,7 +150,7 @@ def test_check_ignore_text_functionality(client, live_server):
# Just to be sure.. set a regular modified change..
set_modified_original_ignore_response()
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data
@@ -167,7 +166,6 @@ def test_check_ignore_text_functionality(client, live_server):
assert b'Deleted' in res.data
def test_check_global_ignore_text_functionality(client, live_server):
sleep_time_for_fetch_thread = 3
# Give the endpoint time to spin up
time.sleep(1)
@@ -198,7 +196,7 @@ def test_check_global_ignore_text_functionality(client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Goto the edit page of the item, add our ignore text
@@ -220,7 +218,7 @@ def test_check_global_ignore_text_functionality(client, live_server):
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# so that we are sure everything is viewed and in a known 'nothing changed' state
res = client.get(url_for("diff_history_page", uuid="first"))
@@ -237,7 +235,7 @@ def test_check_global_ignore_text_functionality(client, live_server):
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
@@ -247,7 +245,7 @@ def test_check_global_ignore_text_functionality(client, live_server):
# Just to be sure.. set a regular modified change that will trigger it
set_modified_original_ignore_response()
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data

16
changedetectionio/tests/test_ignorestatuscode.py
View File

@@ -2,7 +2,7 @@
import time
from flask import url_for
from . util import live_server_setup
from .util import live_server_setup, wait_for_all_checks
def test_setup(live_server):
@@ -40,7 +40,7 @@ def set_some_changed_response():
def test_normal_page_check_works_with_ignore_status_code(client, live_server):
sleep_time_for_fetch_thread = 3
# Give the endpoint time to spin up
time.sleep(1)
@@ -68,15 +68,15 @@ def test_normal_page_check_works_with_ignore_status_code(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
set_some_changed_response()
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
@@ -109,13 +109,13 @@ def test_403_page_check_works_with_ignore_status_code(client, live_server):
# Add our URL to the import page
res = client.post(
url_for("edit_page", uuid="first"),
data={"ignore_status_codes": "y", "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"ignore_status_codes": "y", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# Make a change
set_some_changed_response()
@@ -123,7 +123,7 @@ def test_403_page_check_works_with_ignore_status_code(client, live_server):
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
# It should have 'unviewed' still
# Because it should be looking at only that 'sametext' id

2
changedetectionio/tests/test_jinja2.py
View File

@@ -20,7 +20,7 @@ def test_jinja2_in_url_query(client, live_server):
"date={% now 'Europe/Berlin', '%Y' %}.{% now 'Europe/Berlin', '%m' %}.{% now 'Europe/Berlin', '%d' %}", )
res = client.post(
url_for("form_quick_watch_add"),
data={"url": full_url, "tag": "test"},
data={"url": full_url, "tags": "test"},
follow_redirects=True
)
assert b"Watch added" in res.data

32
changedetectionio/tests/test_jsonpath_jq_selector.py
View File

@@ -208,7 +208,7 @@ def test_check_json_without_filter(client, live_server):
)
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
res = client.get(
url_for("preview_page", uuid="first"),
@@ -238,7 +238,7 @@ def check_json_filter(json_filter, client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
@@ -246,7 +246,7 @@ def check_json_filter(json_filter, client, live_server):
url_for("edit_page", uuid="first"),
data={"include_filters": json_filter,
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
"fetch_backend": "html_requests"
},
@@ -261,14 +261,14 @@ def check_json_filter(json_filter, client, live_server):
assert bytes(escape(json_filter).encode('utf-8')) in res.data
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
# Make a change
set_modified_response()
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(4)
wait_for_all_checks(client)
# It should have 'unviewed' still
res = client.get(url_for("index"))
@@ -306,14 +306,14 @@ def check_json_filter_bool_val(json_filter, client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(3)
wait_for_all_checks(client)
# 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={"include_filters": json_filter,
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
"fetch_backend": "html_requests"
},
@@ -322,14 +322,14 @@ def check_json_filter_bool_val(json_filter, client, live_server):
assert b"Updated watch." in res.data
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
# Make a change
set_modified_response()
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
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
@@ -366,7 +366,7 @@ def check_json_ext_filter(json_filter, client, live_server):
assert b"1 Imported" in res.data
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
# Goto the edit page, add our ignore text
# Add our URL to the import page
@@ -374,7 +374,7 @@ def check_json_ext_filter(json_filter, client, live_server):
url_for("edit_page", uuid="first"),
data={"include_filters": json_filter,
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
"fetch_backend": "html_requests"
},
@@ -389,14 +389,14 @@ def check_json_ext_filter(json_filter, client, live_server):
assert bytes(escape(json_filter).encode('utf-8')) in res.data
# Give the thread time to pick it up
time.sleep(3)
wait_for_all_checks(client)
# Make a change
set_modified_ext_response()
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
time.sleep(4)
wait_for_all_checks(client)
# It should have 'unviewed'
res = client.get(url_for("index"))
@@ -428,14 +428,14 @@ def test_ignore_json_order(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(2)
wait_for_all_checks(client)
with open("test-datastore/endpoint-content.txt", "w") as f:
f.write('{"world" : 123, "hello": 123}')
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(2)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' not in res.data
@@ -446,7 +446,7 @@ def test_ignore_json_order(client, live_server):
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(2)
wait_for_all_checks(client)
res = client.get(url_for("index"))
assert b'unviewed' in res.data

29
changedetectionio/tests/test_notification.py
View File

@@ -3,7 +3,7 @@ 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
from .util import set_original_response, set_modified_response, set_more_modified_response, live_server_setup, wait_for_all_checks
from . util import extract_UUID_from_client
import logging
import base64
@@ -21,10 +21,11 @@ def test_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()
# Give the endpoint time to spin up
time.sleep(3)
time.sleep(1)
# Re 360 - new install should have defaults set
res = client.get(url_for("settings_page"))
@@ -62,13 +63,13 @@ def test_check_notification(client, live_server):
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": ''},
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
time.sleep(3)
wait_for_all_checks(client)
# We write the PNG to disk, but a JPEG should appear in the notification
# Write the last screenshot png
@@ -105,7 +106,7 @@ def test_check_notification(client, live_server):
notification_form_data.update({
"url": test_url,
"tag": "my tag",
"tags": "my tag, my second tag",
"title": "my title",
"headers": "",
"fetch_backend": "html_requests"})
@@ -128,7 +129,7 @@ def test_check_notification(client, live_server):
## Now recheck, and it should have sent the notification
time.sleep(3)
wait_for_all_checks(client)
set_modified_response()
# Trigger a check
@@ -150,7 +151,7 @@ def test_check_notification(client, live_server):
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" 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
@@ -193,11 +194,11 @@ def test_check_notification(client, live_server):
# Trigger a check
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(1)
wait_for_all_checks(client)
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(1)
wait_for_all_checks(client)
client.get(url_for("form_watch_checknow"), follow_redirects=True)
time.sleep(1)
wait_for_all_checks(client)
assert os.path.exists("test-datastore/notification.txt") == False
res = client.get(url_for("notification_logs"))
@@ -209,7 +210,7 @@ def test_check_notification(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "my tag",
"tags": "my tag",
"title": "my title",
"notification_urls": '',
"notification_title": '',
@@ -243,7 +244,7 @@ def test_notification_validation(client, live_server):
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": 'nice one'},
data={"url": test_url, "tags": 'nice one'},
follow_redirects=True
)
@@ -303,13 +304,13 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server):
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": 'nice one'},
data={"url": test_url, "tags": 'nice one'},
follow_redirects=True
)
assert b"Watch added" in res.data
time.sleep(2)
wait_for_all_checks(client)
set_modified_response()
client.get(url_for("form_watch_checknow"), follow_redirects=True)

4
changedetectionio/tests/test_notification_errors.py
View File

@@ -17,7 +17,7 @@ def test_check_notification_error_handling(client, live_server):
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": ''},
data={"url": test_url, "tags": ''},
follow_redirects=True
)
assert b"Watch added" in res.data
@@ -32,7 +32,7 @@ def test_check_notification_error_handling(client, live_server):
"notification_body": "xxxxx",
"notification_format": "Text",
"url": test_url,
"tag": "",
"tags": "",
"title": "",
"headers": "",
"time_between_check-minutes": "180",

32
changedetectionio/tests/test_request.py
View File

@@ -25,7 +25,7 @@ def test_headers_in_request(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(1)
wait_for_all_checks(client)
res = client.post(
url_for("import_page"),
@@ -43,7 +43,7 @@ def test_headers_in_request(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "",
"tags": "",
"fetch_backend": 'html_webdriver' if os.getenv('PLAYWRIGHT_DRIVER_URL') else 'html_requests',
"headers": "xxx:ooo\ncool:yeah\r\ncookie:"+cookie_header},
follow_redirects=True
@@ -95,14 +95,14 @@ def test_body_in_request(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(3)
wait_for_all_checks(client)
# add the first 'version'
res = client.post(
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "",
"tags": "",
"method": "POST",
"fetch_backend": "html_requests",
"body": "something something"},
@@ -110,7 +110,7 @@ def test_body_in_request(client, live_server):
)
assert b"Updated watch." in res.data
time.sleep(3)
wait_for_all_checks(client)
# Now the change which should trigger a change
body_value = 'Test Body Value'
@@ -118,7 +118,7 @@ def test_body_in_request(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "",
"tags": "",
"method": "POST",
"fetch_backend": "html_requests",
"body": body_value},
@@ -126,7 +126,7 @@ def test_body_in_request(client, live_server):
)
assert b"Updated watch." in res.data
time.sleep(3)
wait_for_all_checks(client)
# The service should echo back the body
res = client.get(
@@ -163,7 +163,7 @@ def test_body_in_request(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "",
"tags": "",
"method": "GET",
"fetch_backend": "html_requests",
"body": "invalid"},
@@ -187,7 +187,7 @@ def test_method_in_request(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(2)
wait_for_all_checks(client)
res = client.post(
url_for("import_page"),
data={"urls": test_url},
@@ -195,14 +195,14 @@ def test_method_in_request(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(2)
wait_for_all_checks(client)
# Attempt to add a method which is not valid
res = client.post(
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "",
"tags": "",
"fetch_backend": "html_requests",
"method": "invalid"},
follow_redirects=True
@@ -214,7 +214,7 @@ def test_method_in_request(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "",
"tags": "",
"fetch_backend": "html_requests",
"method": "PATCH"},
follow_redirects=True
@@ -222,7 +222,7 @@ def test_method_in_request(client, live_server):
assert b"Updated watch." in res.data
# Give the thread time to pick up the first version
time.sleep(2)
wait_for_all_checks(client)
# The service should echo back the request verb
res = client.get(
@@ -233,7 +233,7 @@ def test_method_in_request(client, live_server):
# The test call service will return the verb as the body
assert b"PATCH" in res.data
time.sleep(2)
wait_for_all_checks(client)
watches_with_method = 0
with open('test-datastore/url-watches.json') as f:
@@ -265,7 +265,7 @@ def test_headers_textfile_in_request(client, live_server):
)
assert b"1 Imported" in res.data
time.sleep(1)
wait_for_all_checks(client)
# Add some headers to a request
@@ -273,7 +273,7 @@ def test_headers_textfile_in_request(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": test_url,
"tag": "testtag",
"tags": "testtag",
"fetch_backend": 'html_webdriver' if os.getenv('PLAYWRIGHT_DRIVER_URL') else 'html_requests',
"headers": "xxx:ooo\ncool:yeah\r\n"},
follow_redirects=True

4
changedetectionio/tests/test_search.py
View File

@@ -28,7 +28,7 @@ def test_basic_search(client, live_server):
res = client.post(
url_for("edit_page", uuid="first"),
data={"title": "xxx-title", "url": urls[0], "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"title": "xxx-title", "url": urls[0], "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data
@@ -62,7 +62,7 @@ def test_search_in_tag_limit(client, live_server):
# By Title
res = client.post(
url_for("edit_page", uuid="first"),
data={"title": "xxx-title", "url": urls[0].split(' ')[0], "tag": urls[0].split(' ')[1], "headers": "",
data={"title": "xxx-title", "url": urls[0].split(' ')[0], "tags": urls[0].split(' ')[1], "headers": "",
'fetch_backend': "html_requests"},
follow_redirects=True
)

10
changedetectionio/tests/test_security.py
View File

@@ -18,7 +18,7 @@ def test_bad_access(client, live_server):
url_for("edit_page", uuid="first"),
data={
"url": 'javascript:alert(document.domain)',
"tag": "",
"tags": "",
"method": "GET",
"fetch_backend": "html_requests",
"body": ""},
@@ -29,7 +29,7 @@ def test_bad_access(client, live_server):
res = client.post(
url_for("form_quick_watch_add"),
data={"url": ' javascript:alert(123)', "tag": ''},
data={"url": ' javascript:alert(123)', "tags": ''},
follow_redirects=True
)
@@ -37,7 +37,7 @@ def test_bad_access(client, live_server):
res = client.post(
url_for("form_quick_watch_add"),
data={"url": '%20%20%20javascript:alert(123)%20%20', "tag": ''},
data={"url": '%20%20%20javascript:alert(123)%20%20', "tags": ''},
follow_redirects=True
)
@@ -46,7 +46,7 @@ def test_bad_access(client, live_server):
res = client.post(
url_for("form_quick_watch_add"),
data={"url": ' source:javascript:alert(document.domain)', "tag": ''},
data={"url": ' source:javascript:alert(document.domain)', "tags": ''},
follow_redirects=True
)
@@ -56,7 +56,7 @@ def test_bad_access(client, live_server):
client.post(
url_for("form_quick_watch_add"),
data={"url": 'file:///tasty/disk/drive', "tag": ''},
data={"url": 'file:///tasty/disk/drive', "tags": ''},
follow_redirects=True
)
time.sleep(1)

2
changedetectionio/tests/test_share_watch.py
View File

@@ -29,7 +29,7 @@ def test_share_watch(client, live_server):
# Add our URL to the import page
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": include_filters, "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": include_filters, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data

10
changedetectionio/tests/test_source.py
View File

@@ -3,7 +3,7 @@
import time
from flask import url_for
from urllib.request import urlopen
from .util import set_original_response, set_modified_response, live_server_setup
from .util import set_original_response, set_modified_response, live_server_setup, wait_for_all_checks
sleep_time_for_fetch_thread = 3
@@ -42,7 +42,7 @@ def test_check_basic_change_detection_functionality_source(client, live_server):
res = client.get(url_for("form_watch_checknow"), follow_redirects=True)
assert b'1 watches queued for rechecking.' in res.data
time.sleep(5)
wait_for_all_checks(client)
# Now something should be ready, indicated by having a 'unviewed' class
res = client.get(url_for("index"))
@@ -60,7 +60,7 @@ def test_check_basic_change_detection_functionality_source(client, live_server):
# `subtractive_selectors` should still work in `source:` type requests
def test_check_ignore_elements(client, live_server):
set_original_response()
time.sleep(2)
time.sleep(1)
test_url = 'source:'+url_for('test_endpoint', _external=True)
# Add our URL to the import page
res = client.post(
@@ -71,14 +71,14 @@ def test_check_ignore_elements(client, live_server):
assert b"1 Imported" in res.data
time.sleep(sleep_time_for_fetch_thread)
wait_for_all_checks(client)
#####################
# We want <span> and <p> ONLY, but ignore span with .foobar-detection
client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": 'span,p', "url": test_url, "tag": "", "subtractive_selectors": ".foobar-detection", 'fetch_backend': "html_requests"},
data={"include_filters": 'span,p', "url": test_url, "tags": "", "subtractive_selectors": ".foobar-detection", 'fetch_backend': "html_requests"},
follow_redirects=True
)

2
changedetectionio/tests/test_watch_fields_storage.py
View File

@@ -26,7 +26,7 @@ def test_check_watch_field_storage(client, live_server):
"title" : "My title",
"ignore_text" : "ignore this",
"url": test_url,
"tag": "woohoo",
"tags": "woohoo",
"headers": "curl:foo",
'fetch_backend': "html_requests"
},

10
changedetectionio/tests/test_xpath_selector.py
View File

@@ -89,7 +89,7 @@ def test_check_xpath_filter_utf8(client, live_server):
time.sleep(1)
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": filter, "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": filter, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data
@@ -143,7 +143,7 @@ def test_check_xpath_text_function_utf8(client, live_server):
time.sleep(1)
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": filter, "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": filter, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data
@@ -189,7 +189,7 @@ def test_check_markup_xpath_filter_restriction(client, live_server):
# Add our URL to the import page
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": xpath_filter, "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": xpath_filter, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Updated watch." in res.data
@@ -231,7 +231,7 @@ def test_xpath_validation(client, live_server):
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": "/something horrible", "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": "/something horrible", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"is not a valid XPath expression" in res.data
@@ -261,7 +261,7 @@ def test_check_with_prefix_include_filters(client, live_server):
res = client.post(
url_for("edit_page", uuid="first"),
data={"include_filters": "xpath://*[contains(@class, 'sametext')]", "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"},
data={"include_filters": "xpath://*[contains(@class, 'sametext')]", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests"},
follow_redirects=True
)

10
changedetectionio/tests/util.py
View File

@@ -70,6 +70,16 @@ def extract_api_key_from_UI(client):
api_key = m.group(1)
return api_key.strip()
# kinda funky, but works for now
def get_UUID_for_tag_name(client, name):
app_config = client.application.config.get('DATASTORE').data
for uuid, tag in app_config['settings']['application'].get('tags', {}).items():
if name == tag.get('title', '').lower().strip():
return uuid
return None
# kinda funky, but works for now
def extract_rss_token_from_UI(client):
import re

4
changedetectionio/tests/visualselector/test_fetch_data.py
View File

@@ -19,7 +19,7 @@ def test_visual_selector_content_ready(client, live_server):
res = client.post(
url_for("form_quick_watch_add"),
data={"url": test_url, "tag": '', 'edit_and_watch_submit_button': 'Edit > Watch'},
data={"url": test_url, "tags": '', 'edit_and_watch_submit_button': 'Edit > Watch'},
follow_redirects=True
)
assert b"Watch added in Paused state, saving will unpause" in res.data
@@ -28,7 +28,7 @@ def test_visual_selector_content_ready(client, live_server):
url_for("edit_page", uuid="first", unpause_on_save=1),
data={
"url": test_url,
"tag": "",
"tags": "",
"headers": "",
'fetch_backend': "html_webdriver",
'webdriver_js_execute_code': 'document.querySelector("button[name=test-button]").click();'