#!/usr/bin/python3 import time from flask import url_for from . util import live_server_setup def test_setup(live_server): live_server_setup(live_server) def set_original_response(): test_return_data = """
Some initial textWhich is across multiple lines
So let's see what happens. """ with open("test-datastore/endpoint-content.txt", "w") as f: f.write(test_return_data) def set_some_changed_response(): test_return_data = """ Some initial textWhich is across multiple lines, and a new thing too.
So let's see what happens. """ with open("test-datastore/endpoint-content.txt", "w") as f: f.write(test_return_data) def test_normal_page_check_works_with_ignore_status_code(client, live_server): sleep_time_for_fetch_thread = 3 # Give the endpoint time to spin up time.sleep(1) set_original_response() # Goto the settings page, add our ignore text res = client.post( url_for("settings_page"), data={ "requests-minutes_between_check": 180, "application-ignore_status_codes": "y", 'application-fetch_backend': "html_requests" }, follow_redirects=True ) assert b"Settings updated." in res.data # Add our URL to the import page test_url = url_for('test_endpoint', _external=True) res = client.post( url_for("import_page"), data={"urls": test_url}, follow_redirects=True ) assert b"1 Imported" in res.data time.sleep(sleep_time_for_fetch_thread) # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) set_some_changed_response() time.sleep(sleep_time_for_fetch_thread) # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # It should report nothing found (no new 'unviewed' class) res = client.get(url_for("index")) assert b'unviewed' in res.data assert b'/test-endpoint' in res.data # Tests the whole stack works with staus codes ignored def test_403_page_check_works_with_ignore_status_code(client, live_server): sleep_time_for_fetch_thread = 3 set_original_response() # Give the endpoint time to spin up time.sleep(1) # Add our URL to the import page test_url = url_for('test_endpoint', status_code=403, _external=True) res = client.post( url_for("import_page"), data={"urls": test_url}, follow_redirects=True ) assert b"1 Imported" in res.data # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # Goto the edit page, check our ignore option # Add our URL to the import page res = client.post( url_for("edit_page", uuid="first"), data={"ignore_status_codes": "y", "url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"}, follow_redirects=True ) assert b"Updated watch." in res.data # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # Make a change set_some_changed_response() # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # It should have 'unviewed' still # Because it should be looking at only that 'sametext' id res = client.get(url_for("index")) assert b'unviewed' in res.data # Tests the whole stack works with staus codes ignored def test_403_page_check_fails_without_ignore_status_code(client, live_server): sleep_time_for_fetch_thread = 3 set_original_response() # Give the endpoint time to spin up time.sleep(1) # Add our URL to the import page test_url = url_for('test_endpoint', status_code=403, _external=True) res = client.post( url_for("import_page"), data={"urls": test_url}, follow_redirects=True ) assert b"1 Imported" in res.data # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # Goto the edit page, check our ignore option # Add our URL to the import page res = client.post( url_for("edit_page", uuid="first"), data={"url": test_url, "tag": "", "headers": "", 'fetch_backend': "html_requests"}, follow_redirects=True ) assert b"Updated watch." in res.data # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # Make a change set_some_changed_response() # Trigger a check client.get(url_for("api_watch_checknow"), follow_redirects=True) # Give the thread time to pick it up time.sleep(sleep_time_for_fetch_thread) # It should have 'unviewed' still # Because it should be looking at only that 'sametext' id res = client.get(url_for("index")) assert b'Status Code 403' in res.data