Tweak to tests

changedetectionio/__init__.py

@@ -19,6 +19,7 @@ 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
@@ -28,6 +29,8 @@ 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

@@ -0,0 +1,12 @@
+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

@@ -0,0 +1,98 @@
+# 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

@@ -1,16 +0,0 @@
-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

@@ -1,112 +0,0 @@
-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,7 +20,10 @@ 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 'diff_history_page' in request.endpoint and datastore.data['settings']['application'].get('shared_diff_access'):
+        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'):
             return func(*args, **kwargs)
         elif request.method in flask_login.config.EXEMPT_METHODS:
             return func(*args, **kwargs)

changedetectionio/blueprint/browser_steps/__init__.py

@@ -23,6 +23,7 @@ from loguru import logger
 browsersteps_sessions = {}
 io_interface_context = None
 import json
+import base64
 import hashlib
 from flask import Response
 
@@ -33,8 +34,10 @@ 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
 
@@ -101,6 +104,8 @@ 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')
 
@@ -144,6 +149,7 @@ 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/notification.py

@@ -4,7 +4,6 @@ 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")
@@ -18,10 +17,11 @@ def construct_blueprint(datastore: ChangeDetectionStore):
 
         # Watch_uuid could be unset in the case it`s used in tag editor, global settings
         import apprise
-        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)
+        from changedetectionio.apprise_asset import asset
+        apobj = apprise.Apprise(asset=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,6 +90,7 @@ 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,7 +116,8 @@ 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 {'executed_data': EXECUTE_DATA, 'result': result}
+    return result
+
 
 # Load plugins dynamically
 for plugin in plugin_manager.get_plugins():

changedetectionio/conditions/blueprint.py

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

changedetectionio/flask_app.py

@@ -233,8 +233,7 @@ 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:
-                # Handled by static_content handler
+            if request.endpoint and request.endpoint == 'static_content' and request.view_args and request.view_args.get('group') in ['styles', 'js', 'images', 'favicons']:
                 return None
             # Permitted
             elif request.endpoint and 'login' in request.endpoint:
@@ -352,15 +351,11 @@ 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:
-                if not datastore.data['settings']['application'].get('shared_diff_access'):
-                    abort(403)
+                abort(403)
 
             screenshot_filename = "last-screenshot.png" if not request.args.get('error_screenshot') else "last-error-screenshot.png"
 
@@ -409,7 +404,7 @@ def changedetection_app(config=None, datastore_o=None):
 
         # These files should be in our subdirectory
         try:
-            return send_from_directory(f"static/{group}", path=filename)
+            return send_from_directory("static/{}".format(group), path=filename)
         except FileNotFoundError:
             abort(404)
 

changedetectionio/forms.py

@@ -306,10 +306,10 @@ class ValidateAppRiseServers(object):
 
     def __call__(self, form, field):
         import apprise
-        from .apprise_plugin.assets import apprise_asset
-        from .apprise_plugin.custom_handlers import apprise_http_custom_handler  # noqa: F401
+        apobj = apprise.Apprise()
 
-        apobj = apprise.Apprise(asset=apprise_asset)
+        # so that the custom endpoints are registered
+        from .apprise_asset import 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) or os.path.getsize(filepath) == 0:
+        if not os.path.isfile(filepath):
             # 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,9 +4,6 @@ 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': '',
@@ -42,6 +39,10 @@ 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")
@@ -65,12 +66,12 @@ def process_notification(n_object, datastore):
     # raise it as an exception
 
     sent_objs = []
-    from .apprise_plugin.assets import apprise_asset
+    from .apprise_asset import asset
 
     if 'as_async' in n_object:
-        apprise_asset.async_mode = n_object.get('as_async')
+        asset.async_mode = n_object.get('as_async')
 
-    apobj = apprise.Apprise(debug=True, asset=apprise_asset)
+    apobj = apprise.Apprise(debug=True, asset=asset)
 
     if not n_object.get('notification_urls'):
         return None
@@ -111,7 +112,7 @@ def process_notification(n_object, datastore):
                     and not url.startswith('get') \
                     and not url.startswith('delete') \
                     and not url.startswith('put'):
-                url += k + f"avatar_url={APPRISE_AVATAR_URL}"
+                url += k + 'avatar_url=https://raw.githubusercontent.com/dgtlmoon/changedetection.io/master/changedetectionio/static/images/avatar-256x256.png'
 
             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,14 +334,12 @@ 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'):
-            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'):
+            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 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
-            let rule = {
+            const rule = {
                 field: field,
                 operator: operator,
                 value: value
@@ -96,10 +96,6 @@ $(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/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">
-
-                        <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>
+                        <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>
                     </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 1 &amp; 2, 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, 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>

changedetectionio/tests/apprise/test_apprise_asset.py

@@ -1,24 +0,0 @@
-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

@@ -1,211 +0,0 @@
-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,6 +25,7 @@ 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
@@ -43,6 +44,7 @@ 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'
 
     #####################
@@ -97,6 +99,7 @@ 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,11 +60,6 @@ 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"),
@@ -168,7 +163,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": "",
+#                  "application-shared_diff_access": "True",
                   "requests-time_between_check-minutes": 180,
                   'application-fetch_backend': "html_requests"},
             follow_redirects=True
@@ -181,10 +176,6 @@ 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_conditions.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 import json
-import time
+import urllib
 
 from flask import url_for
 from .util import live_server_setup, wait_for_all_checks
@@ -113,7 +113,6 @@ 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/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.get('result'))
+        self.assertTrue(result)
 
 
 if __name__ == '__main__':