Merge branch 'master' of github.com:dgtlmoon/changedetection.io

changedetectionio/__init__.py

@@ -744,6 +744,26 @@ def changedetection_app(config=None, datastore_o=None):
                                  uuid=uuid)
         return output
 
+    @app.route("/api/<string:uuid>/snapshot/current", methods=['GET'])
+    @login_required
+    def api_snapshot(uuid):
+
+        # 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:
+            return abort(400, "No history found for the specified link, bad link?")
+
+        newest = list(watch['history'].keys())[-1]
+        with open(watch['history'][newest], 'r') as f:
+            content = f.read()
+
+        resp = make_response(content)
+        resp.headers['Content-Type'] = 'text/plain'
+        return resp
 
     @app.route("/favicon.ico", methods=['GET'])
     def favicon():

changedetectionio/tests/test_api.py

@@ -0,0 +1,74 @@
+#!/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_response_data(test_return_data):
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write(test_return_data)
+
+
+def test_snapshot_api_detects_change(client, live_server):
+
+    test_return_data = "Some initial text"
+
+    test_return_data_modified = "Some NEW nice initial text"
+
+    sleep_time_for_fetch_thread = 3
+
+    set_response_data(test_return_data)
+
+    # Give the endpoint time to spin up
+    time.sleep(1)
+
+    # 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
+
+    # 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)
+
+    res = client.get(
+        url_for("api_snapshot", uuid="first"),
+        follow_redirects=True
+    )
+
+    assert test_return_data.encode() == res.data
+
+    #  Make a change
+    set_response_data(test_return_data_modified)
+
+    # 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)
+
+    res = client.get(
+        url_for("api_snapshot", uuid="first"),
+        follow_redirects=True
+    )
+
+    assert test_return_data_modified.encode() == res.data
+
+def test_snapshot_api_invalid_uuid(client, live_server):
+
+    res = client.get(
+        url_for("api_snapshot", uuid="invalid"),
+        follow_redirects=True
+    )
+
+    assert res.status_code == 400
+