diff --git a/changedetectionio/diff.py b/changedetectionio/diff.py index d9d37762..31e37efc 100644 --- a/changedetectionio/diff.py +++ b/changedetectionio/diff.py @@ -1,8 +1,10 @@ import difflib from typing import List, Iterator, Union -HTML_REMOVED_STYLE = "background-color: #fadad7; color: #b30000;" -HTML_ADDED_STYLE = "background-color: #eaf2c2; color: #406619;" +# https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050 +HTML_REMOVED_STYLE = "background-color: #ffebe9; color: #82071e" +HTML_ADDED_STYLE = "background-color: #dafbe1; color: #116329;" +HTML_CHANGED_STYLE = "background-color: #ffd8b5; color: #953800;" # These get set to html or telegram type or discord compatible or whatever in handler.py REMOVED_PLACEMARKER_OPEN = '<<>> 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='Test', body_format='html') +# Your handler should receive 'Test' not 'Test' +""" + import json import re from urllib.parse import unquote_plus import requests -from apprise.decorators import notify -from apprise.utils.parse import parse_url as apprise_parse_url +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 loguru import logger from requests.structures import CaseInsensitiveDict @@ -12,13 +63,66 @@ 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: - func = notify(on=method)(func) - # Add support for https, for each supported http method - func = notify(on=f"{method}s")(func) + _register_http_handler(method, func) + _register_http_handler(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") @@ -74,6 +178,8 @@ 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() diff --git a/changedetectionio/notification/handler.py b/changedetectionio/notification/handler.py index f02664dd..6f32eab1 100644 --- a/changedetectionio/notification/handler.py +++ b/changedetectionio/notification/handler.py @@ -8,9 +8,10 @@ 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 + CHANGED_PLACEMARKER_CLOSED, HTML_CHANGED_STYLE from ..notification_service import NotificationContextData +CUSTOM_LINEBREAK_PLACEHOLDER='$$BR$$' def markup_text_links_to_html(body): """ @@ -156,16 +157,17 @@ 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': - n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, f'') + # https://github.com/dgtlmoon/changedetection.io/issues/821#issuecomment-1241837050 + n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, f'') n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, f'') - n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'') + n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'') n_body = n_body.replace(ADDED_PLACEMARKER_CLOSED, f'') # Handle changed/replaced lines (old → new) - n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'') + n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'') n_body = n_body.replace(CHANGED_PLACEMARKER_CLOSED, f'') - n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'') + n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'') n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_CLOSED, f'') - n_body = n_body.replace("\n", '
') + n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n') elif requested_output_format == 'html': n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ') n_body = n_body.replace(REMOVED_PLACEMARKER_CLOSED, '') @@ -175,7 +177,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", '
') + n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n') else: #plaintext etc default n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ') @@ -233,11 +235,6 @@ 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 @@ -258,42 +255,15 @@ 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_to_html'): + if n_object.get('markup_text_links_to_html_links'): 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) @@ -311,6 +281,41 @@ def process_notification(n_object: NotificationContextData, datastore): (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 + + # 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', '').lower() + if watch_mime_type and 'text/' in watch_mime_type and not 'html' in watch_mime_type: + if 'html' in requested_output_format: + from markupsafe import escape + n_body = str(escape(n_body)) + + # 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, '
') + else: + # Just incase + n_body = n_body.replace(CUSTOM_LINEBREAK_PLACEHOLDER, '') + + apobj.add(url) sent_objs.append({'title': n_title, @@ -320,9 +325,9 @@ def process_notification(n_object: NotificationContextData, datastore): apobj.notify( title=n_title, body=n_body, - # `body_format` Tell apprise what format the INPUT is in + # `body_format` Tell apprise what format the INPUT is in, specify a wrong/bad type and it will force skip conversion in apprise # &format= in URL Tell apprise what format the OUTPUT should be in (it can convert between) - body_format=input_format, + body_format=apprise_input_format, # False is not an option for AppRise, must be type None attach=n_object.get('screenshot', None) ) diff --git a/changedetectionio/notification_service.py b/changedetectionio/notification_service.py index c94cc545..e19f314a 100644 --- a/changedetectionio/notification_service.py +++ b/changedetectionio/notification_service.py @@ -31,7 +31,7 @@ class NotificationContextData(dict): 'preview_url': None, 'watch_tag': None, 'watch_title': None, - 'markup_text_to_html': False, # If automatic conversion of plaintext to HTML should happen + 'markup_text_links_to_html_links': False, # If automatic conversion of plaintext to HTML should happen }) # Apply any initial data passed in @@ -131,6 +131,7 @@ class NotificationService: '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: @@ -228,7 +229,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_to_html' 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_links_to_html_links' 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. @@ -244,7 +245,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_to_html': n_format.lower().startswith('html') + 'markup_text_links_to_html_links': n_format.lower().startswith('html') }) if len(watch['notification_urls']): @@ -275,7 +276,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_to_html' 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_links_to_html_links' is not needed # {{{{ }}}} because this will be Jinja2 {{ }} tokens body = f"""Hello, @@ -293,7 +294,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_to_html': n_format.lower().startswith('html') + 'markup_text_links_to_html_links': n_format.lower().startswith('html') }) if len(watch['notification_urls']): diff --git a/changedetectionio/tests/smtp/smtp-test-server.py b/changedetectionio/tests/smtp/smtp-test-server.py index a6e3df66..4bb7fa6e 100755 --- a/changedetectionio/tests/smtp/smtp-test-server.py +++ b/changedetectionio/tests/smtp/smtp-test-server.py @@ -1,51 +1,76 @@ #!/usr/bin/env python3 -import asyncio +import threading +import time from aiosmtpd.controller import Controller -from aiosmtpd.smtp import SMTP +from flask import Flask, Response -# Accept a SMTP message and offer a way to retrieve the last message via TCP Socket +# Accept a SMTP message and offer a way to retrieve the last message via HTTP -last_received_message = b"Nothing" +last_received_message = b"Nothing received yet." +active_smtp_connections = 0 +smtp_lock = threading.Lock() class CustomSMTPHandler: async def handle_DATA(self, server, session, envelope): - 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' + 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('*******************************') + return '250 Message accepted for delivery' + finally: + with smtp_lock: + active_smtp_connections -= 1 -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() +# Simple Flask HTTP server to echo back the last SMTP message +app = Flask(__name__) -async def main(): +@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__": # Start the SMTP server controller = Controller(CustomSMTPHandler(), hostname='0.0.0.0', port=11025) controller.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() + # Start the HTTP server in a separate thread + flask_thread = threading.Thread(target=run_flask, daemon=True) + flask_thread.start() - -if __name__ == "__main__": - asyncio.run(main()) + # Keep the main thread alive + try: + flask_thread.join() + except KeyboardInterrupt: + print("Shutting down...") diff --git a/changedetectionio/tests/smtp/test_notification_smtp.py b/changedetectionio/tests/smtp/test_notification_smtp.py index 69cede3f..0cd7a7cb 100644 --- a/changedetectionio/tests/smtp/test_notification_smtp.py +++ b/changedetectionio/tests/smtp/test_notification_smtp.py @@ -3,7 +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 +from changedetectionio.diff import HTML_REMOVED_STYLE, HTML_ADDED_STYLE, HTML_CHANGED_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 @@ -24,16 +24,14 @@ from changedetectionio.notification import ( def get_last_message_from_smtp_server(): - 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 + 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 logging.info("get_last_message_from_smtp_server..") logging.info(data) - client_socket.close() # close the connection return data @@ -172,7 +170,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": "some text\n" + default_notification_body, + "application-notification_body": "some text\n" + default_notification_body, #some text\n should get
"application-notification_format": 'HTML Color', "requests-time_between_check-minutes": 180, 'application-fetch_backend': "html_requests"}, @@ -225,7 +223,7 @@ 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_REMOVED_STYLE in html_content + assert HTML_CHANGED_STYLE or HTML_REMOVED_STYLE in html_content assert HTML_ADDED_STYLE in html_content assert 'some text
' in html_content @@ -299,7 +297,7 @@ def test_check_notification_markdown_format(client, live_server, measure_memory_ assert '(added) So let\'s see what happens.' 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( @@ -405,10 +407,130 @@ 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 '(removed) So let\'s see what happens.' in html_content # the html part + assert '<!DOCTYPE html' not in html_content + assert 'Test' 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": 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 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 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 + + 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": notification_body, + "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(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() + + + assert 'And let\'s talk about <title> tags\r\n' in text_content + + # 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 'talk about <title>' not in html_content # the html part, should have got marked up to < etc + assert '<br>\r\n(added) And let's talk about <title> tags<br>' in html_content + + delete_all_watches(client) diff --git a/changedetectionio/tests/test_notification.py b/changedetectionio/tests/test_notification.py index 7edffe7a..9692e043 100644 --- a/changedetectionio/tests/test_notification.py +++ b/changedetectionio/tests/test_notification.py @@ -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,8 +485,6 @@ 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"): @@ -533,7 +531,8 @@ def _test_color_notifications(client, notification_body_token): with open("test-datastore/notification.txt", 'r') as f: x = f.read() - assert f'<span style="{HTML_REMOVED_STYLE}">Which is across multiple lines' in x + 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 client.get( diff --git a/changedetectionio/tests/util.py b/changedetectionio/tests/util.py index 3115de8e..403a6bc8 100644 --- a/changedetectionio/tests/util.py +++ b/changedetectionio/tests/util.py @@ -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: