Merge pull request #15 from dgtlmoon/dev

Prepare 0.27
pull/19/head 0.27
dgtlmoon 4 years ago committed by GitHub
commit 35546c331c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -4,11 +4,7 @@
name: changedetection.io
on:
push:
branches: [ master ]
pull_request:
branches: [ master ]
on: [push, pull_request]
jobs:
build:

@ -4,7 +4,7 @@
<img src="https://img.shields.io/docker/pulls/dgtlmoon/changedetection.io" alt="Docker Pulls"/>
</a>
<a href="https://hub.docker.com/r/dgtlmoon/changedetection.io" target="_blank" title="Change detection docker hub">
<img src="https://img.shields.io/docker/v/dgtlmoon/changedetection.io" alt="Change detection latest tag version"/>
<img src="https://img.shields.io/docker/v/dgtlmoon/changedetection.io/0.27" alt="Change detection latest tag version"/>
</a>
## Self-hosted change monitoring of web pages.

@ -2,10 +2,8 @@
# @todo logging
# @todo sort by last_changed
# @todo extra options for url like , verify=False etc.
# @todo enable https://urllib3.readthedocs.io/en/latest/user-guide.html#ssl as option?
# @todo maybe a button to reset all 'last-changed'.. so you can see it clearly when something happens since your last visit
# @todo option for interval day/6 hour/etc
# @todo on change detected, config for calling some API
# @todo make tables responsive!
@ -19,9 +17,16 @@ import os
import timeago
import threading
from threading import Event
import queue
from flask import Flask, render_template, request, send_file, send_from_directory, abort, redirect, url_for
from flask import Flask, render_template, request, send_file, send_from_directory, abort, redirect, url_for
from feedgen.feed import FeedGenerator
from flask import make_response
import datetime
import pytz
datastore = None
@ -39,7 +44,9 @@ app = Flask(__name__, static_url_path="/var/www/change-detection/backen/static")
# Stop browser caching of assets
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
app.config['STOP_THREADS'] = False
app.config.exit = Event()
app.config['NEW_VERSION_AVAILABLE'] = False
# Disables caching of the templates
app.config['TEMPLATES_AUTO_RELOAD'] = True
@ -76,7 +83,7 @@ def _jinja2_filter_datetimestamp(timestamp, format="%Y-%m-%d %H:%M:%S"):
def changedetection_app(config=None, datastore_o=None):
global datastore
datastore = datastore_o
# Hmm
app.config.update(dict(DEBUG=True))
app.config.update(config or {})
@ -112,14 +119,40 @@ def changedetection_app(config=None, datastore_o=None):
sorted_watches.sort(key=lambda x: x['last_changed'], reverse=True)
existing_tags = datastore.get_all_tags()
output = render_template("watch-overview.html",
watches=sorted_watches,
messages=messages,
tags=existing_tags,
active_tag=limit_tag)
rss = request.args.get('rss')
if rss:
fg = FeedGenerator()
fg.title('changedetection.io')
fg.description('Feed description')
fg.link(href='https://changedetection.io')
for watch in sorted_watches:
if not watch['viewed']:
fe = fg.add_entry()
fe.title(watch['url'])
fe.link(href=watch['url'])
fe.description(watch['url'])
fe.guid(watch['uuid'], permalink=False)
dt = datetime.datetime.fromtimestamp(int(watch['newest_history_key']))
dt = dt.replace(tzinfo=pytz.UTC)
fe.pubDate(dt)
response = make_response(fg.rss_str())
response.headers.set('Content-Type', 'application/rss+xml')
return response
else:
output = render_template("watch-overview.html",
watches=sorted_watches,
messages=messages,
tags=existing_tags,
active_tag=limit_tag,
has_unviewed=datastore.data['has_unviewed'])
# Show messages but once.
messages = []
# Show messages but once.
messages = []
return output
@app.route("/scrub", methods=['GET', 'POST'])
@ -151,29 +184,80 @@ def changedetection_app(config=None, datastore_o=None):
return render_template("scrub.html")
@app.route("/edit", methods=['GET', 'POST'])
def edit_page():
# If they edited an existing watch, we need to know to reset the current/previous md5 to include
# the excluded text.
def get_current_checksum_include_ignore_text(uuid):
import hashlib
from backend import fetch_site_status
# Get the most recent one
newest_history_key = datastore.get_val(uuid, 'newest_history_key')
# 0 means that theres only one, so that there should be no 'unviewed' history availabe
if newest_history_key == 0:
newest_history_key = list(datastore.data['watching'][uuid]['history'].keys())[0]
if newest_history_key:
with open(datastore.data['watching'][uuid]['history'][newest_history_key],
encoding='utf-8') as file:
raw_content = file.read()
handler = fetch_site_status.perform_site_check(datastore=datastore)
stripped_content = handler.strip_ignore_text(raw_content,
datastore.data['watching'][uuid]['ignore_text'])
checksum = hashlib.md5(stripped_content).hexdigest()
return checksum
return datastore.data['watching'][uuid]['previous_md5']
@app.route("/edit/<string:uuid>", methods=['GET', 'POST'])
def edit_page(uuid):
global messages
import validators
# More for testing, possible to return the first/only
if uuid == 'first':
uuid = list(datastore.data['watching'].keys()).pop()
if request.method == 'POST':
uuid = request.args.get('uuid')
url = request.form.get('url').strip()
tag = request.form.get('tag').strip()
# Extra headers
form_headers = request.form.get('headers').strip().split("\n")
extra_headers = {}
if form_headers:
for header in form_headers:
if len(header):
parts = header.split(':', 1)
extra_headers.update({parts[0].strip(): parts[1].strip()})
if len(parts) == 2:
extra_headers.update({parts[0].strip(): parts[1].strip()})
update_obj = {'url': url,
'tag': tag,
'headers': extra_headers
}
# Ignore text
form_ignore_text = request.form.get('ignore-text').strip()
ignore_text = []
if len(form_ignore_text):
for text in form_ignore_text.split("\n"):
text = text.strip()
if len(text):
ignore_text.append(text)
datastore.data['watching'][uuid]['ignore_text'] = ignore_text
# Reset the previous_md5 so we process a new snapshot including stripping ignore text.
if len(datastore.data['watching'][uuid]['history']):
update_obj['previous_md5'] = get_current_checksum_include_ignore_text(uuid=uuid)
validators.url(url) # @todo switch to prop/attr/observer
datastore.data['watching'][uuid].update({'url': url,
'tag': tag,
'headers': extra_headers})
datastore.data['watching'][uuid].update(update_obj)
datastore.needs_write = True
messages.append({'class': 'ok', 'message': 'Updated watch.'})
@ -181,8 +265,6 @@ def changedetection_app(config=None, datastore_o=None):
return redirect(url_for('index'))
else:
uuid = request.args.get('uuid')
output = render_template("edit.html", uuid=uuid, watch=datastore.data['watching'][uuid], messages=messages)
return output
@ -235,23 +317,37 @@ def changedetection_app(config=None, datastore_o=None):
messages.append({'class': 'ok', 'message': "{} Imported, {} Skipped.".format(good, len(remaining_urls))})
if len(remaining_urls) == 0:
return redirect(url_for('index'))
else:
output = render_template("import.html",
messages=messages,
remaining="\n".join(remaining_urls)
)
messages = []
if len(remaining_urls) == 0:
# Looking good, redirect to index.
return redirect(url_for('index'))
# Could be some remaining, or we could be on GET
output = render_template("import.html",
messages=messages,
remaining="\n".join(remaining_urls)
)
messages = []
return output
# Clear all statuses, so we do not see the 'unviewed' class
@app.route("/api/mark-all-viewed", methods=['GET'])
def mark_all_viewed():
# Save the current newest history as the most recently viewed
for watch_uuid, watch in datastore.data['watching'].items():
datastore.set_last_viewed(watch_uuid, watch['newest_history_key'])
messages.append({'class': 'ok', 'message': "Cleared all statuses."})
return redirect(url_for('index'))
@app.route("/diff/<string:uuid>", methods=['GET'])
def diff_history_page(uuid):
global messages
# More for testing, possible to return the first/only
if uuid == 'first':
uuid= list(datastore.data['watching'].keys()).pop()
uuid = list(datastore.data['watching'].keys()).pop()
extra_stylesheets = ['/static/css/diff.css']
try:
@ -266,9 +362,9 @@ def changedetection_app(config=None, datastore_o=None):
dates.sort(reverse=True)
dates = [str(i) for i in dates]
if len(dates) < 2:
messages.append({'class': 'error', 'message': "Not enough saved change detection snapshots to produce a report."})
messages.append(
{'class': 'error', 'message': "Not enough saved change detection snapshots to produce a report."})
return redirect(url_for('index'))
# Save the current newest history as the most recently viewed
@ -409,9 +505,38 @@ def changedetection_app(config=None, datastore_o=None):
# @todo handle ctrl break
ticker_thread = threading.Thread(target=ticker_thread_check_time_launch_checks).start()
# Check for new release version
threading.Thread(target=check_for_new_version).start()
return app
# Check for new version and anonymous stats
def check_for_new_version():
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
while not app.config.exit.is_set():
try:
r = requests.post("https://changedetection.io/check-ver.php",
data={'version': datastore.data['version_tag'],
'app_guid': datastore.data['app_guid']},
verify=False)
except:
pass
try:
if "new_version" in r.text:
app.config['NEW_VERSION_AVAILABLE'] = True
except:
pass
# Check daily
app.config.exit.wait(86400)
# Requests for checking on the site use a pool of thread Workers managed by a Queue.
class Worker(threading.Thread):
current_uuid = None
@ -425,16 +550,13 @@ class Worker(threading.Thread):
update_handler = fetch_site_status.perform_site_check(datastore=datastore)
while True:
while not app.config.exit.is_set():
try:
uuid = self.q.get(block=True, timeout=1)
uuid = self.q.get(block=False)
except queue.Empty:
# We have a chance to kill this thread that needs to monitor for new jobs..
# Delays here would be caused by a current response object pending
# @todo switch to threaded response handler
if app.config['STOP_THREADS']:
return
pass
else:
self.current_uuid = uuid
@ -453,10 +575,11 @@ class Worker(threading.Thread):
# A change was detected
datastore.save_history_text(uuid=uuid, contents=contents, result_obj=result)
self.current_uuid = None # Done
self.q.task_done()
app.config.exit.wait(1)
# Thread runner to check every minute, look for new watches to feed into the Queue.
def ticker_thread_check_time_launch_checks():
@ -467,23 +590,19 @@ def ticker_thread_check_time_launch_checks():
new_worker.start()
# Every minute check for new UUIDs to follow up on
while True:
if app.config['STOP_THREADS']:
return
minutes = datastore.data['settings']['requests']['minutes_between_check']
while not app.config.exit.is_set():
running_uuids = []
for t in running_update_threads:
running_uuids.append(t.current_uuid)
# Look at the dataset, find a stale watch to process
minutes = datastore.data['settings']['requests']['minutes_between_check']
threshold = time.time() - (minutes * 60)
for uuid, watch in datastore.data['watching'].items():
if watch['last_checked'] <= time.time() - (minutes * 60):
# @todo maybe update_q.queue is enough?
if watch['last_checked'] <= threshold:
if not uuid in running_uuids and uuid not in update_q.queue:
update_q.put(uuid)
# Should be low so we can break this out in testing
time.sleep(1)
app.config.exit.wait(1)

@ -2,7 +2,8 @@ import time
import requests
import hashlib
from inscriptis import get_text
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# Some common stuff here that can be moved to a base class
class perform_site_check():
@ -11,6 +12,24 @@ class perform_site_check():
super().__init__(*args, **kwargs)
self.datastore = datastore
def strip_ignore_text(self, content, list_ignore_text):
ignore = []
for k in list_ignore_text:
ignore.append(k.encode('utf8'))
output = []
for line in content.splitlines():
line = line.encode('utf8')
# Always ignore blank lines in this mode. (when this function gets called)
if len(line.strip()):
if not any(skip_text in line for skip_text in ignore):
output.append(line)
return "\n".encode('utf8').join(output)
def run(self, uuid):
timestamp = int(time.time()) # used for storage etc too
stripped_text_from_html = False
@ -76,7 +95,15 @@ class perform_site_check():
if not len(r.text):
update_obj["last_error"] = "Empty reply"
fetched_md5 = hashlib.md5(stripped_text_from_html.encode('utf-8')).hexdigest()
# If there's text to skip
# @todo we could abstract out the get_text() to handle this cleaner
if len(self.datastore.data['watching'][uuid]['ignore_text']):
content = self.strip_ignore_text(stripped_text_from_html,
self.datastore.data['watching'][uuid]['ignore_text'])
else:
content = stripped_text_from_html.encode('utf8')
fetched_md5 = hashlib.md5(content).hexdigest()
# could be None or False depending on JSON type
if self.datastore.data['watching'][uuid]['previous_md5'] != fetched_md5:

@ -1,2 +1,12 @@
[pytest]
addopts = --no-start-live-server --live-server-port=5005
addopts = --no-start-live-server --live-server-port=5005
#testpaths = tests pytest_invenio
#live_server_scope = session
filterwarnings =
ignore::DeprecationWarning:urllib3.*:
; logging options
log_cli = 1
log_cli_level = DEBUG
log_cli_format = %(asctime)s %(name)s: %(levelname)s %(message)s

@ -88,11 +88,16 @@ section.content {
margin: 0 3px 0 5px;
}
#check-all-button {
text-align:right;
#post-list-buttons {
text-align: right;
padding: 0px;
margin: 0px;
}
#post-list-buttons li {
display: inline-block;
}
#check-all-button a {
#post-list-buttons a {
border-top-left-radius: initial;
border-top-right-radius: initial;
border-bottom-left-radius: 5px;
@ -243,4 +248,21 @@ footer {
background: #fff;
color: #444;
text-align: center;
}
}
#feed-icon {
vertical-align: middle;
}
#version {
position: absolute;
top: 80px;
right: 0px;
font-size: 8px;
background: #fff;
padding: 10px;
}
#new-version-text a{
color: #e07171;
}

@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<svg xmlns="http://www.w3.org/2000/svg"
id="RSSicon"
viewBox="0 0 8 8" width="256" height="256">
<title>RSS feed icon</title>
<style type="text/css">
.button {stroke: none; fill: orange;}
.symbol {stroke: none; fill: white;}
</style>
<rect class="button" width="8" height="8" rx="1.5" />
<circle class="symbol" cx="2" cy="6" r="1" />
<path class="symbol" d="m 1,4 a 3,3 0 0 1 3,3 h 1 a 4,4 0 0 0 -4,-4 z" />
<path class="symbol" d="m 1,2 a 5,5 0 0 1 5,5 h 1 a 6,6 0 0 0 -6,-6 z" />
</svg>

After

Width:  |  Height:  |  Size: 569 B

@ -22,10 +22,10 @@ class ChangeDetectionStore:
self.datastore_path = datastore_path
self.json_store_path = "{}/url-watches.json".format(self.datastore_path)
self.stop_thread = False
self.__data = {
'note': "Hello! If you change this file manually, please be sure to restart your changedetection.io instance!",
'watching': {},
'tag': '0.261',
'settings': {
'headers': {
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.66 Safari/537.36',
@ -53,7 +53,8 @@ class ChangeDetectionStore:
'previous_md5': "",
'uuid': str(uuid_builder.uuid4()),
'headers': {}, # Extra headers to send
'history': {} # Dict of timestamp and output stripped filename
'history': {}, # Dict of timestamp and output stripped filename
'ignore_text': [] # List of text to ignore when calculating the comparison checksum
}
if path.isfile('/source.txt'):
@ -63,6 +64,7 @@ class ChangeDetectionStore:
self.__data['build_sha'] = f.read()
try:
# @todo retest with ", encoding='utf-8'"
with open(self.json_store_path) as json_file:
from_disk = json.load(json_file)
@ -80,8 +82,7 @@ class ChangeDetectionStore:
# Reinitialise each `watching` with our generic_definition in the case that we add a new var in the future.
# @todo pretty sure theres a python we todo this with an abstracted(?) object!
for uuid, watch in self.data['watching'].items():
for uuid, watch in self.__data['watching'].items():
_blank = deepcopy(self.generic_definition)
_blank.update(watch)
self.__data['watching'].update({uuid: _blank})
@ -98,6 +99,14 @@ class ChangeDetectionStore:
self.add_watch(url='https://www.gov.uk/coronavirus', tag='Covid')
self.add_watch(url='https://changedetection.io', tag='Tech news')
self.__data['version_tag'] = "0.27"
if not 'app_guid' in self.__data:
self.__data['app_guid'] = str(uuid_builder.uuid4())
self.needs_write = True
# Finally start the thread that will manage periodic data saves to JSON
save_data_thread = threading.Thread(target=self.save_datastore).start()
@ -117,7 +126,7 @@ class ChangeDetectionStore:
return 0
def set_last_viewed(self, uuid, timestamp):
self.data['watching'][uuid].update({'last_viewed': str(timestamp)})
self.data['watching'][uuid].update({'last_viewed': int(timestamp)})
self.needs_write = True
def update_watch(self, uuid, update_obj):
@ -139,6 +148,19 @@ class ChangeDetectionStore:
@property
def data(self):
has_unviewed = False
for uuid, v in self.__data['watching'].items():
self.__data['watching'][uuid]['newest_history_key'] = self.get_newest_history_key(uuid)
if int(v['newest_history_key']) <= int(v['last_viewed']):
self.__data['watching'][uuid]['viewed'] = True
else:
self.__data['watching'][uuid]['viewed'] = False
has_unviewed = True
self.__data['has_unviewed'] = has_unviewed
return self.__data
def get_all_tags(self):
@ -156,7 +178,11 @@ class ChangeDetectionStore:
def delete(self, uuid):
with self.lock:
del (self.__data['watching'][uuid])
if uuid == 'all':
self.__data['watching'] = {}
else:
del (self.__data['watching'][uuid])
self.needs_write = True
def url_exists(self, url):

@ -19,9 +19,11 @@
<div class="home-menu pure-menu pure-menu-horizontal pure-menu-fixed">
<a class="pure-menu-heading" href="/"><strong>Change</strong>Detection.io</a>
{% if current_diff_url %}
<a class=current-diff-url href="{{ current_diff_url }}"><span style="max-width: 30%; overflow: hidden;">{{ current_diff_url }}</a>
<a class=current-diff-url href="{{ current_diff_url }}"><span style="max-width: 30%; overflow: hidden;">{{ current_diff_url }}</span></a>
{% else %}
<span id="version-text" class="pure-menu-heading">Version {{ version }}</span>
{% if new_version_available %}
<span id="new-version-text" class="pure-menu-heading"><a href="https://github.com/dgtlmoon/changedetection.io">A new version is available</a></span>
{% endif %}
{% endif %}
<ul class="pure-menu-list">
@ -36,7 +38,8 @@
<a href="/settings" class="pure-menu-link">SETTINGS</a>
</li>
<li class="pure-menu-item"><a class="github-link" href="https://github.com/dgtlmoon/changedetection.io">
<svg class="octicon octicon-mark-github v-align-middle" height="32" viewBox="0 0 16 16" version="1.1"
<svg class="octicon octicon-mark-github v-align-middle" height="32" viewBox="0 0 16 16"
version="1.1"
width="32" aria-hidden="true">
<path fill-rule="evenodd"
d="M8 0C3.58 0 0 3.58 0 8c0 3.54 2.29 6.53 5.47 7.59.4.07.55-.17.55-.38 0-.19-.01-.82-.01-1.49-2.01.37-2.53-.49-2.69-.94-.09-.23-.48-.94-.82-1.13-.28-.15-.68-.52-.01-.53.63-.01 1.08.58 1.23.82.72 1.21 1.87.87 2.33.66.07-.52.28-.87.51-1.07-1.78-.2-3.64-.89-3.64-3.95 0-.87.31-1.59.82-2.15-.08-.2-.36-1.02.08-2.12 0 0 .67-.21 2.2.82.64-.18 1.32-.27 2-.27.68 0 1.36.09 2 .27 1.53-1.04 2.2-.82 2.2-.82.44 1.1.16 1.92.08 2.12.51.56.82 1.27.82 2.15 0 3.07-1.87 3.75-3.65 3.95.29.25.54.73.54 1.48 0 1.07-.01 1.93-.01 2.2 0 .21.15.46.55.38A8.013 8.013 0 0016 8c0-4.42-3.58-8-8-8z"></path>
@ -49,7 +52,7 @@
</ul>
</div>
</div>
<div id="version">v{{ version }}</div>
<section class="content">
<header>
{% block header %}{% endblock %}

@ -4,7 +4,7 @@
<div class="edit-form">
<form class="pure-form pure-form-stacked" action="/edit?uuid={{uuid}}" method="POST">
<form class="pure-form pure-form-stacked" action="/edit/{{uuid}}" method="POST">
<fieldset>
<div class="pure-control-group">
<label for="url">URL</label>
@ -18,10 +18,26 @@
<span class="pure-form-message-inline">Grouping tags, can be a comma separated list.</span>
</div>
<!-- @todo: move to tabs --->
<fieldset class="pure-group">
<label for="ignore-text">Ignore text</label>
<textarea id="ignore-text" name="ignore-text" class="pure-input-1-2" placeholder=""
style="width: 100%;
font-family:monospace;
white-space: pre;
overflow-wrap: normal;
overflow-x: scroll;" rows="5">{% for value in watch.ignore_text %}{{ value }}
{% endfor %}</textarea>
<span class="pure-form-message-inline">Each line will be processed separately as an ignore rule.</span>
</fieldset>
<!-- @todo: move to tabs --->
<fieldset class="pure-group">
<label for="headers">Extra request headers</label>
<textarea id=headers name="headers" class="pure-input-1-2" placeholder="Example
<textarea id="headers" name="headers" class="pure-input-1-2" placeholder="Example
Cookie: foobar
User-Agent: wonderbra 1.0"
style="width: 100%;
@ -33,6 +49,8 @@ User-Agent: wonderbra 1.0"
<br/>
</fieldset>
<div class="pure-control-group">
<button type="submit" class="pure-button pure-button-primary">Save</button>
</div>

@ -15,13 +15,11 @@
<!-- user/pass r = requests.get('https://api.github.com/user', auth=('user', 'pass')) -->
</form>
<div>
<a href="/" class="pure-button button-tag {{'active' if not active_tag }}">All</a>
{% for tag in tags %}
{% if tag == "" %}
<a href="/" class="pure-button button-tag {{'active' if active_tag == tag }}">All</a>
{% else %}
<a href="/?tag={{ tag}}" class="pure-button button-tag {{'active' if active_tag == tag }}">{{ tag }}</a>
{% endif %}
{% if tag != "" %}
<a href="/?tag={{ tag}}" class="pure-button button-tag {{'active' if active_tag == tag }}">{{ tag }}</a>
{% endif %}
{% endfor %}
</div>
@ -64,22 +62,29 @@
<td>
<a href="/api/checknow?uuid={{ watch.uuid}}{% if request.args.get('tag') %}&tag={{request.args.get('tag')}}{% endif %}"
class="pure-button button-small pure-button-primary">Recheck</a>
<a href="/edit?uuid={{ watch.uuid}}" class="pure-button button-small pure-button-primary">Edit</a>
<a href="/edit/{{ watch.uuid}}" class="pure-button button-small pure-button-primary">Edit</a>
{% if watch.history|length >= 2 %}
<a href="/diff/{{ watch.uuid}}" class="pure-button button-small pure-button-primary">Diff</a>
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
<div id="check-all-button">
<a href="/api/checknow{% if active_tag%}?tag={{active_tag}}{%endif%}" class="pure-button button-tag ">Recheck
<ul id="post-list-buttons">
{% if has_unviewed %}
<li>
<a href="/api/mark-all-viewed" class="pure-button button-tag ">Mark all viewed</a>
</li>
{% endif %}
<li>
<a href="/api/checknow{% if active_tag%}?tag={{active_tag}}{%endif%}" class="pure-button button-tag ">Recheck
all {% if active_tag%}in "{{active_tag}}"{%endif%}</a>
</div>
</li>
<li>
<a href="{{ url_for('index', tag=active_tag , rss=true)}}"><img id="feed-icon" src="/static/images/Generic_Feed-icon.svg" height="15px"></a>
</li>
</ul>
</div>
</div>
{% endblock %}

@ -7,17 +7,14 @@ import os
# https://github.com/pallets/flask/blob/1.1.2/examples/tutorial/tests/test_auth.py
# Much better boilerplate than the docs
# https://www.python-boilerplate.com/py3+flask+pytest/
global app
@pytest.fixture(scope='session')
def app(request):
"""Create application for the tests."""
datastore_path = "./test-datastore"
try:
@ -33,11 +30,19 @@ def app(request):
app_config = {'datastore_path': datastore_path}
datastore = store.ChangeDetectionStore(datastore_path=app_config['datastore_path'], include_default_watches=False)
app = changedetection_app(app_config, datastore)
app.config['STOP_THREADS'] = True
def teardown():
datastore.stop_thread = True
app.config['STOP_THREADS'] = True
app.config.exit.set()
try:
os.unlink("{}/url-watches.json".format(datastore_path))
except FileNotFoundError:
# This is fine in the case of a failure.
pass
assert 1 == 1
request.addfinalizer(teardown)
yield app
return app

@ -3,6 +3,21 @@
import time
from flask import url_for
from urllib.request import urlopen
import pytest
sleep_time_for_fetch_thread = 3
def test_setup_liveserver(live_server):
@live_server.app.route('/test-endpoint')
def test_endpoint():
# Tried using a global var here but didn't seem to work, so reading from a file instead.
with open("test-datastore/output.txt", "r") as f:
return f.read()
live_server.start()
assert 1 == 1
def set_original_response():
@ -14,7 +29,6 @@ def set_original_response():
So let's see what happens. </br>
</body>
</html>
"""
with open("test-datastore/output.txt", "w") as f:
@ -30,7 +44,6 @@ def set_modified_response():
So let's see what happens. </br>
</body>
</html>
"""
with open("test-datastore/output.txt", "w") as f:
@ -38,18 +51,8 @@ def set_modified_response():
def test_check_basic_change_detection_functionality(client, live_server):
sleep_time_for_fetch_thread = 5
@live_server.app.route('/test-endpoint')
def test_endpoint():
# Tried using a global var here but didn't seem to work, so reading from a file instead.
with open("test-datastore/output.txt", "r") as f:
return f.read()
set_original_response()
live_server.start()
# Add our URL to the import page
res = client.post(
url_for("import_page"),
@ -91,13 +94,13 @@ def test_check_basic_change_detection_functionality(client, live_server):
assert b'unviewed' in res.data
# Following the 'diff' link, it should no longer display as 'unviewed' even after we recheck it a few times
res = client.get(url_for("diff_history_page", uuid="first") )
res = client.get(url_for("diff_history_page", uuid="first"))
assert b'Compare newest' in res.data
time.sleep(2)
# Do this a few times.. ensures we dont accidently set the status
for n in range(3):
for n in range(2):
client.get(url_for("api_watch_checknow"), follow_redirects=True)
# Give the thread time to pick it up
@ -108,10 +111,13 @@ def test_check_basic_change_detection_functionality(client, live_server):
assert b'unviewed' not in res.data
assert b'test-endpoint' in res.data
set_original_response()
client.get(url_for("api_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
res = client.get(url_for("index"))
assert b'unviewed' in res.data
assert b'unviewed' in res.data
# Cleanup everything
res = client.get(url_for("api_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data

@ -0,0 +1,152 @@
#!/usr/bin/python3
import time
from flask import url_for
from urllib.request import urlopen
import pytest
# Unit test of the stripper
# Always we are dealing in utf-8
def test_strip_text_func():
from backend import fetch_site_status
test_content = """
Some content
is listed here
but sometimes we want to remove the lines.
but not always."""
ignore_lines = ["sometimes"]
fetcher = fetch_site_status.perform_site_check(datastore=False)
stripped_content = fetcher.strip_ignore_text(test_content, ignore_lines)
assert b"sometimes" not in stripped_content
assert b"Some content" in stripped_content
def set_original_ignore_response():
test_return_data = """<html>
<body>
Some initial text</br>
<p>Which is across multiple lines</p>
</br>
So let's see what happens. </br>
</body>
</html>
"""
with open("test-datastore/output.txt", "w") as f:
f.write(test_return_data)
def set_modified_original_ignore_response():
test_return_data = """<html>
<body>
Some NEW nice initial text</br>
<p>Which is across multiple lines</p>
</br>
So let's see what happens. </br>
</body>
</html>
"""
with open("test-datastore/output.txt", "w") as f:
f.write(test_return_data)
# Is the same but includes ZZZZZ, 'ZZZZZ' is the last line in ignore_text
def set_modified_ignore_response():
test_return_data = """<html>
<body>
Some initial text</br>
<p>Which is across multiple lines</p>
<P>ZZZZZ</P>
</br>
So let's see what happens. </br>
</body>
</html>
"""
with open("test-datastore/output.txt", "w") as f:
f.write(test_return_data)
def test_check_ignore_text_functionality(client, live_server):
sleep_time_for_fetch_thread = 3
ignore_text = "XXXXX\nYYYYY\nZZZZZ"
set_original_ignore_response()
# 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)
# Goto the edit page, add our ignore text
# Add our URL to the import page
res = client.post(
url_for("edit_page", uuid="first"),
data={"ignore-text": ignore_text, "url": test_url, "tag": "", "headers": ""},
follow_redirects=True
)
assert b"Updated watch." in res.data
# Check it saved
res = client.get(
url_for("edit_page", uuid="first"),
)
assert bytes(ignore_text.encode('utf-8')) 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)
# It should report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
assert b'unviewed' not in res.data
assert b'/test-endpoint' in res.data
# Make a change
set_modified_ignore_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 report nothing found (no new 'unviewed' class)
res = client.get(url_for("index"))
assert b'unviewed' not in res.data
assert b'/test-endpoint' in res.data
# Just to be sure.. set a regular modified change..
set_modified_original_ignore_response()
client.get(url_for("api_watch_checknow"), follow_redirects=True)
time.sleep(sleep_time_for_fetch_thread)
res = client.get(url_for("index"))
assert b'unviewed' in res.data
res = client.get(url_for("api_delete", uuid="all"), follow_redirects=True)
assert b'Deleted' in res.data

@ -52,7 +52,11 @@ def main(argv):
@app.context_processor
def inject_version():
return dict(version=datastore.data['tag'])
return dict(version=datastore.data['version_tag'])
@app.context_processor
def inject_new_version_available():
return dict(new_version_available=app.config['NEW_VERSION_AVAILABLE'])
if ssl_mode:
# @todo finalise SSL config, but this should get you in the right direction if you need it.

@ -8,4 +8,6 @@ requests
validators
timeago ~=1.0
inscriptis ~= 1.1
feedgen ~= 0.9
pytz
urllib3
Loading…
Cancel
Save