PLG+DOCS: better scaffolding

docs/PLUGINS_DEV.md

@@ -235,7 +235,9 @@ Check your plugin against these repo-wide conventions before opening a PR (verif
 
 - **`RUN` defaults to `"disabled"`.** True for the large majority of plugins; only core maintenance plugins (`csv_backup`, `db_cleanup`, `maintenance`, `vendor_update`) default to `schedule`. A new optional plugin should load disabled until the user configures it.
 - **Pick `RUN_SCHD` from precedent, not an arbitrary value.** Check the closest existing plugin for its schedule (e.g. `pihole_api_scan` uses `*/5 * * * *`) rather than inventing a new cadence — consistency keeps first-time setup predictable across plugins.
-- **`RUN_TIMEOUT` is a subprocess kill-timer, not a per-request budget.** The core plugin runner (`server/plugin.py`) passes this same value as the hard timeout for the *entire* script (`subprocess` `timeout=`). If your script makes multiple sequential network calls (e.g. two upstream instances, or a per-device lookup in a loop), don't also reuse `RUN_TIMEOUT` as each individual call's timeout — one slow call can then consume the whole budget and get the process killed before it writes its result file, silently dropping the entire run.
+- **`RUN_TIMEOUT` is a subprocess kill-timer, not a per-request budget.** The core plugin runner (`server/plugin.py`) passes this same value as the hard timeout for the *entire* script (`subprocess` `timeout=`). If your script makes multiple sequential network calls (e.g. two upstream instances, or a per-device lookup in a loop), don't also reuse `RUN_TIMEOUT` as each individual call's timeout — one slow call can then consume the whole budget and get the process killed before it writes its result file, silently dropping the entire run. Two correct alternatives, depending on the shape of your loop:
+  - **Looping over a config-declared, known-length list** (e.g. a subnets or IPs setting) — mark that `params` entry with `"timeoutMultiplier": true` in `config.json`. The framework then multiplies the *outer* kill-timeout by that list's length before running your script, so each iteration can safely use the full `RUN_TIMEOUT` internally. See `arp_scan/config.json`'s `subnets` param for a working example.
+  - **Looping over a runtime-variable-length collection** (e.g. a notification queue, where length isn't known until the script runs) — `timeoutMultiplier` doesn't apply here since there's no config-declared count. Instead, divide the *inner* per-call timeout down using `plugin_helper.per_item_timeout(run_timeout, item_count)`, so N sequential calls can't collectively exceed the outer budget. See `server/plugins/_publisher_ntfy/ntfy.py`'s notification loop for a working example.
 - **Reuse existing core settings instead of duplicating them.** If NetAlertX already has a concept your plugin needs (e.g. `API_TOKEN` for its own GraphQL/API endpoint), read it with `get_setting_value("API_TOKEN")` rather than adding a plugin-specific `<PREFIX>_API_TOKEN` — see `server/plugins/sync/sync.py` for the pattern.
 - **Keep `description` strings short.** They render directly in the Settings UI. Put implementation rationale and design trade-offs in the plugin's README or code comments, not the UI-facing description.
 - **For "one or more instances of the same thing," use the nested array + popup-form settings pattern**, not a fixed hardcoded count (e.g. "primary"/"secondary"). See `rest_import` (`RSTIMPRT`)'s `imports` setting for a working example — it also gives each instance its own sub-settings (URL, credentials, per-instance flags) for free.

docs/PLUGINS_DEV_DATA_CONTRACT.md

@@ -85,6 +85,8 @@ The library automatically:
 | 11 | `helpVal3` | string | *conditional* | Helper value 3. If used, all help values must be supplied |
 | 12 | `helpVal4` | string | *conditional* | Helper value 4. If used, all help values must be supplied |
 
+> **Gotcha:** `plugin_helper.py`'s `Plugin_Object.__init__` applies `helpVal1-4 or ""` internally, collapsing any falsy value (`0`, `False`, `""`) to `""` - indistinguishable from "not supplied." `watchedValue1-4` are stored as-is with no such coercion. If a real `0` or `False` is a meaningful value you need to preserve (not just "absent"), put it in a `watchedValue*` column instead of a `helpVal*` one.
+
 ## Usage Guide
 
 ### Empty/Null Values

server/plugins/_publisher_ntfy/ntfy.py

@@ -13,7 +13,7 @@ sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"])
 
 import conf  # noqa: E402 [flake8 lint suppression]
 from const import confFileName, logPath  # noqa: E402 [flake8 lint suppression]
-from plugin_helper import Plugin_Objects, handleEmpty  # noqa: E402 [flake8 lint suppression]
+from plugin_helper import Plugin_Objects, handleEmpty, per_item_timeout  # noqa: E402 [flake8 lint suppression]
 from utils.datetime_utils import timeNowUTC  # noqa: E402 [flake8 lint suppression]
 from logger import mylog, Logger  # noqa: E402 [flake8 lint suppression]
 from helper import get_setting_value  # noqa: E402 [flake8 lint suppression]
@@ -55,11 +55,18 @@ def main():
     # Retrieve new notifications
     new_notifications = notifications.getNew()
 
+    # RUN_TIMEOUT is enforced by the core plugin runner as this whole
+    # script's kill-timeout, not a safe per-request timeout - divide it
+    # across the queue so a burst of notifications can't let one slow send()
+    # call consume the whole budget and get the process killed mid-loop.
+    run_timeout = int(get_setting_value('NTFY_RUN_TIMEOUT') or 10)
+    per_call_timeout = per_item_timeout(run_timeout, len(new_notifications))
+
     # Process the new notifications (see the Notifications DB table for structure or check the /php/server/query_json.php?file=table_notifications.json endpoint)
     for notification in new_notifications:
 
         # Send notification
-        response_text, response_status_code = send(notification["HTML"], notification["Text"])
+        response_text, response_status_code = send(notification["HTML"], notification["Text"], per_call_timeout)
 
         # Log result
         plugin_objects.add_object(
@@ -131,11 +138,14 @@ def build_custom_headers(entries, reserved_headers):
 
 
 # -------------------------------------------------------------------------------
-def send(html, text):
+def send(html, text, timeout=None):
 
     response_text = ''
     response_status_code = ''
 
+    if timeout is None:
+        timeout = int(get_setting_value('NTFY_RUN_TIMEOUT') or 10)
+
     # settings
     token = get_setting_value('NTFY_TOKEN')
     user = get_setting_value('NTFY_USER')
@@ -178,7 +188,7 @@ def send(html, text):
             headers = headers,
             params  = url_query_string if url_query_string != '' else None,
             verify  = verify_ssl,
-            timeout = get_setting_value('NTFY_RUN_TIMEOUT')
+            timeout = timeout
         )
 
         response_status_code = response.status_code

server/plugins/_publisher_pushsafer/pushsafer.py

@@ -10,7 +10,7 @@ sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"])
 
 import conf  # noqa: E402 [flake8 lint suppression]
 from const import confFileName, logPath  # noqa: E402 [flake8 lint suppression]
-from plugin_helper import Plugin_Objects, handleEmpty  # noqa: E402 [flake8 lint suppression]
+from plugin_helper import Plugin_Objects, handleEmpty, per_item_timeout  # noqa: E402 [flake8 lint suppression]
 from logger import mylog, Logger  # noqa: E402 [flake8 lint suppression]
 from helper import get_setting_value, hide_string  # noqa: E402 [flake8 lint suppression]
 from utils.datetime_utils import timeNowUTC  # noqa: E402 [flake8 lint suppression]
@@ -52,11 +52,18 @@ def main():
     # Retrieve new notifications
     new_notifications = notifications.getNew()
 
+    # RUN_TIMEOUT is enforced by the core plugin runner as this whole
+    # script's kill-timeout, not a safe per-request timeout - divide it
+    # across the queue so a burst of notifications can't let one slow send()
+    # call consume the whole budget and get the process killed mid-loop.
+    run_timeout = int(get_setting_value("PUSHSAFER_RUN_TIMEOUT") or 10)
+    per_call_timeout = per_item_timeout(run_timeout, len(new_notifications))
+
     # Process the new notifications (see the Notifications DB table for structure or check the /php/server/query_json.php?file=table_notifications.json endpoint)
     for notification in new_notifications:
 
         # Send notification
-        response_text, response_status_code = send(notification["Text"])
+        response_text, response_status_code = send(notification["Text"], per_call_timeout)
 
         # Log result
         plugin_objects.add_object(
@@ -74,13 +81,16 @@ def main():
 
 
 # -------------------------------------------------------------------------------
-def send(text):
+def send(text, timeout=None):
 
     response_text = ''
     response_status_code = ''
 
     token = get_setting_value('PUSHSAFER_TOKEN')
 
+    if timeout is None:
+        timeout = int(get_setting_value("PUSHSAFER_RUN_TIMEOUT") or 10)
+
     mylog('verbose', [f'[{pluginName}] PUSHSAFER_TOKEN: "{hide_string(token)}"'])
 
     try:
@@ -97,7 +107,7 @@ def send(text):
             "ut" : 'Open NetAlertX',
             "k" : token,
         }
-        response = requests.post(url, data=post_fields, timeout=get_setting_value("PUSHSAFER_RUN_TIMEOUT"))
+        response = requests.post(url, data=post_fields, timeout=timeout)
         response_status_code = response.status_code
 
         # Check if the request was successful (status code 200)

server/plugins/_publisher_telegram/tg.py

@@ -11,7 +11,7 @@ sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"])
 
 import conf  # noqa: E402 [flake8 lint suppression]
 from const import confFileName, logPath  # noqa: E402 [flake8 lint suppression]
-from plugin_helper import Plugin_Objects  # noqa: E402 [flake8 lint suppression]
+from plugin_helper import Plugin_Objects, per_item_timeout  # noqa: E402 [flake8 lint suppression]
 from utils.datetime_utils import timeNowUTC  # noqa: E402 [flake8 lint suppression]
 from logger import mylog, Logger  # noqa: E402 [flake8 lint suppression]
 from helper import get_setting_value  # noqa: E402 [flake8 lint suppression]
@@ -53,10 +53,17 @@ def main():
     # Retrieve new notifications
     new_notifications = notifications.getNew()
 
+    # RUN_TIMEOUT is enforced by the core plugin runner as this whole
+    # script's kill-timeout, not a safe per-request timeout - divide it
+    # across the queue so a burst of notifications can't let one slow send()
+    # call consume the whole budget and get the process killed mid-loop.
+    run_timeout = int(get_setting_value('TELEGRAM_RUN_TIMEOUT'))
+    per_call_timeout = per_item_timeout(run_timeout, len(new_notifications))
+
     # Process the new notifications (see the Notifications DB table for structure or check the /php/server/query_json.php?file=table_notifications.json endpoint)
     for notification in new_notifications:
         # Send notification
-        result = send(notification["Text"])
+        result = send(notification["Text"], per_call_timeout)
 
         # Log result
         plugin_objects.add_object(
@@ -79,13 +86,14 @@ def check_config():
 
 
 # -------------------------------------------------------------------------------
-def send(text):
+def send(text, timeout=None):
     """
     Send a Telegram notification.
     """
     limit = get_setting_value('TELEGRAM_SIZE')
-    run_timeout = int(get_setting_value('TELEGRAM_RUN_TIMEOUT'))
-    curl_timeout = str(max(1, run_timeout - 1))
+    if timeout is None:
+        timeout = int(get_setting_value('TELEGRAM_RUN_TIMEOUT'))
+    curl_timeout = str(max(1, timeout - 1))
 
     # Ensure the final payload, including the truncation marker,
     # never exceeds TELEGRAM_SIZE.

server/plugins/adguard_export/script.py

@@ -374,7 +374,7 @@ def main():
     # Read settings
     # ------------------------------------------------------------------
     agrd_url        = get_setting_value("ADGUARDEXP_URL")       or "http://localhost:3000"
-    agrd_user       = get_setting_value("ADGUARDEXP_USER")      or ""
+    agrd_user       = get_setting_value("ADGUARDEXP_USER")      or "admin"
     agrd_pass       = get_setting_value("ADGUARDEXP_PASSWORD")  or ""
     verify_ssl_str      = get_setting_value("ADGUARDEXP_VERIFYSSL")       or "true"
     include_offline_str = get_setting_value("ADGUARDEXP_INCLUDE_OFFLINE") or "true"

server/plugins/adguard_import/adguard_import.py

@@ -66,7 +66,7 @@ def main():
     user = get_setting_value("ADGUARDIMP_USER")
     pw = get_setting_value("ADGUARDIMP_PASS")
     fake_mac_enabled = get_setting_value("ADGUARDIMP_FAKE_MAC")
-    timeout = int(get_setting_value("ADGUARDIMP_RUN_TIMEOUT") or 5)
+    timeout = int(get_setting_value("ADGUARDIMP_RUN_TIMEOUT") or 30)
 
     auth = (user, pw) if user or pw else None
 

server/plugins/nbtscan_scan/nbtscan.py

@@ -36,8 +36,11 @@ plugin_objects = Plugin_Objects(RESULT_FILE)
 def main():
     mylog('verbose', [f'[{pluginName}] In script'])
 
-    # timeout = get_setting_value('NBLOOKUP_RUN_TIMEOUT')
-    timeout = 20
+    # The "ips" param in config.json is marked timeoutMultiplier: true, so the
+    # framework already scales the outer subprocess kill-timeout by device
+    # count - use the real per-device budget here instead of a hardcoded
+    # value that could exceed what the multiplier actually grants.
+    timeout = int(get_setting_value('NBTSCAN_RUN_TIMEOUT') or 10)
 
     # Initialize the Plugin obj output file
     plugin_objects = Plugin_Objects(RESULT_FILE)

server/plugins/nmap_dev_scan/config.json

@@ -44,7 +44,8 @@
       "name": "subnets",
       "type": "setting",
       "value": "SCAN_SUBNETS",
-      "base64": true
+      "base64": true,
+      "timeoutMultiplier": true
     }
   ],
   "settings": [

server/plugins/plugin_helper.py

@@ -264,6 +264,28 @@ def normalize_mac(mac):
     return ':'.join(normalized_parts)
 
 
+# -------------------------------------------------------------------
+def per_item_timeout(run_timeout, item_count, floor=1):
+    """
+    Divide a RUN_TIMEOUT budget evenly across `item_count` sequential
+    operations (e.g. one HTTP call per queued notification) so no single
+    item can consume the whole script's kill-timeout - the core plugin
+    runner (server/plugin.py) enforces RUN_TIMEOUT as the entire
+    subprocess's hard timeout, not a per-call one.
+
+    Returns run_timeout unchanged when there's 0 or 1 items, so the common
+    single-item case sees no behavior change. For a config-declared,
+    known-length list (e.g. a subnets/IPs setting), prefer the config.json
+    "timeoutMultiplier" mechanism instead - it scales the outer timeout up
+    rather than dividing the inner one down. Use this helper for
+    runtime-variable-length loops (e.g. a notification queue) where
+    timeoutMultiplier doesn't apply.
+    """
+    if item_count <= 1:
+        return run_timeout
+    return max(floor, run_timeout // item_count)
+
+
 # -------------------------------------------------------------------
 class Plugin_Object:
     """

test/plugins/test___template.py

@@ -20,21 +20,16 @@ from unittest.mock import MagicMock
 _tmp_log = tempfile.mkdtemp()
 _tmp_db = tempfile.mkdtemp()
 
+_stubbed_module_names = []
+
 
 def _stub(name: str, **attrs):
-    # Additive: several plugin test files stub the same generic module names
-    # (helper, plugin_helper, const, ...) with different attribute subsets.
-    # If another test already registered this name, add whatever attributes
-    # it doesn't have yet instead of skipping outright - a plain skip-if-
-    # present guard makes collection order decide which test's dependencies
-    # win, breaking whichever test runs later in the same pytest session.
-    mod = sys.modules.get(name)
-    if mod is None:
+    if name not in sys.modules:
         mod = types.ModuleType(name)
-        sys.modules[name] = mod
-    for k, v in attrs.items():
-        if not hasattr(mod, k):
+        for k, v in attrs.items():
             setattr(mod, k, v)
+        sys.modules[name] = mod
+        _stubbed_module_names.append(name)
 
 
 _stub("pytz", timezone=lambda tz: tz)
@@ -48,6 +43,12 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "server",
 
 import rename_me  # noqa: E402
 
+# Stops these fake entries from shadowing the real modules for other test
+# files collected later in the same pytest session (rename_me's own
+# module-level `from x import y` bindings are already resolved by now).
+for _name in _stubbed_module_names:
+    sys.modules.pop(_name, None)
+
 
 class TestGetDeviceData:
     def test_returns_the_sample_devices(self):

test/plugins/test_adguard_export.py

@@ -26,21 +26,16 @@ _tmp_log = tempfile.mkdtemp()
 _tmp_data = tempfile.mkdtemp()
 _tmp_db = tempfile.mkdtemp()
 
+_stubbed_module_names = []
+
 
 def _stub(name: str, **attrs):
-    # Additive: several plugin test files stub the same generic module names
-    # (helper, plugin_helper, const, ...) with different attribute subsets.
-    # If another test already registered this name, add whatever attributes
-    # it doesn't have yet instead of skipping outright - a plain skip-if-
-    # present guard makes collection order decide which test's dependencies
-    # win, breaking whichever test runs later in the same pytest session.
-    mod = sys.modules.get(name)
-    if mod is None:
+    if name not in sys.modules:
         mod = types.ModuleType(name)
-        sys.modules[name] = mod
-    for k, v in attrs.items():
-        if not hasattr(mod, k):
+        for k, v in attrs.items():
             setattr(mod, k, v)
+        sys.modules[name] = mod
+        _stubbed_module_names.append(name)
 
 
 _stub("pytz", timezone=lambda tz: tz)
@@ -60,6 +55,9 @@ _stub("models.device_instance", DeviceInstance=MagicMock)
 
 # Stub requests only when it isn't installed (e.g. bare system Python locally).
 # In the container and CI, the real package is present and will be used.
+# Tracked via _stubbed_module_names (not the real package) so it gets popped
+# below like the other stubs, instead of leaking an incomplete fake `requests`
+# (missing .post/.get) to other test files collected later in the same run.
 if "requests" not in sys.modules:
     _req = types.ModuleType("requests")
     _req.Session = MagicMock
@@ -69,6 +67,7 @@ if "requests" not in sys.modules:
     _req.exceptions = _req_exc
     sys.modules["requests"] = _req
     sys.modules["requests.exceptions"] = _req_exc
+    _stubbed_module_names.extend(["requests", "requests.exceptions"])
 
 # ---------------------------------------------------------------------------
 # Import the functions under test (must come after the stubs above).
@@ -87,6 +86,12 @@ from script import (  # noqa: E402
     sync_to_adguard,
 )
 
+# Stops these fake entries from shadowing the real modules for other test
+# files collected later in the same pytest session (script's own
+# module-level `from x import y` bindings are already resolved by now).
+for _name in _stubbed_module_names:
+    sys.modules.pop(_name, None)
+
 
 # ---------------------------------------------------------------------------
 # Helpers

test/plugins/test_ntfy_custom_headers.py

@@ -39,7 +39,7 @@ def _stub(name: str, **attrs):
 _stub("pytz", timezone=lambda tz: tz)
 _stub("conf", tz=None)
 _stub("const", confFileName="app.conf", logPath=_tmp_log)
-_stub("plugin_helper", Plugin_Objects=MagicMock, handleEmpty=lambda v: v)
+_stub("plugin_helper", Plugin_Objects=MagicMock, handleEmpty=lambda v: v, per_item_timeout=lambda run_timeout, count, floor=1: run_timeout)
 _stub("utils")
 _stub("utils.datetime_utils", timeNowUTC=lambda: "2026-01-01 00:00:00")
 _stub("logger", mylog=lambda *a: None, Logger=MagicMock)
@@ -57,6 +57,7 @@ if "requests" not in sys.modules:
     _req.exceptions = _req_exc
     sys.modules["requests"] = _req
     sys.modules["requests.exceptions"] = _req_exc
+    _stubbed_module_names.extend(["requests", "requests.exceptions"])
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "server", "plugins", "_publisher_ntfy"))
 

test/plugins/test_pushsafer.py

@@ -0,0 +1,77 @@
+"""
+Tests for _publisher_pushsafer/pushsafer.py - focused on the per-notification
+timeout wiring (RUN_TIMEOUT divided across a queue via
+plugin_helper.per_item_timeout, instead of reused unchanged per call).
+
+Run from inside the NetAlertX container, or locally - NetAlertX-specific
+modules are stubbed out automatically before the script is imported.
+
+    pytest test/plugins/test_pushsafer.py -v
+"""
+
+import os
+import sys
+import tempfile
+import types
+from unittest.mock import MagicMock, patch
+
+_tmp_log = tempfile.mkdtemp()
+
+_stubbed_module_names = []
+
+
+def _stub(name: str, **attrs):
+    if name not in sys.modules:
+        mod = types.ModuleType(name)
+        for k, v in attrs.items():
+            setattr(mod, k, v)
+        sys.modules[name] = mod
+        _stubbed_module_names.append(name)
+
+
+_stub("pytz", timezone=lambda tz: tz)
+_stub("conf", tz=None)
+_stub("const", confFileName="app.conf", logPath=_tmp_log)
+_stub("plugin_helper", Plugin_Objects=MagicMock, handleEmpty=lambda v: v, per_item_timeout=lambda run_timeout, count, floor=1: run_timeout)
+_stub("logger", mylog=lambda *a: None, Logger=MagicMock)
+_stub("helper", get_setting_value=lambda k: "", hide_string=lambda s: "***")
+_stub("utils")
+_stub("utils.datetime_utils", timeNowUTC=lambda: "2026-01-01 00:00:00")
+_stub("models")
+_stub("models.notification_instance", NotificationInstance=MagicMock)
+_stub("database", DB=MagicMock)
+
+if "requests" not in sys.modules:
+    _req = types.ModuleType("requests")
+    _req.post = MagicMock
+    _req_exc = types.ModuleType("requests.exceptions")
+    _req_exc.RequestException = type("RequestException", (Exception,), {})
+    _req.exceptions = _req_exc
+    sys.modules["requests"] = _req
+    sys.modules["requests.exceptions"] = _req_exc
+    _stubbed_module_names.extend(["requests", "requests.exceptions"])
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "server", "plugins", "_publisher_pushsafer"))
+
+import pushsafer  # noqa: E402
+
+# Stops these fake entries from shadowing the real modules for other test
+# files collected later in the same pytest session (pushsafer's own
+# module-level `from x import y` bindings are already resolved by now).
+for _name in _stubbed_module_names:
+    sys.modules.pop(_name, None)
+
+
+class TestSendTimeout:
+    def test_uses_explicit_timeout_when_given(self):
+        mock_response = MagicMock(status_code=200, text="ok")
+        with patch("pushsafer.requests.post", return_value=mock_response) as mock_post:
+            pushsafer.send("hello", timeout=3)
+        assert mock_post.call_args.kwargs["timeout"] == 3
+
+    def test_falls_back_to_setting_when_timeout_not_given(self):
+        mock_response = MagicMock(status_code=200, text="ok")
+        with patch("pushsafer.get_setting_value", return_value="10"), \
+                patch("pushsafer.requests.post", return_value=mock_response) as mock_post:
+            pushsafer.send("hello")
+        assert mock_post.call_args.kwargs["timeout"] == 10

test/plugins/test_tg.py

@@ -0,0 +1,90 @@
+"""
+Tests for _publisher_telegram/tg.py - focused on the per-notification timeout
+wiring (RUN_TIMEOUT divided across a queue via plugin_helper.per_item_timeout,
+instead of reused unchanged per call).
+
+Run from inside the NetAlertX container, or locally - NetAlertX-specific
+modules are stubbed out automatically before the script is imported.
+
+    pytest test/plugins/test_tg.py -v
+"""
+
+import os
+import sys
+import tempfile
+import types
+from unittest.mock import MagicMock, patch
+
+_tmp_log = tempfile.mkdtemp()
+
+_stubbed_module_names = []
+
+
+def _stub(name: str, **attrs):
+    if name not in sys.modules:
+        mod = types.ModuleType(name)
+        for k, v in attrs.items():
+            setattr(mod, k, v)
+        sys.modules[name] = mod
+        _stubbed_module_names.append(name)
+
+
+_stub("pytz", timezone=lambda tz: tz)
+_stub("conf", tz=None)
+_stub("const", confFileName="app.conf", logPath=_tmp_log)
+_stub("plugin_helper", Plugin_Objects=MagicMock, per_item_timeout=lambda run_timeout, count, floor=1: run_timeout)
+_stub("logger", mylog=lambda *a: None, Logger=MagicMock)
+_stub("helper", get_setting_value=lambda k: "")
+_stub("utils")
+_stub("utils.datetime_utils", timeNowUTC=lambda: "2026-01-01 00:00:00")
+_stub("models")
+_stub("models.notification_instance", NotificationInstance=MagicMock)
+_stub("database", DB=MagicMock)
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "server", "plugins", "_publisher_telegram"))
+
+import tg  # noqa: E402
+
+# Stops these fake entries from shadowing the real modules for other test
+# files collected later in the same pytest session (tg's own module-level
+# `from x import y` bindings are already resolved by now).
+for _name in _stubbed_module_names:
+    sys.modules.pop(_name, None)
+
+
+def _mock_proc(stdout='{"ok": true}'):
+    return MagicMock(stdout=stdout, returncode=0)
+
+
+def _settings(run_timeout="10", size=4096, host="123:ABC", url="123:ABC"):
+    values = {
+        "TELEGRAM_SIZE": size,
+        "TELEGRAM_RUN_TIMEOUT": run_timeout,
+        "TELEGRAM_HOST": host,
+        "TELEGRAM_URL": url,
+    }
+    return lambda key: values.get(key, "")
+
+
+class TestSendTimeout:
+    def test_uses_explicit_timeout_for_curl(self):
+        with patch("tg.get_setting_value", side_effect=_settings(run_timeout="1000")), \
+                patch("tg.subprocess.run", return_value=_mock_proc()) as mock_run:
+            tg.send("hello", timeout=5)
+        cmd = mock_run.call_args.args[0]
+        assert cmd[cmd.index("--connect-timeout") + 1] == "4"  # timeout - 1
+        assert cmd[cmd.index("--max-time") + 1] == "4"
+
+    def test_falls_back_to_setting_when_timeout_not_given(self):
+        with patch("tg.get_setting_value", side_effect=_settings(run_timeout="10")), \
+                patch("tg.subprocess.run", return_value=_mock_proc()) as mock_run:
+            tg.send("hello")
+        cmd = mock_run.call_args.args[0]
+        assert cmd[cmd.index("--connect-timeout") + 1] == "9"  # setting(10) - 1
+
+    def test_curl_timeout_floor_is_one(self):
+        with patch("tg.get_setting_value", side_effect=_settings()), \
+                patch("tg.subprocess.run", return_value=_mock_proc()) as mock_run:
+            tg.send("hello", timeout=1)
+        cmd = mock_run.call_args.args[0]
+        assert cmd[cmd.index("--connect-timeout") + 1] == "1"  # max(1, 1-1)

test/plugins/test_unifi_import.py

@@ -21,21 +21,16 @@ import pytest
 _tmp_log = tempfile.mkdtemp()
 _tmp_db = tempfile.mkdtemp()
 
+_stubbed_module_names = []
+
 
 def _stub(name: str, **attrs):
-    # Additive: several plugin test files stub the same generic module names
-    # (helper, plugin_helper, const, ...) with different attribute subsets.
-    # If another test already registered this name, add whatever attributes
-    # it doesn't have yet instead of skipping outright - a plain skip-if-
-    # present guard makes collection order decide which test's dependencies
-    # win, breaking whichever test runs later in the same pytest session.
-    mod = sys.modules.get(name)
-    if mod is None:
+    if name not in sys.modules:
         mod = types.ModuleType(name)
-        sys.modules[name] = mod
-    for k, v in attrs.items():
-        if not hasattr(mod, k):
+        for k, v in attrs.items():
             setattr(mod, k, v)
+        sys.modules[name] = mod
+        _stubbed_module_names.append(name)
 
 
 _stub("pytz", timezone=lambda tz: tz)
@@ -58,6 +53,7 @@ if "pyunifi" not in sys.modules:
     _pyunifi.controller = _pyunifi_controller
     sys.modules["pyunifi"] = _pyunifi
     sys.modules["pyunifi.controller"] = _pyunifi_controller
+    _stubbed_module_names.extend(["pyunifi", "pyunifi.controller"])
 
 if "urllib3" not in sys.modules:
     _urllib3 = types.ModuleType("urllib3")
@@ -67,6 +63,7 @@ if "urllib3" not in sys.modules:
     _urllib3.exceptions = _urllib3_exc
     sys.modules["urllib3"] = _urllib3
     sys.modules["urllib3.exceptions"] = _urllib3_exc
+    _stubbed_module_names.extend(["urllib3", "urllib3.exceptions"])
 
 # unifi_import's module file is named "script.py", same as several other
 # plugins (e.g. adguard_export) - load it under a private module name
@@ -79,6 +76,12 @@ script = importlib.util.module_from_spec(_spec)
 sys.modules["unifi_import_script"] = script
 _spec.loader.exec_module(script)
 
+# Stops these fake entries from shadowing the real modules for other test
+# files collected later in the same pytest session (script's own
+# module-level `from x import y` bindings are already resolved by now).
+for _name in _stubbed_module_names:
+    sys.modules.pop(_name, None)
+
 _migrate_legacy_lock_file = script._migrate_legacy_lock_file
 check_full_run_state = script.check_full_run_state
 read_lock_file = script.read_lock_file

test/test_plugin_helper.py

@@ -1,4 +1,4 @@
-from server.plugins.plugin_helper import is_mac, normalize_mac
+from server.plugins.plugin_helper import is_mac, normalize_mac, per_item_timeout
 
 
 def test_is_mac_accepts_wildcard():
@@ -27,4 +27,21 @@ def test_normalize_mac_preserves_internet_root():
     # Stays lowercase
     assert normalize_mac("internet") == "internet"
     assert normalize_mac("Internet") == "internet"
-    assert normalize_mac("INTERNET") == "internet"
+    assert normalize_mac("INTERNET") == "internet"
+
+
+def test_per_item_timeout_unchanged_for_zero_or_one_items():
+    # The common case (0 or 1 queued items) must see no behavior change.
+    assert per_item_timeout(10, 0) == 10
+    assert per_item_timeout(10, 1) == 10
+
+
+def test_per_item_timeout_divides_budget_across_items():
+    assert per_item_timeout(10, 5) == 2
+    assert per_item_timeout(9, 2) == 4  # integer division, not rounded
+
+
+def test_per_item_timeout_never_goes_below_floor():
+    # A large queue must not divide the per-item timeout down to 0.
+    assert per_item_timeout(10, 100) == 1
+    assert per_item_timeout(10, 100, floor=2) == 2