Compare commits

..

1 Commits

23 changed files with 162 additions and 611 deletions

View File

@@ -28,7 +28,7 @@ jobs:
test-pypi-package:
name: Test the built package works basically.
name: Test the built 📦 package works basically.
runs-on: ubuntu-latest
needs:
- build
@@ -42,39 +42,18 @@ jobs:
uses: actions/setup-python@v6
with:
python-version: '3.11'
- name: Test that the basic pip built package runs without error
run: |
set -ex
ls -alR
# Install the first wheel found in dist/
WHEEL=$(find dist -type f -name "*.whl" -print -quit)
echo Installing $WHEEL
python3 -m pip install --upgrade pip
python3 -m pip install "$WHEEL"
# Find and install the first .whl file
find dist -type f -name "*.whl" -exec pip3 install {} \; -quit
changedetection.io -d /tmp -p 10000 &
sleep 3
curl --retry-connrefused --retry 6 http://127.0.0.1:10000/static/styles/pure-min.css >/dev/null
curl --retry-connrefused --retry 6 http://127.0.0.1:10000/ >/dev/null
# --- API test ---
# This also means that the docs/api-spec.yml was shipped and could be read
test -f /tmp/url-watches.json
API_KEY=$(jq -r '.. | .api_access_token? // empty' /tmp/url-watches.json)
echo Test API KEY is $API_KEY
curl -X POST "http://127.0.0.1:10000/api/v1/watch" \
-H "x-api-key: ${API_KEY}" \
-H "Content-Type: application/json" \
--show-error --fail \
--retry 6 --retry-delay 1 --retry-connrefused \
-d '{
"url": "https://example.com",
"title": "Example Site Monitor",
"time_between_check": { "hours": 1 }
}'
killall changedetection.io

View File

@@ -54,10 +54,7 @@ jobs:
- name: Spin up ancillary SMTP+Echo message test server
run: |
# Debug SMTP server/echo message back server, telnet 11080 to it should immediately bounce back the most recent message that tried to send (then you can see if cdio tried to send, the format, etc)
# 11025 is the SMTP port for testing
# apprise example would be 'mailto://changedetection@localhost:11025/?to=fff@home.com (it will also echo to STDOUT)
# telnet localhost 11080
# Debug SMTP server/echo message back server
docker run --network changedet-network -d -p 11025:11025 -p 11080:11080 --hostname mailserver test-changedetectionio bash -c 'pip3 install aiosmtpd && python changedetectionio/tests/smtp/smtp-test-server.py'
docker ps

View File

@@ -1,5 +1,4 @@
recursive-include changedetectionio/api *
include docs/api-spec.yaml
recursive-include changedetectionio/blueprint *
recursive-include changedetectionio/conditions *
recursive-include changedetectionio/content_fetchers *

View File

@@ -2,7 +2,7 @@
# Read more https://github.com/dgtlmoon/changedetection.io/wiki
__version__ = '0.50.28'
__version__ = '0.50.24'
from changedetectionio.strtobool import strtobool
from json.decoder import JSONDecodeError

View File

@@ -37,10 +37,6 @@ def get_openapi_spec():
from openapi_core import OpenAPI # Lazy import - saves ~10.7 MB on startup
spec_path = os.path.join(os.path.dirname(__file__), '../../docs/api-spec.yaml')
if not os.path.exists(spec_path):
# Possibly for pip3 packages
spec_path = os.path.join(os.path.dirname(__file__), '../docs/api-spec.yaml')
with open(spec_path, 'r') as f:
spec_dict = yaml.safe_load(f)
_openapi_spec = OpenAPI.from_dict(spec_dict)

View File

@@ -408,9 +408,6 @@ def strip_ignore_text(content, wordlist, mode="content"):
ignored_lines = []
for k in wordlist:
# Skip empty strings to avoid matching everything
if not k or not k.strip():
continue
# Is it a regex?
res = re.search(PERL_STYLE_REGEX, k, re.IGNORECASE)
if res:

View File

@@ -89,8 +89,9 @@ class model(watch_base):
ready_url = jinja_render(template_str=url)
except Exception as e:
logger.critical(f"Invalid URL template for: '{url}' - {str(e)}")
from flask import flash, url_for
from markupsafe import Markup
from flask import (
flash, Markup, url_for
)
message = Markup('<a href="{}#general">The URL {} is invalid and cannot be used, click to edit</a>'.format(
url_for('ui.ui_edit.edit_page', uuid=self.get('uuid')), self.get('url', '')))
flash(message, 'error')

View File

@@ -1,5 +1,6 @@
from changedetectionio.model import default_notification_format_for_watch
ult_notification_format_for_watch = 'System default'
default_notification_format = 'HTML Color'
default_notification_body = '{{watch_url}} had a change.\n---\n{{diff}}\n---\n'
default_notification_title = 'ChangeDetection.io Notification - {{watch_url}}'
@@ -7,10 +8,10 @@ default_notification_title = 'ChangeDetection.io Notification - {{watch_url}}'
# The values (markdown etc) are from apprise NotifyFormat,
# But to avoid importing the whole heavy module just use the same strings here.
valid_notification_formats = {
'Plain Text': 'text',
'Text': 'text',
'Markdown': 'markdown',
'HTML': 'html',
'HTML Color': 'htmlcolor',
'Markdown to HTML': 'markdown',
# Used only for editing a watch (not for global)
default_notification_format_for_watch: default_notification_format_for_watch
}

View File

@@ -70,7 +70,6 @@ def apprise_http_custom_handler(
title: str,
notify_type: str,
meta: dict,
body_format: str = None,
*args,
**kwargs,
) -> bool:

View File

@@ -1,134 +1,17 @@
import time
import apprise
from apprise import NotifyFormat
from loguru import logger
from urllib.parse import urlparse
from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
from .apprise_plugin.custom_handlers import SUPPORTED_HTTP_METHODS
from ..notification_service import NotificationContextData
def markup_text_links_to_html(body):
"""
Convert plaintext to HTML with clickable links.
Uses Jinja2's escape and Markup for XSS safety.
"""
from linkify_it import LinkifyIt
from markupsafe import Markup, escape
linkify = LinkifyIt()
# Match URLs in the ORIGINAL text (before escaping)
matches = linkify.match(body)
if not matches:
# No URLs, just escape everything
return Markup(escape(body))
result = []
last_index = 0
# Process each URL match
for match in matches:
# Add escaped text before the URL
if match.index > last_index:
text_part = body[last_index:match.index]
result.append(escape(text_part))
# Add the link with escaped URL (both in href and display)
url = match.url
result.append(Markup(f'<a href="{escape(url)}">{escape(url)}</a>'))
last_index = match.last_index
# Add remaining escaped text
if last_index < len(body):
result.append(escape(body[last_index:]))
# Join all parts
return str(Markup(''.join(str(part) for part in result)))
def notification_format_align_with_apprise(n_format : str):
"""
Correctly align changedetection's formats with apprise's formats
Probably these are the same - but good to be sure.
These set the expected OUTPUT format type
:param n_format:
:return:
"""
if n_format.lower().startswith('html'):
# Apprise only knows 'html' not 'htmlcolor' etc, which shouldnt matter here
n_format = NotifyFormat.HTML.value
elif n_format.lower().startswith('markdown'):
# probably the same but just to be safe
n_format = NotifyFormat.MARKDOWN.value
elif n_format.lower().startswith('text'):
# probably the same but just to be safe
n_format = NotifyFormat.TEXT.value
else:
n_format = NotifyFormat.TEXT.value
return n_format
def apply_service_tweaks(url, n_body, n_title):
# Re 323 - Limit discord length to their 2000 char limit total or it wont send.
# Because different notifications may require different pre-processing, run each sequentially :(
# 2000 bytes minus -
# 200 bytes for the overhead of the _entire_ json payload, 200 bytes for {tts, wait, content} etc headers
# Length of URL - Incase they specify a longer custom avatar_url
# So if no avatar_url is specified, add one so it can be correctly calculated into the total payload
parsed = urlparse(url)
k = '?' if not parsed.query else '&'
if not 'avatar_url' in url \
and not url.startswith('mail') \
and not url.startswith('post') \
and not url.startswith('get') \
and not url.startswith('delete') \
and not url.startswith('put'):
url += k + f"avatar_url={APPRISE_AVATAR_URL}"
if url.startswith('tgram://'):
# Telegram only supports a limit subset of HTML, remove the '<br>' we place in.
# re https://github.com/dgtlmoon/changedetection.io/issues/555
# @todo re-use an existing library we have already imported to strip all non-allowed tags
n_body = n_body.replace('<br>', '\n')
n_body = n_body.replace('</br>', '\n')
# real limit is 4096, but minus some for extra metadata
payload_max_size = 3600
body_limit = max(0, payload_max_size - len(n_title))
n_title = n_title[0:payload_max_size]
n_body = n_body[0:body_limit]
elif url.startswith('discord://') or url.startswith('https://discordapp.com/api/webhooks') or url.startswith(
'https://discord.com/api'):
# real limit is 2000, but minus some for extra metadata
payload_max_size = 1700
body_limit = max(0, payload_max_size - len(n_title))
n_title = n_title[0:payload_max_size]
n_body = n_body[0:body_limit]
return url, n_body, n_title
def process_notification(n_object: NotificationContextData, datastore):
from changedetectionio.jinja2_custom import render as jinja_render
from . import default_notification_format_for_watch, default_notification_format, valid_notification_formats
# be sure its registered
from .apprise_plugin.custom_handlers import apprise_http_custom_handler
# Create list of custom handler protocols (both http and https versions)
custom_handler_protocols = [f"{method}://" for method in SUPPORTED_HTTP_METHODS]
custom_handler_protocols += [f"{method}s://" for method in SUPPORTED_HTTP_METHODS]
has_custom_handler = any(
url.startswith(tuple(custom_handler_protocols))
for url in n_object['notification_urls']
)
if not isinstance(n_object, NotificationContextData):
raise TypeError(f"Expected NotificationContextData, got {type(n_object)}")
@@ -139,25 +22,18 @@ def process_notification(n_object: NotificationContextData, datastore):
# Insert variables into the notification content
notification_parameters = create_notification_parameters(n_object, datastore)
requested_output_format = valid_notification_formats.get(
n_format = valid_notification_formats.get(
n_object.get('notification_format', default_notification_format),
valid_notification_formats[default_notification_format],
)
# If we arrived with 'System default' then look it up
if requested_output_format == default_notification_format_for_watch and datastore.data['settings']['application'].get('notification_format') != default_notification_format_for_watch:
if n_format == default_notification_format_for_watch and datastore.data['settings']['application'].get('notification_format') != default_notification_format_for_watch:
# Initially text or whatever
requested_output_format = datastore.data['settings']['application'].get('notification_format', valid_notification_formats[default_notification_format]).lower()
requested_output_format = notification_format_align_with_apprise(n_format=requested_output_format)
n_format = datastore.data['settings']['application'].get('notification_format', valid_notification_formats[default_notification_format])
logger.trace(f"Complete notification body including Jinja and placeholders calculated in {time.time() - now:.2f}s")
# If we have custom handlers, use invalid format to prevent conversion
# Otherwise use the proper format
if has_custom_handler:
input_format = 'raw-no-convert'
# https://github.com/caronc/apprise/wiki/Development_LogCapture
# Anything higher than or equal to WARNING (which covers things like Connection errors)
# raise it as an exception
@@ -174,41 +50,11 @@ def process_notification(n_object: NotificationContextData, datastore):
with apprise.LogCapture(level=apprise.logging.DEBUG) as logs:
for url in n_object['notification_urls']:
parsed_url = urlparse(url)
prefix_add_to_url = '?' if not parsed_url.query else '&'
# Get the notification body from datastore
n_body = jinja_render(template_str=n_object.get('notification_body', ''), **notification_parameters)
if n_object.get('markup_text_to_html'):
n_body = markup_text_links_to_html(body=n_body)
# This actually means we request "Markdown to HTML"
if requested_output_format == NotifyFormat.MARKDOWN.value:
output_format = NotifyFormat.HTML.value
input_format = NotifyFormat.MARKDOWN.value
if not 'format=' in url.lower():
url = f"{url}{prefix_add_to_url}format={output_format}"
# Deviation from apprise.
# No conversion, its like they want to send raw HTML but we add linebreaks
elif requested_output_format == NotifyFormat.HTML.value:
# same in and out means apprise wont try to convert
input_format = output_format = NotifyFormat.HTML.value
if n_format.lower().startswith('html'):
n_body = n_body.replace("\n", '<br>')
if not 'format=' in url.lower():
url = f"{url}{prefix_add_to_url}format={output_format}"
else:
# Nothing to be done, leave it as plaintext
# `body_format` Tell apprise what format the INPUT is in
# &format= in URL Tell apprise what format the OUTPUT should be in (it can convert between)
input_format = output_format = NotifyFormat.TEXT.value
if not 'format=' in url.lower():
url = f"{url}{prefix_add_to_url}format={output_format}"
if has_custom_handler:
input_format='raw-no-convert'
n_title = jinja_render(template_str=n_object.get('notification_title', ''), **notification_parameters)
@@ -224,28 +70,74 @@ def process_notification(n_object: NotificationContextData, datastore):
logger.info(f">> Process Notification: AppRise notifying {url}")
url = jinja_render(template_str=url, **notification_parameters)
(url, n_body, n_title) = apply_service_tweaks(url=url, n_body=n_body, n_title=n_title)
# Re 323 - Limit discord length to their 2000 char limit total or it wont send.
# Because different notifications may require different pre-processing, run each sequentially :(
# 2000 bytes minus -
# 200 bytes for the overhead of the _entire_ json payload, 200 bytes for {tts, wait, content} etc headers
# Length of URL - Incase they specify a longer custom avatar_url
# So if no avatar_url is specified, add one so it can be correctly calculated into the total payload
k = '?' if not '?' in url else '&'
if not 'avatar_url' in url \
and not url.startswith('mail') \
and not url.startswith('post') \
and not url.startswith('get') \
and not url.startswith('delete') \
and not url.startswith('put'):
url += k + f"avatar_url={APPRISE_AVATAR_URL}"
if url.startswith('tgram://'):
# Telegram only supports a limit subset of HTML, remove the '<br>' we place in.
# re https://github.com/dgtlmoon/changedetection.io/issues/555
# @todo re-use an existing library we have already imported to strip all non-allowed tags
n_body = n_body.replace('<br>', '\n')
n_body = n_body.replace('</br>', '\n')
# real limit is 4096, but minus some for extra metadata
payload_max_size = 3600
body_limit = max(0, payload_max_size - len(n_title))
n_title = n_title[0:payload_max_size]
n_body = n_body[0:body_limit]
elif url.startswith('discord://') or url.startswith('https://discordapp.com/api/webhooks') or url.startswith(
'https://discord.com/api'):
# real limit is 2000, but minus some for extra metadata
payload_max_size = 1700
body_limit = max(0, payload_max_size - len(n_title))
n_title = n_title[0:payload_max_size]
n_body = n_body[0:body_limit]
elif url.startswith('mailto'):
# Apprise will default to HTML, so we need to override it
# So that whats' generated in n_body is in line with what is going to be sent.
# https://github.com/caronc/apprise/issues/633#issuecomment-1191449321
if not 'format=' in url and (n_format == 'Text' or n_format == 'Markdown'):
prefix = '?' if not '?' in url else '&'
# Apprise format is lowercase text https://github.com/caronc/apprise/issues/633
n_format = n_format.lower()
url = f"{url}{prefix}format={n_format}"
# If n_format == HTML, then apprise email should default to text/html and we should be sending HTML only
apobj.add(url)
sent_objs.append({'title': n_title,
'body': n_body,
'url': url})
'url': url,
'body_format': n_format})
# Blast off the notifications tht are set in .add()
apobj.notify(
title=n_title,
body=n_body,
# `body_format` Tell apprise what format the INPUT is in
# &format= in URL Tell apprise what format the OUTPUT should be in (it can convert between)
body_format=input_format,
body_format=n_format,
# False is not an option for AppRise, must be type None
attach=n_object.get('screenshot', None)
)
# Returns empty string if nothing found, multi-line string otherwise
log_value = logs.getvalue()
if log_value and ('WARNING' in log_value or 'ERROR' in log_value):
if log_value and 'WARNING' in log_value or 'ERROR' in log_value:
logger.critical(log_value)
raise Exception(log_value)

View File

@@ -9,8 +9,6 @@ for both sync and async workers
from loguru import logger
import time
from changedetectionio.notification import default_notification_format
# What is passed around as notification context, also used as the complete list of valid {{ tokens }}
class NotificationContextData(dict):
def __init__(self, initial_data=None, **kwargs):
@@ -30,8 +28,7 @@ class NotificationContextData(dict):
'diff_url': None,
'preview_url': None,
'watch_tag': None,
'watch_title': None,
'markup_text_to_html': False, # If automatic conversion of plaintext to HTML should happen
'watch_title': None
})
# Apply any initial data passed in
@@ -228,25 +225,12 @@ class NotificationService:
if not watch:
return
n_format = self.datastore.data['settings']['application'].get('notification_format', default_notification_format)
filter_list = ", ".join(watch['include_filters'])
# @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_to_html' is not needed
body = f"""Hello,
Your configured CSS/xPath filters of '{filter_list}' for {{{{watch_url}}}} did not appear on the page after {threshold} attempts.
It's possible the page changed layout and the filter needs updating ( Try the 'Visual Selector' tab )
Edit link: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}
Thanks - Your omniscient changedetection.io installation.
"""
n_object = NotificationContextData({
'notification_title': 'Changedetection.io - Alert - CSS/xPath filter was not present in the page',
'notification_body': body,
'notification_format': n_format,
'markup_text_to_html': n_format.lower().startswith('html')
'notification_body': "Your configured CSS/xPath filters of '{}' for {{{{watch_url}}}} did not appear on the page after {} attempts, did the page change layout?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\nThanks - Your omniscient changedetection.io installation :)\n".format(
", ".join(watch['include_filters']),
threshold),
'notification_format': 'text'
})
if len(watch['notification_urls']):
@@ -275,27 +259,13 @@ Thanks - Your omniscient changedetection.io installation.
if not watch:
return
threshold = self.datastore.data['settings']['application'].get('filter_failure_notification_threshold_attempts')
n_format = self.datastore.data['settings']['application'].get('notification_format', default_notification_format).lower()
step = step_n + 1
# @todo - This could be a markdown template on the disk, apprise will convert the markdown to HTML+Plaintext parts in the email, and then 'markup_text_to_html' is not needed
# {{{{ }}}} because this will be Jinja2 {{ }} tokens
body = f"""Hello,
Your configured browser step at position {step} for the web page watch {{{{watch_url}}}} did not appear on the page after {threshold} attempts, did the page change layout?
The element may have moved and needs editing, or does it need a delay added?
Edit link: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}
Thanks - Your omniscient changedetection.io installation.
"""
n_object = NotificationContextData({
'notification_title': f"Changedetection.io - Alert - Browser step at position {step} could not be run",
'notification_body': body,
'notification_format': n_format,
'markup_text_to_html': n_format.lower().startswith('html')
'notification_title': "Changedetection.io - Alert - Browser step at position {} could not be run".format(step_n+1),
'notification_body': "Your configured browser step at position {} for {{{{watch_url}}}} "
"did not appear on the page after {} attempts, did the page change layout? "
"Does it need a delay added?\n\nLink: {{{{base_url}}}}/edit/{{{{watch_uuid}}}}\n\n"
"Thanks - Your omniscient changedetection.io installation :)\n".format(step_n+1, threshold),
'notification_format': 'text'
})
if len(watch['notification_urls']):

View File

@@ -324,13 +324,13 @@ class ContentProcessor:
append_pretty_line_formatting=not self.watch.is_source_type_url
)
# Raise error if filter returned nothing
if not filtered_content.strip():
raise FilterNotFoundInResponse(
msg=self.filter_config.include_filters,
screenshot=self.fetcher.screenshot,
xpath_data=self.fetcher.xpath_data
)
# Raise error if filter returned nothing
if not filtered_content.strip():
raise FilterNotFoundInResponse(
msg=self.filter_config.include_filters,
screenshot=self.fetcher.screenshot,
xpath_data=self.fetcher.xpath_data
)
return filtered_content

View File

@@ -15,7 +15,7 @@ find tests/test_*py -type f|while read test_name
do
echo "TEST RUNNING $test_name"
# REMOVE_REQUESTS_OLD_SCREENSHOTS disabled so that we can write a screenshot and send it in test_notifications.py without a real browser
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest -vv -s --maxfail=1 --tb=long $test_name
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest $test_name
done
echo "RUNNING WITH BASE_URL SET"
@@ -23,20 +23,20 @@ echo "RUNNING WITH BASE_URL SET"
# Now re-run some tests with BASE_URL enabled
# Re #65 - Ability to include a link back to the installation, in the notification.
export BASE_URL="https://really-unique-domain.io"
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest -vv -s --maxfail=1 tests/test_notification.py
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest tests/test_notification.py
# Re-run with HIDE_REFERER set - could affect login
export HIDE_REFERER=True
pytest -vv -s --maxfail=1 tests/test_access_control.py
pytest tests/test_access_control.py
# Re-run a few tests that will trigger brotli based storage
export SNAPSHOT_BROTLI_COMPRESSION_THRESHOLD=5
pytest -vv -s --maxfail=1 tests/test_access_control.py
pytest tests/test_access_control.py
REMOVE_REQUESTS_OLD_SCREENSHOTS=false pytest tests/test_notification.py
pytest -vv -s --maxfail=1 tests/test_backend.py
pytest -vv -s --maxfail=1 tests/test_rss.py
pytest -vv -s --maxfail=1 tests/test_unique_lines.py
pytest tests/test_backend.py
pytest tests/test_rss.py
pytest tests/test_unique_lines.py
# Try high concurrency
FETCH_WORKERS=130 pytest tests/test_history_consistency.py -v -l

View File

@@ -228,36 +228,26 @@ class ChangeDetectionStore:
d['settings']['application']['active_base_url'] = active_base_url.strip('" ')
return d
from pathlib import Path
def delete_path(self, path: Path):
import shutil
"""Delete a file or directory tree, including the path itself."""
if not path.exists():
return
if path.is_file() or path.is_symlink():
path.unlink(missing_ok=True) # deletes a file or symlink
else:
shutil.rmtree(path, ignore_errors=True) # deletes dir *and* its contents
# Delete a single watch by UUID
def delete(self, uuid):
import pathlib
import shutil
with self.lock:
if uuid == 'all':
self.__data['watching'] = {}
time.sleep(1) # Mainly used for testing to allow all items to flush before running next test
# GitHub #30 also delete history records
for uuid in self.data['watching']:
path = pathlib.Path(os.path.join(self.datastore_path, uuid))
if os.path.exists(path):
self.delete(uuid)
shutil.rmtree(path)
else:
path = pathlib.Path(os.path.join(self.datastore_path, uuid))
if os.path.exists(path):
self.delete_path(path)
shutil.rmtree(path)
del self.data['watching'][uuid]
self.needs_write_urgent = True

View File

@@ -1,15 +1,14 @@
import json
import os
import time
import re
from flask import url_for
from email import message_from_string
from email.policy import default as email_policy
from changedetectionio.diff import REMOVED_STYLE, ADDED_STYLE
from changedetectionio.tests.util import set_original_response, set_modified_response, set_more_modified_response, live_server_setup, \
wait_for_all_checks, \
set_longer_modified_response, delete_all_watches
from changedetectionio.tests.util import extract_UUID_from_client
import logging
import base64
# NOTE - RELIES ON mailserver as hostname running, see github build recipes
smtp_test_server = 'mailserver'
@@ -51,7 +50,7 @@ def test_check_notification_email_formats_default_HTML(client, live_server, meas
url_for("settings.settings_page"),
data={"application-notification_urls": notification_url,
"application-notification_title": "fallback-title " + default_notification_title,
"application-notification_body": "some text\nfallback-body<br> " + default_notification_body,
"application-notification_body": "fallback-body<br> " + default_notification_body,
"application-notification_format": 'HTML',
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
@@ -78,229 +77,19 @@ def test_check_notification_email_formats_default_HTML(client, live_server, meas
time.sleep(3)
msg_raw = get_last_message_from_smtp_server()
assert len(msg_raw) >= 1
msg = get_last_message_from_smtp_server()
assert len(msg) >= 1
# Parse the email properly using Python's email library
msg = message_from_string(msg_raw, policy=email_policy)
# The email should have two bodies (multipart/alternative with text/plain and text/html)
assert msg.is_multipart()
assert msg.get_content_type() == 'multipart/alternative'
# Get the parts
parts = list(msg.iter_parts())
assert len(parts) == 2
# First part should be text/plain (the auto-generated plaintext version)
text_part = parts[0]
assert text_part.get_content_type() == 'text/plain'
text_content = text_part.get_content()
assert '(added) So let\'s see what happens.\r\n' in text_content # The plaintext part
assert 'fallback-body\r\n' in text_content # The plaintext part
# Second part should be text/html
html_part = parts[1]
assert html_part.get_content_type() == 'text/html'
html_content = html_part.get_content()
assert 'some text<br>' in html_content # We converted \n from the notification body
assert 'fallback-body<br>' in html_content # kept the original <br>
assert '(added) So let\'s see what happens.<br>' in html_content # the html part
delete_all_watches(client)
def test_check_notification_plaintext_format(client, live_server, measure_memory_usage):
set_original_response()
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
#####################
# Set this up for when we remove the notification from the watch, it should fallback with these details
res = client.post(
url_for("settings.settings_page"),
data={"application-notification_urls": notification_url,
"application-notification_title": "fallback-title " + default_notification_title,
"application-notification_body": "some text\n" + default_notification_body,
"application-notification_format": 'Plain Text',
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Settings updated." in res.data
# Add a watch and trigger a HTTP POST
test_url = url_for('test_endpoint', _external=True)
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
time.sleep(2)
set_longer_modified_response()
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
time.sleep(3)
msg_raw = get_last_message_from_smtp_server()
assert len(msg_raw) >= 1
# Parse the email properly using Python's email library
msg = message_from_string(msg_raw, policy=email_policy)
# The email should be plain text only (not multipart)
assert not msg.is_multipart()
assert msg.get_content_type() == 'text/plain'
# Get the plain text content
text_content = msg.get_content()
assert '(added) So let\'s see what happens.\r\n' in text_content # The plaintext part
# Should NOT contain HTML
assert '<br>' not in text_content # We should not have HTML in plain text
delete_all_watches(client)
def test_check_notification_html_color_format(client, live_server, measure_memory_usage):
set_original_response()
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
#####################
# Set this up for when we remove the notification from the watch, it should fallback with these details
res = client.post(
url_for("settings.settings_page"),
data={"application-notification_urls": notification_url,
"application-notification_title": "fallback-title " + default_notification_title,
"application-notification_body": "some text\n" + default_notification_body,
"application-notification_format": 'HTML Color',
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Settings updated." in res.data
# Add a watch and trigger a HTTP POST
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": test_url, "tags": 'nice one'},
follow_redirects=True
)
assert b"Watch added" in res.data
wait_for_all_checks(client)
set_longer_modified_response()
time.sleep(2)
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
time.sleep(3)
msg_raw = get_last_message_from_smtp_server()
assert len(msg_raw) >= 1
# Parse the email properly using Python's email library
msg = message_from_string(msg_raw, policy=email_policy)
# The email should have two bodies (multipart/alternative with text/plain and text/html)
assert msg.is_multipart()
assert msg.get_content_type() == 'multipart/alternative'
# Get the parts
parts = list(msg.iter_parts())
assert len(parts) == 2
# First part should be text/plain (the auto-generated plaintext version)
text_part = parts[0]
assert text_part.get_content_type() == 'text/plain'
text_content = text_part.get_content()
assert 'So let\'s see what happens.\r\n' in text_content # The plaintext part
assert '(added)' not in text_content # Because apprise only dumb converts the html to text
# Second part should be text/html with color styling
html_part = parts[1]
assert html_part.get_content_type() == 'text/html'
html_content = html_part.get_content()
assert REMOVED_STYLE in html_content
assert ADDED_STYLE in html_content
assert 'some text<br>' in html_content
delete_all_watches(client)
def test_check_notification_markdown_format(client, live_server, measure_memory_usage):
set_original_response()
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
#####################
# Set this up for when we remove the notification from the watch, it should fallback with these details
res = client.post(
url_for("settings.settings_page"),
data={"application-notification_urls": notification_url,
"application-notification_title": "fallback-title " + default_notification_title,
"application-notification_body": "*header*\n\nsome text\n" + default_notification_body,
"application-notification_format": 'Markdown to HTML',
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
follow_redirects=True
)
assert b"Settings updated." in res.data
# Add a watch and trigger a HTTP POST
test_url = url_for('test_endpoint', _external=True)
res = client.post(
url_for("ui.ui_views.form_quick_watch_add"),
data={"url": test_url, "tags": 'nice one'},
follow_redirects=True
)
assert b"Watch added" in res.data
wait_for_all_checks(client)
set_longer_modified_response()
time.sleep(2)
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
wait_for_all_checks(client)
time.sleep(3)
msg_raw = get_last_message_from_smtp_server()
assert len(msg_raw) >= 1
# Parse the email properly using Python's email library
msg = message_from_string(msg_raw, policy=email_policy)
# The email should have two bodies (multipart/alternative with text/plain and text/html)
assert msg.is_multipart()
assert msg.get_content_type() == 'multipart/alternative'
# Get the parts
parts = list(msg.iter_parts())
assert len(parts) == 2
# First part should be text/plain (the auto-generated plaintext version)
text_part = parts[0]
assert text_part.get_content_type() == 'text/plain'
text_content = text_part.get_content()
assert '(added) So let\'s see what happens.\r\n' in text_content # The plaintext part
# Second part should be text/html and roughly converted from markdown to HTML
html_part = parts[1]
assert html_part.get_content_type() == 'text/html'
html_content = html_part.get_content()
assert '<p><em>header</em></p>' in html_content
assert '(added) So let\'s see what happens.<br' in html_content
# The email should have two bodies, and the text/html part should be <br>
assert 'Content-Type: text/plain' in msg
assert '(added) So let\'s see what happens.\r\n' in msg # The plaintext part with \r\n
assert 'Content-Type: text/html' in msg
assert '(added) So let\'s see what happens.<br>' in msg # the html part
delete_all_watches(client)
def test_check_notification_email_formats_default_Text_override_HTML(client, live_server, measure_memory_usage):
## live_server_setup(live_server) # Setup on conftest per function
# HTML problems? see this
# https://github.com/caronc/apprise/issues/633
@@ -326,7 +115,7 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
data={"application-notification_urls": notification_url,
"application-notification_title": "fallback-title " + default_notification_title,
"application-notification_body": notification_body,
"application-notification_format": 'Plain Text',
"application-notification_format": 'Text',
"requests-time_between_check-minutes": 180,
'application-fetch_backend': "html_requests"},
follow_redirects=True
@@ -350,21 +139,15 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
wait_for_all_checks(client)
time.sleep(3)
msg_raw = get_last_message_from_smtp_server()
assert len(msg_raw) >= 1
msg = get_last_message_from_smtp_server()
assert len(msg) >= 1
# with open('/tmp/m.txt', 'w') as f:
# f.write(msg_raw)
# Parse the email properly using Python's email library
msg = message_from_string(msg_raw, policy=email_policy)
# f.write(msg)
# The email should not have two bodies, should be TEXT only
assert not msg.is_multipart()
assert msg.get_content_type() == 'text/plain'
# Get the plain text content
text_content = msg.get_content()
assert '(added) So let\'s see what happens.\r\n' in text_content # The plaintext part
assert 'Content-Type: text/plain' in msg
assert '(added) So let\'s see what happens.\r\n' in msg # The plaintext part with \r\n
set_original_response()
# Now override as HTML format
@@ -381,34 +164,18 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
wait_for_all_checks(client)
time.sleep(3)
msg_raw = get_last_message_from_smtp_server()
assert len(msg_raw) >= 1
msg = get_last_message_from_smtp_server()
assert len(msg) >= 1
# Parse the email properly using Python's email library
msg = message_from_string(msg_raw, policy=email_policy)
# The email should have two bodies (multipart/alternative)
assert msg.is_multipart()
assert msg.get_content_type() == 'multipart/alternative'
# Get the parts
parts = list(msg.iter_parts())
assert len(parts) == 2
# First part should be text/plain
text_part = parts[0]
assert text_part.get_content_type() == 'text/plain'
text_content = text_part.get_content()
assert '(removed) So let\'s see what happens.\r\n' in text_content # The plaintext part
# Second part should be text/html
html_part = parts[1]
assert html_part.get_content_type() == 'text/html'
html_content = html_part.get_content()
assert '(removed) So let\'s see what happens.<br>' in html_content # the html part
# The email should have two bodies, and the text/html part should be <br>
assert 'Content-Type: text/plain' in msg
assert '(removed) So let\'s see what happens.\r\n' in msg # The plaintext part with \n
assert 'Content-Type: text/html' in msg
assert '(removed) So let\'s see what happens.<br>' in msg # the html part
# https://github.com/dgtlmoon/changedetection.io/issues/2103
assert '<h1>Test</h1>' in html_content
assert '&lt;' not in html_content
assert '<h1>Test</h1>' in msg
assert '&lt;' not in msg
assert 'Content-Type: text/html' in msg
delete_all_watches(client)

View File

@@ -86,16 +86,14 @@ def test_filter_doesnt_exist_then_exists_should_get_notification(client, live_se
"Diff Full: {{diff_full}}\n"
"Diff as Patch: {{diff_patch}}\n"
":-)",
"notification_format": 'Plain Text'}
"notification_format": "Text"}
notification_form_data.update({
"url": test_url,
"tags": "my tag",
"title": "my title",
"headers": "",
# preprended with extra filter that intentionally doesn't match any entry,
# notification should still be sent even if first filter does not match (PR#3516)
"include_filters": ".non-matching-selector\n.ticket-available",
"include_filters": '.ticket-available',
"fetch_backend": "html_requests",
"time_between_check_use_default": "y"})

View File

@@ -1,8 +1,10 @@
import os
import time
from loguru import logger
from flask import url_for
from .util import set_original_response, wait_for_all_checks, wait_for_notification_endpoint_output
from ..notification import valid_notification_formats
from .util import set_original_response, live_server_setup, extract_UUID_from_client, wait_for_all_checks, \
wait_for_notification_endpoint_output
from changedetectionio.model import App
def set_response_with_filter():
@@ -21,14 +23,13 @@ def set_response_with_filter():
f.write(test_return_data)
return None
def run_filter_test(client, live_server, content_filter, app_notification_format):
def run_filter_test(client, live_server, content_filter):
# Response WITHOUT the filter ID element
set_original_response()
live_server.app.config['DATASTORE'].data['settings']['application']['notification_format'] = app_notification_format
# Goto the edit page, add our ignore text
notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'post')
notification_url = url_for('test_notification_endpoint', _external=True).replace('http', 'json')
# Add our URL to the import page
test_url = url_for('test_endpoint', _external=True)
@@ -63,7 +64,7 @@ def run_filter_test(client, live_server, content_filter, app_notification_format
"Diff Full: {{diff_full}}\n"
"Diff as Patch: {{diff_patch}}\n"
":-)",
"notification_format": 'Plain Text',
"notification_format": "Text",
"fetch_backend": "html_requests",
"filter_failure_notification_send": 'y',
"time_between_check_use_default": "y",
@@ -126,23 +127,8 @@ def run_filter_test(client, live_server, content_filter, app_notification_format
with open("test-datastore/notification.txt", 'r') as f:
notification = f.read()
assert 'Your configured CSS/xPath filters' in notification
# Text (or HTML conversion) markup to make the notifications a little nicer should have worked
if app_notification_format.startswith('html'):
# apprise should have used sax-escape (&#39; instead of &quot;, " etc), lets check it worked
from apprise.conversion import convert_between
from apprise.common import NotifyFormat
escaped_filter = convert_between(NotifyFormat.TEXT, NotifyFormat.HTML, content_filter)
assert escaped_filter in notification or escaped_filter.replace('&quot;', '&#34;') in notification
assert 'a href="' in notification # Quotes should still be there so the link works
else:
assert 'a href' not in notification
assert content_filter in notification
assert 'CSS/xPath filter was not present in the page' in notification
assert content_filter.replace('"', '\\"') in notification
# Remove it and prove that it doesn't trigger when not expected
# It should register a change, but no 'filter not found'
@@ -173,20 +159,14 @@ def run_filter_test(client, live_server, content_filter, app_notification_format
os.unlink("test-datastore/notification.txt")
def test_check_include_filters_failure_notification(client, live_server, measure_memory_usage):
# # live_server_setup(live_server) # Setup on conftest per function
run_filter_test(client=client, live_server=live_server, content_filter='#nope-doesnt-exist', app_notification_format=valid_notification_formats.get('HTML Color'))
# Check markup send conversion didnt affect plaintext preference
run_filter_test(client=client, live_server=live_server, content_filter='#nope-doesnt-exist', app_notification_format=valid_notification_formats.get('Plain Text'))
# # live_server_setup(live_server) # Setup on conftest per function
run_filter_test(client, live_server,'#nope-doesnt-exist')
def test_check_xpath_filter_failure_notification(client, live_server, measure_memory_usage):
# # live_server_setup(live_server) # Setup on conftest per function
run_filter_test(client=client, live_server=live_server, content_filter='//*[@id="nope-doesnt-exist"]', app_notification_format=valid_notification_formats.get('HTML Color'))
# # live_server_setup(live_server) # Setup on conftest per function
run_filter_test(client, live_server, '//*[@id="nope-doesnt-exist"]')
# Test that notification is never sent
def test_basic_markup_from_text(client, live_server, measure_memory_usage):
# Test the notification error templates convert to HTML if needed (link activate)
from ..notification.handler import markup_text_links_to_html
x = markup_text_links_to_html("hello https://google.com")
assert 'a href' in x

View File

@@ -195,7 +195,7 @@ def test_group_tag_notification(client, live_server, measure_memory_usage):
"Diff as Patch: {{diff_patch}}\n"
":-)",
"notification_screenshot": True,
"notification_format": 'Plain Text',
"notification_format": "Text",
"title": "test-tag"}
res = client.post(

View File

@@ -2,8 +2,7 @@
# coding=utf-8
import time
from flask import url_for
from markupsafe import escape
from flask import url_for, escape
from . util import live_server_setup, wait_for_all_checks, delete_all_watches
import pytest
jq_support = True

View File

@@ -101,7 +101,7 @@ def test_check_notification(client, live_server, measure_memory_usage):
"Diff as Patch: {{diff_patch}}\n"
":-)",
"notification_screenshot": True,
"notification_format": 'Plain Text'}
"notification_format": "Text"}
notification_form_data.update({
"url": test_url,
@@ -267,7 +267,7 @@ def test_notification_validation(client, live_server, measure_memory_usage):
# data={"notification_urls": 'json://localhost/foobar',
# "notification_title": "",
# "notification_body": "",
# "notification_format": 'Plain Text',
# "notification_format": "Text",
# "url": test_url,
# "tag": "my tag",
# "title": "my title",
@@ -383,7 +383,7 @@ def test_notification_custom_endpoint_and_jinja2(client, live_server, measure_me
assert 'second: hello world "space"' in notification_headers.lower()
# Should always be automatically detected as JSON content type even when we set it as 'Plain Text' (default)
# Should always be automatically detected as JSON content type even when we set it as 'Text' (default)
assert os.path.isfile("test-datastore/notification-content-type.txt")
with open("test-datastore/notification-content-type.txt", 'r') as f:
assert 'application/json' in f.read()
@@ -541,7 +541,9 @@ def _test_color_notifications(client, notification_body_token):
follow_redirects=True
)
# Just checks the format of the colour notifications was correct
def test_html_color_notifications(client, live_server, measure_memory_usage):
_test_color_notifications(client, '{{diff}}')
_test_color_notifications(client, '{{diff_full}}')

View File

@@ -30,7 +30,7 @@ def test_check_notification_error_handling(client, live_server, measure_memory_u
data={"notification_urls": f"{broken_notification_url}\r\n{working_notification_url}",
"notification_title": "xxx",
"notification_body": "xxxxx",
"notification_format": 'Plain Text',
"notification_format": "Text",
"url": test_url,
"tags": "",
"title": "",

View File

@@ -10,14 +10,14 @@ flask_restful
flask_cors # For the Chrome extension to operate
janus # Thread-safe async/sync queue bridge
flask_wtf~=1.2
flask~=3.1
flask~=2.3
flask-socketio~=5.5.1
python-socketio~=5.13.0
python-engineio~=4.12.3
inscriptis~=2.2
pytz
timeago~=1.0
validators~=0.35
validators~=0.21
# Set these versions together to avoid a RequestsDependencyWarning
@@ -42,9 +42,6 @@ jsonpath-ng~=1.5.3
# Notification library
apprise==1.9.5
# Lightweight URL linkifier for notifications
linkify-it-py
# - Needed for apprise/spush, and maybe others? hopefully doesnt trigger a rust compile.
# - Requires extra wheel for rPi, adds build time for arm/v8 which is not in piwheels
# Pinned to 43.0.1 for ARM compatibility (45.x may not have pre-built ARM wheels)
@@ -56,7 +53,7 @@ cryptography==44.0.1
paho-mqtt!=2.0.*
# Used for CSS filtering, JSON extraction from HTML
beautifulsoup4>=4.0.0,<=4.14.2
beautifulsoup4>=4.0.0,<=4.13.5
# XPath filtering, lxml is required by bs4 anyway, but put it here to be safe.
# #2328 - 5.2.0 and 5.2.1 had extra CPU flag CFLAGS set which was not compatible on older hardware
@@ -66,10 +63,14 @@ lxml >=4.8.0,<6,!=5.2.0,!=5.2.1
# XPath 2.0-3.1 support - 4.2.0 had issues, 4.1.5 stable
# Consider updating to latest stable version periodically
elementpath==5.0.4
elementpath==4.1.5
selenium~=4.31.0
# https://github.com/pallets/werkzeug/issues/2985
# Maybe related to pytest?
werkzeug==3.0.6
# Templating, so far just in the URLs but in the future can be for the notifications also
jinja2~=3.1
arrow

View File

@@ -5,8 +5,6 @@ import re
import sys
from setuptools import setup, find_packages
from setuptools.command.build_py import build_py
import shutil
here = os.path.abspath(os.path.dirname(__file__))
@@ -24,20 +22,6 @@ def find_version(*file_paths):
raise RuntimeError("Unable to find version string.")
class BuildPyCommand(build_py):
"""Custom build command to copy api-spec.yaml to the package."""
def run(self):
build_py.run(self)
# Ensure the docs directory exists in the build output
docs_dir = os.path.join(self.build_lib, 'changedetectionio', 'docs')
os.makedirs(docs_dir, exist_ok=True)
# Copy api-spec.yaml to the package
shutil.copy(
os.path.join(here, 'docs', 'api-spec.yaml'),
os.path.join(docs_dir, 'api-spec.yaml')
)
install_requires = open('requirements.txt').readlines()
setup(
@@ -53,10 +37,9 @@ setup(
scripts=["changedetection.py"],
author='dgtlmoon',
url='https://changedetection.io',
packages=find_packages(include=['changedetectionio', 'changedetectionio.*']),
packages=['changedetectionio'],
include_package_data=True,
install_requires=install_requires,
cmdclass={'build_py': BuildPyCommand},
license="Apache License 2.0",
python_requires=">= 3.10",
classifiers=['Intended Audience :: Customer Service',