From 23dde28399c6c4873c8d6fdf80dae24faee47986 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Fri, 20 May 2022 12:40:32 +0200 Subject: [PATCH] More test coverage --- changedetectionio/api_v1.py | 15 ++-- changedetectionio/tests/test_api.py | 110 +++++++++++++++++++++++++++- 2 files changed, 117 insertions(+), 8 deletions(-) diff --git a/changedetectionio/api_v1.py b/changedetectionio/api_v1.py index 9f443924e..9c711d9b3 100644 --- a/changedetectionio/api_v1.py +++ b/changedetectionio/api_v1.py @@ -33,7 +33,7 @@ class Watch(Resource): abort(400, message='No watch exists with the UUID of {}'.format(uuid)) self.datastore.delete(uuid) - return '', 204 + return 'OK', 204 class WatchHistory(Resource): def __init__(self, **kwargs): @@ -90,23 +90,26 @@ class CreateWatch(Resource): if not validators.url(json_data['url'].strip()): return "Invalid or unsupported URL", 400 - new_uuid = self.datastore.add_watch(url=json_data['url'].strip(), tag=tag) + extras = {'title': json_data['title'].strip()} if json_data.get('title') else {} + + new_uuid = self.datastore.add_watch(url=json_data['url'].strip(), tag=tag, extras=extras) self.update_q.put(new_uuid) return {'uuid': new_uuid}, 201 # Return concise list of available watches and some very basic info # curl http://localhost:4000/api/v1/watch|python -mjson.tool - # ?recheck=all to recheck all + # ?recheck_all=1 to recheck all def get(self): list = {} for k, v in self.datastore.data['watching'].items(): list[k] = {'url': v['url'], 'title': v['title'], 'last_checked': v['last_checked'], - 'last_changed': v['last_changed']} + 'last_changed': v['last_changed'], + 'last_error' : v['last_error']} - if request.args.get('recheck'): - for uuid in self.datastore.data['watching'].items(): + if request.args.get('recheck_all'): + for uuid in self.datastore.data['watching'].keys(): self.update_q.put(uuid) return {'status':"OK"}, 200 diff --git a/changedetectionio/tests/test_api.py b/changedetectionio/tests/test_api.py index f74c166ff..99dc42d06 100644 --- a/changedetectionio/tests/test_api.py +++ b/changedetectionio/tests/test_api.py @@ -8,6 +8,43 @@ import json import uuid +def set_original_response(): + 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("test-datastore/endpoint-content.txt", "w") as f: + f.write(test_return_data) + return None + + +def set_modified_response(): + 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("test-datastore/endpoint-content.txt", "w") as f: + f.write(test_return_data) + + return None + + def is_valid_uuid(val): try: uuid.UUID(str(val)) @@ -19,8 +56,12 @@ def is_valid_uuid(val): def test_api_simple(client, live_server): live_server_setup(live_server) + # Create a watch + set_original_response() watch_uuid = None + # Validate bad URL + test_url = url_for('test_endpoint', _external=True) res = client.post( url_for("createwatch"), data=json.dumps({"url": "h://xxxxxxxxxom"}), @@ -29,15 +70,80 @@ def test_api_simple(client, live_server): ) assert res.status_code == 400 + # Create new res = client.post( url_for("createwatch"), - data=json.dumps({"url": "https://nice.com"}), + data=json.dumps({"url": test_url, 'tag': "One, Two", "title": "My test URL"}), headers={'content-type': 'application/json'}, follow_redirects=True ) s = json.loads(res.data) assert is_valid_uuid(s['uuid']) watch_uuid = s['uuid'] - assert res.status_code == 201 + time.sleep(3) + + # Verify its in the list and that recheck worked + res = client.get( + url_for("createwatch") + ) + assert watch_uuid in json.loads(res.data).keys() + before_recheck_info = json.loads(res.data)[watch_uuid] + assert before_recheck_info['last_checked'] != 0 + assert before_recheck_info['title'] == 'My test URL' + + set_modified_response() + # Trigger recheck of all ?recheck_all=1 + client.get( + url_for("createwatch", recheck_all='1') + ) + time.sleep(3) + + # Did the recheck fire? + res = client.get( + url_for("createwatch") + ) + after_recheck_info = json.loads(res.data)[watch_uuid] + assert after_recheck_info['last_checked'] != before_recheck_info['last_checked'] + assert after_recheck_info['last_changed'] != 0 + + # Check history index list + res = client.get( + url_for("watchhistory", uuid=watch_uuid) + ) + history = json.loads(res.data) + assert len(history) == 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(history.keys())[-1]) + ) + 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') + ) + assert b'which has this one new line' in res.data + + # Fetch the whole watch + res = client.get( + url_for("watch", uuid=watch_uuid) + ) + watch = json.loads(res.data) + # @todo how to handle None/default global values? + assert watch['history_n'] == 2, "Found replacement history section, which is in its own API" + + # Finally delete the watch + res = client.delete( + url_for("watch", uuid=watch_uuid) + ) + assert res.status_code == 204 + + # Check via a relist + res = client.get( + url_for("createwatch") + ) + watch_list = json.loads(res.data) + assert len(watch_list) == 0, "Watch list should be empty"