mirror of
				https://github.com/dgtlmoon/changedetection.io.git
				synced 2025-10-31 22:57:18 +00:00 
			
		
		
		
	Compare commits
	
		
			7 Commits
		
	
	
		
			regex-filt
			...
			0.50.31
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
|   | e44853c439 | ||
|   | 3830bec891 | ||
|   | 88ab663330 | ||
|   | 68335b95c3 | ||
|   | 7bbfa0ef32 | ||
|   | e233d52931 | ||
|   | 181d32e82a | 
| @@ -2,7 +2,7 @@ | ||||
|  | ||||
| # Read more https://github.com/dgtlmoon/changedetection.io/wiki | ||||
|  | ||||
| __version__ = '0.50.29' | ||||
| __version__ = '0.50.31' | ||||
|  | ||||
| from changedetectionio.strtobool import strtobool | ||||
| from json.decoder import JSONDecodeError | ||||
|   | ||||
| @@ -1,21 +1,25 @@ | ||||
| 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_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;" | ||||
|  | ||||
| # These get set to html or telegram type or discord compatible or whatever in handler.py | ||||
| REMOVED_PLACEMARKER_OPEN = '<<<removed_PLACEMARKER_OPEN' | ||||
| REMOVED_PLACEMARKER_CLOSED = '<<<removed_PLACEMARKER_CLOSED' | ||||
| # Something that cant get escaped to HTML by accident | ||||
| 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,10 +1,61 @@ | ||||
| """ | ||||
| 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.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() | ||||
|   | ||||
| @@ -8,8 +8,9 @@ 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 | ||||
| from ..notification_service import NotificationContextData | ||||
|     CHANGED_PLACEMARKER_CLOSED, HTML_CHANGED_STYLE, HTML_CHANGED_INTO_STYLE | ||||
| from ..notification_service import NotificationContextData, CUSTOM_LINEBREAK_PLACEHOLDER | ||||
|  | ||||
|  | ||||
|  | ||||
| def markup_text_links_to_html(body): | ||||
| @@ -104,7 +105,8 @@ 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>') | ||||
| @@ -128,6 +130,7 @@ 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, | ||||
| @@ -156,16 +159,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'<span style="{HTML_REMOVED_STYLE}">') | ||||
|         # 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_CLOSED, f'</span>') | ||||
|         n_body = n_body.replace(ADDED_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}">') | ||||
|         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_CLOSED, f'</span>') | ||||
|         # Handle changed/replaced lines (old → new) | ||||
|         n_body = n_body.replace(CHANGED_PLACEMARKER_OPEN, f'<span style="{HTML_REMOVED_STYLE}">') | ||||
|         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_CLOSED, f'</span>') | ||||
|         n_body = n_body.replace(CHANGED_INTO_PLACEMARKER_OPEN, f'<span style="{HTML_ADDED_STYLE}">') | ||||
|         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_CLOSED, f'</span>') | ||||
|         n_body = n_body.replace("\n", '<br>') | ||||
|         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 +179,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", '<br>') | ||||
|         n_body = n_body.replace('\n', f'{CUSTOM_LINEBREAK_PLACEHOLDER}\n') | ||||
|  | ||||
|     else: #plaintext etc default | ||||
|         n_body = n_body.replace(REMOVED_PLACEMARKER_OPEN, '(removed) ') | ||||
| @@ -198,15 +202,6 @@ 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)}") | ||||
|  | ||||
| @@ -233,11 +228,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 +248,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) | ||||
|  | ||||
| @@ -309,20 +272,54 @@ 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) | ||||
|  | ||||
|             apobj.add(url) | ||||
|             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, '') | ||||
|  | ||||
|             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 | ||||
|             # `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) | ||||
|         ) | ||||
|   | ||||
| @@ -11,27 +11,32 @@ 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_url': 'https://WATCH-PLACE-HOLDER/', | ||||
|             'base_url': None, | ||||
|             'diff_url': None, | ||||
|             'preview_url': None, | ||||
|             'watch_mime_type': None, | ||||
|             'watch_tag': None, | ||||
|             'watch_title': None, | ||||
|             'markup_text_to_html': False, # If automatic conversion of plaintext to HTML should happen | ||||
|             'watch_url': 'https://WATCH-PLACE-HOLDER/', | ||||
|         }) | ||||
|  | ||||
|         # Apply any initial data passed in | ||||
| @@ -92,24 +97,13 @@ 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 = line_feed_sep.join(triggered_text) | ||||
|                 triggered_text = CUSTOM_LINEBREAK_PLACEHOLDER.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" | ||||
| @@ -121,16 +115,17 @@ class NotificationService: | ||||
|  | ||||
|         n_object.update({ | ||||
|             'current_snapshot': snapshot_contents, | ||||
|             '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), | ||||
|             '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), | ||||
|             '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: | ||||
| @@ -228,7 +223,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 +239,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 +270,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 +288,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']): | ||||
|   | ||||
| @@ -1,51 +1,108 @@ | ||||
| #!/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 | ||||
| from email import message_from_bytes | ||||
| from email.policy import default | ||||
|  | ||||
| # 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('*******************************') | ||||
|  | ||||
|             # 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 | ||||
|  | ||||
|  | ||||
| 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...") | ||||
|   | ||||
| @@ -3,7 +3,8 @@ 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.notification_service import NotificationContextData | ||||
| 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 | ||||
| @@ -14,6 +15,8 @@ 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, | ||||
| @@ -24,16 +27,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 +173,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": f"some text\n{default_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"}, | ||||
| @@ -225,8 +226,9 @@ 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 '<' not in html_content | ||||
|  | ||||
|     assert 'some text<br>' in html_content | ||||
|     delete_all_watches(client) | ||||
| @@ -299,7 +301,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 | ||||
| @@ -334,7 +336,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', _external=True) | ||||
|     test_url = url_for('test_endpoint',content_type="text/html", _external=True) | ||||
|     res = client.post( | ||||
|         url_for("ui.ui_views.form_quick_watch_add"), | ||||
|         data={"url": test_url, "tags": 'nice one'}, | ||||
| @@ -343,6 +345,7 @@ 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) | ||||
| @@ -365,7 +368,10 @@ 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( | ||||
| @@ -405,10 +411,210 @@ 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.<br>' 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 '<!DOCTYPE html' in html_content # Our original template is working correctly | ||||
|  | ||||
|     # 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,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( | ||||
|   | ||||
| @@ -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