mirror of
https://github.com/dgtlmoon/changedetection.io.git
synced 2025-11-07 10:07:00 +00:00
Compare commits
3 Commits
0.50.31
...
regex-filt
| Author | SHA1 | Date | |
|---|---|---|---|
|
0e9f01639a | ||
|
|
b236c8b24b | ||
|
|
27cedc9fa4 |
@@ -2,7 +2,7 @@
|
||||
|
||||
# Read more https://github.com/dgtlmoon/changedetection.io/wiki
|
||||
|
||||
__version__ = '0.50.31'
|
||||
__version__ = '0.50.29'
|
||||
|
||||
from changedetectionio.strtobool import strtobool
|
||||
from json.decoder import JSONDecodeError
|
||||
|
||||
@@ -1,25 +1,21 @@
|
||||
import difflib
|
||||
from typing import List, Iterator, Union
|
||||
|
||||
# https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050
|
||||
HTML_ADDED_STYLE = "background-color: #d2f7c2; color: #255d00;"
|
||||
HTML_CHANGED_INTO_STYLE = "background-color: #dafbe1; color: #116329;"
|
||||
HTML_CHANGED_STYLE = "background-color: #ffd6cc; color: #7a2000;"
|
||||
HTML_REMOVED_STYLE = "background-color: #ffebe9; color: #82071e;"
|
||||
HTML_REMOVED_STYLE = "background-color: #fadad7; color: #b30000;"
|
||||
HTML_ADDED_STYLE = "background-color: #eaf2c2; color: #406619;"
|
||||
|
||||
# These get set to html or telegram type or discord compatible or whatever in handler.py
|
||||
# Something that cant get escaped to HTML by accident
|
||||
REMOVED_PLACEMARKER_OPEN = '@removed_PLACEMARKER_OPEN'
|
||||
REMOVED_PLACEMARKER_CLOSED = '@removed_PLACEMARKER_CLOSED'
|
||||
REMOVED_PLACEMARKER_OPEN = '<<<removed_PLACEMARKER_OPEN'
|
||||
REMOVED_PLACEMARKER_CLOSED = '<<<removed_PLACEMARKER_CLOSED'
|
||||
|
||||
ADDED_PLACEMARKER_OPEN = '@added_PLACEMARKER_OPEN'
|
||||
ADDED_PLACEMARKER_CLOSED = '@added_PLACEMARKER_CLOSED'
|
||||
ADDED_PLACEMARKER_OPEN = '<<<added_PLACEMARKER_OPEN'
|
||||
ADDED_PLACEMARKER_CLOSED = '<<<added_PLACEMARKER_CLOSED'
|
||||
|
||||
CHANGED_PLACEMARKER_OPEN = '@changed_PLACEMARKER_OPEN'
|
||||
CHANGED_PLACEMARKER_CLOSED = '@changed_PLACEMARKER_CLOSED'
|
||||
CHANGED_PLACEMARKER_OPEN = '<<<changed_PLACEMARKER_OPEN'
|
||||
CHANGED_PLACEMARKER_CLOSED = '<<<changed_PLACEMARKER_CLOSED'
|
||||
|
||||
CHANGED_INTO_PLACEMARKER_OPEN = '@changed_into_PLACEMARKER_OPEN'
|
||||
CHANGED_INTO_PLACEMARKER_CLOSED = '@changed_into_PLACEMARKER_CLOSED'
|
||||
CHANGED_INTO_PLACEMARKER_OPEN = '<<<changed_into_PLACEMARKER_OPEN'
|
||||
CHANGED_INTO_PLACEMARKER_CLOSED = '<<<changed_into_PLACEMARKER_CLOSED'
|
||||
|
||||
def same_slicer(lst: List[str], start: int, end: int) -> List[str]:
|
||||
"""Return a slice of the list, or a single element if start == end."""
|
||||
|
||||
@@ -1,61 +1,10 @@
|
||||
"""
|
||||
Custom Apprise HTTP Handlers with format= Parameter Support
|
||||
|
||||
IMPORTANT: This module works around a limitation in Apprise's @notify decorator.
|
||||
|
||||
THE PROBLEM:
|
||||
-------------
|
||||
When using Apprise's @notify decorator to create custom notification handlers, the
|
||||
decorator creates a CustomNotifyPlugin that uses parse_url(..., simple=True) to parse
|
||||
URLs. This simple parsing mode does NOT extract the format= query parameter from the URL
|
||||
and set it as a top-level parameter that NotifyBase.__init__ can use to set notify_format.
|
||||
|
||||
As a result:
|
||||
1. URL: post://example.com/webhook?format=html
|
||||
2. Apprise parses this and sees format=html in qsd (query string dictionary)
|
||||
3. But it does NOT extract it and pass it to NotifyBase.__init__
|
||||
4. NotifyBase defaults to notify_format=TEXT
|
||||
5. When you call apobj.notify(body="<html>...", body_format="html"):
|
||||
- Apprise sees: input format = html, output format (notify_format) = text
|
||||
- Apprise calls convert_between("html", "text", body)
|
||||
- This strips all HTML tags, leaving only plain text
|
||||
6. Your custom handler receives stripped plain text instead of HTML
|
||||
|
||||
THE SOLUTION:
|
||||
-------------
|
||||
Instead of using the @notify decorator directly, we:
|
||||
1. Manually register custom plugins using plugins.N_MGR.add()
|
||||
2. Create a CustomHTTPHandler class that extends CustomNotifyPlugin
|
||||
3. Override __init__ to extract format= from qsd and set it as kwargs['format']
|
||||
4. Call NotifyBase.__init__ which properly sets notify_format from kwargs['format']
|
||||
5. Set up _default_args like CustomNotifyPlugin does for compatibility
|
||||
|
||||
This ensures that when format=html is in the URL:
|
||||
- notify_format is set to HTML
|
||||
- Apprise sees: input format = html, output format = html
|
||||
- No conversion happens (convert_between returns content unchanged)
|
||||
- Your custom handler receives the original HTML intact
|
||||
|
||||
TESTING:
|
||||
--------
|
||||
To verify this works:
|
||||
>>> apobj = apprise.Apprise()
|
||||
>>> apobj.add('post://localhost:5005/test?format=html')
|
||||
>>> for server in apobj:
|
||||
... print(server.notify_format) # Should print: html (not text)
|
||||
>>> apobj.notify(body='<span>Test</span>', body_format='html')
|
||||
# Your handler should receive '<span>Test</span>' not 'Test'
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from urllib.parse import unquote_plus
|
||||
|
||||
import requests
|
||||
from apprise import plugins
|
||||
from apprise.decorators.base import CustomNotifyPlugin
|
||||
from apprise.utils.parse import parse_url as apprise_parse_url, url_assembly
|
||||
from apprise.utils.logic import dict_full_update
|
||||
from apprise.decorators import notify
|
||||
from apprise.utils.parse import parse_url as apprise_parse_url
|
||||
from loguru import logger
|
||||
from requests.structures import CaseInsensitiveDict
|
||||
|
||||
@@ -63,66 +12,13 @@ SUPPORTED_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head"}
|
||||
|
||||
|
||||
def notify_supported_methods(func):
|
||||
"""Register custom HTTP method handlers that properly support format= parameter."""
|
||||
for method in SUPPORTED_HTTP_METHODS:
|
||||
_register_http_handler(method, func)
|
||||
_register_http_handler(f"{method}s", func)
|
||||
func = notify(on=method)(func)
|
||||
# Add support for https, for each supported http method
|
||||
func = notify(on=f"{method}s")(func)
|
||||
return func
|
||||
|
||||
|
||||
def _register_http_handler(schema, send_func):
|
||||
"""Register a custom HTTP handler that extracts format= from URL query parameters."""
|
||||
|
||||
# Parse base URL
|
||||
base_url = f"{schema}://"
|
||||
base_args = apprise_parse_url(base_url, default_schema=schema, verify_host=False, simple=True)
|
||||
|
||||
class CustomHTTPHandler(CustomNotifyPlugin):
|
||||
secure_protocol = schema
|
||||
service_name = f"Custom HTTP - {schema.upper()}"
|
||||
_base_args = base_args
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
# Extract format from qsd and set it as a top-level kwarg
|
||||
# This allows NotifyBase.__init__ to properly set notify_format
|
||||
if 'qsd' in kwargs and 'format' in kwargs['qsd']:
|
||||
kwargs['format'] = kwargs['qsd']['format']
|
||||
|
||||
# Call NotifyBase.__init__ (skip CustomNotifyPlugin.__init__)
|
||||
super(CustomNotifyPlugin, self).__init__(**kwargs)
|
||||
|
||||
# Set up _default_args like CustomNotifyPlugin does
|
||||
self._default_args = {}
|
||||
kwargs.pop("secure", None)
|
||||
dict_full_update(self._default_args, self._base_args)
|
||||
dict_full_update(self._default_args, kwargs)
|
||||
self._default_args["url"] = url_assembly(**self._default_args)
|
||||
|
||||
__send = staticmethod(send_func)
|
||||
|
||||
def send(self, body, title="", notify_type="info", *args, **kwargs):
|
||||
"""Call the custom send function."""
|
||||
try:
|
||||
result = self.__send(
|
||||
body, title, notify_type,
|
||||
*args,
|
||||
meta=self._default_args,
|
||||
**kwargs
|
||||
)
|
||||
return True if result is None else bool(result)
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Exception in custom HTTP handler: {e}")
|
||||
return False
|
||||
|
||||
# Register the plugin
|
||||
plugins.N_MGR.add(
|
||||
plugin=CustomHTTPHandler,
|
||||
schemas=schema,
|
||||
send_func=send_func,
|
||||
url=base_url,
|
||||
)
|
||||
|
||||
|
||||
def _get_auth(parsed_url: dict) -> str | tuple[str, str]:
|
||||
user: str | None = parsed_url.get("user")
|
||||
password: str | None = parsed_url.get("password")
|
||||
@@ -178,8 +74,6 @@ def apprise_http_custom_handler(
|
||||
*args,
|
||||
**kwargs,
|
||||
) -> bool:
|
||||
|
||||
|
||||
url: str = meta.get("url")
|
||||
schema: str = meta.get("schema")
|
||||
method: str = re.sub(r"s$", "", schema).upper()
|
||||
|
||||
@@ -8,9 +8,8 @@ from .apprise_plugin.assets import apprise_asset, APPRISE_AVATAR_URL
|
||||
from .apprise_plugin.custom_handlers import SUPPORTED_HTTP_METHODS
|
||||
from ..diff import HTML_REMOVED_STYLE, REMOVED_PLACEMARKER_OPEN, REMOVED_PLACEMARKER_CLOSED, ADDED_PLACEMARKER_OPEN, HTML_ADDED_STYLE, \
|
||||
ADDED_PLACEMARKER_CLOSED, CHANGED_INTO_PLACEMARKER_OPEN, CHANGED_INTO_PLACEMARKER_CLOSED, CHANGED_PLACEMARKER_OPEN, \
|
||||
CHANGED_PLACEMARKER_CLOSED, HTML_CHANGED_STYLE, HTML_CHANGED_INTO_STYLE
|
||||
from ..notification_service import NotificationContextData, CUSTOM_LINEBREAK_PLACEHOLDER
|
||||
|
||||
CHANGED_PLACEMARKER_CLOSED
|
||||
from ..notification_service import NotificationContextData
|
||||
|
||||
|
||||
def markup_text_links_to_html(body):
|
||||
@@ -105,8 +104,7 @@ def apply_service_tweaks(url, n_body, n_title, requested_output_format):
|
||||
# @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')
|
||||
n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '\n')
|
||||
|
||||
|
||||
# Use strikethrough for removed content, bold for added content
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '<s>')
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, '</s>')
|
||||
@@ -130,7 +128,6 @@ def apply_service_tweaks(url, n_body, n_title, requested_output_format):
|
||||
# Discord doesn't support HTML, replace <br> with newlines
|
||||
n_body = n_body.strip().replace('<br>', '\n')
|
||||
n_body = n_body.replace('</br>', '\n')
|
||||
n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '\n')
|
||||
|
||||
# Don't replace placeholders or truncate here - let the custom Discord plugin handle it
|
||||
# The plugin will use embeds (6000 char limit across all embeds) if placeholders are present,
|
||||
@@ -159,17 +156,16 @@ def apply_service_tweaks(url, n_body, n_title, requested_output_format):
|
||||
|
||||
# Is not discord/tgram and they want htmlcolor
|
||||
elif requested_output_format == 'htmlcolor':
|
||||
# https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}" role="deletion" aria-label="Removed text" title="Removed text">')
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}">')
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, f'</span>')
|
||||
n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}" role="insertion" aria-label="Added text" title="Added text">')
|
||||
n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}">')
|
||||
n_body = n_body.replace(ADDED_PLACEMARKER_CLOSED, f'</span>')
|
||||
# Handle changed/replaced lines (old → new)
|
||||
n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'<span style="{HTML_CHANGED_STYLE}" role="note" aria-label="Changed text" title="Changed text">')
|
||||
n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}">')
|
||||
n_body = n_body.replace(CHANGED_PLACEMARKER_CLOSED, f'</span>')
|
||||
n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'<span style="{HTML_CHANGED_INTO_STYLE}" role="note" aria-label="Changed into" title="Changed into">')
|
||||
n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}">')
|
||||
n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_CLOSED, f'</span>')
|
||||
n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n')
|
||||
n_body = n_body.replace("\n", '<br>')
|
||||
elif requested_output_format == 'html':
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ')
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, '')
|
||||
@@ -179,7 +175,7 @@ def apply_service_tweaks(url, n_body, n_title, requested_output_format):
|
||||
n_body = n_body.replace(CHANGED_PLACEMARKER_CLOSED, f'')
|
||||
n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'(into) ')
|
||||
n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_CLOSED, f'')
|
||||
n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n')
|
||||
n_body = n_body.replace("\n", '<br>')
|
||||
|
||||
else: #plaintext etc default
|
||||
n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ')
|
||||
@@ -202,6 +198,15 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
# Register custom Discord plugin
|
||||
from .apprise_plugin.discord import NotifyDiscordCustom
|
||||
|
||||
# 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)}")
|
||||
|
||||
@@ -228,6 +233,11 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
|
||||
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
|
||||
@@ -248,15 +258,42 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
if not n_object.get('notification_urls'):
|
||||
return None
|
||||
|
||||
with (apprise.LogCapture(level=apprise.logging.DEBUG) as logs):
|
||||
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_links_to_html_links'):
|
||||
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 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)
|
||||
|
||||
@@ -272,54 +309,20 @@ def process_notification(n_object: NotificationContextData, datastore):
|
||||
logger.info(f">> Process Notification: AppRise notifying {url}")
|
||||
url = jinja_render(template_str=url, **notification_parameters)
|
||||
|
||||
# If it's a plaintext document, and they want HTML type email/alerts, so it needs to be escaped
|
||||
watch_mime_type = n_object.get('watch_mime_type')
|
||||
if watch_mime_type and 'text/' in watch_mime_type.lower() and not 'html' in watch_mime_type.lower():
|
||||
if 'html' in requested_output_format:
|
||||
from markupsafe import escape
|
||||
n_body = str(escape(n_body))
|
||||
|
||||
(url, n_body, n_title) = apply_service_tweaks(url=url, n_body=n_body, n_title=n_title, requested_output_format=requested_output_format_original)
|
||||
|
||||
apprise_input_format = "NO-THANKS-WE-WILL-MANAGE-ALL-OF-THIS"
|
||||
|
||||
if not 'format=' in url:
|
||||
parsed_url = urlparse(url)
|
||||
prefix_add_to_url = '?' if not parsed_url.query else '&'
|
||||
|
||||
# THIS IS THE TRICK HOW TO DISABLE APPRISE DOING WEIRD AUTO-CONVERSION WITH BREAKING BR TAGS ETC
|
||||
if 'html' in requested_output_format:
|
||||
url = f"{url}{prefix_add_to_url}format={NotifyFormat.HTML.value}"
|
||||
apprise_input_format = NotifyFormat.HTML.value
|
||||
elif 'text' in requested_output_format:
|
||||
url = f"{url}{prefix_add_to_url}format={NotifyFormat.TEXT.value}"
|
||||
apprise_input_format = NotifyFormat.TEXT.value
|
||||
|
||||
elif requested_output_format == NotifyFormat.MARKDOWN.value:
|
||||
# This actually means we request "Markdown to HTML", we want HTML output
|
||||
url = f"{url}{prefix_add_to_url}format={NotifyFormat.HTML.value}"
|
||||
requested_output_format = NotifyFormat.HTML.value
|
||||
apprise_input_format = NotifyFormat.MARKDOWN.value
|
||||
|
||||
|
||||
# Could have arrived at any stage, so we dont end up running .escape on it
|
||||
if 'html' in requested_output_format:
|
||||
n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '<br>\n')
|
||||
else:
|
||||
# Just incase
|
||||
n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '')
|
||||
apobj.add(url)
|
||||
|
||||
sent_objs.append({'title': n_title,
|
||||
'body': n_body,
|
||||
'url': url})
|
||||
apobj.add(url)
|
||||
|
||||
apobj.notify(
|
||||
title=n_title,
|
||||
body=n_body,
|
||||
# `body_format` Tell apprise what format the INPUT is in, specify a wrong/bad type and it will force skip conversion in apprise
|
||||
# `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=apprise_input_format,
|
||||
body_format=input_format,
|
||||
# False is not an option for AppRise, must be type None
|
||||
attach=n_object.get('screenshot', None)
|
||||
)
|
||||
|
||||
@@ -11,32 +11,27 @@ import time
|
||||
|
||||
from changedetectionio.notification import default_notification_format
|
||||
|
||||
# This gets modified on notification time (handler.py) depending on the required notification output
|
||||
CUSTOM_LINEBREAK_PLACEHOLDER='@BR@'
|
||||
|
||||
|
||||
# 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):
|
||||
super().__init__({
|
||||
'base_url': None,
|
||||
'current_snapshot': None,
|
||||
'diff': None,
|
||||
'diff_added': None,
|
||||
'diff_full': None,
|
||||
'diff_patch': None,
|
||||
'diff_removed': None,
|
||||
'diff_url': None,
|
||||
'markup_text_links_to_html_links': False, # If automatic conversion of plaintext to HTML should happen
|
||||
'notification_timestamp': time.time(),
|
||||
'preview_url': None,
|
||||
'screenshot': None,
|
||||
'triggered_text': None,
|
||||
'uuid': 'XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX', # Converted to 'watch_uuid' in create_notification_parameters
|
||||
'watch_mime_type': None,
|
||||
'watch_url': 'https://WATCH-PLACE-HOLDER/',
|
||||
'base_url': None,
|
||||
'diff_url': None,
|
||||
'preview_url': None,
|
||||
'watch_tag': None,
|
||||
'watch_title': None,
|
||||
'watch_url': 'https://WATCH-PLACE-HOLDER/',
|
||||
'markup_text_to_html': False, # If automatic conversion of plaintext to HTML should happen
|
||||
})
|
||||
|
||||
# Apply any initial data passed in
|
||||
@@ -97,13 +92,24 @@ class NotificationService:
|
||||
if n_object.get('notification_format') == default_notification_format_for_watch:
|
||||
n_object['notification_format'] = self.datastore.data['settings']['application'].get('notification_format')
|
||||
|
||||
# HTML needs linebreak, but MarkDown and Text can use a linefeed
|
||||
if n_object.get('notification_format') == 'HTML':
|
||||
line_feed_sep = "<br>"
|
||||
# Snapshot will be plaintext on the disk, convert to some kind of HTML
|
||||
snapshot_contents = snapshot_contents.replace('\n', line_feed_sep)
|
||||
elif n_object.get('notification_format') == 'HTML Color':
|
||||
line_feed_sep = "<br>"
|
||||
# Snapshot will be plaintext on the disk, convert to some kind of HTML
|
||||
snapshot_contents = snapshot_contents.replace('\n', line_feed_sep)
|
||||
else:
|
||||
line_feed_sep = "\n"
|
||||
|
||||
triggered_text = ''
|
||||
if len(trigger_text):
|
||||
from . import html_tools
|
||||
triggered_text = html_tools.get_triggered_text(content=snapshot_contents, trigger_text=trigger_text)
|
||||
if triggered_text:
|
||||
triggered_text = CUSTOM_LINEBREAK_PLACEHOLDER.join(triggered_text)
|
||||
triggered_text = line_feed_sep.join(triggered_text)
|
||||
|
||||
# Could be called as a 'test notification' with only 1 snapshot available
|
||||
prev_snapshot = "Example text: example test\nExample text: change detection is cool\nExample text: some more examples\n"
|
||||
@@ -115,17 +121,16 @@ class NotificationService:
|
||||
|
||||
n_object.update({
|
||||
'current_snapshot': snapshot_contents,
|
||||
'diff': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=CUSTOM_LINEBREAK_PLACEHOLDER),
|
||||
'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False, line_feed_sep=CUSTOM_LINEBREAK_PLACEHOLDER),
|
||||
'diff_full': diff.render_diff(prev_snapshot, current_snapshot, include_equal=True, line_feed_sep=CUSTOM_LINEBREAK_PLACEHOLDER),
|
||||
'diff_patch': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=CUSTOM_LINEBREAK_PLACEHOLDER, patch_format=True),
|
||||
'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False, line_feed_sep=CUSTOM_LINEBREAK_PLACEHOLDER),
|
||||
'diff': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep),
|
||||
'diff_added': diff.render_diff(prev_snapshot, current_snapshot, include_removed=False, line_feed_sep=line_feed_sep),
|
||||
'diff_full': diff.render_diff(prev_snapshot, current_snapshot, include_equal=True, line_feed_sep=line_feed_sep),
|
||||
'diff_patch': diff.render_diff(prev_snapshot, current_snapshot, line_feed_sep=line_feed_sep, patch_format=True),
|
||||
'diff_removed': diff.render_diff(prev_snapshot, current_snapshot, include_added=False, line_feed_sep=line_feed_sep),
|
||||
'screenshot': watch.get_screenshot() if watch and watch.get('notification_screenshot') else None,
|
||||
'triggered_text': triggered_text,
|
||||
'uuid': watch.get('uuid') if watch else None,
|
||||
'watch_url': watch.get('url') if watch else None,
|
||||
'watch_uuid': watch.get('uuid') if watch else None,
|
||||
'watch_mime_type': watch.get('content-type')
|
||||
})
|
||||
|
||||
if watch:
|
||||
@@ -223,7 +228,7 @@ class NotificationService:
|
||||
|
||||
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_links_to_html_links' is not needed
|
||||
# @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.
|
||||
@@ -239,7 +244,7 @@ Thanks - Your omniscient changedetection.io installation.
|
||||
'notification_title': 'Changedetection.io - Alert - CSS/xPath filter was not present in the page',
|
||||
'notification_body': body,
|
||||
'notification_format': n_format,
|
||||
'markup_text_links_to_html_links': n_format.lower().startswith('html')
|
||||
'markup_text_to_html': n_format.lower().startswith('html')
|
||||
})
|
||||
|
||||
if len(watch['notification_urls']):
|
||||
@@ -270,7 +275,7 @@ Thanks - Your omniscient changedetection.io installation.
|
||||
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_links_to_html_links' is not needed
|
||||
# @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,
|
||||
@@ -288,7 +293,7 @@ Thanks - Your omniscient changedetection.io installation.
|
||||
'notification_title': f"Changedetection.io - Alert - Browser step at position {step} could not be run",
|
||||
'notification_body': body,
|
||||
'notification_format': n_format,
|
||||
'markup_text_links_to_html_links': n_format.lower().startswith('html')
|
||||
'markup_text_to_html': n_format.lower().startswith('html')
|
||||
})
|
||||
|
||||
if len(watch['notification_urls']):
|
||||
|
||||
@@ -1,108 +1,51 @@
|
||||
#!/usr/bin/env python3
|
||||
import threading
|
||||
import time
|
||||
import asyncio
|
||||
from aiosmtpd.controller import Controller
|
||||
from flask import Flask, Response
|
||||
from email import message_from_bytes
|
||||
from email.policy import default
|
||||
from aiosmtpd.smtp import SMTP
|
||||
|
||||
# Accept a SMTP message and offer a way to retrieve the last message via HTTP
|
||||
# Accept a SMTP message and offer a way to retrieve the last message via TCP Socket
|
||||
|
||||
last_received_message = b"Nothing received yet."
|
||||
active_smtp_connections = 0
|
||||
smtp_lock = threading.Lock()
|
||||
last_received_message = b"Nothing"
|
||||
|
||||
|
||||
class CustomSMTPHandler:
|
||||
async def handle_DATA(self, server, session, envelope):
|
||||
global last_received_message, active_smtp_connections
|
||||
|
||||
with smtp_lock:
|
||||
active_smtp_connections += 1
|
||||
|
||||
try:
|
||||
last_received_message = envelope.content
|
||||
print('Receiving message from:', session.peer)
|
||||
print('Message addressed from:', envelope.mail_from)
|
||||
print('Message addressed to :', envelope.rcpt_tos)
|
||||
print('Message length :', len(envelope.content))
|
||||
print('*******************************')
|
||||
print(envelope.content.decode('utf8'))
|
||||
print('*******************************')
|
||||
|
||||
# Parse the email message
|
||||
msg = message_from_bytes(envelope.content, policy=default)
|
||||
|
||||
# Write parts to files based on content type
|
||||
if msg.is_multipart():
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type()
|
||||
payload = part.get_payload(decode=True)
|
||||
|
||||
if payload:
|
||||
if content_type == 'text/plain':
|
||||
with open('/tmp/last.txt', 'wb') as f:
|
||||
f.write(payload)
|
||||
print(f'Written text/plain part to /tmp/last.txt')
|
||||
elif content_type == 'text/html':
|
||||
with open('/tmp/last.html', 'wb') as f:
|
||||
f.write(payload)
|
||||
print(f'Written text/html part to /tmp/last.html')
|
||||
else:
|
||||
# Single part message
|
||||
content_type = msg.get_content_type()
|
||||
payload = msg.get_payload(decode=True)
|
||||
|
||||
if payload:
|
||||
if content_type == 'text/plain' or content_type.startswith('text/'):
|
||||
with open('/tmp/last.txt', 'wb') as f:
|
||||
f.write(payload)
|
||||
print(f'Written single part message to /tmp/last.txt')
|
||||
|
||||
return '250 Message accepted for delivery'
|
||||
finally:
|
||||
with smtp_lock:
|
||||
active_smtp_connections -= 1
|
||||
global last_received_message
|
||||
last_received_message = envelope.content
|
||||
print('Receiving message from:', session.peer)
|
||||
print('Message addressed from:', envelope.mail_from)
|
||||
print('Message addressed to :', envelope.rcpt_tos)
|
||||
print('Message length :', len(envelope.content))
|
||||
print(envelope.content.decode('utf8'))
|
||||
return '250 Message accepted for delivery'
|
||||
|
||||
|
||||
# Simple Flask HTTP server to echo back the last SMTP message
|
||||
app = Flask(__name__)
|
||||
class EchoServerProtocol(asyncio.Protocol):
|
||||
def connection_made(self, transport):
|
||||
global last_received_message
|
||||
self.transport = transport
|
||||
peername = transport.get_extra_info('peername')
|
||||
print('Incoming connection from {}'.format(peername))
|
||||
self.transport.write(last_received_message)
|
||||
|
||||
last_received_message = b''
|
||||
self.transport.close()
|
||||
|
||||
|
||||
@app.route('/')
|
||||
def echo_last_message():
|
||||
global last_received_message, active_smtp_connections
|
||||
|
||||
# Wait for any in-progress SMTP connections to complete
|
||||
max_wait = 5 # Maximum 5 seconds
|
||||
wait_interval = 0.05 # Check every 50ms
|
||||
elapsed = 0
|
||||
|
||||
while elapsed < max_wait:
|
||||
with smtp_lock:
|
||||
if active_smtp_connections == 0:
|
||||
break
|
||||
time.sleep(wait_interval)
|
||||
elapsed += wait_interval
|
||||
|
||||
return Response(last_received_message, mimetype='text/plain')
|
||||
|
||||
|
||||
def run_flask():
|
||||
app.run(host='0.0.0.0', port=11080, debug=False, use_reloader=False)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
async def main():
|
||||
# Start the SMTP server
|
||||
controller = Controller(CustomSMTPHandler(), hostname='0.0.0.0', port=11025)
|
||||
controller.start()
|
||||
|
||||
# Start the HTTP server in a separate thread
|
||||
flask_thread = threading.Thread(target=run_flask, daemon=True)
|
||||
flask_thread.start()
|
||||
# Start the TCP Echo server
|
||||
loop = asyncio.get_running_loop()
|
||||
server = await loop.create_server(
|
||||
lambda: EchoServerProtocol(),
|
||||
'0.0.0.0', 11080
|
||||
)
|
||||
async with server:
|
||||
await server.serve_forever()
|
||||
|
||||
# Keep the main thread alive
|
||||
try:
|
||||
flask_thread.join()
|
||||
except KeyboardInterrupt:
|
||||
print("Shutting down...")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
||||
@@ -3,8 +3,7 @@ from flask import url_for
|
||||
from email import message_from_string
|
||||
from email.policy import default as email_policy
|
||||
|
||||
from changedetectionio.diff import HTML_REMOVED_STYLE, HTML_ADDED_STYLE, HTML_CHANGED_STYLE
|
||||
from changedetectionio.notification_service import NotificationContextData
|
||||
from changedetectionio.diff import HTML_REMOVED_STYLE, HTML_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
|
||||
@@ -15,8 +14,6 @@ import logging
|
||||
# NOTE - RELIES ON mailserver as hostname running, see github build recipes
|
||||
smtp_test_server = 'mailserver'
|
||||
|
||||
ALL_MARKUP_TOKENS = ''.join(f"TOKEN: '{t}'\n{{{{{t}}}}}\n" for t in NotificationContextData().keys())
|
||||
|
||||
from changedetectionio.notification import (
|
||||
default_notification_body,
|
||||
default_notification_format,
|
||||
@@ -27,14 +24,16 @@ from changedetectionio.notification import (
|
||||
|
||||
|
||||
def get_last_message_from_smtp_server():
|
||||
import requests
|
||||
time.sleep(1) # wait for any smtp connects to die off
|
||||
port = 11080 # HTTP server port number
|
||||
# Make HTTP GET request to Flask server
|
||||
response = requests.get(f'http://{smtp_test_server}:{port}/')
|
||||
data = response.text
|
||||
import socket
|
||||
port = 11080 # socket server port number
|
||||
|
||||
client_socket = socket.socket() # instantiate
|
||||
client_socket.connect((smtp_test_server, port)) # connect to the server
|
||||
|
||||
data = client_socket.recv(50024).decode() # receive response
|
||||
logging.info("get_last_message_from_smtp_server..")
|
||||
logging.info(data)
|
||||
client_socket.close() # close the connection
|
||||
return data
|
||||
|
||||
|
||||
@@ -173,7 +172,7 @@ def test_check_notification_html_color_format(client, live_server, measure_memor
|
||||
url_for("settings.settings_page"),
|
||||
data={"application-notification_urls": notification_url,
|
||||
"application-notification_title": "fallback-title " + default_notification_title,
|
||||
"application-notification_body": f"some text\n{default_notification_body}\nMore output test\n{ALL_MARKUP_TOKENS}",
|
||||
"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"},
|
||||
@@ -226,9 +225,8 @@ def test_check_notification_html_color_format(client, live_server, measure_memor
|
||||
html_part = parts[1]
|
||||
assert html_part.get_content_type() == 'text/html'
|
||||
html_content = html_part.get_content()
|
||||
assert HTML_CHANGED_STYLE or HTML_REMOVED_STYLE in html_content
|
||||
assert HTML_REMOVED_STYLE in html_content
|
||||
assert HTML_ADDED_STYLE in html_content
|
||||
assert '<' not in html_content
|
||||
|
||||
assert 'some text<br>' in html_content
|
||||
delete_all_watches(client)
|
||||
@@ -301,7 +299,7 @@ def test_check_notification_markdown_format(client, live_server, measure_memory_
|
||||
assert '(added) So let\'s see what happens.<br' in html_content
|
||||
delete_all_watches(client)
|
||||
|
||||
# Custom notification body with HTML, that is either sent as HTML or rendered to plaintext and sent
|
||||
|
||||
def test_check_notification_email_formats_default_Text_override_HTML(client, live_server, measure_memory_usage):
|
||||
|
||||
# HTML problems? see this
|
||||
@@ -336,7 +334,7 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
|
||||
assert b"Settings updated." in res.data
|
||||
|
||||
# Add a watch and trigger a HTTP POST
|
||||
test_url = url_for('test_endpoint',content_type="text/html", _external=True)
|
||||
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'},
|
||||
@@ -345,7 +343,6 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
|
||||
|
||||
assert b"Watch added" in res.data
|
||||
|
||||
#################################### FIRST SITUATION, PLAIN TEXT NOTIFICATION IS WANTED BUT WE HAVE HTML IN OUR TEMPLATE AND CONTENT ##########
|
||||
wait_for_all_checks(client)
|
||||
set_longer_modified_response()
|
||||
time.sleep(2)
|
||||
@@ -368,10 +365,7 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
|
||||
# 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 '<!DOCTYPE html>' in text_content # even tho they added html, they selected plaintext so it should have not got converted
|
||||
|
||||
|
||||
#################################### SECOND SITUATION, HTML IS CORRECTLY PASSED THROUGH TO THE EMAIL ####################
|
||||
set_original_response()
|
||||
# Now override as HTML format
|
||||
res = client.post(
|
||||
@@ -411,210 +405,10 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
|
||||
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.' in html_content # the html part
|
||||
assert '<!DOCTYPE html' not in html_content
|
||||
assert '<!DOCTYPE html' in html_content # Our original template is working correctly
|
||||
assert '(removed) So let\'s see what happens.<br>' in html_content # the html part
|
||||
|
||||
# https://github.com/dgtlmoon/changedetection.io/issues/2103
|
||||
assert '<h1>Test</h1>' in html_content
|
||||
assert '<' not in html_content
|
||||
|
||||
delete_all_watches(client)
|
||||
|
||||
def test_check_plaintext_document_plaintext_notification_smtp(client, live_server, measure_memory_usage):
|
||||
"""When following a plaintext document, notification in Plain Text format is sent correctly"""
|
||||
|
||||
with open("test-datastore/endpoint-content.txt", "w") as f:
|
||||
f.write("Some nice plain text\nwhich we add some extra data\nover here\n")
|
||||
|
||||
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
|
||||
notification_body = f"""{default_notification_body}"""
|
||||
|
||||
#####################
|
||||
# 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": f"{notification_body}\nMore output test\n{ALL_MARKUP_TOKENS}",
|
||||
"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 our URL to the import page
|
||||
test_url = url_for('test_endpoint', content_type="text/plain", _external=True)
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Change the content
|
||||
with open("test-datastore/endpoint-content.txt", "w") as f:
|
||||
f.write("Some nice plain text\nwhich we add some extra data\nAnd let's talk about <title> tags\nover here\n")
|
||||
|
||||
|
||||
time.sleep(1)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(get_last_message_from_smtp_server(), policy=email_policy)
|
||||
|
||||
assert not msg.is_multipart()
|
||||
assert msg.get_content_type() == 'text/plain'
|
||||
body = msg.get_content()
|
||||
# nothing is escaped, raw html stuff in text/plain
|
||||
assert 'talk about <title> tags' in body
|
||||
assert '(added)' in body
|
||||
assert '<br' not in body
|
||||
assert '<' not in body
|
||||
|
||||
delete_all_watches(client)
|
||||
|
||||
def test_check_plaintext_document_html_notifications(client, live_server, measure_memory_usage):
|
||||
"""When following a plaintext document, notification in Plain Text format is sent correctly"""
|
||||
|
||||
with open("test-datastore/endpoint-content.txt", "w") as f:
|
||||
f.write("Some nice plain text\nwhich we add some extra data\nover here\n")
|
||||
|
||||
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
|
||||
notification_body = f"""{default_notification_body}"""
|
||||
|
||||
#####################
|
||||
# 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": f"{notification_body}\nMore output test\n{ALL_MARKUP_TOKENS}",
|
||||
"application-notification_format": 'HTML',
|
||||
"requests-time_between_check-minutes": 180,
|
||||
'application-fetch_backend': "html_requests"},
|
||||
follow_redirects=True
|
||||
)
|
||||
assert b"Settings updated." in res.data
|
||||
|
||||
# Add our URL to the import page
|
||||
test_url = url_for('test_endpoint', content_type="text/plain", _external=True)
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Change the content
|
||||
with open("test-datastore/endpoint-content.txt", "w") as f:
|
||||
f.write("Some nice plain text\nwhich we add some extra data\nAnd let's talk about <title> tags\nover here\n")
|
||||
|
||||
|
||||
time.sleep(2)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(get_last_message_from_smtp_server(), 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()
|
||||
html_part = parts[1]
|
||||
assert html_part.get_content_type() == 'text/html'
|
||||
html_content = html_part.get_content()
|
||||
|
||||
|
||||
assert 'And let\'s talk about <title> tags\r\n' in text_content
|
||||
assert '<br' not in text_content
|
||||
assert '<span' not in text_content
|
||||
|
||||
|
||||
assert 'talk about <title>' not in html_content # the html part, should have got marked up to < etc
|
||||
assert 'talk about <title>' in html_content
|
||||
# Should be the HTML, but not HTML Color
|
||||
assert 'background-color' not in html_content
|
||||
assert '<br>\r\n(added) And let's talk about <title> tags<br>' in html_content
|
||||
assert '<br' not in html_content
|
||||
|
||||
delete_all_watches(client)
|
||||
|
||||
|
||||
def test_check_plaintext_document_html_color_notifications(client, live_server, measure_memory_usage):
|
||||
"""When following a plaintext document, notification in Plain Text format is sent correctly"""
|
||||
|
||||
with open("test-datastore/endpoint-content.txt", "w") as f:
|
||||
f.write("Some nice plain text\nwhich we add some extra data\nover here\n")
|
||||
|
||||
notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
|
||||
notification_body = f"""{default_notification_body}"""
|
||||
|
||||
#####################
|
||||
# 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": f"{notification_body}\nMore output test\n{ALL_MARKUP_TOKENS}",
|
||||
"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 our URL to the import page
|
||||
test_url = url_for('test_endpoint', content_type="text/plain", _external=True)
|
||||
uuid = client.application.config.get('DATASTORE').add_watch(url=test_url)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Change the content
|
||||
with open("test-datastore/endpoint-content.txt", "w") as f:
|
||||
f.write("Some nice plain text\nwhich we add some extra data\nAnd let's talk about <title> tags\nover here\n")
|
||||
|
||||
time.sleep(1)
|
||||
client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
|
||||
wait_for_all_checks(client)
|
||||
|
||||
# Parse the email properly using Python's email library
|
||||
msg = message_from_string(get_last_message_from_smtp_server(), 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()
|
||||
html_part = parts[1]
|
||||
assert html_part.get_content_type() == 'text/html'
|
||||
html_content = html_part.get_content()
|
||||
|
||||
|
||||
assert 'And let\'s talk about <title> tags\r\n' in text_content
|
||||
assert '<br' not in text_content
|
||||
assert '<span' not in text_content
|
||||
|
||||
assert 'talk about <title>' not in html_content # the html part, should have got marked up to < etc
|
||||
assert 'talk about <title>' in html_content
|
||||
# Should be the HTML, but not HTML Color
|
||||
assert 'background-color' in html_content
|
||||
assert '(added) And let' not in html_content
|
||||
assert '<br' not in html_content
|
||||
assert '<br>' in html_content
|
||||
|
||||
delete_all_watches(client)
|
||||
|
||||
@@ -16,7 +16,7 @@ from changedetectionio.notification import (
|
||||
default_notification_title,
|
||||
valid_notification_formats,
|
||||
)
|
||||
from ..diff import HTML_CHANGED_STYLE
|
||||
|
||||
|
||||
|
||||
# Hard to just add more live server URLs when one test is already running (I think)
|
||||
@@ -485,6 +485,8 @@ def test_global_send_test_notification(client, live_server, measure_memory_usage
|
||||
|
||||
def _test_color_notifications(client, notification_body_token):
|
||||
|
||||
from changedetectionio.diff import HTML_ADDED_STYLE, HTML_REMOVED_STYLE
|
||||
|
||||
set_original_response()
|
||||
|
||||
if os.path.isfile("test-datastore/notification.txt"):
|
||||
@@ -531,8 +533,7 @@ def _test_color_notifications(client, notification_body_token):
|
||||
|
||||
with open("test-datastore/notification.txt", 'r') as f:
|
||||
x = f.read()
|
||||
s = f'<span style="{HTML_CHANGED_STYLE}" role="note" aria-label="Changed text" title="Changed text">Which is across multiple lines'
|
||||
assert s in x
|
||||
assert f'<span style="{HTML_REMOVED_STYLE}">Which is across multiple lines' in x
|
||||
|
||||
|
||||
client.get(
|
||||
|
||||
@@ -244,7 +244,7 @@ def new_live_server_setup(live_server):
|
||||
return request.method
|
||||
|
||||
# Where we POST to as a notification, also use a space here to test URL escaping is OK across all tests that use this. ( #2868 )
|
||||
@live_server.app.route('/test_notification_endpoint', methods=['POST', 'GET'])
|
||||
@live_server.app.route('/test_notification endpoint', methods=['POST', 'GET'])
|
||||
def test_notification_endpoint():
|
||||
|
||||
with open("test-datastore/notification.txt", "wb") as f:
|
||||
|
||||
Reference in New Issue
Block a user