Adding delay random fail

changedetectionio/__init__.py

@@ -19,7 +19,6 @@ from changedetectionio import store
 from changedetectionio.flask_app import changedetection_app
 from loguru import logger
 
-
 # Only global so we can access it in the signal handler
 app = None
 datastore = None
@@ -29,8 +28,6 @@ def get_version():
 
 # Parent wrapper or OS sends us a SIGTERM/SIGINT, do everything required for a clean shutdown
 def sigshutdown_handler(_signo, _stack_frame):
-    global app
-    global datastore
     name = signal.Signals(_signo).name
     logger.critical(f'Shutdown: Got Signal - {name} ({_signo}), Saving DB to disk and calling shutdown')
     datastore.sync_to_json()

changedetectionio/apprise_asset.py

@@ -1,12 +0,0 @@
-from changedetectionio import apprise_plugin
-import apprise
-
-# Create our AppriseAsset and populate it with some of our new values:
-# https://github.com/caronc/apprise/wiki/Development_API#the-apprise-asset-object
-asset = apprise.AppriseAsset(
-   image_url_logo='https://raw.githubusercontent.com/dgtlmoon/changedetection.io/master/changedetectionio/static/images/avatar-256x256.png'
-)
-
-asset.app_id = "changedetection.io"
-asset.app_desc = "ChangeDetection.io best and simplest website monitoring and change detection"
-asset.app_url = "https://changedetection.io"

changedetectionio/apprise_plugin/__init__.py

@@ -1,98 +0,0 @@
-# include the decorator
-from apprise.decorators import notify
-from loguru import logger
-from requests.structures import CaseInsensitiveDict
-
-
-@notify(on="delete")
-@notify(on="deletes")
-@notify(on="get")
-@notify(on="gets")
-@notify(on="post")
-@notify(on="posts")
-@notify(on="put")
-@notify(on="puts")
-def apprise_custom_api_call_wrapper(body, title, notify_type, *args, **kwargs):
-    import requests
-    import json
-    import re
-
-    from urllib.parse import unquote_plus
-    from apprise.utils.parse import parse_url as apprise_parse_url
-
-    url = kwargs['meta'].get('url')
-    schema = kwargs['meta'].get('schema').lower().strip()
-
-    # Choose POST, GET etc from requests
-    method =  re.sub(rf's$', '', schema)
-    requests_method = getattr(requests, method)
-
-    params = CaseInsensitiveDict({}) # Added to requests
-    auth = None
-    has_error = False
-
-    # Convert /foobar?+some-header=hello to proper header dictionary
-    results = apprise_parse_url(url)
-
-    # Add our headers that the user can potentially over-ride if they wish
-    # to to our returned result set and tidy entries by unquoting them
-    headers = CaseInsensitiveDict({unquote_plus(x): unquote_plus(y)
-               for x, y in results['qsd+'].items()})
-
-    # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
-    # In Apprise, it relies on prefixing each request arg with "-", because it uses say &method=update as a flag for apprise
-    # but here we are making straight requests, so we need todo convert this against apprise's logic
-    for k, v in results['qsd'].items():
-        if not k.strip('+-') in results['qsd+'].keys():
-            params[unquote_plus(k)] = unquote_plus(v)
-
-    # Determine Authentication
-    auth = ''
-    if results.get('user') and results.get('password'):
-        auth = (unquote_plus(results.get('user')), unquote_plus(results.get('user')))
-    elif results.get('user'):
-        auth = (unquote_plus(results.get('user')))
-
-    # If it smells like it could be JSON and no content-type was already set, offer a default content type.
-    if body and '{' in body[:100] and not headers.get('Content-Type'):
-        json_header = 'application/json; charset=utf-8'
-        try:
-            # Try if it's JSON
-            json.loads(body)
-            headers['Content-Type'] = json_header
-        except ValueError as e:
-            logger.warning(f"Could not automatically add '{json_header}' header to the notification because the document failed to parse as JSON: {e}")
-            pass
-
-    # POSTS -> HTTPS etc
-    if schema.lower().endswith('s'):
-        url = re.sub(rf'^{schema}', 'https', results.get('url'))
-    else:
-        url = re.sub(rf'^{schema}', 'http', results.get('url'))
-
-    status_str = ''
-    try:
-        r = requests_method(url,
-          auth=auth,
-          data=body.encode('utf-8') if type(body) is str else body,
-          headers=headers,
-          params=params
-        )
-
-        if not (200 <= r.status_code < 300):
-            status_str = f"Error sending '{method.upper()}' request to {url} - Status: {r.status_code}: '{r.reason}'"
-            logger.error(status_str)
-            has_error = True
-        else:
-            logger.info(f"Sent '{method.upper()}' request to {url}")
-            has_error = False
-
-    except requests.RequestException as e:
-        status_str = f"Error sending '{method.upper()}' request to {url} - {str(e)}"
-        logger.error(status_str)
-        has_error = True
-
-    if has_error:
-        raise TypeError(status_str)
-
-    return True

changedetectionio/apprise_plugin/assets.py

@@ -0,0 +1,16 @@
+from apprise import AppriseAsset
+
+# Refer to:
+# https://github.com/caronc/apprise/wiki/Development_API#the-apprise-asset-object
+
+APPRISE_APP_ID = "changedetection.io"
+APPRISE_APP_DESC = "ChangeDetection.io best and simplest website monitoring and change detection"
+APPRISE_APP_URL = "https://changedetection.io"
+APPRISE_AVATAR_URL = "https://raw.githubusercontent.com/dgtlmoon/changedetection.io/master/changedetectionio/static/images/avatar-256x256.png"
+
+apprise_asset = AppriseAsset(
+    app_id=APPRISE_APP_ID,
+    app_desc=APPRISE_APP_DESC,
+    app_url=APPRISE_APP_URL,
+    image_url_logo=APPRISE_AVATAR_URL,
+)

changedetectionio/apprise_plugin/custom_handlers.py

@@ -0,0 +1,112 @@
+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 loguru import logger
+from requests.structures import CaseInsensitiveDict
+
+SUPPORTED_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head"}
+
+
+def notify_supported_methods(func):
+    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)
+    return func
+
+
+def _get_auth(parsed_url: dict) -> str | tuple[str, str]:
+    user: str | None = parsed_url.get("user")
+    password: str | None = parsed_url.get("password")
+
+    if user is not None and password is not None:
+        return (unquote_plus(user), unquote_plus(password))
+
+    if user is not None:
+        return unquote_plus(user)
+
+    return ""
+
+
+def _get_headers(parsed_url: dict, body: str) -> CaseInsensitiveDict:
+    headers = CaseInsensitiveDict(
+        {unquote_plus(k).title(): unquote_plus(v) for k, v in parsed_url["qsd+"].items()}
+    )
+
+    # If Content-Type is not specified, guess if the body is a valid JSON
+    if headers.get("Content-Type") is None:
+        try:
+            json.loads(body)
+            headers["Content-Type"] = "application/json; charset=utf-8"
+        except Exception:
+            pass
+
+    return headers
+
+
+def _get_params(parsed_url: dict) -> CaseInsensitiveDict:
+    # https://github.com/caronc/apprise/wiki/Notify_Custom_JSON#get-parameter-manipulation
+    # In Apprise, it relies on prefixing each request arg with "-", because it uses say &method=update as a flag for apprise
+    # but here we are making straight requests, so we need todo convert this against apprise's logic
+    params = CaseInsensitiveDict(
+        {
+            unquote_plus(k): unquote_plus(v)
+            for k, v in parsed_url["qsd"].items()
+            if k.strip("-") not in parsed_url["qsd-"]
+            and k.strip("+") not in parsed_url["qsd+"]
+        }
+    )
+
+    return params
+
+
+@notify_supported_methods
+def apprise_http_custom_handler(
+    body: str,
+    title: str,
+    notify_type: str,
+    meta: dict,
+    *args,
+    **kwargs,
+) -> bool:
+    url: str = meta.get("url")
+    schema: str = meta.get("schema")
+    method: str = re.sub(r"s$", "", schema).upper()
+
+    # Convert /foobar?+some-header=hello to proper header dictionary
+    parsed_url: dict[str, str | dict | None] | None = apprise_parse_url(url)
+    if parsed_url is None:
+        return False
+
+    auth = _get_auth(parsed_url=parsed_url)
+    headers = _get_headers(parsed_url=parsed_url, body=body)
+    params = _get_params(parsed_url=parsed_url)
+
+    url = re.sub(rf"^{schema}", "https" if schema.endswith("s") else "http", parsed_url.get("url"))
+
+    try:
+        response = requests.request(
+            method=method,
+            url=url,
+            auth=auth,
+            headers=headers,
+            params=params,
+            data=body.encode("utf-8") if isinstance(body, str) else body,
+        )
+
+        response.raise_for_status()
+
+        logger.info(f"Successfully sent custom notification to {url}")
+        return True
+
+    except requests.RequestException as e:
+        logger.error(f"Remote host error while sending custom notification to {url}: {e}")
+        return False
+
+    except Exception as e:
+        logger.error(f"Unexpected error occurred while sending custom notification to {url}: {e}")
+        return False

changedetectionio/auth_decorator.py

@@ -20,10 +20,7 @@ def login_optionally_required(func):
         has_password_enabled = datastore.data['settings']['application'].get('password') or os.getenv("SALTED_PASS", False)
 
         # Permitted
-        if request.endpoint and 'static_content' in request.endpoint and request.view_args and request.view_args.get('group') == 'styles':
-            return func(*args, **kwargs)
-        # Permitted
-        elif request.endpoint and 'diff_history_page' in request.endpoint and datastore.data['settings']['application'].get('shared_diff_access'):
+        if request.endpoint and 'diff_history_page' in request.endpoint and datastore.data['settings']['application'].get('shared_diff_access'):
             return func(*args, **kwargs)
         elif request.method in flask_login.config.EXEMPT_METHODS:
             return func(*args, **kwargs)

changedetectionio/blueprint/browser_steps/__init__.py

@@ -23,7 +23,6 @@ from loguru import logger
 browsersteps_sessions = {}
 io_interface_context = None
 import json
-import base64
 import hashlib
 from flask import Response
 
@@ -34,10 +33,8 @@ def construct_blueprint(datastore: ChangeDetectionStore):
         from . import nonContext
         from . import browser_steps
         import time
-        global browsersteps_sessions
         global io_interface_context
 
-
         # We keep the playwright session open for many minutes
         keepalive_seconds = int(os.getenv('BROWSERSTEPS_MINUTES_KEEPALIVE', 10)) * 60
 
@@ -104,8 +101,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
         # A new session was requested, return sessionID
 
         import uuid
-        global browsersteps_sessions
-
         browsersteps_session_id = str(uuid.uuid4())
         watch_uuid = request.args.get('uuid')
 
@@ -149,7 +144,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
     def browsersteps_ui_update():
         import base64
         import playwright._impl._errors
-        global browsersteps_sessions
         from changedetectionio.blueprint.browser_steps import browser_steps
 
         remaining =0

changedetectionio/blueprint/ui/__init__.py

@@ -96,12 +96,13 @@ def construct_blueprint(datastore: ChangeDetectionStore, update_q, running_updat
             uuid = list(datastore.data['watching'].keys()).pop()
 
         new_uuid = datastore.clone(uuid)
-        if new_uuid:
-            if not datastore.data['watching'].get(uuid).get('paused'):
-                update_q.put(queuedWatchMetaData.PrioritizedItem(priority=5, item={'uuid': new_uuid}))
-            flash('Cloned.')
 
-        return redirect(url_for('watchlist.index'))
+        if not datastore.data['watching'].get(uuid).get('paused'):
+            update_q.put(queuedWatchMetaData.PrioritizedItem(priority=5, item={'uuid': new_uuid}))
+
+        flash('Cloned, you are editing the new watch.')
+
+        return redirect(url_for("ui.ui_edit.edit_page", uuid=new_uuid))
 
     @ui_blueprint.route("/checknow", methods=['GET'])
     @login_optionally_required

changedetectionio/blueprint/ui/notification.py

@@ -4,6 +4,7 @@ from loguru import logger
 
 from changedetectionio.store import ChangeDetectionStore
 from changedetectionio.auth_decorator import login_optionally_required
+from changedetectionio.notification import process_notification
 
 def construct_blueprint(datastore: ChangeDetectionStore):
     notification_blueprint = Blueprint('ui_notification', __name__, template_folder="../ui/templates")
@@ -17,11 +18,10 @@ def construct_blueprint(datastore: ChangeDetectionStore):
 
         # Watch_uuid could be unset in the case it`s used in tag editor, global settings
         import apprise
-        from changedetectionio.apprise_asset import asset
-        apobj = apprise.Apprise(asset=asset)
+        from ...apprise_plugin.assets import apprise_asset
+        from ...apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
+        apobj = apprise.Apprise(asset=apprise_asset)
 
-        # so that the custom endpoints are registered
-        from changedetectionio.apprise_plugin import apprise_custom_api_call_wrapper
         is_global_settings_form = request.args.get('mode', '') == 'global-settings'
         is_group_settings_form = request.args.get('mode', '') == 'group-settings'
 
@@ -90,7 +90,6 @@ def construct_blueprint(datastore: ChangeDetectionStore):
 
             n_object['as_async'] = False
             n_object.update(watch.extra_notification_token_values())
-            from changedetectionio.notification import process_notification
             sent_obj = process_notification(n_object, datastore)
 
         except Exception as e:

changedetectionio/conditions/__init__.py

@@ -116,8 +116,7 @@ def execute_ruleset_against_all_plugins(current_watch_uuid: str, application_dat
             if not jsonLogic(logic=ruleset, data=EXECUTE_DATA, operations=CUSTOM_OPERATIONS):
                 result = False
 
-    return result
-
+    return {'executed_data': EXECUTE_DATA, 'result': result}
 
 # Load plugins dynamically
 for plugin in plugin_manager.get_plugins():

changedetectionio/conditions/blueprint.py

@@ -67,7 +67,8 @@ def construct_blueprint(datastore):
 
             return jsonify({
                 'status': 'success',
-                'result': result,
+                'result': result.get('result'),
+                'data': result.get('executed_data'),
                 'message': 'Condition passes' if result else 'Condition does not pass'
             })
 

changedetectionio/flask_app.py

@@ -233,7 +233,8 @@ def changedetection_app(config=None, datastore_o=None):
 
         if has_password_enabled and not flask_login.current_user.is_authenticated:
             # Permitted
-            if request.endpoint and request.endpoint == 'static_content' and request.view_args and request.view_args.get('group') in ['styles', 'js', 'images', 'favicons']:
+            if request.endpoint and request.endpoint == 'static_content' and request.view_args:
+                # Handled by static_content handler
                 return None
             # Permitted
             elif request.endpoint and 'login' in request.endpoint:
@@ -351,11 +352,15 @@ def changedetection_app(config=None, datastore_o=None):
     @app.route("/static/<string:group>/<string:filename>", methods=['GET'])
     def static_content(group, filename):
         from flask import make_response
+        import re
+        group = re.sub(r'[^\w.-]+', '', group.lower())
+        filename = re.sub(r'[^\w.-]+', '', filename.lower())
 
         if group == 'screenshot':
             # Could be sensitive, follow password requirements
             if datastore.data['settings']['application']['password'] and not flask_login.current_user.is_authenticated:
-                abort(403)
+                if not datastore.data['settings']['application'].get('shared_diff_access'):
+                    abort(403)
 
             screenshot_filename = "last-screenshot.png" if not request.args.get('error_screenshot') else "last-error-screenshot.png"
 
@@ -404,7 +409,7 @@ def changedetection_app(config=None, datastore_o=None):
 
         # These files should be in our subdirectory
         try:
-            return send_from_directory("static/{}".format(group), path=filename)
+            return send_from_directory(f"static/{group}", path=filename)
         except FileNotFoundError:
             abort(404)
 

changedetectionio/forms.py

@@ -306,10 +306,10 @@ class ValidateAppRiseServers(object):
 
     def __call__(self, form, field):
         import apprise
-        apobj = apprise.Apprise()
+        from .apprise_plugin.assets import apprise_asset
+        from .apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
 
-        # so that the custom endpoints are registered
-        from .apprise_asset import asset
+        apobj = apprise.Apprise(asset=apprise_asset)
 
         for server_url in field.data:
             url = server_url.strip()

changedetectionio/model/Watch.py

@@ -575,7 +575,7 @@ class model(watch_base):
         import brotli
         filepath = os.path.join(self.watch_data_dir, 'last-fetched.br')
 
-        if not os.path.isfile(filepath):
+        if not os.path.isfile(filepath) or os.path.getsize(filepath) == 0:
             # If a previous attempt doesnt yet exist, just snarf the previous snapshot instead
             dates = list(self.history.keys())
             if len(dates):

changedetectionio/notification.py

@@ -4,6 +4,9 @@ from apprise import NotifyFormat
 import apprise
 from loguru import logger
 
+from .apprise_plugin.assets import APPRISE_AVATAR_URL
+from .apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
+from .safe_jinja import render as jinja_render
 
 valid_tokens = {
     'base_url': '',
@@ -39,10 +42,6 @@ valid_notification_formats = {
 
 
 def process_notification(n_object, datastore):
-    # so that the custom endpoints are registered
-    from changedetectionio.apprise_plugin import apprise_custom_api_call_wrapper
-
-    from .safe_jinja import render as jinja_render
     now = time.time()
     if n_object.get('notification_timestamp'):
         logger.trace(f"Time since queued {now-n_object['notification_timestamp']:.3f}s")
@@ -66,12 +65,12 @@ def process_notification(n_object, datastore):
     # raise it as an exception
 
     sent_objs = []
-    from .apprise_asset import asset
+    from .apprise_plugin.assets import apprise_asset
 
     if 'as_async' in n_object:
-        asset.async_mode = n_object.get('as_async')
+        apprise_asset.async_mode = n_object.get('as_async')
 
-    apobj = apprise.Apprise(debug=True, asset=asset)
+    apobj = apprise.Apprise(debug=True, asset=apprise_asset)
 
     if not n_object.get('notification_urls'):
         return None
@@ -112,7 +111,7 @@ def process_notification(n_object, datastore):
                     and not url.startswith('get') \
                     and not url.startswith('delete') \
                     and not url.startswith('put'):
-                url += k + 'avatar_url=https://raw.githubusercontent.com/dgtlmoon/changedetection.io/master/changedetectionio/static/images/avatar-256x256.png'
+                url += k + f"avatar_url={APPRISE_AVATAR_URL}"
 
             if url.startswith('tgram://'):
                 # Telegram only supports a limit subset of HTML, remove the '<br>' we place in.

changedetectionio/processors/text_json_diff/processor.py

@@ -334,12 +334,14 @@ class perform_site_check(difference_detection_processor):
 
         # And check if 'conditions' will let this pass through
         if watch.get('conditions') and watch.get('conditions_match_logic'):
-            if not execute_ruleset_against_all_plugins(current_watch_uuid=watch.get('uuid'),
-                                                application_datastruct=self.datastore.data,
-                                                ephemeral_data={
-                                                    'text': stripped_text_from_html
-                                                }
-                                                ):
+            conditions_result = execute_ruleset_against_all_plugins(current_watch_uuid=watch.get('uuid'),
+                                                                    application_datastruct=self.datastore.data,
+                                                                    ephemeral_data={
+                                                                        'text': stripped_text_from_html
+                                                                    }
+                                                                    )
+
+            if not conditions_result.get('result'):
                 # Conditions say "Condition not met" so we block it.
                 blocked = True
 

changedetectionio/static/js/conditions.js

@@ -52,7 +52,7 @@ $(document).ready(function () {
 
             
             // Create a rule object
-            const rule = {
+            let rule = {
                 field: field,
                 operator: operator,
                 value: value
@@ -96,6 +96,10 @@ $(document).ready(function () {
                 contentType: false, // Let the browser set the correct content type
                 success: function (response) {
                     if (response.status === "success") {
+                        if(rule['field'] !== "page_filtered_text") {
+                            // A little debug helper for the user
+                            $('#verify-state-text').text(`${rule['field']} was value "${response.data[rule['field']]}"`)
+                        }
                         if (response.result) {
                             alert("✅ Condition PASSES verification against current snapshot!");
                         } else {

changedetectionio/store.py

@@ -251,8 +251,14 @@ class ChangeDetectionStore:
     # Clone a watch by UUID
     def clone(self, uuid):
         url = self.data['watching'][uuid].get('url')
-        extras = self.data['watching'][uuid]
+        extras = deepcopy(self.data['watching'][uuid])
         new_uuid = self.add_watch(url=url, extras=extras)
+        watch = self.data['watching'][new_uuid]
+
+        if self.data['settings']['application'].get('extract_title_as_title') or watch['extract_title_as_title']:
+            # Because it will be recalculated on the next fetch
+            self.data['watching'][new_uuid]['title'] = None
+
         return new_uuid
 
     def url_exists(self, url):
@@ -363,7 +369,6 @@ class ChangeDetectionStore:
         new_watch.ensure_data_dir_exists()
         self.__data['watching'][new_uuid] = new_watch
 
-
         if write_to_disk_now:
             self.sync_to_json()
 

changedetectionio/templates/edit.html

@@ -305,9 +305,9 @@ Math: {{ 1 + 1 }}") }}
                     {{ render_field(form.conditions_match_logic) }}
                     {{ render_fieldlist_of_formfields_as_table(form.conditions) }}
                     <div class="pure-form-message-inline">
-                        <br>
-                        Use the verify (✓) button to test if a condition passes against the current snapshot.<br><br>
-                        Did you know that <strong>conditions</strong> can be extended with your own custom plugin? tutorials coming soon!<br>
+
+                        <p id="verify-state-text">Use the verify (✓) button to test if a condition passes against the current snapshot.</p>
+                       Read a quick tutorial about <a href="https://changedetection.io/tutorial/conditional-actions-web-page-changes">using conditional web page changes here</a>.<br>
                     </div>
                 </div>
             </div>
@@ -337,7 +337,7 @@ xpath://body/div/span[contains(@class, 'example-class')]",
                         {% if '/text()' in  field %}
                           <span class="pure-form-message-inline"><strong>Note!: //text() function does not work where the &lt;element&gt; contains &lt;![CDATA[]]&gt;</strong></span><br>
                         {% endif %}
-                        <span class="pure-form-message-inline">One CSS, xPath, JSON Path/JQ selector per line, <i>any</i> rules that matches will be used.<br>
+                        <span class="pure-form-message-inline">One CSS, xPath 1 &amp; 2, JSON Path/JQ selector per line, <i>any</i> rules that matches will be used.<br>
                         <span data-target="#advanced-help-selectors" class="toggle-show pure-button button-tag button-xsmall">Show advanced help and tips</span><br>
                     <ul id="advanced-help-selectors" style="display: none;">
                         <li>CSS - Limit text to this CSS rule, only text matching this CSS rule is included.</li>
@@ -588,10 +588,10 @@ keyword") }}
                     {{ render_button(form.save_button) }}
                     <a href="{{url_for('ui.form_delete', uuid=uuid)}}"
                        class="pure-button button-small button-error ">Delete</a>
-                    <a href="{{url_for('ui.clear_watch_history', uuid=uuid)}}"
-                       class="pure-button button-small button-error ">Clear History</a>
+                    {% if watch.history_n %}<a href="{{url_for('ui.clear_watch_history', uuid=uuid)}}"
+                       class="pure-button button-small button-error ">Clear History</a>{% endif %}
                     <a href="{{url_for('ui.form_clone', uuid=uuid)}}"
-                       class="pure-button button-small ">Create Copy</a>
+                       class="pure-button button-small ">Clone &amp; Edit</a>
                 </div>
             </div>
         </form>

changedetectionio/tests/apprise/test_apprise_asset.py

@@ -0,0 +1,24 @@
+import pytest
+from apprise import AppriseAsset
+
+from changedetectionio.apprise_asset import (
+    APPRISE_APP_DESC,
+    APPRISE_APP_ID,
+    APPRISE_APP_URL,
+    APPRISE_AVATAR_URL,
+)
+
+
+@pytest.fixture(scope="function")
+def apprise_asset() -> AppriseAsset:
+    from changedetectionio.apprise_asset import apprise_asset
+
+    return apprise_asset
+
+
+def test_apprise_asset_init(apprise_asset: AppriseAsset):
+    assert isinstance(apprise_asset, AppriseAsset)
+    assert apprise_asset.app_id == APPRISE_APP_ID
+    assert apprise_asset.app_desc == APPRISE_APP_DESC
+    assert apprise_asset.app_url == APPRISE_APP_URL
+    assert apprise_asset.image_url_logo == APPRISE_AVATAR_URL

changedetectionio/tests/apprise/test_apprise_custom_api_call.py

@@ -0,0 +1,211 @@
+import json
+from unittest.mock import patch
+
+import pytest
+import requests
+from apprise.utils.parse import parse_url as apprise_parse_url
+
+from ...apprise_plugin.custom_handlers import (
+    _get_auth,
+    _get_headers,
+    _get_params,
+    apprise_http_custom_handler,
+    SUPPORTED_HTTP_METHODS,
+)
+
+
+@pytest.mark.parametrize(
+    "url,expected_auth",
+    [
+        ("get://user:pass@localhost:9999", ("user", "pass")),
+        ("get://user@localhost:9999", "user"),
+        ("get://localhost:9999", ""),
+        ("get://user%20name:pass%20word@localhost:9999", ("user name", "pass word")),
+    ],
+)
+def test_get_auth(url, expected_auth):
+    """Test authentication extraction with various URL formats."""
+    parsed_url = apprise_parse_url(url)
+    assert _get_auth(parsed_url) == expected_auth
+
+
+@pytest.mark.parametrize(
+    "url,body,expected_content_type",
+    [
+        (
+            "get://localhost:9999?+content-type=application/xml",
+            "test",
+            "application/xml",
+        ),
+        ("get://localhost:9999", '{"key": "value"}', "application/json; charset=utf-8"),
+        ("get://localhost:9999", "plain text", None),
+        ("get://localhost:9999?+content-type=text/plain", "test", "text/plain"),
+    ],
+)
+def test_get_headers(url, body, expected_content_type):
+    """Test header extraction and content type detection."""
+    parsed_url = apprise_parse_url(url)
+    headers = _get_headers(parsed_url, body)
+
+    if expected_content_type:
+        assert headers.get("Content-Type") == expected_content_type
+
+
+@pytest.mark.parametrize(
+    "url,expected_params",
+    [
+        ("get://localhost:9999?param1=value1", {"param1": "value1"}),
+        ("get://localhost:9999?param1=value1&-param2=ignored", {"param1": "value1"}),
+        ("get://localhost:9999?param1=value1&+header=test", {"param1": "value1"}),
+        (
+            "get://localhost:9999?encoded%20param=encoded%20value",
+            {"encoded param": "encoded value"},
+        ),
+    ],
+)
+def test_get_params(url, expected_params):
+    """Test parameter extraction with URL encoding and exclusion logic."""
+    parsed_url = apprise_parse_url(url)
+    params = _get_params(parsed_url)
+    assert dict(params) == expected_params
+
+
+@pytest.mark.parametrize(
+    "url,schema,method",
+    [
+        ("get://localhost:9999", "get", "GET"),
+        ("post://localhost:9999", "post", "POST"),
+        ("delete://localhost:9999", "delete", "DELETE"),
+    ],
+)
+@patch("requests.request")
+def test_apprise_custom_api_call_success(mock_request, url, schema, method):
+    """Test successful API calls with different HTTP methods and schemas."""
+    mock_request.return_value.raise_for_status.return_value = None
+
+    meta = {"url": url, "schema": schema}
+    result = apprise_http_custom_handler(
+        body="test body", title="Test Title", notify_type="info", meta=meta
+    )
+
+    assert result is True
+    mock_request.assert_called_once()
+
+    call_args = mock_request.call_args
+    assert call_args[1]["method"] == method.upper()
+    assert call_args[1]["url"].startswith("http")
+
+
+@patch("requests.request")
+def test_apprise_custom_api_call_with_auth(mock_request):
+    """Test API call with authentication."""
+    mock_request.return_value.raise_for_status.return_value = None
+
+    url = "get://user:pass@localhost:9999/secure"
+    meta = {"url": url, "schema": "get"}
+
+    result = apprise_http_custom_handler(
+        body=json.dumps({"key": "value"}),
+        title="Secure Test",
+        notify_type="info",
+        meta=meta,
+    )
+
+    assert result is True
+    mock_request.assert_called_once()
+    call_args = mock_request.call_args
+    assert call_args[1]["auth"] == ("user", "pass")
+
+
+@pytest.mark.parametrize(
+    "exception_type,expected_result",
+    [
+        (requests.RequestException, False),
+        (requests.HTTPError, False),
+        (Exception, False),
+    ],
+)
+@patch("requests.request")
+def test_apprise_custom_api_call_failure(mock_request, exception_type, expected_result):
+    """Test various failure scenarios."""
+    url = "get://localhost:9999/error"
+    meta = {"url": url, "schema": "get"}
+
+    # Simulate different types of exceptions
+    mock_request.side_effect = exception_type("Error occurred")
+
+    result = apprise_http_custom_handler(
+        body="error body", title="Error Test", notify_type="error", meta=meta
+    )
+
+    assert result == expected_result
+
+
+def test_invalid_url_parsing():
+    """Test handling of invalid URL parsing."""
+    meta = {"url": "invalid://url", "schema": "invalid"}
+    result = apprise_http_custom_handler(
+        body="test", title="Invalid URL", notify_type="info", meta=meta
+    )
+
+    assert result is False
+
+
+@pytest.mark.parametrize(
+    "schema,expected_method",
+    [
+        (http_method, http_method.upper())
+        for http_method in SUPPORTED_HTTP_METHODS
+    ],
+)
+@patch("requests.request")
+def test_http_methods(mock_request, schema, expected_method):
+    """Test all supported HTTP methods."""
+    mock_request.return_value.raise_for_status.return_value = None
+
+    url = f"{schema}://localhost:9999"
+
+    result = apprise_http_custom_handler(
+        body="test body",
+        title="Test Title",
+        notify_type="info",
+        meta={"url": url, "schema": schema},
+    )
+
+    assert result is True
+    mock_request.assert_called_once()
+
+    call_args = mock_request.call_args
+    assert call_args[1]["method"] == expected_method
+
+
+@pytest.mark.parametrize(
+    "input_schema,expected_method",
+    [
+        (f"{http_method}s", http_method.upper())
+        for http_method in SUPPORTED_HTTP_METHODS
+    ],
+)
+@patch("requests.request")
+def test_https_method_conversion(
+    mock_request, input_schema, expected_method
+):
+    """Validate that methods ending with 's' use HTTPS and correct HTTP method."""
+    mock_request.return_value.raise_for_status.return_value = None
+
+    url = f"{input_schema}://localhost:9999"
+
+    result = apprise_http_custom_handler(
+        body="test body",
+        title="Test Title",
+        notify_type="info",
+        meta={"url": url, "schema": input_schema},
+    )
+
+    assert result is True
+    mock_request.assert_called_once()
+
+    call_args = mock_request.call_args
+
+    assert call_args[1]["method"] == expected_method
+    assert call_args[1]["url"].startswith("https")

changedetectionio/tests/smtp/test_notification_smtp.py

@@ -25,7 +25,6 @@ def test_setup(live_server):
 
 def get_last_message_from_smtp_server():
     import socket
-    global smtp_test_server
     port = 11080  # socket server port number
 
     client_socket = socket.socket()  # instantiate
@@ -44,7 +43,6 @@ def test_check_notification_email_formats_default_HTML(client, live_server, meas
     # live_server_setup(live_server)
     set_original_response()
 
-    global smtp_test_server
     notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
 
     #####################
@@ -99,7 +97,6 @@ def test_check_notification_email_formats_default_Text_override_HTML(client, liv
     # https://github.com/caronc/apprise/issues/633
 
     set_original_response()
-    global smtp_test_server
     notification_url = f'mailto://changedetection@{smtp_test_server}:11025/?to=fff@home.com'
     notification_body = f"""<!DOCTYPE html>
 <html lang="en">

changedetectionio/tests/test_access_control.py

@@ -60,6 +60,11 @@ def test_check_access_control(app, client, live_server):
         res = c.get(url_for('static_content', group='styles', filename='404-testetest.css'))
         assert res.status_code == 404
 
+        # Access to screenshots should be limited by 'shared_diff_access'
+        path = url_for('static_content', group='screenshot', filename='random-uuid-that-will-404.png', _external=True)
+        res = c.get(path)
+        assert res.status_code == 404
+
         # Check wrong password does not let us in
         res = c.post(
             url_for("login"),
@@ -163,7 +168,7 @@ def test_check_access_control(app, client, live_server):
             url_for("settings.settings_page"),
             data={"application-password": "foobar",
                   # Should be disabled
-#                  "application-shared_diff_access": "True",
+                  "application-shared_diff_access": "",
                   "requests-time_between_check-minutes": 180,
                   'application-fetch_backend': "html_requests"},
             follow_redirects=True
@@ -176,6 +181,10 @@ def test_check_access_control(app, client, live_server):
         # Should be logged out
         assert b"Login" in res.data
 
+        # Access to screenshots should be limited by 'shared_diff_access'
+        res = c.get(url_for('static_content', group='screenshot', filename='random-uuid-that-will-403.png'))
+        assert res.status_code == 403
+
         # The diff page should return something valid when logged out
         res = c.get(url_for("ui.ui_views.diff_history_page", uuid="first"))
         assert b'Random content' not in res.data

changedetectionio/tests/test_clone.py

@@ -2,29 +2,39 @@
 
 import time
 from flask import url_for
-from . util import live_server_setup
+from .util import live_server_setup, wait_for_all_checks
 
 
-
-def test_trigger_functionality(client, live_server, measure_memory_usage):
+def test_clone_functionality(client, live_server, measure_memory_usage):
 
     live_server_setup(live_server)
+    with open("test-datastore/endpoint-content.txt", "w") as f:
+        f.write("<html><body>Some content</body></html>")
 
-    # Give the endpoint time to spin up
-    time.sleep(1)
+    test_url = url_for('test_endpoint', _external=True)
 
     # Add our URL to the import page
     res = client.post(
         url_for("imports.import_page"),
-        data={"urls": "https://changedetection.io"},
+        data={"urls": test_url},
         follow_redirects=True
     )
     assert b"1 Imported" in res.data
+    wait_for_all_checks(client)
 
+    # So that we can be sure the same history doesnt carry over
+    time.sleep(1)
 
     res = client.get(
         url_for("ui.form_clone", uuid="first"),
         follow_redirects=True
     )
+    existing_uuids = set()
 
-    assert b"Cloned." in res.data
+    for uuid, watch in live_server.app.config['DATASTORE'].data['watching'].items():
+        new_uuids = set(watch.history.keys())
+        duplicates = existing_uuids.intersection(new_uuids)
+        assert len(duplicates) == 0
+        existing_uuids.update(new_uuids)
+
+    assert b"Cloned" in res.data

changedetectionio/tests/test_conditions.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 import json
-import urllib
+import time
 
 from flask import url_for
 from .util import live_server_setup, wait_for_all_checks
@@ -113,6 +113,7 @@ def test_conditions_with_text_and_number(client, live_server):
     client.get(url_for("ui.form_watch_checknow"), follow_redirects=True)
     wait_for_all_checks(client)
 
+    time.sleep(2)
     # 75 is > 20 and < 100 and contains "5"
     res = client.get(url_for("watchlist.index"))
     assert b'unviewed' in res.data

changedetectionio/tests/test_group.py

@@ -273,6 +273,7 @@ def test_limit_tag_ui(client, live_server, measure_memory_usage):
     assert b'Deleted' in res.data
     res = client.get(url_for("tags.delete_all"), follow_redirects=True)
     assert b'All tags deleted' in res.data
+
 def test_clone_tag_on_import(client, live_server, measure_memory_usage):
     #live_server_setup(live_server)
     test_url = url_for('test_endpoint', _external=True)
@@ -292,6 +293,7 @@ def test_clone_tag_on_import(client, live_server, measure_memory_usage):
     res = client.get(url_for("ui.form_clone", uuid=watch_uuid), follow_redirects=True)
 
     assert b'Cloned' in res.data
+    res = client.get(url_for("watchlist.index"))
     # 2 times plus the top link to tag
     assert res.data.count(b'test-tag') == 3
     assert res.data.count(b'another-tag') == 3
@@ -317,8 +319,9 @@ def test_clone_tag_on_quickwatchform_add(client, live_server, measure_memory_usa
 
     watch_uuid = next(iter(live_server.app.config['DATASTORE'].data['watching']))
     res = client.get(url_for("ui.form_clone", uuid=watch_uuid), follow_redirects=True)
-
     assert b'Cloned' in res.data
+
+    res = client.get(url_for("watchlist.index"))
     # 2 times plus the top link to tag
     assert res.data.count(b'test-tag') == 3
     assert res.data.count(b'another-tag') == 3

changedetectionio/tests/unit/test_conditions.py

@@ -75,7 +75,7 @@ class TestTriggerConditions(unittest.TestCase):
                                                      ephemeral_data={'text': "I saw 500 people at a rock show"})
 
         # @todo - now we can test that 'Extract number' increased more than X since last time
-        self.assertTrue(result)
+        self.assertTrue(result.get('result'))
 
 
 if __name__ == '__main__':