@ -1,5 +1,6 @@
import os
import time
from loguru import logger
from flask import url_for
from . util import set_original_response , live_server_setup , extract_UUID_from_client , wait_for_all_checks , \
wait_for_notification_endpoint_output
@ -27,6 +28,12 @@ def run_filter_test(client, live_server, content_filter):
# Response WITHOUT the filter ID element
set_original_response ( )
# Goto the edit page, add our ignore text
notification_url = url_for ( ' test_notification_endpoint ' , _external = True ) . replace ( ' http ' , ' json ' )
# Add our URL to the import page
test_url = url_for ( ' test_endpoint ' , _external = True )
# cleanup for the next
client . get (
url_for ( " form_delete " , uuid = " all " ) ,
@ -35,28 +42,20 @@ def run_filter_test(client, live_server, content_filter):
if os . path . isfile ( " test-datastore/notification.txt " ) :
os . unlink ( " test-datastore/notification.txt " )
# Add our URL to the import page
test_url = url_for ( ' test_endpoint ' , _external = True )
res = client . post (
url_for ( " form_quick_watch_add " ) ,
data = { " url " : test_url , " tags " : ' ' } ,
url_for ( " import_page " ) ,
data = { " url s " : test_url } ,
follow_redirects = True
)
assert b " Watch added " in res . data
# Give the thread time to pick up the first version
assert b " 1 Imported " in res . data
wait_for_all_checks ( client )
# Goto the edit page, add our ignore text
# Add our URL to the import page
url = url_for ( ' test_notification_endpoint ' , _external = True )
notification_url = url . replace ( ' http ' , ' json ' )
uuid = extract_UUID_from_client ( client )
print ( " >>>> Notification URL: " + notification_url )
assert live_server . app . config [ ' DATASTORE ' ] . data [ ' watching ' ] [ uuid ] [ ' consecutive_filter_failures ' ] == 0 , " No filter = No filter failure "
# Just a regular notification setting, this will be used by the special 'filter not found' notification
notification_form_data = { " notification_urls " : notification_url ,
watch_data = { " notification_urls " : notification_url ,
" notification_title " : " New ChangeDetection.io Notification - {{ watch_url}} " ,
" notification_body " : " BASE URL: {{ base_url}} \n "
" Watch URL: {{ watch_url}} \n "
@ -70,49 +69,63 @@ def run_filter_test(client, live_server, content_filter):
" Diff Full: {{ diff_full}} \n "
" Diff as Patch: {{ diff_patch}} \n "
" :-) " ,
" notification_format " : " Text " }
notification_form_data . update ( {
" url " : test_url ,
" notification_format " : " Text " ,
" fetch_backend " : " html_requests " ,
" filter_failure_notification_send " : ' y ' ,
" headers " : " " ,
" tags " : " my tag " ,
" title " : " my title 123 " ,
" headers " : " " ,
" filter_failure_notification_send " : ' y ' ,
" include_filters " : content_filter ,
" fetch_backend " : " html_requests " } )
" time_between_check-hours " : 5 , # So that the queue runner doesnt also put it in
" url " : test_url ,
}
# A POST here will also reset the filter failure counter (filter_failure_notification_threshold_attempts)
res = client . post (
url_for ( " edit_page " , uuid = " first " ) ,
data = notification_form _data,
url_for ( " edit_page " , uuid = uuid ) ,
data = watch _data,
follow_redirects = True
)
assert b " Updated watch. " in res . data
wait_for_all_checks ( client )
assert live_server . app . config [ ' DATASTORE ' ] . data [ ' watching ' ] [ uuid ] [ ' consecutive_filter_failures ' ] == 0 , " No filter = No filter failure "
# Now the notification should not exist, because we didnt reach the threshold
# Now add a filter, because recheck hours == 5, ONLY pressing of the [edit] or [recheck all] should trigger
watch_data [ ' include_filters ' ] = content_filter
res = client . post (
url_for ( " edit_page " , uuid = uuid ) ,
data = watch_data ,
follow_redirects = True
)
assert b " Updated watch. " in res . data
# It should have checked once so far and given this error (because we hit SAVE)
wait_for_all_checks ( client )
assert not os . path . isfile ( " test-datastore/notification.txt " )
# Hitting [save] would have triggered a recheck, and we have a filter, so this would be ONE failure
assert live_server . app . config [ ' DATASTORE ' ] . data [ ' watching ' ] [ uuid ] [ ' consecutive_filter_failures ' ] == 1 , " Should have been checked once "
# recheck it up to just before the threshold, including the fact that in the previous POST it would have rechecked (and incremented)
for i in range ( 0 , App . _FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT - 2 ) :
# Add 4 more checks
checked = 0
ATTEMPT_THRESHOLD_SETTING = live_server . app . config [ ' DATASTORE ' ] . data [ ' settings ' ] [ ' application ' ] . get ( ' filter_failure_notification_threshold_attempts ' , 0 )
for i in range ( 0 , ATTEMPT_THRESHOLD_SETTING - 2 ) :
checked + = 1
client . get ( url_for ( " form_watch_checknow " ) , follow_redirects = True )
wait_for_all_checks ( client )
time . sleep ( 2 ) # delay for apprise to fire
assert not os . path . isfile ( " test-datastore/notification.txt " ) , f " test-datastore/notification.txt should not exist - Attempt { i } when threshold is { App . _FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT } "
# We should see something in the frontend
res = client . get ( url_for ( " index " ) )
assert b ' Warning, no filters were found ' in res . data
assert not os . path . isfile ( " test-datastore/notification.txt " )
assert live_server . app . config [ ' DATASTORE ' ] . data [ ' watching ' ] [ uuid ] [ ' consecutive_filter_failures ' ] == 5
# One more check should trigger the _FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT threshold
client . get ( url_for ( " form_watch_checknow " ) , follow_redirects = True )
wait_for_all_checks ( client )
wait_for_notification_endpoint_output ( )
# Now it should exist and contain our "filter not found" alert
assert os . path . isfile ( " test-datastore/notification.txt " )
with open ( " test-datastore/notification.txt " , ' r ' ) as f :
notification = f . read ( )
@ -125,7 +138,7 @@ def run_filter_test(client, live_server, content_filter):
set_response_with_filter ( )
# Try several times, it should NOT have 'filter not found'
for i in range ( 0 , A pp. _FILTER_FAILURE_THRESHOLD_ATTEMPTS_DEFAULT ) :
for i in range ( 0 , A TTEMPT_THRESHOLD_SETTING + 2 ) :
client . get ( url_for ( " form_watch_checknow " ) , follow_redirects = True )
wait_for_all_checks ( client )
@ -138,9 +151,6 @@ def run_filter_test(client, live_server, content_filter):
assert not ' CSS/xPath filter was not present in the page ' in notification
# 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
@ -155,9 +165,11 @@ def test_setup(live_server):
live_server_setup ( live_server )
def test_check_include_filters_failure_notification ( client , live_server , measure_memory_usage ) :
# live_server_setup(live_server)
run_filter_test ( client , live_server , ' #nope-doesnt-exist ' )
def test_check_xpath_filter_failure_notification ( client , live_server , measure_memory_usage ) :
# live_server_setup(live_server)
run_filter_test ( client , live_server , ' //*[@id= " nope-doesnt-exist " ] ' )
# Test that notification is never sent
dgtlmoon
3 months ago
committed by
GitHub