lee/changedetection.io
1
0
Fork 0
mirror of https://github.com/dgtlmoon/changedetection.io.git synced 2026-07-08 08:21:14 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

test improvements

Browse Source
This commit is contained in:
dgtlmoon
2026-02-01 18:18:22 +01:00
parent 424e4ec1aa
commit cf32bf5f47
62 changed files with 271 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changedetectionio/blueprint/tags/__init__.py
-4
View File
@@ -137,8 +137,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
@login_optionally_required
def form_tag_edit(uuid):
from changedetectionio.blueprint.tags.form import group_restock_settings_form
if uuid == 'first':
uuid = list(datastore.data['settings']['application']['tags'].keys()).pop()
default = datastore.data['settings']['application']['tags'].get(uuid)
if not default:
@@ -199,8 +197,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
@login_optionally_required
def form_tag_edit_submit(uuid):
from changedetectionio.blueprint.tags.form import group_restock_settings_form
if uuid == 'first':
uuid = list(datastore.data['settings']['application']['tags'].keys()).pop()
default = datastore.data['settings']['application']['tags'].get(uuid)
changedetectionio/blueprint/ui/__init__.py
+1 -9
View File
@@ -227,9 +227,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_pool,
flash(gettext('The watch by UUID {} does not exist.').format(uuid), 'error')
return redirect(url_for('watchlist.index'))
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
datastore.delete(uuid)
flash(gettext('Deleted.'))
@@ -239,9 +236,7 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_pool,
@login_optionally_required
def form_clone():
uuid = request.args.get('uuid')
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
new_uuid = datastore.clone(uuid)
@@ -367,9 +362,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, worker_pool,
import json
from copy import deepcopy
# more for testing
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
# copy it to memory as trim off what we dont need (history)
watch = deepcopy(datastore.data['watching'].get(uuid))
changedetectionio/blueprint/ui/diff.py
+3 -12
View File
@@ -83,9 +83,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
If a processor doesn't have a difference module, falls back to text_json_diff.
"""
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
try:
watch = datastore.data['watching'][uuid]
@@ -144,9 +141,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
Each processor implements processors/{type}/extract.py::render_form()
If a processor doesn't have an extract module, falls back to text_json_diff.
"""
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
try:
watch = datastore.data['watching'][uuid]
@@ -200,9 +195,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
Each processor implements processors/{type}/extract.py::process_extraction()
If a processor doesn't have an extract module, falls back to text_json_diff.
"""
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
try:
watch = datastore.data['watching'][uuid]
@@ -267,9 +260,7 @@ def construct_blueprint(datastore: ChangeDetectionStore):
- /diff/{uuid}/processor-asset/after
- /diff/{uuid}/processor-asset/rendered_diff
"""
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
try:
watch = datastore.data['watching'][uuid]
changedetectionio/blueprint/ui/edit.py
-3
View File
@@ -35,9 +35,6 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, queuedWatchMe
flash(gettext("No watches to edit"), "error")
return redirect(url_for('watchlist.index'))
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
if not uuid in datastore.data['watching']:
flash(gettext("No watch with the UUID {} found.").format(uuid), "error")
return redirect(url_for('watchlist.index'))
changedetectionio/blueprint/ui/preview.py
-7
View File
@@ -26,9 +26,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
Each processor implements processors/{type}/preview.py::render()
If a processor doesn't have a preview module, falls back to default text preview.
"""
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
try:
watch = datastore.data['watching'][uuid]
@@ -150,10 +147,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
"""
from flask import make_response
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
try:
watch = datastore.data['watching'][uuid]
except KeyError:
changedetectionio/tests/conftest.py
+1 -1
View File
@@ -162,7 +162,7 @@ def prepare_test_function(live_server, datastore_path):
CRITICAL for xdist per-test isolation:
- Reuses the SAME datastore instance (so blueprint references stay valid)
- Clears all watches and state for a clean slate
- First watch will get uuid="first"
- First watch will get uuid=uuid
"""
routes = [rule.rule for rule in live_server.app.url_map.iter_rules()]
if '/test-random-content-endpoint' not in routes:
changedetectionio/tests/custom_browser_url/test_custom_browser_url.py
+3 -3
View File
@@ -37,13 +37,13 @@ def do_test(client, live_server, make_test_use_extra_browser=False):
# So the name should appear in the edit page under "Request" > "Fetch Method"
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
follow_redirects=True
)
assert b'custom browser URL' in res.data
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
# 'run_customer_browser_url_tests.sh' will search for this string to know if we hit the right browser container or not
"url": "https://changedetection.io/ci-test.html?custom-browser-search-string=1",
@@ -66,7 +66,7 @@ def do_test(client, live_server, make_test_use_extra_browser=False):
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'cool it works' in res.data
changedetectionio/tests/fetchers/test_content.py
+1 -1
View File
@@ -36,7 +36,7 @@ def test_fetch_webdriver_content(client, live_server, measure_memory_usage, data
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
logging.getLogger().info("Looking for correct fetched HTML (text) from server")
changedetectionio/tests/fetchers/test_custom_js_before_content.py
+1 -1
View File
@@ -21,7 +21,7 @@ def test_execute_custom_js(client, live_server, measure_memory_usage, datastore_
assert b"Watch added in Paused state, saving will unpause" in res.data
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data={
"url": test_url,
"tags": "",
changedetectionio/tests/proxy_list/test_multiple_proxy.py
+1 -1
View File
@@ -19,7 +19,7 @@ def test_preferred_proxy(client, live_server, measure_memory_usage, datastore_pa
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data={
"include_filters": "",
"fetch_backend": 'html_webdriver' if os.getenv('PLAYWRIGHT_DRIVER_URL') else 'html_requests',
changedetectionio/tests/proxy_list/test_noproxy.py
+1 -1
View File
@@ -13,7 +13,7 @@ def test_noproxy_option(client, live_server, measure_memory_usage, datastore_pat
# Should only be available when a proxy is setup
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1))
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1))
assert b'No proxy' not in res.data
# Setup a proxy
changedetectionio/tests/proxy_list/test_proxy_noconnect.py
+1 -1
View File
@@ -48,7 +48,7 @@ def test_proxy_noconnect_custom(client, live_server, measure_memory_usage, datas
}
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data=options,
follow_redirects=True
)
changedetectionio/tests/proxy_list/test_select_custom_proxy.py
+1 -1
View File
@@ -40,7 +40,7 @@ def test_select_custom(client, live_server, measure_memory_usage, datastore_path
assert b'Proxy Authentication Required' not in res.data
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
# We should see something via proxy
changedetectionio/tests/proxy_socks5/test_socks5_proxy.py
+3 -3
View File
@@ -52,13 +52,13 @@ def test_socks5(client, live_server, measure_memory_usage, datastore_path):
assert b"Watch added in Paused state, saving will unpause" in res.data
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
)
# check the proxy is offered as expected
assert b'ui-0socks5proxy' in res.data
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data={
"include_filters": "",
"fetch_backend": 'html_webdriver' if os.getenv('PLAYWRIGHT_DRIVER_URL') else 'html_requests',
@@ -74,7 +74,7 @@ def test_socks5(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/proxy_socks5/test_socks5_proxy_sources.py
+3 -3
View File
@@ -39,13 +39,13 @@ def test_socks5_from_proxiesjson_file(client, live_server, measure_memory_usage,
assert b"Watch added in Paused state, saving will unpause" in res.data
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
)
# check the proxy is offered as expected
assert b'name="proxy" type="radio" value="socks5proxy"' in res.data
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data={
"include_filters": "",
"fetch_backend": 'html_webdriver' if os.getenv('PLAYWRIGHT_DRIVER_URL') else 'html_requests',
@@ -61,7 +61,7 @@ def test_socks5_from_proxiesjson_file(client, live_server, measure_memory_usage,
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/smtp/test_notification_smtp.py
+1 -1
View File
@@ -416,7 +416,7 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
# Now override as HTML format
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"notification_format": 'html',
changedetectionio/tests/test_access_control.py
+3 -9
View File
@@ -12,13 +12,7 @@ def test_check_access_control(app, client, live_server, measure_memory_usage, da
assert b"Remove password" not in res.data
# add something that we can hit via diff page later
res = c.post(
url_for("imports.import_page"),
data={"urls": url_for('test_random_content_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=url_for('test_random_content_endpoint', _external=True))
# causes a 'Popped wrong request context.' error when client. is accessed?
wait_for_all_checks(client)
@@ -45,7 +39,7 @@ def test_check_access_control(app, client, live_server, measure_memory_usage, da
assert b"Login" in res.data
# The diff page should return something valid when logged out
res = c.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = c.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
assert b'Random content' in res.data
# access to assets should work (check_authentication)
@@ -181,5 +175,5 @@ def test_check_access_control(app, client, live_server, measure_memory_usage, da
assert res.status_code == 403
# The diff page should return something valid when logged out
res = c.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = c.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
assert b'Random content' not in res.data
changedetectionio/tests/test_add_replace_remove_filter.py
+2 -2
View File
@@ -57,7 +57,7 @@ def test_check_removed_line_contains_trigger(client, live_server, measure_memory
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": 'The golden line',
"url": test_url,
'fetch_backend': "html_requests",
@@ -145,7 +145,7 @@ def test_check_add_line_contains_trigger(client, live_server, measure_memory_usa
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": 'Oh yes please',
"url": test_url,
'processor': 'text_json_diff',
changedetectionio/tests/test_api.py
+1 -1
View File
@@ -253,7 +253,7 @@ def test_api_simple(client, live_server, measure_memory_usage, datastore_path):
assert watch.get('viewed') == False
# Loading the most recent snapshot should force viewed to become true
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"), follow_redirects=True)
client.get(url_for("ui.ui_diff.diff_history_page", uuid=watch_uuid), follow_redirects=True)
time.sleep(3)
# Fetch the whole watch again, viewed should be true
changedetectionio/tests/test_auth.py
+2 -2
View File
@@ -18,7 +18,7 @@ def test_basic_auth(client, live_server, measure_memory_usage, datastore_path):
time.sleep(1)
# Check form validation
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -26,7 +26,7 @@ def test_basic_auth(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_backend.py
+6 -6
View File
@@ -44,7 +44,7 @@ def test_check_basic_change_detection_functionality(client, live_server, measure
# Check HTML conversion detected and workd
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
# Check this class does not appear (that we didnt see the actual source)
@@ -90,7 +90,7 @@ def test_check_basic_change_detection_functionality(client, live_server, measure
# Check the [preview] pulls the right one
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'which has this one new line' in res.data
@@ -272,7 +272,7 @@ got it\r\n
### check the front end
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b"some random text that should be split by line\n" in res.data
@@ -332,7 +332,7 @@ got it\r\n
### check the front end
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -377,7 +377,7 @@ def test_plaintext_even_if_xml_content(client, live_server, measure_memory_usage
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -404,7 +404,7 @@ def test_plaintext_even_if_xml_content_and_can_apply_filters(client, live_server
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_clone.py
+1 -1
View File
@@ -23,7 +23,7 @@ def test_clone_functionality(client, live_server, measure_memory_usage, datastor
time.sleep(1)
res = client.get(
url_for("ui.form_clone", uuid="first"),
url_for("ui.form_clone", uuid=uuid),
follow_redirects=True
)
existing_uuids = set()
changedetectionio/tests/test_conditions.py
+2 -2
View File
@@ -69,7 +69,7 @@ def test_conditions_with_text_and_number(client, live_server, measure_memory_usa
# 1. The page filtered text must contain "5" (first digit of value)
# 2. The extracted number should be >= 20 and <= 100
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"fetch_backend": "html_requests",
@@ -230,7 +230,7 @@ def test_wordcount_conditions_plugin(client, live_server, measure_memory_usage,
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
# Assert the word count is counted correctly
changedetectionio/tests/test_css_selector.py
+5 -5
View File
@@ -91,7 +91,7 @@ def test_check_markup_include_filters_restriction(client, live_server, measure_m
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": include_filters, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -99,7 +99,7 @@ def test_check_markup_include_filters_restriction(client, live_server, measure_m
time.sleep(1)
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
assert bytes(include_filters.encode('utf-8')) in res.data
@@ -142,7 +142,7 @@ def test_check_multiple_filters(client, live_server, measure_memory_usage, datas
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": include_filters,
"url": test_url,
"tags": "",
@@ -157,7 +157,7 @@ def test_check_multiple_filters(client, live_server, measure_memory_usage, datas
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -193,7 +193,7 @@ def test_filter_is_empty_help_suggestion(client, live_server, measure_memory_usa
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": include_filters,
"url": test_url,
"tags": "",
changedetectionio/tests/test_element_removal.py
+6 -8
View File
@@ -155,17 +155,15 @@ def test_element_removal_full(client, live_server, measure_memory_usage, datasto
# Add our URL to the import page
test_url = url_for("test_endpoint", _external=True)
res = client.post(
url_for("imports.import_page"), data={"urls": test_url}, follow_redirects=True
)
assert b"1 Imported" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
wait_for_all_checks(client)
# Goto the edit page, add the filter data
# Not sure why \r needs to be added - absent of the #changetext this is not necessary
subtractive_selectors_data = "header\r\nfooter\r\nnav\r\n#changetext"
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"subtractive_selectors": subtractive_selectors_data,
"url": test_url,
@@ -181,7 +179,7 @@ def test_element_removal_full(client, live_server, measure_memory_usage, datasto
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
assert bytes(subtractive_selectors_data.encode("utf-8")) in res.data
@@ -192,7 +190,7 @@ def test_element_removal_full(client, live_server, measure_memory_usage, datasto
wait_for_all_checks(client)
# so that we set the state to 'has-unread-changes' after all the edits
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
# Make a change to header/footer/nav
set_modified_response(datastore_path=datastore_path)
@@ -239,7 +237,7 @@ body > table > tr:nth-child(3) > td:nth-child(3)""",
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_encoding.py
+2 -2
View File
@@ -41,7 +41,7 @@ def test_check_encoding_detection(client, live_server, measure_memory_usage, dat
assert live_server.app.config['DATASTORE'].data['watching'][uuid]['content-type'] == "text/html"
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -63,7 +63,7 @@ def test_check_encoding_detection_missing_content_type_header(client, live_serve
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_errorhandling.py
+3 -8
View File
@@ -34,7 +34,7 @@ def _runner_test_http_errors(client, live_server, http_code, expected_text, data
# Error viewing tabs should appear
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -90,13 +90,8 @@ def test_low_level_errors_clear_correctly(client, live_server, measure_memory_us
# Add our URL to the import page
test_url = url_for('test_endpoint', _external=True)
uuidA = client.application.config.get('DATASTORE').add_watch(url="https://dfkjasdkfjaidjfsdajfksdajfksdjfDOESNTEXIST.com")
res = client.post(
url_for("imports.import_page"),
data={"urls": "https://dfkjasdkfjaidjfsdajfksdajfksdjfDOESNTEXIST.com"},
follow_redirects=True
)
assert b"1 Imported" in res.data
wait_for_all_checks(client)
# We should see the DNS error
@@ -113,7 +108,7 @@ def test_low_level_errors_clear_correctly(client, live_server, measure_memory_us
# Update with what should work
client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuidA),
data={
"url": test_url,
"fetch_backend": "html_requests",
changedetectionio/tests/test_extract_csv.py
+3 -9
View File
@@ -18,15 +18,9 @@ def test_check_extract_text_from_diff(client, live_server, measure_memory_usage,
# live_server_setup(live_server) # Setup on conftest per function
# Add our URL to the import page
res = client.post(
url_for("imports.import_page"),
data={"urls": url_for('test_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=url_for('test_endpoint', _external=True))
wait_for_all_checks(client)
res = client.get(url_for("ui.ui_diff.diff_history_page_extract_GET", uuid="first"))
res = client.get(url_for("ui.ui_diff.diff_history_page_extract_GET", uuid=uuid))
assert res.status_code == 200
assert b'extract_regex' in res.data
@@ -46,7 +40,7 @@ def test_check_extract_text_from_diff(client, live_server, measure_memory_usage,
res = client.post(
url_for("ui.ui_diff.diff_history_page_extract_POST", uuid="first"),
url_for("ui.ui_diff.diff_history_page_extract_POST", uuid=uuid),
data={"extract_regex": "Now it's ([0-9\.]+)",
"extract_submit_button": "Extract as CSV"},
follow_redirects=False
changedetectionio/tests/test_extract_regex.py
+4 -4
View File
@@ -85,7 +85,7 @@ def test_check_filter_multiline(client, live_server, measure_memory_usage, datas
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": '',
# Test a regex and a plaintext
'extract_text': '/something.+?6 billion.+?lines/si\r\nand this should be',
@@ -107,7 +107,7 @@ def test_check_filter_multiline(client, live_server, measure_memory_usage, datas
assert b'not at the start of the expression' not in res.data
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
# Plaintext that doesnt look like a regex should match also
@@ -137,7 +137,7 @@ def test_check_filter_and_regex_extract(client, live_server, measure_memory_usag
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
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\r\n/issue1828.+?2022/i',
"url": test_url,
@@ -174,7 +174,7 @@ def test_check_filter_and_regex_extract(client, live_server, measure_memory_usag
# Check HTML conversion detected and workd
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_filter_exist_changes.py
+2 -9
View File
@@ -51,15 +51,8 @@ def test_filter_doesnt_exist_then_exists_should_get_notification(client, live_se
# Give the endpoint time to spin up
time.sleep(1)
set_response_without_filter(datastore_path=datastore_path)
# 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": 'cinema'},
follow_redirects=True
)
assert b"Watch added" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
# Give the thread time to pick up the first version
time.sleep(3)
@@ -100,7 +93,7 @@ def test_filter_doesnt_exist_then_exists_should_get_notification(client, live_se
"time_between_check_use_default": "y"})
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data=notification_form_data,
follow_redirects=True
)
changedetectionio/tests/test_group.py
+14 -18
View File
@@ -51,31 +51,27 @@ def test_setup_group_tag(client, live_server, measure_memory_usage, datastore_pa
)
assert b"Tag added" in res.data
assert b"test-tag" in res.data
tag_uuid = get_UUID_for_tag_name(client, name="test-tag")
res = client.post(
url_for("tags.form_tag_edit_submit", uuid="first"),
url_for("tags.form_tag_edit_submit", uuid=tag_uuid),
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")
url_for("tags.form_tag_edit", uuid=tag_uuid)
)
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("imports.import_page"),
data={"urls": test_url + "?first-imported=1 test-tag, extra-import-tag"},
follow_redirects=True
)
assert b"1 Imported" in res.data
test_url = url_for('test_endpoint', _external=True)+"?first-imported=1"
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, tag="test-tag, extra-import-tag")
res = client.get(url_for("watchlist.index"))
assert b'import-tag' in res.data
@@ -94,14 +90,14 @@ def test_setup_group_tag(client, live_server, measure_memory_usage, datastore_pa
assert b'Warning, no filters were found' not in res.data
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'Should be only this' in res.data
assert b'And never this' not in res.data
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
follow_redirects=True
)
# 2307 the UI notice should appear in the placeholder
@@ -343,17 +339,17 @@ def test_order_of_filters_tag_filter_and_watch_filter(client, live_server, measu
'#only-this',
'#only-this',
]
tag_uuid = get_UUID_for_tag_name(client, name="test-tag-keep-order")
res = client.post(
url_for("tags.form_tag_edit_submit", uuid="first"),
url_for("tags.form_tag_edit_submit", uuid=tag_uuid),
data={"name": "test-tag-keep-order",
"include_filters": '\n'.join(tag_filters) },
follow_redirects=True
)
assert b"Updated" in res.data
tag_uuid = get_UUID_for_tag_name(client, name="test-tag-keep-order")
res = client.get(
url_for("tags.form_tag_edit", uuid="first")
url_for("tags.form_tag_edit", uuid=tag_uuid)
)
assert b"#only-this" in res.data
@@ -409,7 +405,7 @@ def test_order_of_filters_tag_filter_and_watch_filter(client, live_server, measu
]
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": '\n'.join(filters),
"url": test_url,
"tags": "test-tag-keep-order",
@@ -422,7 +418,7 @@ def test_order_of_filters_tag_filter_and_watch_filter(client, live_server, measu
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_history_consistency.py
+2 -2
View File
@@ -162,7 +162,7 @@ def test_check_text_history_view(client, live_server, measure_memory_usage, data
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
assert b'test-one' in res.data
assert b'test-two' in res.data
@@ -174,7 +174,7 @@ def test_check_text_history_view(client, live_server, measure_memory_usage, data
wait_for_all_checks(client)
# It should remember the last viewed time, so the first difference is not shown
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
assert b'test-three' in res.data
assert b'test-two' in res.data
assert b'test-one' not in res.data
changedetectionio/tests/test_ignore_text.py
+5 -5
View File
@@ -107,7 +107,7 @@ def test_check_ignore_text_functionality(client, live_server, measure_memory_usa
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"ignore_text": ignore_text, "url": test_url, 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -115,7 +115,7 @@ def test_check_ignore_text_functionality(client, live_server, measure_memory_usa
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
assert bytes(ignore_text.encode('utf-8')) in res.data
@@ -131,7 +131,7 @@ def test_check_ignore_text_functionality(client, live_server, measure_memory_usa
assert b'/test-endpoint' in res.data
res = client.get(url_for("ui.ui_preview.preview_page", uuid="first"))
res = client.get(url_for("ui.ui_preview.preview_page", uuid=uuid))
# nothing ignored because none of the text matched
assert b'ignored_line_numbers = []' in res.data
@@ -158,7 +158,7 @@ def test_check_ignore_text_functionality(client, live_server, measure_memory_usa
res = client.get(url_for("watchlist.index"))
assert b'has-unread-changes' in res.data
res = client.get(url_for("ui.ui_preview.preview_page", uuid="first"))
res = client.get(url_for("ui.ui_preview.preview_page", uuid=uuid))
# SHOULD BE be in the preview, it was added in set_modified_original_ignore_response()
# and we have "new ignore stuff" in ignore_text
@@ -203,7 +203,7 @@ def _run_test_global_ignore(client, datastore_path, as_source=False, extra_ignor
#Adding some ignore text should not trigger a change
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"ignore_text": "something irrelevent but just to check", "url": test_url, 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
changedetectionio/tests/test_ignorehyperlinks.py
+3 -7
View File
@@ -62,11 +62,7 @@ def test_render_anchor_tag_content_true(client, live_server, measure_memory_usag
# Add our URL to the import page
test_url = url_for("test_endpoint", _external=True)
res = client.post(
url_for("imports.import_page"), data={"urls": test_url},
follow_redirects=True
)
assert b"1 Imported" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
wait_for_all_checks(client)
# Trigger a check
@@ -82,7 +78,7 @@ def test_render_anchor_tag_content_true(client, live_server, measure_memory_usag
wait_for_all_checks(client)
# We should not see the rendered anchor tag
res = client.get(url_for("ui.ui_preview.preview_page", uuid="first"))
res = client.get(url_for("ui.ui_preview.preview_page", uuid=uuid))
assert '(/modified_link)' not in res.data.decode()
# Goto the settings page, ENABLE render anchor tag
@@ -106,7 +102,7 @@ def test_render_anchor_tag_content_true(client, live_server, measure_memory_usag
# check that the anchor tag content is rendered
res = client.get(url_for("ui.ui_preview.preview_page", uuid="first"))
res = client.get(url_for("ui.ui_preview.preview_page", uuid=uuid))
assert '(/modified_link)' in res.data.decode()
# since the link has changed, and we chose to render anchor tag content,
changedetectionio/tests/test_ignorestatuscode.py
+1 -1
View File
@@ -100,7 +100,7 @@ def test_403_page_check_works_with_ignore_status_code(client, live_server, measu
# Goto the edit page, check our ignore option
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"ignore_status_codes": "y", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
changedetectionio/tests/test_import.py
+2 -2
View File
@@ -95,8 +95,8 @@ def test_import_distillio(client, live_server, measure_memory_usage, datastore_p
assert b"Unable to read JSON file, was it broken?" not in res.data
assert b"1 Imported from Distill.io" in res.data
res = client.get( url_for("ui.ui_edit.edit_page", uuid="first"))
uuids = list(client.application.config.get('DATASTORE').data['watching'])
res = client.get( url_for("ui.ui_edit.edit_page", uuid=uuids[0]))
assert b"https://unraid.net/blog" in res.data
assert b"Unraid | News" in res.data
changedetectionio/tests/test_jinja2.py
+6 -14
View File
@@ -20,17 +20,13 @@ def test_jinja2_in_url_query(client, live_server, measure_memory_usage, datastor
# because url_for() will URL-encode the var, but we dont here
full_url = "{}?{}".format(test_url,
"date={% now 'Europe/Berlin', '%Y' %}.{% now 'Europe/Berlin', '%m' %}.{% now 'Europe/Berlin', '%d' %}", )
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": full_url, "tags": "test"},
follow_redirects=True
)
assert b"Watch added" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=full_url)
wait_for_all_checks(client)
# It should report nothing found (no new 'has-unread-changes' class)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'date=2' in res.data
@@ -46,17 +42,13 @@ def test_jinja2_time_offset_in_url_query(client, live_server, measure_memory_usa
# This should work now with our custom TimeExtension
full_url = "{}?{}".format(test_url,
"timestamp={% now 'utc' - 'minutes=11', '%Y-%m-%d %H:%M' %}", )
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": full_url, "tags": "test"},
follow_redirects=True
)
assert b"Watch added" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=full_url)
wait_for_all_checks(client)
# Verify the URL was processed correctly (should not have errors)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
# Should have a valid timestamp in the response
changedetectionio/tests/test_jsonpath_jq_selector.py
+7 -6
View File
@@ -208,9 +208,9 @@ def test_check_json_without_filter(client, live_server, measure_memory_usage, da
# Give the thread time to pick it up
wait_for_all_checks(client)
time.sleep(1)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -289,7 +289,7 @@ def check_json_filter_bool_val(json_filter, client, live_server, datastore_path)
# Give the thread time to pick it up
wait_for_all_checks(client)
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
# But the change should be there, tho its hard to test the change was detected because it will show old and new versions
assert b'false' in res.data
@@ -353,6 +353,7 @@ def check_json_ext_filter(json_filter, client, live_server, datastore_path):
watch = live_server.app.config['DATASTORE'].data['watching'][uuid]
dates = list(watch.history.keys())
time.sleep(1)
snapshot_contents = watch.get_history_snapshot(timestamp=dates[0])
assert snapshot_contents[0] == '['
@@ -361,7 +362,7 @@ def check_json_ext_filter(json_filter, client, live_server, datastore_path):
res = client.get(url_for("watchlist.index"))
assert b'has-unread-changes' in res.data
res = client.get(url_for("ui.ui_preview.preview_page", uuid="first"))
res = client.get(url_for("ui.ui_preview.preview_page", uuid=uuid))
# We should never see 'ForSale' because we are selecting on 'Sold' in the rule,
# But we should know it triggered ('has-unread-changes' assert above)
@@ -371,7 +372,7 @@ def check_json_ext_filter(json_filter, client, live_server, datastore_path):
# And the difference should have both?
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
assert b'ForSale' in res.data
assert b'Sold' in res.data
@@ -432,7 +433,7 @@ def test_correct_header_detect(client, live_server, measure_memory_usage, datast
assert b'No parsable JSON found in this document' not in res.data
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_notification.py
+3 -3
View File
@@ -121,7 +121,7 @@ def test_check_notification(client, live_server, measure_memory_usage, datastore
"time_between_check_use_default": "y"})
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data=notification_form_data,
follow_redirects=True
)
@@ -131,7 +131,7 @@ def test_check_notification(client, live_server, measure_memory_usage, datastore
# 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"))
url_for("ui.ui_edit.edit_page", uuid=uuid))
assert bytes(notification_url.encode('utf-8')) in res.data
assert bytes("New ChangeDetection.io Notification".encode('utf-8')) in res.data
@@ -227,7 +227,7 @@ def test_check_notification(client, live_server, measure_memory_usage, datastore
set_original_response(datastore_path=datastore_path)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "my tag",
changedetectionio/tests/test_notification_errors.py
+2 -7
View File
@@ -11,12 +11,7 @@ def test_check_notification_error_handling(client, live_server, measure_memory_u
# Set a URL and fetch it, then set a notification URL which is going to give errors
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
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
wait_for_all_checks(client)
set_modified_response(datastore_path=datastore_path)
@@ -25,7 +20,7 @@ def test_check_notification_error_handling(client, live_server, measure_memory_u
broken_notification_url = "jsons://broken-url-xxxxxxxx123/test"
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
# A URL with errors should not block the one that is working
data={"notification_urls": f"{broken_notification_url}\r\n{working_notification_url}",
"notification_title": "xxx",
changedetectionio/tests/test_obfuscations.py
+1 -1
View File
@@ -33,7 +33,7 @@ def test_obfuscations(client, live_server, measure_memory_usage, datastore_path)
# Check HTML conversion detected and workd
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_pdf.py
+2 -2
View File
@@ -56,7 +56,7 @@ def test_fetch_pdf(client, live_server, measure_memory_usage, datastore_path):
# The original checksum should be not be here anymore (cdio adds it to the bottom of the text)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -64,7 +64,7 @@ def test_fetch_pdf(client, live_server, measure_memory_usage, datastore_path):
assert changed_md5.encode('utf-8') in res.data
res = client.get(
url_for("ui.ui_diff.diff_history_page", uuid="first"),
url_for("ui.ui_diff.diff_history_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_preview_endpoints.py
+3 -3
View File
@@ -20,7 +20,7 @@ def test_fetch_pdf(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -50,7 +50,7 @@ def test_fetch_pdf(client, live_server, measure_memory_usage, datastore_path):
# The original checksum should be not be here anymore (cdio adds it to the bottom of the text)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -58,7 +58,7 @@ def test_fetch_pdf(client, live_server, measure_memory_usage, datastore_path):
assert changed_md5.encode('utf-8') in res.data
res = client.get(
url_for("ui.ui_diff.diff_history_page", uuid="first"),
url_for("ui.ui_diff.diff_history_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/test_request.py
+16 -16
View File
@@ -31,7 +31,7 @@ def test_headers_in_request(client, live_server, measure_memory_usage, datastore
# Add some headers to a request
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "",
@@ -48,7 +48,7 @@ def test_headers_in_request(client, live_server, measure_memory_usage, datastore
# The service should echo back the request headers
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -92,7 +92,7 @@ def test_body_in_request(client, live_server, measure_memory_usage, datastore_pa
# add the first 'version'
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "",
@@ -110,7 +110,7 @@ def test_body_in_request(client, live_server, measure_memory_usage, datastore_pa
body_value = 'Test Body Value {{ 1+1 }}'
body_value_formatted = 'Test Body Value 2'
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "",
@@ -126,7 +126,7 @@ def test_body_in_request(client, live_server, measure_memory_usage, datastore_pa
# The service should echo back the body
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -157,7 +157,7 @@ def test_body_in_request(client, live_server, measure_memory_usage, datastore_pa
# Attempt to add a body with a GET method
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "",
@@ -190,7 +190,7 @@ def test_method_in_request(client, live_server, measure_memory_usage, datastore_
# Attempt to add a method which is not valid
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "",
@@ -203,7 +203,7 @@ def test_method_in_request(client, live_server, measure_memory_usage, datastore_
# Add a properly formatted body
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "",
@@ -219,7 +219,7 @@ def test_method_in_request(client, live_server, measure_memory_usage, datastore_
# The service should echo back the request verb
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -265,7 +265,7 @@ def test_ua_global_override(client, live_server, measure_memory_usage, datastore
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -276,7 +276,7 @@ def test_ua_global_override(client, live_server, measure_memory_usage, datastore
# Add some headers to a request
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "testtag",
@@ -289,7 +289,7 @@ def test_ua_global_override(client, live_server, measure_memory_usage, datastore
assert b"Updated watch." in res.data
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b"agent-from-watch" in res.data
@@ -342,7 +342,7 @@ def test_headers_textfile_in_request(client, live_server, measure_memory_usage,
# Add some headers to a request
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"tags": "testtag",
@@ -375,7 +375,7 @@ def test_headers_textfile_in_request(client, live_server, measure_memory_usage,
if os.getenv('FAST_PUPPETEER_CHROME_FETCHER'):
time.sleep(6)
res = client.get(url_for("ui.ui_edit.edit_page", uuid="first"))
res = client.get(url_for("ui.ui_edit.edit_page", uuid=uuid))
assert b"Extra headers file found and will be added to this watch" in res.data
# Not needed anymore
@@ -384,7 +384,7 @@ def test_headers_textfile_in_request(client, live_server, measure_memory_usage,
# The service should echo back the request verb
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -414,7 +414,7 @@ def test_headers_validation(client, live_server, measure_memory_usage, datastore
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"fetch_backend": 'html_requests',
changedetectionio/tests/test_restock_itemprop.py
+14 -19
View File
@@ -3,7 +3,8 @@ import os
import time
from flask import url_for
from .util import live_server_setup, wait_for_all_checks, wait_for_notification_endpoint_output, extract_UUID_from_client, delete_all_watches
from .util import live_server_setup, wait_for_all_checks, wait_for_notification_endpoint_output, extract_UUID_from_client, \
delete_all_watches, get_UUID_for_tag_name
from ..notification import default_notification_format
instock_props = [
@@ -83,11 +84,7 @@ def test_itemprop_price_change(client, live_server, measure_memory_usage, datast
test_url = url_for('test_endpoint', _external=True)
set_original_response(props_markup=instock_props[0], price="190.95", datastore_path=datastore_path)
client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": test_url, "tags": 'restock tests', 'processor': 'restock_diff'},
follow_redirects=True
)
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, tag='restock tests', extras={'processor': 'restock_diff'})
# A change in price, should trigger a change by default
wait_for_all_checks(client)
@@ -108,7 +105,7 @@ def test_itemprop_price_change(client, live_server, measure_memory_usage, datast
# turning off price change trigger, but it should show the new price, with no change notification
set_original_response(props_markup=instock_props[0], price='120.45', datastore_path=datastore_path)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"restock_settings-follow_price_changes": "", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -129,11 +126,9 @@ def _run_test_minmax_limit(client, extra_watch_edit_form, datastore_path):
test_url = url_for('test_endpoint', _external=True)
set_original_response(props_markup=instock_props[0], price="950.95", datastore_path=datastore_path)
client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": test_url, "tags": 'restock tests', 'processor': 'restock_diff'},
follow_redirects=True
)
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, tag='restock tests', extras={'processor': 'restock_diff'})
wait_for_all_checks(client)
data = {
@@ -146,7 +141,7 @@ def _run_test_minmax_limit(client, extra_watch_edit_form, datastore_path):
}
data.update(extra_watch_edit_form)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data=data,
follow_redirects=True
)
@@ -211,18 +206,18 @@ def test_restock_itemprop_minmax(client, live_server, measure_memory_usage, data
_run_test_minmax_limit(client, extra_watch_edit_form=extras, datastore_path=datastore_path)
def test_restock_itemprop_with_tag(client, live_server, measure_memory_usage, datastore_path):
tag_name = "test-tag"
res = client.post(
url_for("tags.form_tag_add"),
data={"name": "test-tag"},
data={"name": tag_name},
follow_redirects=True
)
assert b"Tag added" in res.data
tag_uuid = get_UUID_for_tag_name(client, name=tag_name)
res = client.post(
url_for("tags.form_tag_edit_submit", uuid="first"),
data={"name": "test-tag",
url_for("tags.form_tag_edit_submit", uuid=tag_uuid),
data={"name": tag_name,
"restock_settings-follow_price_changes": "y",
"restock_settings-price_change_min": 900.0,
"restock_settings-price_change_max": 1100.10,
@@ -232,7 +227,7 @@ def test_restock_itemprop_with_tag(client, live_server, measure_memory_usage, da
)
extras = {
"tags": "test-tag"
"tags": tag_name
}
_run_test_minmax_limit(client, extra_watch_edit_form=extras,datastore_path=datastore_path)
changedetectionio/tests/test_rss.py
+2 -2
View File
@@ -124,7 +124,7 @@ def test_basic_cdata_rss_markup(client, live_server, measure_memory_usage, datas
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'CDATA' not in res.data
@@ -166,7 +166,7 @@ def test_rss_xpath_filtering(client, live_server, measure_memory_usage, datastor
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'CDATA' not in res.data
changedetectionio/tests/test_rss_single_watch.py
+5 -2
View File
@@ -6,7 +6,8 @@ import xml.etree.ElementTree as ET
from flask import url_for
from .restock.test_restock import set_original_response
from .util import live_server_setup, wait_for_all_checks, extract_rss_token_from_UI, extract_UUID_from_client, delete_all_watches, set_modified_response
from .util import live_server_setup, wait_for_all_checks, extract_rss_token_from_UI, extract_UUID_from_client, delete_all_watches, \
set_modified_response, get_UUID_for_tag_name
from ..notification import default_notification_format
@@ -290,6 +291,8 @@ def test_rss_single_watch_follow_notification_body(client, live_server, measure_
# Add our URL to the import page
test_url = url_for('test_endpoint', _external=True)
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, tag="RSS-Custom")
tag_uuid = get_UUID_for_tag_name(client=client, name="rss-custom")
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
@@ -316,7 +319,7 @@ def test_rss_single_watch_follow_notification_body(client, live_server, measure_
## Edit the tag notification_body, it should cascade up and become the RSS output
res = client.post(
url_for("tags.form_tag_edit_submit", uuid="first"),
url_for("tags.form_tag_edit_submit", uuid=tag_uuid),
data={"name": "rss-custom",
"notification_body": 'Hello from the group/tag level'},
follow_redirects=True
changedetectionio/tests/test_scheduler.py
+6 -6
View File
@@ -65,13 +65,13 @@ def test_check_basic_scheduler_functionality(client, live_server, measure_memory
data.update(scheduler_data)
time.sleep(1)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data=data,
follow_redirects=True
)
assert b"Updated watch." in res.data
res = client.get(url_for("ui.ui_edit.edit_page", uuid="first"))
res = client.get(url_for("ui.ui_edit.edit_page", uuid=uuid))
assert b"Pacific/Kiritimati" in res.data, "Should be Pacific/Kiritimati in placeholder data"
# "Edit" should not trigger a check because it's not enabled in the schedule.
@@ -143,13 +143,13 @@ def test_check_basic_global_scheduler_functionality(client, live_server, measure
# UI Sanity check
res = client.get(url_for("ui.ui_edit.edit_page", uuid="first"))
res = client.get(url_for("ui.ui_edit.edit_page", uuid=uuid))
assert b"Pacific/Kiritimati" in res.data, "Should be Pacific/Kiritimati in placeholder data"
#### HITTING SAVE SHOULD NOT TRIGGER A CHECK
last_check = live_server.app.config['DATASTORE'].data['watching'][uuid]['last_checked']
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
"fetch_backend": "html_requests",
@@ -179,7 +179,7 @@ def test_validation_time_interval_field(client, live_server, measure_memory_usag
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": 'The golden line',
"url": test_url,
'fetch_backend': "html_requests",
@@ -194,7 +194,7 @@ def test_validation_time_interval_field(client, live_server, measure_memory_usag
# Now set atleast something
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": 'The golden line',
"url": test_url,
'fetch_backend': "html_requests",
changedetectionio/tests/test_search.py
+19 -37
View File
@@ -1,73 +1,55 @@
from flask import url_for
from .util import set_original_response, set_modified_response, live_server_setup
import time
def test_basic_search(client, live_server, measure_memory_usage, datastore_path):
uuidA = client.application.config.get('DATASTORE').add_watch(url='https://localhost:12300?first-result=1')
uuidB = client.application.config.get('DATASTORE').add_watch(url='https://localhost:5000?second-result=1')
urls = ['https://localhost:12300?first-result=1',
'https://localhost:5000?second-result=1'
]
res = client.post(
url_for("imports.import_page"),
data={"urls": "\r\n".join(urls)},
follow_redirects=True
)
assert b"2 Imported" in res.data
# By URL
res = client.get(url_for("watchlist.index") + "?q=first-res")
assert urls[0].encode('utf-8') in res.data
assert urls[1].encode('utf-8') not in res.data
assert uuidA.encode('utf-8') in res.data
assert uuidB.encode('utf-8') not in res.data
# By Title
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
data={"title": "xxx-title", "url": urls[0], "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
url_for("ui.ui_edit.edit_page", uuid=uuidA),
data={"title": "xxx-title", "url": 'https://localhost:12300?first-result=1', "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
assert b"Updated watch." in res.data
res = client.get(url_for("watchlist.index") + "?q=xxx-title")
assert urls[0].encode('utf-8') in res.data
assert urls[1].encode('utf-8') not in res.data
assert uuidA.encode('utf-8') in res.data
assert uuidB.encode('utf-8') not in res.data
def test_search_in_tag_limit(client, live_server, measure_memory_usage, datastore_path):
urls = ['https://localhost:12300?first-result=1 tag-one',
'https://localhost:5000?second-result=1 tag-two'
]
res = client.post(
url_for("imports.import_page"),
data={"urls": "\r\n".join(urls)},
follow_redirects=True
)
assert b"2 Imported" in res.data
uuidA = client.application.config.get('DATASTORE').add_watch(url='https://localhost:12300?first-result=1', tag='tag-one')
uuidB = client.application.config.get('DATASTORE').add_watch(url='https://localhost:5000?second-result=1', tag='tag-two')
# By URL
res = client.get(url_for("watchlist.index") + "?q=first-res")
# Split because of the import tag separation
assert urls[0].split(' ')[0].encode('utf-8') in res.data, urls[0].encode('utf-8')
assert urls[1].split(' ')[0].encode('utf-8') not in res.data, urls[0].encode('utf-8')
assert 'tag-one'.encode('utf-8') in res.data
# @todo filter from results?
#assert 'tag-two'.encode('utf-8') not in res.data
# By Title
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
data={"title": "xxx-title", "url": urls[0].split(' ')[0], "tags": urls[0].split(' ')[1], "headers": "",
url_for("ui.ui_edit.edit_page", uuid=uuidA),
data={"title": "xxx-title", "url": 'https://localhost:12300?first-result=1', "tags": "tag-one", "headers": "",
'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
assert b"Updated watch." in res.data
res = client.get(url_for("watchlist.index") + "?q=xxx-title")
assert urls[0].split(' ')[0].encode('utf-8') in res.data, urls[0].encode('utf-8')
assert urls[1].split(' ')[0].encode('utf-8') not in res.data, urls[0].encode('utf-8')
assert "tag-one".encode('utf-8') in res.data
# @todo filter from results?
# assert "tag-two".encode('utf-8') not in res.data
changedetectionio/tests/test_security.py
+14 -32
View File
@@ -26,18 +26,14 @@ def set_original_response(datastore_path):
def test_bad_access(client, live_server, measure_memory_usage, datastore_path):
res = client.post(
url_for("imports.import_page"),
data={"urls": 'https://localhost'},
follow_redirects=True
)
assert b"1 Imported" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url='https://localhost')
wait_for_all_checks(client)
# Attempt to add a body with a GET method
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": 'javascript:alert(document.domain)',
"tags": "",
@@ -85,11 +81,9 @@ def test_bad_access(client, live_server, measure_memory_usage, datastore_path):
def _runner_test_various_file_slash(client, file_uri):
client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": file_uri, "tags": ''},
follow_redirects=True
)
test_url = url_for('test_endpoint', _external=True)
uuid = client.application.config.get('DATASTORE').add_watch(url=file_uri)
wait_for_all_checks(client)
res = client.get(url_for("watchlist.index"))
@@ -101,7 +95,7 @@ def _runner_test_various_file_slash(client, file_uri):
if file_uri.startswith('file:///'):
# This one should be the full qualified path to the file and should get the contents of this file
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'_runner_test_various_file_slash' in res.data
@@ -164,18 +158,11 @@ def test_xss(client, live_server, measure_memory_usage, datastore_path):
def test_xss_watch_last_error(client, live_server, measure_memory_usage, datastore_path):
set_original_response(datastore_path=datastore_path)
# Add our URL to the import page
res = client.post(
url_for("imports.import_page"),
data={"urls": url_for('test_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
test_url = url_for('test_endpoint', _external=True)
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"include_filters": '<a href="https://foobar"></a><script>alert(123);</script>',
"url": url_for('test_endpoint', _external=True),
@@ -360,13 +347,8 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
# Add a watch first
set_original_response(datastore_path=datastore_path)
res = client.post(
url_for("imports.import_page"),
data={"urls": url_for('test_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
wait_for_all_checks(client)
test_url = url_for('test_endpoint', _external=True)
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
# Set a password
password = "test123"
@@ -380,7 +362,7 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
# Try to access a protected page (edit page for first watch)
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
follow_redirects=False
)
@@ -397,7 +379,7 @@ def test_login_redirect_from_protected_page(client, live_server, measure_memory_
# The redirect parameter should be preserved in the login form
# It should contain the edit page URL
assert b'name="redirect"' in res.data
assert b'value="/edit/first"' in res.data or b'value="%2Fedit%2Ffirst"' in res.data
assert f'value="/edit/{uuid}'.encode('utf8') in res.data or f'value="%2Fedit%2F{uuid}'.encode('utf-8') in res.data
# Now login with correct password and the redirect parameter
res = client.post(
changedetectionio/tests/test_share_watch.py
+5 -12
View File
@@ -12,8 +12,6 @@ sleep_time_for_fetch_thread = 3
def test_share_watch(client, live_server, measure_memory_usage, datastore_path):
set_original_response(datastore_path=datastore_path)
# live_server_setup(live_server) # Setup on conftest per function
test_url = url_for('test_endpoint', _external=True)
include_filters = ".nice-filter"
@@ -24,20 +22,20 @@ def test_share_watch(client, live_server, measure_memory_usage, datastore_path):
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": include_filters, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
assert b"Updated watch." in res.data
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
assert bytes(include_filters.encode('utf-8')) in res.data
# click share the link
res = client.get(
url_for("ui.form_share_put_watch", uuid="first"),
url_for("ui.form_share_put_watch", uuid=uuid),
follow_redirects=True
)
@@ -53,20 +51,15 @@ def test_share_watch(client, live_server, measure_memory_usage, datastore_path):
delete_all_watches(client)
# Add our URL to the import page
res = client.post(
url_for("imports.import_page"),
data={"urls": share_link_search.group(1)},
follow_redirects=True
)
uuid = client.application.config.get('DATASTORE').add_watch(url=share_link_search.group(1))
assert b"1 Imported" in res.data
# Now hit edit, we should see what we expect
# that the import fetched the meta-data
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
assert bytes(include_filters.encode('utf-8')) in res.data
changedetectionio/tests/test_source.py
+4 -4
View File
@@ -22,7 +22,7 @@ def test_check_basic_change_detection_functionality_source(client, live_server,
# Check HTML conversion detected and workd
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -43,7 +43,7 @@ def test_check_basic_change_detection_functionality_source(client, live_server,
assert b'has-unread-changes' in res.data
res = client.get(
url_for("ui.ui_diff.diff_history_page", uuid="first"),
url_for("ui.ui_diff.diff_history_page", uuid=uuid),
follow_redirects=True
)
# With diff-match-patch, HTML tags are properly tokenized and excluded from diff spans
@@ -67,7 +67,7 @@ def test_check_ignore_elements(client, live_server, measure_memory_usage, datast
# We want <span> and <p> ONLY, but ignore span with .foobar-detection
client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": 'span,p', "url": test_url, "tags": "", "subtractive_selectors": ".foobar-detection", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -75,7 +75,7 @@ def test_check_ignore_elements(client, live_server, measure_memory_usage, datast
time.sleep(sleep_time_for_fetch_thread)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
assert b'foobar-detection' not in res.data
changedetectionio/tests/test_trigger.py
+5 -5
View File
@@ -76,7 +76,7 @@ def test_trigger_functionality(client, live_server, measure_memory_usage, datast
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": trigger_text,
"ignore_text": "and more",
"url": test_url,
@@ -89,14 +89,14 @@ def test_trigger_functionality(client, live_server, measure_memory_usage, datast
wait_for_all_checks(client)
# Check it saved
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
)
assert bytes(trigger_text.encode('utf-8')) in res.data
# so that we set the state to 'has-unread-changes' after all the edits
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
# Trigger a check
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
@@ -129,11 +129,11 @@ def test_trigger_functionality(client, live_server, measure_memory_usage, datast
# https://github.com/dgtlmoon/changedetection.io/issues/616
# Apparently the actual snapshot that contains the trigger never shows
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
res = client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
assert b'Add to cart' in res.data
# Check the preview/highlighter, we should be able to see what we triggered on, but it should be highlighted
res = client.get(url_for("ui.ui_preview.preview_page", uuid="first"))
res = client.get(url_for("ui.ui_preview.preview_page", uuid=uuid))
assert b'ignored_line_numbers = [8]' in res.data
# We should be able to see what we triggered on
changedetectionio/tests/test_trigger_regex.py
+2 -2
View File
@@ -43,7 +43,7 @@ def test_trigger_regex_functionality(client, live_server, measure_memory_usage,
### test regex
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": '/something \d{3}/',
"url": test_url,
"fetch_backend": "html_requests",
@@ -52,7 +52,7 @@ def test_trigger_regex_functionality(client, live_server, measure_memory_usage,
)
wait_for_all_checks(client)
# so that we set the state to 'has-unread-changes' after all the edits
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
f.write("some new noise")
changedetectionio/tests/test_trigger_regex_with_filter.py
+2 -2
View File
@@ -43,7 +43,7 @@ def test_trigger_regex_functionality_with_filter(client, live_server, measure_me
### test regex with filter
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"trigger_text": "/cool.stuff/",
"url": test_url,
"include_filters": '#in-here',
@@ -55,7 +55,7 @@ def test_trigger_regex_functionality_with_filter(client, live_server, measure_me
# Give the thread time to pick it up
time.sleep(sleep_time_for_fetch_thread)
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid))
# Check that we have the expected text.. but it's not in the css filter we want
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
changedetectionio/tests/test_ui.py
+9 -21
View File
@@ -40,7 +40,7 @@ def test_recheck_time_field_validation_single_watch(client, live_server, measure
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
'fetch_backend': "html_requests",
@@ -59,7 +59,7 @@ def test_recheck_time_field_validation_single_watch(client, live_server, measure
# Now set some time
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
'fetch_backend': "html_requests",
@@ -78,7 +78,7 @@ def test_recheck_time_field_validation_single_watch(client, live_server, measure
# Now set to use defaults
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": test_url,
'fetch_backend': "html_requests",
@@ -184,14 +184,7 @@ def test_page_title_listing_behaviour(client, live_server, measure_memory_usage,
assert b"Settings updated." in res.data
# Add our URL to the import page
res = client.post(
url_for("imports.import_page"),
data={"urls": url_for('test_endpoint', _external=True)},
follow_redirects=True
)
assert b"1 Imported" in res.data
uuid = client.application.config.get('DATASTORE').add_watch(url=url_for('test_endpoint', _external=True))
wait_for_all_checks(client)
# We see the URL only, no title/description was manually entered
@@ -201,7 +194,7 @@ def test_page_title_listing_behaviour(client, live_server, measure_memory_usage,
# Now 'my title' should override
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": url_for('test_endpoint', _external=True),
"title": "my title",
@@ -229,7 +222,7 @@ def test_page_title_listing_behaviour(client, live_server, measure_memory_usage,
# Remove page title description override and it should fall back to title
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={
"url": url_for('test_endpoint', _external=True),
"title": "",
@@ -250,14 +243,9 @@ def test_ui_viewed_unread_flag(client, live_server, measure_memory_usage, datast
set_original_response(datastore_path=datastore_path, extra_title="custom html")
# Add our URL to the import page
res = client.post(
url_for("imports.import_page"),
data={"urls": url_for('test_endpoint', _external=True)+"\r\n"+url_for('test_endpoint', _external=True)},
follow_redirects=True
)
uuidA = client.application.config.get('DATASTORE').add_watch(url= url_for('test_endpoint', _external=True))
uuidB = client.application.config.get('DATASTORE').add_watch(url=url_for('test_endpoint', _external=True))
assert b"2 Imported" in res.data
wait_for_all_checks(client)
set_modified_response(datastore_path=datastore_path)
@@ -269,7 +257,7 @@ def test_ui_viewed_unread_flag(client, live_server, measure_memory_usage, datast
assert res.data.count(b'data-watch-uuid') == 2
# one should now be viewed, but two in total still
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"))
client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuidA))
res = client.get(url_for("watchlist.index"))
assert b'<span id="unread-tab-counter">1</span>' in res.data
assert res.data.count(b'data-watch-uuid') == 2
changedetectionio/tests/test_unique_lines.py
+5 -5
View File
@@ -86,7 +86,7 @@ def test_unique_lines_functionality(client, live_server, measure_memory_usage, d
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"check_unique_lines": "y",
"url": test_url,
"fetch_backend": "html_requests",
@@ -130,7 +130,7 @@ def test_sort_lines_functionality(client, live_server, measure_memory_usage, dat
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"sort_text_alphabetically": "n",
"url": test_url,
"fetch_backend": "html_requests",
@@ -152,7 +152,7 @@ def test_sort_lines_functionality(client, live_server, measure_memory_usage, dat
assert b'has-unread-changes' in res.data
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -176,7 +176,7 @@ def test_extra_filters(client, live_server, measure_memory_usage, datastore_path
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"remove_duplicate_lines": "y",
"trim_text_whitespace": "y",
"sort_text_alphabetically": "", # leave this OFF for testing
@@ -195,7 +195,7 @@ def test_extra_filters(client, live_server, measure_memory_usage, datastore_path
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first")
url_for("ui.ui_preview.preview_page", uuid=uuid)
)
assert res.data.count(b"see what happens.") == 1
changedetectionio/tests/test_watch_fields_storage.py
+2 -2
View File
@@ -16,7 +16,7 @@ def test_check_watch_field_storage(client, live_server, measure_memory_usage, da
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={ "notification_urls": "json://127.0.0.1:30000\r\njson://128.0.0.1\r\n",
"time_between_check-minutes": 126,
"include_filters" : ".fooclass",
@@ -33,7 +33,7 @@ def test_check_watch_field_storage(client, live_server, measure_memory_usage, da
assert b"Updated watch." in res.data
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
follow_redirects=True
)
# checks that we dont get an error when using blank lines in the field value
changedetectionio/tests/test_xpath_selector.py
+22 -28
View File
@@ -118,7 +118,7 @@ def test_check_xpath_filter_utf8(client, live_server, measure_memory_usage, data
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": filter, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -167,7 +167,7 @@ def test_check_xpath_text_function_utf8(client, live_server, measure_memory_usag
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": filter, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -178,7 +178,7 @@ def test_check_xpath_text_function_utf8(client, live_server, measure_memory_usag
# The service should echo back the request headers
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -204,7 +204,7 @@ def test_check_markup_xpath_filter_restriction(client, live_server, measure_memo
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": xpath_filter, "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -214,7 +214,7 @@ def test_check_markup_xpath_filter_restriction(client, live_server, measure_memo
wait_for_all_checks(client)
# view it/reset state back to viewed
client.get(url_for("ui.ui_diff.diff_history_page", uuid="first"), follow_redirects=True)
client.get(url_for("ui.ui_diff.diff_history_page", uuid=uuid), follow_redirects=True)
# Make a change
set_modified_response(datastore_path=datastore_path)
@@ -237,7 +237,7 @@ def test_xpath_validation(client, live_server, measure_memory_usage, datastore_p
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "/something horrible", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -253,7 +253,7 @@ def test_xpath23_prefix_validation(client, live_server, measure_memory_usage, da
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "xpath:/something horrible", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -298,7 +298,7 @@ def test_xpath1_lxml(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "xpath1://title/text()", "url": test_url, "tags": "", "headers": "",
'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
@@ -310,7 +310,7 @@ def test_xpath1_lxml(client, live_server, measure_memory_usage, datastore_path):
assert b'_ElementStringResult' not in res.data # tested with 5.1.1 when it was removed and 5.1.0
assert b'Exception' not in res.data
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -328,7 +328,7 @@ def test_xpath1_validation(client, live_server, measure_memory_usage, datastore_
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "xpath1:/something horrible", "url": test_url, "tags": "", "headers": "", 'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
)
@@ -349,7 +349,7 @@ def test_check_with_prefix_include_filters(client, live_server, measure_memory_u
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "xpath://*[contains(@class, 'sametext')]", "url": test_url, "tags": "", "headers": "",
'fetch_backend': "html_requests", "time_between_check_use_default": "y"},
follow_redirects=True
@@ -359,7 +359,7 @@ def test_check_with_prefix_include_filters(client, live_server, measure_memory_u
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -396,7 +396,7 @@ def test_various_rules(client, live_server, measure_memory_usage, datastore_path
for r in ['//div', '//a', 'xpath://div', 'xpath://a']:
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": r,
"url": test_url,
"tags": "",
@@ -423,7 +423,7 @@ def test_xpath_20(client, live_server, measure_memory_usage, datastore_path):
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "//*[contains(@class, 'sametext')]|//*[contains(@class, 'changetext')]",
"url": test_url,
"tags": "",
@@ -437,7 +437,7 @@ def test_xpath_20(client, live_server, measure_memory_usage, datastore_path):
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -457,7 +457,7 @@ def test_xpath_20_function_count(client, live_server, measure_memory_usage, data
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "xpath:count(//div) * 123456789987654321",
"url": test_url,
"tags": "",
@@ -471,7 +471,7 @@ def test_xpath_20_function_count(client, live_server, measure_memory_usage, data
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -490,7 +490,7 @@ def test_xpath_20_function_count2(client, live_server, measure_memory_usage, dat
wait_for_all_checks(client)
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
data={"include_filters": "/html/body/count(div) * 123456789987654321",
"url": test_url,
"tags": "",
@@ -504,7 +504,7 @@ def test_xpath_20_function_count2(client, live_server, measure_memory_usage, dat
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
@@ -551,16 +551,11 @@ def _subtest_xpath_rss(client, datastore_path, content_type='text/html'):
# Add our URL to the import page
test_url = url_for('test_endpoint', content_type=content_type, _external=True)
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
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
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url, extras={'paused': True})
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data={
"url": test_url,
"include_filters": "xpath://item",
@@ -571,11 +566,10 @@ def _subtest_xpath_rss(client, datastore_path, content_type='text/html'):
follow_redirects=True
)
assert b"unpaused" in res.data
wait_for_all_checks(client)
res = client.get(
url_for("ui.ui_preview.preview_page", uuid="first"),
url_for("ui.ui_preview.preview_page", uuid=uuid),
follow_redirects=True
)
changedetectionio/tests/visualselector/test_fetch_data.py
+2 -2
View File
@@ -77,7 +77,7 @@ def test_visual_selector_content_ready(client, live_server, measure_memory_usage
# Some options should be enabled
# @todo - in the future, the visibility should be toggled by JS from the request type setting
res = client.get(
url_for("ui.ui_edit.edit_page", uuid="first"),
url_for("ui.ui_edit.edit_page", uuid=uuid),
follow_redirects=True
)
assert b'notification_screenshot' in res.data
@@ -103,7 +103,7 @@ def test_basic_browserstep(client, live_server, measure_memory_usage, datastore_
assert b"Watch added in Paused state, saving will unpause" in res.data
res = client.post(
url_for("ui.ui_edit.edit_page", uuid="first", unpause_on_save=1),
url_for("ui.ui_edit.edit_page", uuid=uuid, unpause_on_save=1),
data={
"url": test_url,
"tags": "",
changedetectionio/worker.py
+11
View File
@@ -67,6 +67,17 @@ async def async_update_worker(worker_id, q, notification_q, app, datastore, exec
logger.info(f"Worker {worker_id} idle and reached max runtime ({runtime:.0f}s), restarting")
return "restart"
continue
except RuntimeError as e:
# Handle executor shutdown gracefully - this is expected during shutdown
if "cannot schedule new futures after shutdown" in str(e):
# Executor shut down - exit gracefully without logging in pytest
if not IN_PYTEST:
logger.debug(f"Worker {worker_id} detected executor shutdown, exiting")
break
# Other RuntimeError - log and continue
logger.error(f"Worker {worker_id} runtime error: {e}")
await asyncio.sleep(0.1)
continue
except Exception as e:
# Handle expected Empty exception from queue timeout
import queue
changedetectionio/worker_pool.py
+12 -1
View File
@@ -338,7 +338,7 @@ def queue_item_async_safe(update_q, item, silent=False):
def shutdown_workers():
"""Shutdown all async workers brutally - no delays, no waiting"""
global worker_threads
global worker_threads, queue_executor
# Check if we're in pytest environment - if so, be more gentle with logging
import os
@@ -354,6 +354,17 @@ def shutdown_workers():
# Clear immediately - threads are daemon and will die
worker_threads.clear()
# Shutdown the queue executor to prevent "cannot schedule new futures after shutdown" errors
# This must happen AFTER workers are stopped to avoid race conditions
if queue_executor:
try:
queue_executor.shutdown(wait=False)
if not in_pytest:
logger.debug("Queue executor shut down")
except Exception as e:
if not in_pytest:
logger.warning(f"Error shutting down queue executor: {e}")
if not in_pytest:
logger.info("Async workers brutal shutdown complete")