#!/usr/bin/env python3
import time
from flask import url_for
from .util import live_server_setup, wait_for_all_checks, delete_all_watches
import os
import json
import uuid
def set_original_response(datastore_path):
test_return_data = """
Some initial text
Which is across multiple lines
So let's see what happens.
Some text thats the same
Some text that will change
"""
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
f.write(test_return_data)
return None
def set_modified_response(datastore_path):
test_return_data = """
Some initial text
which has this one new line
So let's see what happens.
Some text thats the same
Some text that changes
"""
with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f:
f.write(test_return_data)
return None
def is_valid_uuid(val):
try:
uuid.UUID(str(val))
return True
except ValueError:
return False
# def test_setup(client, live_server, measure_memory_usage, datastore_path):
# live_server_setup(live_server) # Setup on conftest per function
def test_api_simple(client, live_server, measure_memory_usage, datastore_path):
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
# Create a watch
set_original_response(datastore_path=datastore_path)
# Validate bad URL
test_url = url_for('test_endpoint', _external=True )
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": "h://xxxxxxxxxom"}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
follow_redirects=True
)
assert res.status_code == 400
# Create new
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": test_url, 'tag': "One, Two", "title": "My test URL"}),
headers={'content-type': 'application/json', 'x-api-key': api_key},
follow_redirects=True
)
assert is_valid_uuid(res.json.get('uuid'))
watch_uuid = res.json.get('uuid')
assert res.status_code == 201
wait_for_all_checks(client)
# Verify its in the list and that recheck worked
res = client.get(
url_for("createwatch", tag="OnE"),
headers={'x-api-key': api_key}
)
assert watch_uuid in res.json.keys()
before_recheck_info = res.json[watch_uuid]
assert before_recheck_info['last_checked'] != 0
#705 `last_changed` should be zero on the first check
assert before_recheck_info['last_changed'] == 0
assert before_recheck_info['title'] == 'My test URL'
# Check the limit by tag doesnt return anything when nothing found
res = client.get(
url_for("createwatch", tag="Something else entirely"),
headers={'x-api-key': api_key}
)
assert len(res.json) == 0
time.sleep(1)
wait_for_all_checks(client)
set_modified_response(datastore_path=datastore_path)
# Trigger recheck of all ?recheck_all=1
client.get(
url_for("createwatch", recheck_all='1'),
headers={'x-api-key': api_key},
)
wait_for_all_checks(client)
time.sleep(1)
# Did the recheck fire?
res = client.get(
url_for("createwatch"),
headers={'x-api-key': api_key},
)
after_recheck_info = res.json[watch_uuid]
assert after_recheck_info['last_checked'] != before_recheck_info['last_checked']
assert after_recheck_info['last_changed'] != 0
# #2877 When run in a slow fetcher like playwright etc
assert after_recheck_info['last_changed'] == after_recheck_info['last_checked']
# Check history index list
res = client.get(
url_for("watchhistory", uuid=watch_uuid),
headers={'x-api-key': api_key},
)
watch_history = res.json
assert len(res.json) == 2, "Should have two history entries (the original and the changed)"
# Fetch a snapshot by timestamp, check the right one was found
res = client.get(
url_for("watchsinglehistory", uuid=watch_uuid, timestamp=list(res.json.keys())[-1]),
headers={'x-api-key': api_key},
)
assert b'which has this one new line' in res.data
# Fetch a snapshot by 'latest'', check the right one was found
res = client.get(
url_for("watchsinglehistory", uuid=watch_uuid, timestamp='latest'),
headers={'x-api-key': api_key},
)
assert b'which has this one new line' in res.data
assert b'Which is across multiple lines' in res.data
# Test html format
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?format=html',
headers={'x-api-key': api_key},
)
assert res.status_code == 200
assert b'
' in res.data
# Test markdown format
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?format=markdown',
headers={'x-api-key': api_key},
)
assert res.status_code == 200
# Test new diff preference parameters
# Test removed=false (should hide removed content)
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?removed=false',
headers={'x-api-key': api_key},
)
# Should not contain removed content indicator
assert b'(removed)' not in res.data
# Should still contain added content
assert b'(added)' in res.data or b'which has this one new line' in res.data
# Test added=false (should hide added content)
# Note: The test data has replacements, not pure additions, so we test differently
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?added=false&replaced=false',
headers={'x-api-key': api_key},
)
# With both added and replaced disabled, should have minimal content
# Should not contain added indicators
assert b'(added)' not in res.data
# Test replaced=false (should hide replaced/changed content)
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?replaced=false',
headers={'x-api-key': api_key},
)
# Should not contain changed content indicator
assert b'(changed)' not in res.data
# Test type=diffWords for word-level diff
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?type=diffWords&format=htmlcolor',
headers={'x-api-key': api_key},
)
# Should contain HTML formatted diff
assert res.status_code == 200
assert len(res.data) > 0
# Test combined parameters: show only additions with word diff
res = client.get(
url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp='previous', to_timestamp='latest')+'?removed=false&replaced=false&type=diffWords',
headers={'x-api-key': api_key},
)
assert res.status_code == 200
# Should not contain removed or changed markers
assert b'(removed)' not in res.data
assert b'(changed)' not in res.data
# Fetch the whole watch
res = client.get(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key}
)
watch = res.json
# @todo how to handle None/default global values?
assert watch['history_n'] == 2, "Found replacement history section, which is in its own API"
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)
time.sleep(3)
# Fetch the whole watch again, viewed should be true
res = client.get(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key}
)
watch = res.json
assert watch.get('viewed') == True
# basic systeminfo check
res = client.get(
url_for("systeminfo"),
headers={'x-api-key': api_key},
)
assert res.json.get('watch_count') == 1
assert res.json.get('uptime') > 0.5
######################################################
# Mute and Pause, check it worked
res = client.get(
url_for("watch", uuid=watch_uuid, paused='paused'),
headers={'x-api-key': api_key}
)
assert b'OK' in res.data
res = client.get(
url_for("watch", uuid=watch_uuid, muted='muted'),
headers={'x-api-key': api_key}
)
assert b'OK' in res.data
res = client.get(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key}
)
assert res.json.get('paused') == True
assert res.json.get('notification_muted') == True
# Now unpause, unmute
res = client.get(
url_for("watch", uuid=watch_uuid, muted='unmuted'),
headers={'x-api-key': api_key}
)
assert b'OK' in res.data
res = client.get(
url_for("watch", uuid=watch_uuid, paused='unpaused'),
headers={'x-api-key': api_key}
)
assert b'OK' in res.data
res = client.get(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key}
)
assert res.json.get('paused') == 0
assert res.json.get('notification_muted') == 0
######################################################
# Finally delete the watch
res = client.delete(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key},
)
assert res.status_code == 204
# Check via a relist
res = client.get(
url_for("createwatch"),
headers={'x-api-key': api_key}
)
assert len(res.json) == 0, "Watch list should be empty"
def test_access_denied(client, live_server, measure_memory_usage, datastore_path):
# `config_api_token_enabled` Should be On by default
res = client.get(
url_for("createwatch")
)
assert res.status_code == 403
res = client.get(
url_for("createwatch"),
headers={'x-api-key': "something horrible"}
)
assert res.status_code == 403
# Disable config_api_token_enabled and it should work
res = client.post(
url_for("settings.settings_page"),
data={
"requests-time_between_check-minutes": 180,
"application-fetch_backend": "html_requests",
"application-api_access_token_enabled": ""
},
follow_redirects=True
)
assert b"Settings updated." in res.data
res = client.get(
url_for("createwatch")
)
assert res.status_code == 200
# Cleanup everything
delete_all_watches(client)
res = client.post(
url_for("settings.settings_page"),
data={
"requests-time_between_check-minutes": 180,
"application-fetch_backend": "html_requests",
"application-api_access_token_enabled": "y"
},
follow_redirects=True
)
assert b"Settings updated." in res.data
def test_api_watch_PUT_update(client, live_server, measure_memory_usage, datastore_path):
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
# Create a watch
set_original_response(datastore_path=datastore_path)
test_url = url_for('test_endpoint', _external=True)
# Create new
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": test_url,
'tag': "One, Two",
"title": "My test URL",
'headers': {'cookie': 'yum'},
"conditions": [
{
"field": "page_filtered_text",
"operator": "contains_regex",
"value": "." # contains anything
}
],
"conditions_match_logic": "ALL",
}
),
headers={'content-type': 'application/json', 'x-api-key': api_key},
follow_redirects=True
)
assert res.status_code == 201
wait_for_all_checks(client)
# Get a listing, it will be the first one
res = client.get(
url_for("createwatch"),
headers={'x-api-key': api_key}
)
watch_uuid = list(res.json.keys())[0]
assert not res.json[watch_uuid].get('viewed'), 'A newly created watch can only be unviewed'
# Check in the edit page just to be sure
res = client.get(
url_for("ui.ui_edit.edit_page", uuid=watch_uuid),
)
assert b"cookie: yum" in res.data, "'cookie: yum' found in 'headers' section"
assert b"One" in res.data, "Tag 'One' was found"
assert b"Two" in res.data, "Tag 'Two' was found"
# HTTP PUT ( UPDATE an existing watch )
res = client.put(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key, 'content-type': 'application/json'},
data=json.dumps({
"title": "new title",
'time_between_check': {'minutes': 552},
'headers': {'cookie': 'all eaten'},
'last_viewed': int(time.time())
}),
)
assert res.status_code == 200, "HTTP PUT update was sent OK"
# HTTP GET single watch, title should be updated
res = client.get(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key}
)
assert res.json.get('title') == 'new title'
assert res.json.get('viewed'), 'With the timestamp greater than "changed" a watch can be updated to viewed'
# Check in the edit page just to be sure
res = client.get(
url_for("ui.ui_edit.edit_page", uuid=watch_uuid),
)
assert b"new title" in res.data, "new title found in edit page"
assert b"552" in res.data, "552 minutes found in edit page"
assert b"One" in res.data, "Tag 'One' was found"
assert b"Two" in res.data, "Tag 'Two' was found"
assert b"cookie: all eaten" in res.data, "'cookie: all eaten' found in 'headers' section"
######################################################
# HTTP PUT try a field that doesn't exist
# HTTP PUT an update
res = client.put(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key, 'content-type': 'application/json'},
data=json.dumps({"title": "new title", "some other field": "uh oh"}),
)
assert res.status_code == 400, "Should get error 400 when we give a field that doesnt exist"
# Message will come from `flask_expects_json`
assert b'Additional properties are not allowed' in res.data
# Try a XSS URL
res = client.put(
url_for("watch", uuid=watch_uuid),
headers={'x-api-key': api_key, 'content-type': 'application/json'},
data=json.dumps({
'url': 'javascript:alert(document.domain)'
}),
)
assert res.status_code == 400
# Cleanup everything
delete_all_watches(client)
def test_api_import(client, live_server, measure_memory_usage, datastore_path):
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
res = client.post(
url_for("import") + "?tag=import-test",
data='https://website1.com\r\nhttps://website2.com',
# We removed 'content-type': 'text/plain', the Import API should assume this if none is set #3547 #3542
headers={'x-api-key': api_key},
follow_redirects=True
)
assert res.status_code == 200
assert len(res.json) == 2
res = client.get(url_for("watchlist.index"))
assert b"https://website1.com" in res.data
assert b"https://website2.com" in res.data
# Should see the new tag in the tag/groups list
res = client.get(url_for('tags.tags_overview_page'))
assert b'import-test' in res.data
def test_api_conflict_UI_password(client, live_server, measure_memory_usage, datastore_path):
api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
# Enable password check and diff page access bypass
res = client.post(
url_for("settings.settings_page"),
data={"application-password": "foobar", # password is now set! API should still work!
"application-api_access_token_enabled": "y",
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Password protection enabled." in res.data
# Create a watch
set_original_response(datastore_path=datastore_path)
test_url = url_for('test_endpoint', _external=True)
# Create new
res = client.post(
url_for("createwatch"),
data=json.dumps({"url": test_url, "title": "My test URL" }),
headers={'content-type': 'application/json', 'x-api-key': api_key},
follow_redirects=True
)
assert res.status_code == 201
wait_for_all_checks(client)
url = url_for("createwatch")
# Get a listing, it will be the first one
res = client.get(
url,
headers={'x-api-key': api_key}
)
assert res.status_code == 200
assert len(res.json)