Merge pull request 'feat(notify): configurable plain/html summary formatting' (#64) from dev-notification-format into main

Reviewed-on: #64

README.md

@@ -67,8 +67,19 @@ delete_torrents = false
 
 [pushover]
 use_pushover = false
-po_key = "<key>>"
+po_key = "<key>"
-po_token = "<token>>"
+po_token = "<token>"
+# Notification formatting
+# format: "plain" (aligned text) or "html" (bold labels, Pushover HTML)
+format = "plain"
+use_emoji = false
+include_hostname = false
+notify_on_empty = true
+# Optional: -2..2. Priority 2 also needs retry/expire (defaults shown).
+# priority = 0
+# sound = "pushover"
+# retry = 60
+# expire = 3600
 
 [apprise]
 use_apprise = false
@@ -119,3 +130,50 @@ for example:
 ```
 10/26/2026 10:10:00 PM [INFO] qbit-maid:tor_processor:24 - Submitted ["My Torrent..."] for deletion.
 ```
+
+## Notifications
+
+Pushover and Apprise share one summary builder, so both report the same
+numbers. The title states the outcome, the body gives the scanned total, the
+actionable counters, then the tracker split.
+
+`format = "plain"` (default) produces aligned text:
+
+```
+qbit-maid: 7 torrents removed
+
+Scanned 42 torrents in 12s
+
+Removed          7
+Orphaned         4
+Ignored          5
+Premature        3
+
+Protected       10
+Non-protected   20
+```
+
+`format = "html"` bolds the labels and counts using Pushover's HTML subset
+(`html=1` is set automatically). Apprise has no HTML support, so it always
+receives the plain body.
+
+Setting `use_emoji = true` switches both styles to `emoji label: value` rows:
+
+```
+🗑️ Removed: 7
+👻 Orphaned: 4
+⏭️ Ignored: 5
+🐣 Premature: 3
+```
+
+Other options:
+
+- `include_hostname` — add a `Host:` line, useful when several instances notify
+  the same device.
+- `notify_on_empty` — set to `false` to skip the push when nothing was removed
+  or orphaned.
+- `priority` / `sound` — passed through to Pushover. Priority `2` automatically
+  adds `retry` and `expire`.
+
+When `delete_torrents = false` the summary says `Would remove` instead of
+`Removed`, so a dry run is never mistaken for a real one.

config.toml.example

@@ -23,8 +23,19 @@ delete_torrents = false
 
 [pushover]
 use_pushover = false
-po_key = "<key>>"
+po_key = "<key>"
-po_token = "<token>>"
+po_token = "<token>"
+# Notification formatting
+# format: "plain" (aligned text) or "html" (bold labels, Pushover HTML)
+format = "plain"
+use_emoji = false
+include_hostname = false
+notify_on_empty = true
+# Optional: -2..2. Priority 2 also needs retry/expire (defaults shown).
+# priority = 0
+# sound = "pushover"
+# retry = 60
+# expire = 3600
 
 [apprise]
 use_apprise = false

qlogging.py

@@ -1,4 +1,5 @@
 import logging
+import socket
 from logging.handlers import RotatingFileHandler
 
 
@@ -9,16 +10,14 @@ def tor_notify(self):
         self.poc = self.po.Pushover(self.po_token)
 
 def tor_notify_apprise(self, req_obj, app_obj):
-    """Use apprise"""
+    """Use apprise. Apprise has no HTML support, so always use the plain style."""
-    body = f"   Total: {self.total_torrents}\n\
+    opts = get_notify_options(self)
-    Premature: {self.preme_tor_counter}\n\
+    title, body = build_summary(
-    Ignored: {self.ignored_counter}\n\
+        self,
-    Protected: {self.c[self.tracker_protected_tag]}\n\
+        style="plain",
-    Non-protected: {self.c[self.tracker_non_protected_tag]}\n\
+        use_emoji=opts["use_emoji"],
-    Orphaned: {self.up_tor_counter}\n\
+        include_hostname=opts["include_hostname"],
-    Marked for deletion: {len(self.torrent_hash_delete_list)}\n\
+    )
-    {self.extm}"
-    title = "--- qbit-maid summary ---"
     app_obj(req_obj, self.apprise_host, self.apprise_port, self.apprise_aurls, title, body)
 
 def tornotifytest(self):
@@ -41,15 +40,42 @@ def print_processor(self):
     self.tl.info(f'Marked for deletion: {len(self.torrent_hash_delete_list)}')
 
 def tor_notify_summary(self):
-    """Main notification method when the app is used in an automated fashion"""
+    """Main notification method when the app is used in an automated fashion.
-    self.poc.message(self.po_key, f"   Total: {self.total_torrents}\n\
+
-    Premature: {self.preme_tor_counter}\n\
+    Formatting is controlled by the ``[pushover]`` config section: ``format``
-    Ignored: {self.ignored_counter}\n\
+    (plain|html), ``use_emoji``, ``include_hostname``, ``notify_on_empty``,
-    Protected: {self.c[self.tracker_protected_tag]}\n\
+    ``priority``, ``sound``, ``retry`` and ``expire``.
-    Non-protected: {self.c[self.tracker_non_protected_tag]}\n\
+    """
-    Orphaned: {self.up_tor_counter}\n\
+    logger = _logger(self)
-    Marked for deletion: {len(self.torrent_hash_delete_list)}\n\
+    opts = get_notify_options(self)
-    {self.extm}", title="--- qbit-maid summary ---")
+    title, body = build_summary(
+        self,
+        style=opts["style"],
+        use_emoji=opts["use_emoji"],
+        include_hostname=opts["include_hostname"],
+    )
+
+    if not opts["notify_on_empty"]:
+        stats = summary_stats(self)
+        if not stats["removed"] and not stats["orphaned"]:
+            logger.info("Nothing removed or orphaned, skipping notification.")
+            return
+
+    kwargs = {}
+    if opts["style"] == "html":
+        kwargs["html"] = 1
+    if opts["priority"] is not None:
+        kwargs["priority"] = opts["priority"]
+        if opts["priority"] == 2:
+            kwargs["retry"] = opts["retry"]
+            kwargs["expire"] = opts["expire"]
+    if opts["sound"]:
+        kwargs["sound"] = opts["sound"]
+
+    try:
+        self.poc.message(self.po_key, body, title=title, **kwargs)
+    except Exception as e:  # never let a notification failure abort a run
+        logger.error(f"Failed to send pushover notification: {e}")
 
 def list_first_tor(self, index=0):
     """Only lists the first torrent"""
@@ -77,6 +103,104 @@ def get_script_runtime(self):
     if self.use_apprise:
         self.extm = f"Execution time: [{elapsed_time}]"
 
+# ---------------------------------------------------------------------------
+# Notification formatting
+# ---------------------------------------------------------------------------
+NOTIFY_STYLES = ("plain", "html")
+DEFAULT_NOTIFY_STYLE = "plain"
+LABEL_WIDTH = 14
+VALUE_WIDTH = 4
+
+NOTIFY_EMOJI = {
+    "removed": "\U0001F5D1\uFE0F",
+    "orphaned": "\U0001F47B",
+    "ignored": "\u23ED\uFE0F",
+    "premature": "\U0001F423",
+    "protected": "\U0001F6E1\uFE0F",
+    "non_protected": "\U0001F310",
+}
+
+
+def _logger(self):
+    """Return the configured logger, or a bare one for ad-hoc/test use."""
+    return getattr(self, "tl", logging.getLogger(LOGGER_NAME))
+
+
+def format_runtime(value):
+    """Format a timedelta as ``12s``, ``1m 04s`` or ``1h 02m 03s``."""
+    try:
+        total = int(value.total_seconds())
+    except AttributeError:
+        return str(value)
+    if total < 0:
+        total = 0
+    hours, remainder = divmod(total, 3600)
+    minutes, seconds = divmod(remainder, 60)
+    if hours:
+        return f"{hours}h {minutes:02d}m {seconds:02d}s"
+    if minutes:
+        return f"{minutes}m {seconds:02d}s"
+    return f"{seconds}s"
+
+
+def get_notify_options(self):
+    """Read notification options from ``[pushover]``, falling back to defaults."""
+    config = getattr(self, "config", None) or {}
+    po = config.get("pushover", None) or {}
+    style = str(po.get("format", DEFAULT_NOTIFY_STYLE) or "").lower()
+    if style not in NOTIFY_STYLES:
+        style = DEFAULT_NOTIFY_STYLE
+    priority = po.get("priority")
+    return {
+        "style": style,
+        "use_emoji": bool(po.get("use_emoji", False)),
+        "include_hostname": bool(po.get("include_hostname", False)),
+        "notify_on_empty": bool(po.get("notify_on_empty", True)),
+        "priority": None if priority is None else int(priority),
+        "sound": po.get("sound"),
+        "retry": int(po.get("retry", 60)),
+        "expire": int(po.get("expire", 3600)),
+    }
+
+
+def summary_runtime(self):
+    """Runtime string for the summary, with a fallback to ``self.extm``."""
+    start = getattr(self, "st", None)
+    end = getattr(self, "et", None)
+    if start is not None and end is not None:
+        return format_runtime(end - start)
+    extm = getattr(self, "extm", "") or ""
+    if "[" in extm and "]" in extm:
+        return extm[extm.index("[") + 1:extm.rindex("]")]
+    return "unknown"
+
+
+def summary_stats(self):
+    """Collect the counters shown in a notification summary."""
+    counter = getattr(self, "c", None)
+    protected_tag = getattr(self, "tracker_protected_tag", "") or ""
+    non_protected_tag = getattr(self, "tracker_non_protected_tag", "") or ""
+
+    def tag_count(tag):
+        if counter is None or not tag:
+            return 0
+        try:
+            return counter[tag]
+        except (KeyError, TypeError):
+            return 0
+
+    return {
+        "total": getattr(self, "total_torrents", 0),
+        "removed": len(getattr(self, "torrent_hash_delete_list", None) or []),
+        "orphaned": getattr(self, "up_tor_counter", 0),
+        "ignored": getattr(self, "ignored_counter", 0),
+        "premature": getattr(self, "preme_tor_counter", 0),
+        "protected": tag_count(protected_tag),
+        "non_protected": tag_count(non_protected_tag),
+        "runtime": summary_runtime(self),
+    }
+
+
 def send_ping(self, req_obj, healthcheck_url):
     try:
         req_obj.get(healthcheck_url, timeout=5)
@@ -87,6 +211,82 @@ def debug_torrent_list(self):
     self.tl.debug(self.torrent_list)
 
 
+def _notify_row(label, value, style, use_emoji, key):
+    """Render one label/value line for the chosen style."""
+    emoji = NOTIFY_EMOJI.get(key, "") if use_emoji else ""
+    if use_emoji:
+        prefix = f"{emoji} " if emoji else ""
+        if style == "html":
+            return f"{prefix}<b>{label}</b>: {value}"
+        return f"{prefix}{label}: {value}"
+    if style == "html":
+        pad = " " * max(LABEL_WIDTH - len(label), 1)
+        return f"<b>{label}</b>{pad}{value:>{VALUE_WIDTH}}"
+    return f"{label:<{LABEL_WIDTH}}{value:>{VALUE_WIDTH}}"
+
+
+def build_summary(self, style=DEFAULT_NOTIFY_STYLE, use_emoji=False,
+                  include_hostname=False):
+    """Build the ``(title, body)`` pair used by every notifier.
+
+    Result-first layout: the title states what happened, the body gives the
+    scanned total, the actionable counters, then the tracker split.
+
+    Args:
+        style: ``"plain"`` for aligned text, ``"html"`` for Pushover HTML.
+        use_emoji: Prefix each row with an emoji marker.
+        include_hostname: Add a ``Host:`` line above the summary.
+
+    Returns:
+        A ``(title, body)`` tuple of strings.
+    """
+    if style not in NOTIFY_STYLES:
+        style = DEFAULT_NOTIFY_STYLE
+
+    stats = summary_stats(self)
+    delete_enabled = bool(getattr(self, "delete_torrents", False))
+
+    # Title: say what happened. Pushover does not render HTML in titles.
+    removed = stats["removed"]
+    noun = "torrent" if removed == 1 else "torrents"
+    if removed:
+        action = "removed" if delete_enabled else "would be removed"
+        title = f"qbit-maid: {removed} {noun} {action}"
+    elif delete_enabled:
+        title = "qbit-maid: nothing removed"
+    else:
+        title = "qbit-maid: nothing to remove"
+
+    lines = []
+    if include_hostname:
+        host = socket.gethostname()
+        if style == "html":
+            lines.append(f"<b>Host:</b> {host}")
+        else:
+            lines.append(f"Host: {host}")
+        lines.append("")
+
+    total = stats["total"]
+    scanned_noun = "torrent" if total == 1 else "torrents"
+    if style == "html":
+        lines.append(f"Scanned <b>{total}</b> {scanned_noun} in <b>{stats['runtime']}</b>")
+    else:
+        lines.append(f"Scanned {total} {scanned_noun} in {stats['runtime']}")
+    lines.append("")
+
+    removed_label = "Removed" if delete_enabled else "Would remove"
+    lines.append(_notify_row(removed_label, removed, style, use_emoji, "removed"))
+    lines.append(_notify_row("Orphaned", stats["orphaned"], style, use_emoji, "orphaned"))
+    lines.append(_notify_row("Ignored", stats["ignored"], style, use_emoji, "ignored"))
+    lines.append(_notify_row("Premature", stats["premature"], style, use_emoji, "premature"))
+    lines.append("")
+    lines.append(_notify_row("Protected", stats["protected"], style, use_emoji, "protected"))
+    lines.append(_notify_row("Non-protected", stats["non_protected"], style,
+                             use_emoji, "non_protected"))
+
+    return title, "\n".join(lines)
+
+
 # ---------------------------------------------------------------------------
 # New logging infrastructure (Phase 1 & 3)
 # ---------------------------------------------------------------------------

test_qbitmaid.py

@@ -1,8 +1,61 @@
+import datetime
+import logging
 import unittest
+from collections import Counter
+
 import requests as r
 from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag
 from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor
-from qlogging import send_ping
+from qlogging import (
+    send_ping,
+    build_summary,
+    format_runtime,
+    get_notify_options,
+    summary_stats,
+    tor_notify_apprise,
+    tor_notify_summary,
+)
+
+
+class FakeQbt:
+    """Minimal stand-in for the Qbt object, holding only what notifiers read."""
+
+    def __init__(self, **overrides):
+        self.config = {"pushover": {}}
+        self.total_torrents = 42
+        self.torrent_hash_delete_list = ["a"] * 7
+        self.up_tor_counter = 4
+        self.ignored_counter = 5
+        self.preme_tor_counter = 3
+        self.tracker_protected_tag = "ipt"
+        self.tracker_non_protected_tag = "public"
+        self.c = Counter({"ipt": 10, "public": 20})
+        self.delete_torrents = True
+        self.st = datetime.datetime(2026, 10, 27, 9, 54, 0)
+        self.et = datetime.datetime(2026, 10, 27, 9, 54, 12)
+        self.po_key = "USERKEY"
+        self.poc = StubPushover()
+        self.apprise_host = "localhost"
+        self.apprise_port = 8088
+        self.apprise_aurls = "mailto://user:pass@example.com"
+        self.tl = logging.getLogger("qbit-maid-test")
+        for key, value in overrides.items():
+            setattr(self, key, value)
+
+
+class StubPushover:
+    """Records message() calls instead of hitting the Pushover API."""
+
+    def __init__(self):
+        self.calls = []
+
+    def message(self, user, body, **kwargs):
+        self.calls.append({"user": user, "body": body, "kwargs": kwargs})
+
+
+class FailingPushover:
+    def message(self, *args, **kwargs):
+        raise RuntimeError("pushover is down")
 
 class TestQbitmaid(unittest.TestCase):
     def test_ispreme_sanity(self):
@@ -107,6 +160,204 @@ class TestQbitmaid(unittest.TestCase):
         send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb")
         
 
+    # --- notification formatting -------------------------------------------
+
+    def test_format_runtime(self):
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s")
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s")
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=64)), "1m 04s")
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s")
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s")
+
+    def test_summary_stats(self):
+        stats = summary_stats(FakeQbt())
+        self.assertEqual(stats["total"], 42)
+        self.assertEqual(stats["removed"], 7)
+        self.assertEqual(stats["orphaned"], 4)
+        self.assertEqual(stats["ignored"], 5)
+        self.assertEqual(stats["premature"], 3)
+        self.assertEqual(stats["protected"], 10)
+        self.assertEqual(stats["non_protected"], 20)
+        self.assertEqual(stats["runtime"], "12s")
+
+    def test_summary_stats_missing_attributes(self):
+        class Bare:
+            pass
+
+        stats = summary_stats(Bare())
+        self.assertEqual(stats["total"], 0)
+        self.assertEqual(stats["removed"], 0)
+        self.assertEqual(stats["runtime"], "unknown")
+
+    def test_build_summary_plain(self):
+        title, body = build_summary(FakeQbt(), style="plain")
+        self.assertEqual(title, "qbit-maid: 7 torrents removed")
+        self.assertEqual(
+            body,
+            "Scanned 42 torrents in 12s\n"
+            "\n"
+            "Removed          7\n"
+            "Orphaned         4\n"
+            "Ignored          5\n"
+            "Premature        3\n"
+            "\n"
+            "Protected       10\n"
+            "Non-protected   20",
+        )
+
+    def test_build_summary_plain_has_no_leading_indent(self):
+        _, body = build_summary(FakeQbt(), style="plain")
+        for line in body.splitlines():
+            self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}")
+
+    def test_build_summary_html(self):
+        title, body = build_summary(FakeQbt(), style="html")
+        self.assertEqual(title, "qbit-maid: 7 torrents removed")
+        self.assertIn("Scanned <b>42</b> torrents in <b>12s</b>", body)
+        self.assertIn("<b>Removed</b>", body)
+        self.assertIn("<b>Non-protected</b>", body)
+        self.assertNotIn("---", title)
+
+    def test_build_summary_emoji(self):
+        _, plain = build_summary(FakeQbt(), style="plain", use_emoji=True)
+        self.assertIn("\U0001F5D1\uFE0F Removed: 7", plain)
+        self.assertIn("\U0001F47B Orphaned: 4", plain)
+        _, html = build_summary(FakeQbt(), style="html", use_emoji=True)
+        self.assertIn("\U0001F5D1\uFE0F <b>Removed</b>: 7", html)
+
+    def test_build_summary_delete_disabled(self):
+        title, body = build_summary(FakeQbt(delete_torrents=False))
+        self.assertEqual(title, "qbit-maid: 7 torrents would be removed")
+        self.assertIn("Would remove", body)
+        self.assertNotIn("Removed", body)
+
+    def test_build_summary_nothing_to_do(self):
+        title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0))
+        self.assertEqual(title, "qbit-maid: nothing removed")
+        title, _ = build_summary(
+            FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False)
+        )
+        self.assertEqual(title, "qbit-maid: nothing to remove")
+
+    def test_build_summary_singular(self):
+        title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1))
+        self.assertEqual(title, "qbit-maid: 1 torrent removed")
+
+    def test_build_summary_hostname(self):
+        _, body = build_summary(FakeQbt(), include_hostname=True)
+        self.assertTrue(body.startswith("Host: "))
+
+    def test_build_summary_unknown_style_falls_back(self):
+        _, body = build_summary(FakeQbt(), style="markdown")
+        self.assertNotIn("<b>", body)
+
+    def test_get_notify_options_defaults(self):
+        opts = get_notify_options(FakeQbt())
+        self.assertEqual(opts["style"], "plain")
+        self.assertFalse(opts["use_emoji"])
+        self.assertFalse(opts["include_hostname"])
+        self.assertTrue(opts["notify_on_empty"])
+        self.assertIsNone(opts["priority"])
+        self.assertIsNone(opts["sound"])
+
+    def test_get_notify_options_overrides(self):
+        fake = FakeQbt(config={"pushover": {
+            "format": "HTML",
+            "use_emoji": True,
+            "include_hostname": True,
+            "notify_on_empty": False,
+            "priority": 2,
+            "sound": "siren",
+            "retry": 30,
+            "expire": 600,
+        }})
+        opts = get_notify_options(fake)
+        self.assertEqual(opts["style"], "html")
+        self.assertTrue(opts["use_emoji"])
+        self.assertTrue(opts["include_hostname"])
+        self.assertFalse(opts["notify_on_empty"])
+        self.assertEqual(opts["priority"], 2)
+        self.assertEqual(opts["sound"], "siren")
+        self.assertEqual(opts["retry"], 30)
+        self.assertEqual(opts["expire"], 600)
+
+    def test_get_notify_options_invalid_style(self):
+        fake = FakeQbt(config={"pushover": {"format": "bogus"}})
+        self.assertEqual(get_notify_options(fake)["style"], "plain")
+
+    def test_tor_notify_summary_plain_payload(self):
+        fake = FakeQbt()
+        tor_notify_summary(fake)
+        self.assertEqual(len(fake.poc.calls), 1)
+        call = fake.poc.calls[0]
+        self.assertEqual(call["user"], "USERKEY")
+        self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed")
+        self.assertNotIn("html", call["kwargs"])
+
+    def test_tor_notify_summary_html_payload(self):
+        fake = FakeQbt(config={"pushover": {"format": "html"}})
+        tor_notify_summary(fake)
+        call = fake.poc.calls[0]
+        self.assertEqual(call["kwargs"]["html"], 1)
+        self.assertIn("<b>", call["body"])
+
+    def test_tor_notify_summary_priority_two_adds_retry_expire(self):
+        fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}})
+        tor_notify_summary(fake)
+        kwargs = fake.poc.calls[0]["kwargs"]
+        self.assertEqual(kwargs["priority"], 2)
+        self.assertEqual(kwargs["retry"], 30)
+        self.assertEqual(kwargs["expire"], 600)
+
+    def test_tor_notify_summary_priority_one_has_no_retry(self):
+        fake = FakeQbt(config={"pushover": {"priority": 1}})
+        tor_notify_summary(fake)
+        kwargs = fake.poc.calls[0]["kwargs"]
+        self.assertEqual(kwargs["priority"], 1)
+        self.assertNotIn("retry", kwargs)
+
+    def test_tor_notify_summary_skips_when_empty(self):
+        fake = FakeQbt(
+            config={"pushover": {"notify_on_empty": False}},
+            torrent_hash_delete_list=[],
+            up_tor_counter=0,
+        )
+        tor_notify_summary(fake)
+        self.assertEqual(fake.poc.calls, [])
+
+    def test_tor_notify_summary_sends_when_empty_by_default(self):
+        fake = FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)
+        tor_notify_summary(fake)
+        self.assertEqual(len(fake.poc.calls), 1)
+
+    def test_tor_notify_summary_swallows_send_errors(self):
+        fake = FakeQbt()
+        fake.poc = FailingPushover()
+        tor_notify_summary(fake)  # must not raise
+
+    def test_tor_notify_apprise_uses_plain_body(self):
+        captured = {}
+
+        def app_obj(req_obj, host, port, aurls, title, body):
+            captured.update(host=host, port=port, aurls=aurls, title=title, body=body)
+
+        fake = FakeQbt(config={"pushover": {"format": "html"}})
+        tor_notify_apprise(fake, None, app_obj)
+        self.assertEqual(captured["title"], "qbit-maid: 7 torrents removed")
+        self.assertNotIn("<b>", captured["body"])
+        self.assertEqual(captured["port"], 8088)
+
+    def test_pushover_and_apprise_bodies_match(self):
+        fake = FakeQbt()
+        tor_notify_summary(fake)
+        captured = {}
+        tor_notify_apprise(
+            fake, None,
+            lambda req, host, port, aurls, title, body: captured.update(title=title, body=body),
+        )
+        self.assertEqual(fake.poc.calls[0]["body"], captured["body"])
+        self.assertEqual(fake.poc.calls[0]["kwargs"]["title"], captured["title"])
+
     # def test__sanity(self):
     #     pass