fix: resolve the seven bugs pinned by the test audit (#67)

- read [ignored_tags] instead of [ignored_domains], so tag-based ignoring
  actually works
- match ignored tags as whole tags, not substrings ('ipt' no longer matches
  'iptorrents')
- send the login-error notification through the Pushover client instead of
  the non-existent module-level send_message()
- handle an empty torrent list instead of raising IndexError before any
  summary is sent
- tag a new torrent once instead of once per code path
- honour toml_path without DOCKER in AppriseClient
- close replaced log handlers so the file descriptor is released

Also add the missing [telemetry] section to config.toml.example, which
qbit-maid.py requires at startup.

The tests that pinned each bug now assert the fixed behaviour, and the
suite still reports 100% coverage.

.coveragerc

@@ -0,0 +1,18 @@
+[run]
+source = .
+branch = False
+omit =
+    pushover.py
+    test_*.py
+    qbt_fakes.py
+    */venv/*
+    /tmp/*
+
+[report]
+# Entry-point guards are not reachable from a test process; they are covered
+# by the container smoke test instead.
+exclude_lines =
+    pragma: no cover
+    if __name__ == .__main__.:
+show_missing = True
+skip_covered = False

.gitea/workflows/test.yml

@@ -8,6 +8,26 @@ on:
   pull_request:
 
 jobs:
+  unit-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Install dependencies
+        # Mirrors the Dockerfile: the container installs qbittorrent-api
+        # unpinned, so requirements.txt is not used here.
+        # --break-system-packages: the runner image ships a Debian-packaged
+        # pip, so a plain install (or `pip install --upgrade pip`) fails
+        # with "RECORD file not found".
+        run: |
+          python3 -m pip install --break-system-packages requests qbittorrent-api coverage
+      - name: Run the full suite with coverage
+        run: |
+          python3 -m coverage run --rcfile=.coveragerc -m unittest \
+            test_qbitmaid test_write_csv test_qlist test_qprocess \
+            test_qlogging_extra test_apprise test_qbt
+          python3 -m coverage report -m --fail-under=100
+
   komodo-deploy-script:
     runs-on: ubuntu-latest
     steps:

AppriseClient.py

@@ -11,19 +11,16 @@ def apprise_notify(req_obj, host, port, aurls, title, body):
 class AppriseClient:
     def __init__(self):
         self.config = ''
-        try:
+        if os.environ.get("DOCKER"):
-            if os.environ["DOCKER"]:
+            self.host = os.environ["host"]
-                self.host = os.environ["host"]
+            self.port = os.environ["port"]
-                self.port = os.environ["port"]
+            self.aurls = os.environ["aurls"]
-                self.aurls = os.environ["aurls"]
+            self.title = os.environ["title"]
-                self.title = os.environ["title"]
+            self.body = os.environ["body"]
-                self.body = os.environ["body"]
+        if os.environ.get("toml_path"):
-            if os.environ["toml_path"]:
+            config_file_path=os.environ["toml_path"]
-                config_file_path=os.environ["toml_path"]
+            with open(config_file_path, 'rb') as c:
-                with open(config_file_path, 'rb') as c:
+                self.config = load(c)
-                    self.config = load(c)
-        except:
-            KeyError
         if os.path.exists('./config.toml'):
             config_file_path = './config.toml'
             with open(config_file_path, 'rb') as c:
@@ -37,4 +34,4 @@ class AppriseClient:
         self.apprise_response = apprise_notify(r,self.host,self.port,self.aurls,self.title,self.body)  
 
 if __name__ == "__main__":
-    AppriseClient()
+    AppriseClient()

README.md

@@ -137,35 +137,49 @@ 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:
+The default summary looks like this:
 
 ```
 qbit-maid: 7 torrents removed
 
 Scanned 42 torrents in 12s
 
-Removed          7
+🗑️ 7 removed  ·  👻 4 orphaned
-Orphaned         4
+⏭️ 5 ignored  ·  🐣 3 premature
-Ignored          5
-Premature        3
 
-Protected       10
+🛡️ 10 protected  ·  🌐 20 public
-Non-protected   20
 ```
 
-`format = "html"` bolds the labels and counts using Pushover's HTML subset
+Each counter is `emoji value label`, grouped two per line and joined with a
-(`html=1` is set automatically). Apprise has no HTML support, so it always
+middle dot. The value leads so the numbers are the first thing the eye lands
-receives the plain body.
+on, and the separator survives any font — Pushover renders the body in a
+proportional font, so space-padded columns collapse into a ragged list on a
+phone.
 
-Setting `use_emoji = true` switches both styles to `emoji label: value` rows:
+`format = "html"` bolds the values using Pushover's HTML subset (`html=1` is
+set automatically). Apprise has no HTML support, so it always receives the
+plain body.
+
+`use_emoji` defaults to `true`. Set it to `false` for a plain-text variant:
 
 ```
-🗑️ Removed: 7
+Scanned 42 torrents in 12s
-👻 Orphaned: 4
+
-⏭️ Ignored: 5
+7 removed  ·  4 orphaned
-🐣 Premature: 3
+5 ignored  ·  3 premature
+
+10 protected  ·  20 public
 ```
 
+To preview the result without sending a push, render the real
+`build_summary()` output as a phone-width mockup:
+
+```
+python3 docs/notify-preview.py /tmp/notify.png
+```
+
+Pass `--mock` to render the design sketches instead of the live output.
+
 Other options:
 
 - `include_hostname` — add a `Host:` line, useful when several instances notify

config.toml.example

@@ -26,9 +26,12 @@ use_pushover = false
 po_key = "<key>"
 po_token = "<token>"
 # Notification formatting
-# format: "plain" (aligned text) or "html" (bold labels, Pushover HTML)
+# format: "plain" (grouped text) or "html" (bold values, Pushover HTML)
+# Counters render as "emoji value label", grouped two per line and joined
+# with a middle dot, because Pushover uses a proportional font and padded
+# columns do not align.
 format = "plain"
-use_emoji = false
+use_emoji = true
 include_hostname = false
 notify_on_empty = true
 # Optional: -2..2. Priority 2 also needs retry/expire (defaults shown).
@@ -47,6 +50,10 @@ aurls = 'mailto://user:pass@gmail.com'
 enable_dragnet = false
 dragnet_outfile = "./orphaned.csv"
 
+[telemetry]
+enable_telemetry = false
+telemetry_outfile = "./telemetry.csv"
+
 [ignored_categories]
 tech = "tech"
 books = "books"

docs/notify-preview.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""Render notification layouts as a phone-width PNG mockup.
+
+Pushover renders the body in a proportional font, so space-padded columns do
+not line up on a phone. These mockups use a proportional font on purpose so
+the preview matches what the user actually sees.
+
+By default the cards are rendered from the real ``build_summary()`` output, so
+the preview is ground truth rather than a hand-written approximation. Pass
+``--mock`` to render the built-in candidate sketches instead.
+
+Usage: python3 docs/notify-preview.py [--mock] [out.png]
+"""
+import datetime
+import html
+import os
+import subprocess
+import sys
+import tempfile
+from collections import Counter
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+STATS = {
+    "total": 42,
+    "removed": 7,
+    "orphaned": 4,
+    "ignored": 5,
+    "premature": 3,
+    "protected": 10,
+    "non_protected": 20,
+    "runtime": "12s",
+}
+
+EMOJI = {
+    "removed": "\U0001F5D1\uFE0F",
+    "orphaned": "\U0001F47B",
+    "ignored": "\u23ED\uFE0F",
+    "premature": "\U0001F423",
+    "protected": "\U0001F6E1\uFE0F",
+    "non_protected": "\U0001F310",
+}
+
+
+def b(text):
+    return f"<b>{text}</b>"
+
+
+# --- candidate layouts -----------------------------------------------------
+# Each returns (title, body_html). Body uses Pushover's HTML subset.
+
+def current_table(s):
+    """Today's layout: space-padded columns."""
+    rows = [
+        ("Removed", s["removed"]),
+        ("Orphaned", s["orphaned"]),
+        ("Ignored", s["ignored"]),
+        ("Premature", s["premature"]),
+    ]
+    tail = [("Protected", s["protected"]), ("Non-protected", s["non_protected"])]
+    out = [f"Scanned {s['total']} torrents in {s['runtime']}", ""]
+    out += [f"{b(lbl)}{' ' * max(14 - len(lbl), 1)}{val:>4}" for lbl, val in rows]
+    out.append("")
+    out += [f"{b(lbl)}{' ' * max(14 - len(lbl), 1)}{val:>4}" for lbl, val in tail]
+    return "qbit-maid: 7 torrents removed", "\n".join(out)
+
+
+def compact_inline(s):
+    """One line per group, middle-dot separated."""
+    out = [
+        f"Scanned {s['total']} torrents in {s['runtime']}",
+        "",
+        f"{b('Removed')} {s['removed']}  \u00b7  {b('Orphaned')} {s['orphaned']}",
+        f"{b('Ignored')} {s['ignored']}  \u00b7  {b('Premature')} {s['premature']}",
+        "",
+        f"{b('Protected')} {s['protected']}  \u00b7  {b('Non-protected')} {s['non_protected']}",
+    ]
+    return "qbit-maid: 7 torrents removed", "\n".join(out)
+
+
+def emoji_rows(s):
+    """Emoji-led rows, no padding."""
+    out = [f"Scanned {s['total']} torrents in {s['runtime']}", ""]
+    for key, label in (("removed", "Removed"), ("orphaned", "Orphaned"),
+                       ("ignored", "Ignored"), ("premature", "Premature")):
+        out.append(f"{EMOJI[key]} {b(label)} {s[key]}")
+    out.append("")
+    for key, label in (("protected", "Protected"), ("non_protected", "Non-protected")):
+        out.append(f"{EMOJI[key]} {b(label)} {s[key]}")
+    return "qbit-maid: 7 torrents removed", "\n".join(out)
+
+
+def hero_line(s):
+    """Actionable numbers up front, everything else compact below."""
+    out = [
+        f"{b(str(s['removed']))} removed  \u00b7  {b(str(s['orphaned']))} orphaned",
+        f"{s['total']} torrents scanned in {s['runtime']}",
+        "",
+        f"Ignored {s['ignored']}  \u00b7  Premature {s['premature']}",
+        f"Protected {s['protected']}  \u00b7  Non-protected {s['non_protected']}",
+    ]
+    return "qbit-maid: 7 torrents removed", "\n".join(out)
+
+
+def emoji_inline(s):
+    """Compact inline, emoji instead of bold labels."""
+    out = [
+        f"Scanned {s['total']} torrents in {s['runtime']}",
+        "",
+        f"{EMOJI['removed']} {s['removed']} removed  \u00b7  {EMOJI['orphaned']} {s['orphaned']} orphaned",
+        f"{EMOJI['ignored']} {s['ignored']} ignored  \u00b7  {EMOJI['premature']} {s['premature']} premature",
+        "",
+        f"{EMOJI['protected']} {s['protected']} protected  \u00b7  {EMOJI['non_protected']} {s['non_protected']} public",
+    ]
+    return "qbit-maid: 7 torrents removed", "\n".join(out)
+
+
+MOCK_CANDIDATES = [
+    ("A \u2014 current (padded table)", current_table),
+    ("B \u2014 compact inline", compact_inline),
+    ("C \u2014 emoji rows", emoji_rows),
+    ("D \u2014 hero line", hero_line),
+    ("E \u2014 emoji inline", emoji_inline),
+]
+
+
+class Sample:
+    """Minimal stand-in for the Qbt object, holding only what notifiers read."""
+
+    def __init__(self, **overrides):
+        self.config = {"pushover": {}}
+        self.total_torrents = STATS["total"]
+        self.torrent_hash_delete_list = ["a"] * STATS["removed"]
+        self.up_tor_counter = STATS["orphaned"]
+        self.ignored_counter = STATS["ignored"]
+        self.preme_tor_counter = STATS["premature"]
+        self.tracker_protected_tag = "ipt"
+        self.tracker_non_protected_tag = "public"
+        self.c = Counter({"ipt": STATS["protected"],
+                          "public": STATS["non_protected"]})
+        self.delete_torrents = True
+        self.st = datetime.datetime(2026, 10, 27, 9, 54, 0)
+        self.et = datetime.datetime(2026, 10, 27, 9, 54, 12)
+        for key, value in overrides.items():
+            setattr(self, key, value)
+
+
+def real_candidates():
+    """Render the actual build_summary() output for each supported style."""
+    from qlogging import build_summary
+
+    out = []
+    for style in ("plain", "html"):
+        for emoji in (False, True):
+            title, body = build_summary(
+                Sample(), style=style, use_emoji=emoji, include_hostname=False
+            )
+            label = f"{style}" + (" + emoji" if emoji else "")
+            out.append((label, title, body))
+    return out
+
+
+def card(name, title, body):
+    body_html = "<br>".join(body.split("\n"))
+    return f"""
+    <div class="card">
+      <div class="app">PUSHOVER \u00b7 {html.escape(name)}</div>
+      <div class="title">{html.escape(title)}</div>
+      <div class="body">{body_html}</div>
+    </div>
+    """
+
+
+def build_html(mock=False):
+    if mock:
+        cards = "\n".join(card(n, *fn(STATS)) for n, fn in MOCK_CANDIDATES)
+    else:
+        cards = "\n".join(card(n, t, b) for n, t, b in real_candidates())
+    return f"""<!doctype html>
+<html><head><meta charset="utf-8"><style>
+  body {{ margin:0; padding:18px; background:#1b1d21;
+          font-family:-apple-system,"Segoe UI",Roboto,Helvetica,Arial,sans-serif; }}
+  .grid {{ display:grid; grid-template-columns:1fr 1fr; gap:16px; }}
+  .card {{ background:#2b2f36; border-radius:14px; padding:14px 16px;
+           box-shadow:0 2px 10px rgba(0,0,0,.45); }}
+  .app {{ font-size:10px; letter-spacing:.14em; color:#8b929c; margin-bottom:6px; }}
+  .title {{ font-size:15px; font-weight:600; color:#f2f4f7; margin-bottom:8px; }}
+  .body {{ font-size:14px; line-height:1.45; color:#d7dbe0; white-space:normal; }}
+  .body b {{ color:#ffffff; }}
+</style></head><body><div class="grid">{cards}</div></body></html>
+"""
+
+
+def main():
+    args = [a for a in sys.argv[1:] if not a.startswith("--")]
+    mock = "--mock" in sys.argv[1:]
+    out = args[0] if args else "/tmp/notify-preview.png"
+    workdir = tempfile.mkdtemp(prefix="notify-preview-")
+    page = os.path.join(workdir, "preview.html")
+    with open(page, "w", encoding="utf-8") as fh:
+        fh.write(build_html(mock=mock))
+    subprocess.run([
+        "docker", "run", "--rm", "-v", f"{workdir}:/w", "-w", "/w",
+        "--entrypoint", "google-chrome", "dgtlmoon/sockpuppetbrowser:latest",
+        "--headless", "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage",
+        "--hide-scrollbars", "--window-size=880,760",
+        "--screenshot=/w/preview.png", "file:///w/preview.html",
+    ], check=True, capture_output=True)
+    subprocess.run(["cp", os.path.join(workdir, "preview.png"), out], check=True)
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

qbit-maid.py

@@ -89,7 +89,7 @@ class Qbt:
         self.tracker_whitelist = self.config["ignored_domains"]
 
         #ignored_tags
-        self.ignored_tags = self.config["ignored_domains"]
+        self.ignored_tags = self.config["ignored_tags"]
 
         #healthcheck
         self.use_healthcheck = self.config["healthcheck"]["use_healthcheck"]
@@ -117,7 +117,7 @@ class Qbt:
             self.tl.info('Connected.')
         except qbittorrentapi.APIError as e:
             self.tl.exception(e)
-            self.po.send_message(e, title="qbit-maid API ERROR")
+            tor_notify_error(self, e)
         # Pulling all torrent data
         self.torrent_list = self.qbt_client.torrents_info()
         #Main process block

qbt_fakes.py

@@ -0,0 +1,158 @@
+"""Shared fakes for the qbit-maid test suite.
+
+This is test scaffolding, not production code, so it is excluded from the
+coverage report (see ``.coveragerc``).
+"""
+import datetime
+import logging
+from collections import Counter
+
+# Keep test output clean: the pipeline logs warnings for ignored torrents.
+_test_logger = logging.getLogger("qbit-maid-test")
+_test_logger.addHandler(logging.NullHandler())
+_test_logger.propagate = False
+
+
+# ---------------------------------------------------------------------------
+# qbittorrentapi stand-in
+# ---------------------------------------------------------------------------
+class FakeApp:
+    version = "v4.6.2"
+    web_api_version = "2.9.3"
+
+
+class FakeAPIError(Exception):
+    """Stands in for qbittorrentapi.APIError."""
+
+
+class FakeQbtClient:
+    """Records calls instead of talking to a real qBittorrent instance."""
+
+    def __init__(self, torrents=None, login_error=None, **kwargs):
+        self.app = FakeApp()
+        self._torrents = list(torrents or [])
+        self._login_error = login_error
+        self.init_kwargs = kwargs
+        self.added_tags = []
+        self.deleted_tags = []
+        self.deleted = []
+        self.logged_in = False
+
+    def auth_log_in(self):
+        if self._login_error is not None:
+            raise self._login_error
+        self.logged_in = True
+
+    def torrents_info(self):
+        return list(self._torrents)
+
+    def torrents_add_tags(self, tags, torrent_hash):
+        self.added_tags.append((tags, torrent_hash))
+
+    def torrents_delete_tags(self, tags):
+        self.deleted_tags.append(list(tags))
+
+    def torrents_delete(self, delete_files, hashes):
+        self.deleted.append((delete_files, list(hashes)))
+
+
+def torrent(name="Example.Torrent.Release", tracker="https://tracker.example/announce",
+            category="general", tags="", seeding_time=999999, state="uploading",
+            ratio=2.0, time_active=999999, infohash="abc123", added_on=1):
+    """Build a torrent dict shaped like a qbittorrentapi response."""
+    return {
+        "name": name,
+        "tracker": tracker,
+        "category": category,
+        "tags": tags,
+        "seeding_time": seeding_time,
+        "state": state,
+        "ratio": ratio,
+        "time_active": time_active,
+        "infohash_v1": infohash,
+        "hash": infohash,
+        "added_on": added_on,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Pushover stand-ins
+# ---------------------------------------------------------------------------
+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")
+
+
+# ---------------------------------------------------------------------------
+# The Qbt object
+# ---------------------------------------------------------------------------
+class FakeQbt:
+    """Stand-in for the Qbt object.
+
+    Defaults match the notifier tests (42 scanned, 7 removed, 4 orphaned).
+    Pipeline tests override ``torrent_list`` and ``qbt_client``.
+    """
+
+    def __init__(self, **overrides):
+        # --- notifier surface ---
+        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")
+
+        # --- pipeline surface ---
+        self.torrent_list = []
+        self.tracker_list = []
+        self.qbt_client = FakeQbtClient()
+        self.cat_whitelist = {}
+        self.tracker_whitelist = {"tracker.example": "tracker.example"}
+        self.ignored_tags = {}
+        self.min_age = 432000
+        self.max_age = 2419200
+        self.enable_telemetry = False
+        self.telemetry_outfile = "telemetry.csv"
+        self.enable_dragnet = False
+        self.dragnet_outfile = "orphaned.csv"
+        self.cv = None
+        self.use_pushover = False
+        self.use_apprise = False
+        self.extm = ""
+
+        for key, value in overrides.items():
+            setattr(self, key, value)
+
+    def reset_counters(self):
+        """Zero the pipeline counters.
+
+        The constructor defaults are the notifier fixture (42 scanned, 7
+        removed, ...), which pipeline tests must not inherit.
+        """
+        self.up_tor_counter = 0
+        self.preme_tor_counter = 0
+        self.ignored_counter = 0
+        self.torrent_hash_delete_list = []
+        return self

qlist.py

@@ -22,24 +22,22 @@ def build_tor_list(self):
                 self.ignored_counter += 1
                 continue
             if is_tag_blank(torrent['tags']):
-                self.tl.debug(f'Tagging new torrent: ["{torrent["name"][0:20]}..."] {torrent["tracker"]}hash: {torrent["hash"]}')
+                tag = torrent_tag_for(
-                if is_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
+                    torrent['tracker'],
-                    self.qbt_client.torrents_add_tags(self.tracker_protected_tag,torrent['hash'])
+                    self.tracker_protected_tag,
-                elif is_not_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
+                    self.tracker_non_protected_tag,
-                    self.qbt_client.torrents_add_tags(self.tracker_non_protected_tag,torrent['hash'])
+                    self.tracker_whitelist.values(),
+                )
+                if tag:
+                    self.tl.debug(f'Tagging new torrent: ["{torrent["name"][0:20]}..."] {torrent["tracker"]}hash: {torrent["hash"]}')
+                    self.qbt_client.torrents_add_tags(tag, torrent['hash'])
             if is_preme(torrent['seeding_time'], self.min_age):
                 self.preme_tor_counter += 1
                 self.tl.debug(f'Premature torrent: ["{torrent["name"][0:20]}..."] Seconds Seeded: [{torrent["seeding_time"]}] hash: {torrent["hash"]}')
                 continue
             elif is_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
-                if is_tag_blank(torrent['tags']):
-                    self.qbt_client.torrents_add_tags(self.tracker_protected_tag,torrent['hash'])
-                    self.tl.debug(f'Tagging Protected torrent: ["{torrent["name"][0:20]}..."] {torrent["tracker"]}hash: {torrent["hash"]}')
                 self.tracker_list.append(torrent)
             elif is_not_protected_tracker(torrent['tracker'], self.tracker_whitelist.values()):
-                if is_tag_blank(torrent['tags']):
-                    self.qbt_client.torrents_add_tags(self.tracker_non_protected_tag,torrent['hash'])
-                    self.tl.debug(f'Tagging Non-protected torrent: ["{torrent["name"][0:20]}..."] {torrent["tracker"]}hash: {torrent["hash"]}')
                 self.tracker_list.append(torrent)
 
 
@@ -67,11 +65,26 @@ def is_not_protected_tracker(tracker, trackerlist):
     if tracker.split('/')[2] not in trackerlist:
         return True
 
+def torrent_tag_for(tracker, protected_tag, non_protected_tag, trackerlist):
+    """Return the tag a torrent should carry, or None for an unknown tracker."""
+    if is_protected_tracker(tracker, trackerlist):
+        return protected_tag
+    if is_not_protected_tracker(tracker, trackerlist):
+        return non_protected_tag
+    return None
+
 def is_tag_blank(tag):
     if tag == '':
         return True
 
 def is_ignored_tag(igtags, tortags):
-    for igt in igtags:
+    """True when the torrent carries one of the configured ignored tags.
-        if igt in tortags:
+
-            return True
+    qBittorrent returns tags as a comma-separated string, so the comparison is
+    per tag: a configured ``ipt`` must not match a torrent tagged
+    ``iptorrents``.
+    """
+    if not tortags:
+        return False
+    torrent_tags = {tag.strip() for tag in str(tortags).split(',') if tag.strip()}
+    return any(igt in torrent_tags for igt in igtags)

qlogging.py

@@ -77,9 +77,30 @@ def tor_notify_summary(self):
     except Exception as e:  # never let a notification failure abort a run
         logger.error(f"Failed to send pushover notification: {e}")
 
+def tor_notify_error(self, error, title="qbit-maid API ERROR"):
+    """Report an error through pushover when it is configured.
+
+    The vendored ``pushover`` module only exposes the ``Pushover`` class, so the
+    message has to go through the client built by :func:`tor_notify`.
+    """
+    logger = _logger(self)
+    logger.error(f"{title}: {error}")
+    if not getattr(self, "use_pushover", False):
+        return
+    client = getattr(self, "poc", None)
+    if client is None:
+        return
+    try:
+        client.message(self.po_key, str(error), title=title)
+    except Exception as exc:  # never let a notification failure abort a run
+        logger.error(f"Failed to send pushover error notification: {exc}")
+
 def list_first_tor(self, index=0):
     """Only lists the first torrent"""
     self.tl.debug('First torrent in the list:')
+    if not self.torrent_list:
+        self.tl.debug('No torrents found.')
+        return
     torrent = self.torrent_list[index]
     for k,v in torrent.items():
          self.tl.debug(f'{k}:  {v}')
@@ -108,8 +129,10 @@ def get_script_runtime(self):
 # ---------------------------------------------------------------------------
 NOTIFY_STYLES = ("plain", "html")
 DEFAULT_NOTIFY_STYLE = "plain"
-LABEL_WIDTH = 14
+# Pushover renders the body in a proportional font, so space-padded columns
-VALUE_WIDTH = 4
+# collapse into a ragged list. Rows are therefore joined with a visible
+# separator instead of being aligned with spaces.
+ROW_SEPARATOR = "  \u00b7  "
 
 NOTIFY_EMOJI = {
     "removed": "\U0001F5D1\uFE0F",
@@ -153,7 +176,7 @@ def get_notify_options(self):
     priority = po.get("priority")
     return {
         "style": style,
-        "use_emoji": bool(po.get("use_emoji", False)),
+        "use_emoji": bool(po.get("use_emoji", True)),
         "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),
@@ -211,30 +234,45 @@ def debug_torrent_list(self):
     self.tl.debug(self.torrent_list)
 
 
-def _notify_row(label, value, style, use_emoji, key):
+def _stat_row(pairs, style, use_emoji):
-    """Render one label/value line for the chosen style."""
+    """Render one line of ``emoji value label`` pairs joined by a middle dot.
-    emoji = NOTIFY_EMOJI.get(key, "") if use_emoji else ""
+
-    if use_emoji:
+    The value leads and the label trails, so the numbers are the first thing
+    the eye lands on even though the font is proportional.
+
+    Args:
+        pairs: Sequence of ``(emoji_key, label, value)`` tuples.
+        style: ``"plain"`` or ``"html"``.
+        use_emoji: Prefix each pair with its emoji marker.
+
+    Returns:
+        A single line of text, e.g.
+        ``\U0001F5D1\uFE0F 7 removed  \u00b7  \U0001F47B 4 orphaned``.
+    """
+    parts = []
+    for key, label, value in pairs:
+        emoji = NOTIFY_EMOJI.get(key, "") if use_emoji else ""
         prefix = f"{emoji} " if emoji else ""
         if style == "html":
-            return f"{prefix}<b>{label}</b>: {value}"
+            parts.append(f"{prefix}<b>{value}</b> {label}")
-        return f"{prefix}{label}: {value}"
+        else:
-    if style == "html":
+            parts.append(f"{prefix}{value} {label}")
-        pad = " " * max(LABEL_WIDTH - len(label), 1)
+    return ROW_SEPARATOR.join(parts)
-        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,
+def build_summary(self, style=DEFAULT_NOTIFY_STYLE, use_emoji=True,
                   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.
+    scanned total, the actionable counters, then the tracker split. Counters
+    are grouped two-per-line as ``emoji value label`` and separated by a middle
+    dot, because Pushover renders a proportional font and space-padded columns
+    do not line up.
 
     Args:
-        style: ``"plain"`` for aligned text, ``"html"`` for Pushover HTML.
+        style: ``"plain"`` for grouped text, ``"html"`` for Pushover HTML.
-        use_emoji: Prefix each row with an emoji marker.
+        use_emoji: Prefix each counter with an emoji marker.
         include_hostname: Add a ``Host:`` line above the summary.
 
     Returns:
@@ -274,15 +312,20 @@ def build_summary(self, style=DEFAULT_NOTIFY_STYLE, use_emoji=False,
         lines.append(f"Scanned {total} {scanned_noun} in {stats['runtime']}")
     lines.append("")
 
-    removed_label = "Removed" if delete_enabled else "Would remove"
+    removed_label = "removed" if delete_enabled else "would be removed"
-    lines.append(_notify_row(removed_label, removed, style, use_emoji, "removed"))
+    lines.append(_stat_row([
-    lines.append(_notify_row("Orphaned", stats["orphaned"], style, use_emoji, "orphaned"))
+        ("removed", removed_label, removed),
-    lines.append(_notify_row("Ignored", stats["ignored"], style, use_emoji, "ignored"))
+        ("orphaned", "orphaned", stats["orphaned"]),
-    lines.append(_notify_row("Premature", stats["premature"], style, use_emoji, "premature"))
+    ], style, use_emoji))
+    lines.append(_stat_row([
+        ("ignored", "ignored", stats["ignored"]),
+        ("premature", "premature", stats["premature"]),
+    ], style, use_emoji))
     lines.append("")
-    lines.append(_notify_row("Protected", stats["protected"], style, use_emoji, "protected"))
+    lines.append(_stat_row([
-    lines.append(_notify_row("Non-protected", stats["non_protected"], style,
+        ("protected", "protected", stats["protected"]),
-                             use_emoji, "non_protected"))
+        ("non_protected", "public", stats["non_protected"]),
+    ], style, use_emoji))
 
     return title, "\n".join(lines)
 
@@ -315,6 +358,7 @@ def setup_logging(use_log: bool, log_path: str, log_level: str,
     # Make setup idempotent: clear any handlers from a previous call
     for handler in list(logger.handlers):
         logger.removeHandler(handler)
+        handler.close()
 
     if not use_log:
         logger.addHandler(logging.NullHandler())

test_apprise.py

@@ -0,0 +1,168 @@
+"""Tests for AppriseClient — the HTTP poster and its config resolution."""
+import os
+import shutil
+import tempfile
+import unittest
+from unittest import mock
+
+import AppriseClient as apprise_module
+from AppriseClient import AppriseClient, apprise_notify
+
+
+class FakeResponse:
+    status_code = 200
+
+
+class FakeRequests:
+    def __init__(self):
+        self.calls = []
+
+    def post(self, url, json=None, verify=None):
+        self.calls.append({"url": url, "json": json, "verify": verify})
+        return FakeResponse()
+
+
+class TestAppriseNotify(unittest.TestCase):
+    def test_posts_the_expected_payload(self):
+        req = FakeRequests()
+        response = apprise_notify(
+            req, "apprise.local", 8088, "mailto://u:p@example.com", "Title", "Body"
+        )
+        self.assertIsInstance(response, FakeResponse)
+        self.assertEqual(len(req.calls), 1)
+        call = req.calls[0]
+        self.assertEqual(call["url"], "http://apprise.local:8088/notify/")
+        self.assertEqual(call["json"], {
+            "urls": "mailto://u:p@example.com",
+            "title": "Title",
+            "body": "Body",
+        })
+        self.assertFalse(call["verify"])
+
+
+class TestAppriseClientConfig(unittest.TestCase):
+    ENV_KEYS = ("DOCKER", "host", "port", "aurls", "title", "body", "toml_path")
+
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self._cwd = os.getcwd()
+        os.chdir(self.tmp)
+        self.addCleanup(os.chdir, self._cwd)
+
+        saved = {key: os.environ.get(key) for key in self.ENV_KEYS}
+        for key in self.ENV_KEYS:
+            os.environ.pop(key, None)
+        self.addCleanup(self._restore_env, saved)
+
+        self.captured = {}
+        patcher = mock.patch.object(
+            apprise_module, "apprise_notify", self._fake_notify
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _restore_env(self, saved):
+        for key, value in saved.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
+    def _fake_notify(self, req_obj, host, port, aurls, title, body):
+        self.captured.update(host=host, port=port, aurls=aurls, title=title, body=body)
+        return "sent"
+
+    def write_config(self, **overrides):
+        values = {
+            "host": "config-host",
+            "port": 9999,
+            "aurls": "mailto://config@example.com",
+            "title": "Config Title",
+            "body": "Config Body",
+        }
+        values.update(overrides)
+        with open("config.toml", "w", encoding="utf-8") as fh:
+            fh.write(
+                "[apprise]\n"
+                f'host = "{values["host"]}"\n'
+                f"port = {values["port"]}\n"
+                f'aurls = "{values["aurls"]}"\n'
+                f'title = "{values["title"]}"\n'
+                f'body = "{values["body"]}"\n'
+            )
+
+    def test_reads_values_from_environment(self):
+        os.environ.update({
+            "DOCKER": "1",
+            "host": "env-host",
+            "port": "1234",
+            "aurls": "mailto://env@example.com",
+            "title": "Env Title",
+            "body": "Env Body",
+        })
+        AppriseClient()
+        self.assertEqual(self.captured["host"], "env-host")
+        self.assertEqual(self.captured["port"], "1234")
+        self.assertEqual(self.captured["title"], "Env Title")
+
+    def test_reads_values_from_config_file(self):
+        self.write_config()
+        AppriseClient()
+        self.assertEqual(self.captured["host"], "config-host")
+        self.assertEqual(self.captured["port"], 9999)
+        self.assertEqual(self.captured["aurls"], "mailto://config@example.com")
+        self.assertEqual(self.captured["title"], "Config Title")
+        self.assertEqual(self.captured["body"], "Config Body")
+
+    def test_config_file_overrides_environment(self):
+        os.environ.update({
+            "DOCKER": "1",
+            "host": "env-host",
+            "port": "1234",
+            "aurls": "mailto://env@example.com",
+            "title": "Env Title",
+            "body": "Env Body",
+        })
+        self.write_config()
+        AppriseClient()
+        self.assertEqual(self.captured["host"], "config-host")
+        self.assertEqual(self.captured["title"], "Config Title")
+
+    def _write_elsewhere_config(self):
+        config_path = os.path.join(self.tmp, "elsewhere.toml")
+        with open(config_path, "w", encoding="utf-8") as fh:
+            fh.write(
+                "[apprise]\n"
+                'host = "toml-host"\n'
+                "port = 4321\n"
+                'aurls = "mailto://toml@example.com"\n'
+                'title = "Toml Title"\n'
+                'body = "Toml Body"\n'
+            )
+        return config_path
+
+    def test_toml_path_is_read_when_docker_is_set(self):
+        os.environ.update({
+            "DOCKER": "1",
+            "host": "env-host",
+            "port": "1234",
+            "aurls": "mailto://env@example.com",
+            "title": "Env Title",
+            "body": "Env Body",
+        })
+        os.environ["toml_path"] = self._write_elsewhere_config()
+        AppriseClient()
+        self.assertEqual(self.captured["host"], "toml-host")
+        self.assertEqual(self.captured["port"], 4321)
+
+    def test_toml_path_is_read_without_docker(self):
+        os.environ["toml_path"] = self._write_elsewhere_config()
+        AppriseClient()
+        self.assertEqual(self.captured["host"], "toml-host")
+        self.assertEqual(self.captured["port"], 4321)
+        self.assertEqual(self.captured["title"], "Toml Title")
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qbitmaid.py

@@ -1,368 +1,407 @@
-import datetime
+import datetime
-import logging
+import unittest
-import unittest
+
-from collections import Counter
+from qbt_fakes import FailingPushover, FakeQbt, StubPushover
-
+from qlist import (
-import requests as r
+    is_cat_ignored,
-from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag
+    is_ignored_tag,
-from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor
+    is_not_protected_tracker,
-from qlogging import (
+    is_preme,
-    send_ping,
+    is_protected_tracker,
-    build_summary,
+    is_tag_blank,
-    format_runtime,
+    is_tracker_blank,
-    get_notify_options,
+)
-    summary_stats,
+from qlogging import (
-    tor_notify_apprise,
+    build_summary,
-    tor_notify_summary,
+    format_runtime,
-)
+    get_notify_options,
-
+    send_ping,
-
+    summary_stats,
-class FakeQbt:
+    tor_notify_apprise,
-    """Minimal stand-in for the Qbt object, holding only what notifiers read."""
+    tor_notify_summary,
-
+)
-    def __init__(self, **overrides):
+from qprocess import (
-        self.config = {"pushover": {}}
+    is_downloading,
-        self.total_torrents = 42
+    is_not_protected_tor,
-        self.torrent_hash_delete_list = ["a"] * 7
+    is_old_tor,
-        self.up_tor_counter = 4
+    is_protected_over_ratio,
-        self.ignored_counter = 5
+    is_protected_under_ratio,
-        self.preme_tor_counter = 3
+)
-        self.tracker_protected_tag = "ipt"
+
-        self.tracker_non_protected_tag = "public"
+
-        self.c = Counter({"ipt": 10, "public": 20})
+class RecordingRequests:
-        self.delete_torrents = True
+    """Stands in for the requests module without touching the network."""
-        self.st = datetime.datetime(2026, 10, 27, 9, 54, 0)
+
-        self.et = datetime.datetime(2026, 10, 27, 9, 54, 12)
+    class RequestException(Exception):
-        self.po_key = "USERKEY"
+        pass
-        self.poc = StubPushover()
+
-        self.apprise_host = "localhost"
+    def __init__(self):
-        self.apprise_port = 8088
+        self.calls = []
-        self.apprise_aurls = "mailto://user:pass@example.com"
+        self.timeouts = []
-        self.tl = logging.getLogger("qbit-maid-test")
+
-        for key, value in overrides.items():
+    def get(self, url, timeout=None):
-            setattr(self, key, value)
+        self.calls.append(url)
-
+        self.timeouts.append(timeout)
-
+        return "ok"
-class StubPushover:
+
-    """Records message() calls instead of hitting the Pushover API."""
+
-
+class TestPredicates(unittest.TestCase):
-    def __init__(self):
+    def test_is_preme(self):
-        self.calls = []
+        self.assertTrue(is_preme(1, 1))  # equal to min_age counts as premature
-
+        self.assertTrue(is_preme(1, 2))
-    def message(self, user, body, **kwargs):
+        self.assertFalse(is_preme(2, 1))
-        self.calls.append({"user": user, "body": body, "kwargs": kwargs})
+
-
+    def test_is_cat_ignored(self):
-
+        cats = ["a", "b", "c"]
-class FailingPushover:
+        for cat in cats:
-    def message(self, *args, **kwargs):
+            self.assertTrue(is_cat_ignored(cat, cats))
-        raise RuntimeError("pushover is down")
+        self.assertFalse(is_cat_ignored("d", cats))
-
+        self.assertFalse(is_cat_ignored(1, cats))
-class TestQbitmaid(unittest.TestCase):
+        self.assertFalse(is_cat_ignored(1.0, cats))
-    def test_ispreme_sanity(self):
+
-        self.assertTrue(is_preme(1,1))
+    def test_is_tracker_blank(self):
-        self.assertTrue(is_preme(1,2))
+        self.assertTrue(is_tracker_blank(""))
-        self.assertFalse(is_preme(2,1))
+        self.assertFalse(is_tracker_blank("a"))
-    
+        self.assertFalse(is_tracker_blank(1))
-    def test_ispreme(self):
+        self.assertFalse(is_tracker_blank(1.0))
-        pass
+
-
+    def test_is_protected_tracker(self):
-    def test_iscatignored_sanity(self):
+        whitelist = ["a.com", "b.me", "c.io"]
-        self.assertTrue(is_cat_ignored('a', ['a','b','c']))
+        self.assertTrue(is_protected_tracker("https://a.com/", whitelist))
-        self.assertTrue(is_cat_ignored('b', ['a','b','c']))
+        self.assertFalse(is_protected_tracker("https://google.com/", whitelist))
-        self.assertTrue(is_cat_ignored('c', ['a','b','c']))
+        self.assertFalse(is_protected_tracker("https://d.com", whitelist))
-        self.assertFalse(is_cat_ignored('d', ['a','b','c']))
+        self.assertFalse(is_protected_tracker("", whitelist))
-        self.assertFalse(is_cat_ignored(1, ['a','b','d']))
+
-        self.assertFalse(is_cat_ignored(1.0000000, ['a','b','c']))
+    def test_is_not_protected_tracker(self):
-
+        whitelist = ["a.com", "b.me", "c.io"]
-    def test_iscatignored(self):
+        self.assertFalse(is_not_protected_tracker("https://a.com/", whitelist))
-        pass
+        self.assertTrue(is_not_protected_tracker("https://google.com/", whitelist))
-
+        self.assertTrue(is_not_protected_tracker("https://d.com", whitelist))
-    def test_istrackerblank_sanity(self):
+        self.assertFalse(is_not_protected_tracker("", whitelist))
-        self.assertTrue(is_tracker_blank(''))
+
-        self.assertFalse(is_tracker_blank('a'))
+    def test_is_tag_blank(self):
-        self.assertFalse(is_tracker_blank(1))
+        self.assertTrue(is_tag_blank(""))
-        self.assertFalse(is_tracker_blank(1.00000000))
+        for value in ("a", 1, 1.0001, False, True):
-
+            self.assertFalse(is_tag_blank(value))
-    def test_istrackerblank(self):
+
-        pass
+    def test_is_downloading(self):
-
+        self.assertTrue(is_downloading("downloading"))
-    def test_isprotectedtracker_sanity(self):
+        self.assertFalse(is_downloading("DOWNLOADING"))
-        self.assertTrue(is_protected_tracker('https://a.com/',['a.com','b.me','c.io']))
+        self.assertFalse(is_downloading("dOwNlOaDiNg"))
-        self.assertFalse(is_protected_tracker('https://google.com/',['a.com','b.me','c.io']))
+        self.assertFalse(is_downloading("uploading"))
-        self.assertFalse(is_protected_tracker('https://d.com',['a.com','b.me','c.io']))
+
-
+    def test_is_protected_under_ratio(self):
-    def test_isprotectedtracker(self):
+        self.assertTrue(is_protected_under_ratio(0.5, 1, "a", "a,b,c"))
-        pass
+        self.assertFalse(is_protected_under_ratio(1.5, 1, "a", "a,b,c"))
-
+        self.assertFalse(is_protected_under_ratio(0.5, 1, "z", "a,b,c"))
-    def test_isnotprotectedtracker_sanity(self):
+
-        self.assertFalse(is_not_protected_tracker('https://a.com/',['a.com','b.me','c.io']))
+    def test_is_old_tor(self):
-        self.assertTrue(is_not_protected_tracker('https://google.com/',['a.com','b.me','c.io']))
+        self.assertTrue(is_old_tor(1, 1))
-        self.assertTrue(is_not_protected_tracker('https://d.com',['a.com','b.me','c.io']))
+        self.assertTrue(is_old_tor(2, 1))
-
+        self.assertFalse(is_old_tor(1, 2))
-    def test_isnotprotectedtracker(self):
+
-        pass
+    def test_is_protected_over_ratio(self):
-
+        self.assertTrue(is_protected_over_ratio(2, 1, "a", "a,b,c"))
-    def test_istagblank(self):
+        self.assertFalse(is_protected_over_ratio(0.5, 1, "a", "a,b,c"))
-        self.assertTrue(is_tag_blank(''))
+        self.assertFalse(is_protected_over_ratio(2, 1, "z", "a,b,c"))
-        self.assertFalse(is_tag_blank('a'))
+
-        self.assertFalse(is_tag_blank(1))
+    def test_is_not_protected_tor(self):
-        self.assertFalse(is_tag_blank(1.0001))
+        self.assertTrue(is_not_protected_tor("a", "a,b,c"))
-        self.assertFalse(is_tag_blank(False))
+        self.assertFalse(is_not_protected_tor("z", "a,b,c"))
-        self.assertFalse(is_tag_blank(True))
+
-
+    def test_is_ignored_tag(self):
-    def test_isdownloading_sanity(self):
+        self.assertTrue(is_ignored_tag(["save"], "save,public,ipt"))
-        self.assertTrue(is_downloading('downloading'))
+        self.assertFalse(is_ignored_tag(["save"], "public,ipt"))
-    
+
-    def test_isdownloading(self):
+    def test_is_ignored_tag_matches_whole_tags_only(self):
-        self.assertFalse(is_downloading('DOWNLOADING'))
+        """A configured tag must match a whole tag, not a substring of one.
-        self.assertFalse(is_downloading('dOwNlOaDiNg'))
+
-
+        'ipt' must not match 'iptorrents' and 'save' must not match 'unsaved'.
-    def test_isprotectedunderratio_sanity(self):
+        """
-        self.assertTrue(is_protected_under_ratio(0.5,1,'a','a,b,c'))
+        self.assertFalse(is_ignored_tag(["ipt"], "iptorrents"))
-
+        self.assertFalse(is_ignored_tag(["save"], "unsaved"))
-    def test_isprotectedunderratio(self):
+        self.assertTrue(is_ignored_tag(["ipt"], "save,ipt"))
-        pass
+        self.assertTrue(is_ignored_tag(["save"], " save , public"))
-
+        self.assertFalse(is_ignored_tag(["save"], ""))
-    def test_isoldtor_sanity(self):
+
-        self.assertFalse(is_old_tor(1,2))
+
-
+class TestSendPing(unittest.TestCase):
-    def test_isoldtor(self):
+    def test_send_ping_requests_the_url(self):
-        self.assertTrue(is_old_tor(1,1))
+        req = RecordingRequests()
-        self.assertTrue(is_old_tor(2,1))
+        send_ping(FakeQbt(), req, "https://example.com/ping/x")
-        self.assertFalse(is_old_tor(1,2))
+        self.assertEqual(req.calls, ["https://example.com/ping/x"])
-
+
-    def test_isprotectedoverratio_sanity(self):
+    def test_send_ping_uses_a_five_second_timeout(self):
-        self.assertTrue(is_protected_over_ratio(2,1,'a','a,b,c'))
+        req = RecordingRequests()
-
+        send_ping(FakeQbt(), req, "https://example.com/ping/x")
-    def test_isprotectedoverratio(self):
+        self.assertEqual(req.timeouts, [5])
-        pass
+
-
+
-    def test_isnonprotectedtor_sanity(self):
+# --- notification formatting -----------------------------------------------
-        self.assertTrue(is_not_protected_tor('a','a,b,c'))
+
-
+class TestFormatRuntime(unittest.TestCase):
-    def test_isnonprotectedtor(self):
+    def test_format_runtime(self):
-        pass
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s")
-
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s")
-    def test_isignoredtag_sanity(self):
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=64)), "1m 04s")
-        self.assertTrue(is_ignored_tag(['a','b','c'], 'first,second,third,a'))
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s")
-    
+        self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s")
-    def test_isignoredtag(self):
+
-        self.assertTrue(is_ignored_tag(['save'], 'save,public,ipt'))
+
-
+class TestSummaryStats(unittest.TestCase):
-    def test_sendpingstart_sanity(self):
+    def test_summary_stats(self):
-        send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb" + "/start")
+        stats = summary_stats(FakeQbt())
-        url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb"
+        self.assertEqual(stats["total"], 42)
-        send_ping(self, r, url)
+        self.assertEqual(stats["removed"], 7)
-
+        self.assertEqual(stats["orphaned"], 4)
-    def test_sendping_start(self):
+        self.assertEqual(stats["ignored"], 5)
-        url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb/"
+        self.assertEqual(stats["premature"], 3)
-        send_ping(self, r, url.strip("/") + "/start")
+        self.assertEqual(stats["protected"], 10)
-        send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb")
+        self.assertEqual(stats["non_protected"], 20)
-        
+        self.assertEqual(stats["runtime"], "12s")
-
+
-    # --- notification formatting -------------------------------------------
+    def test_summary_stats_missing_attributes(self):
-
+        class Bare:
-    def test_format_runtime(self):
+            pass
-        self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s")
+
-        self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s")
+        stats = summary_stats(Bare())
-        self.assertEqual(format_runtime(datetime.timedelta(seconds=64)), "1m 04s")
+        self.assertEqual(stats["total"], 0)
-        self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s")
+        self.assertEqual(stats["removed"], 0)
-        self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s")
+        self.assertEqual(stats["runtime"], "unknown")
-
+
-    def test_summary_stats(self):
+
-        stats = summary_stats(FakeQbt())
+class TestBuildSummary(unittest.TestCase):
-        self.assertEqual(stats["total"], 42)
+    def test_build_summary_plain(self):
-        self.assertEqual(stats["removed"], 7)
+        title, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
-        self.assertEqual(stats["orphaned"], 4)
+        self.assertEqual(title, "qbit-maid: 7 torrents removed")
-        self.assertEqual(stats["ignored"], 5)
+        self.assertEqual(
-        self.assertEqual(stats["premature"], 3)
+            body,
-        self.assertEqual(stats["protected"], 10)
+            "Scanned 42 torrents in 12s\n"
-        self.assertEqual(stats["non_protected"], 20)
+            "\n"
-        self.assertEqual(stats["runtime"], "12s")
+            "7 removed  \u00b7  4 orphaned\n"
-
+            "5 ignored  \u00b7  3 premature\n"
-    def test_summary_stats_missing_attributes(self):
+            "\n"
-        class Bare:
+            "10 protected  \u00b7  20 public",
-            pass
+        )
-
+
-        stats = summary_stats(Bare())
+    def test_build_summary_plain_emoji(self):
-        self.assertEqual(stats["total"], 0)
+        _, body = build_summary(FakeQbt(), style="plain", use_emoji=True)
-        self.assertEqual(stats["removed"], 0)
+        self.assertEqual(
-        self.assertEqual(stats["runtime"], "unknown")
+            body,
-
+            "Scanned 42 torrents in 12s\n"
-    def test_build_summary_plain(self):
+            "\n"
-        title, body = build_summary(FakeQbt(), style="plain")
+            "\U0001F5D1\uFE0F 7 removed  \u00b7  \U0001F47B 4 orphaned\n"
-        self.assertEqual(title, "qbit-maid: 7 torrents removed")
+            "\u23ED\uFE0F 5 ignored  \u00b7  \U0001F423 3 premature\n"
-        self.assertEqual(
+            "\n"
-            body,
+            "\U0001F6E1\uFE0F 10 protected  \u00b7  \U0001F310 20 public",
-            "Scanned 42 torrents in 12s\n"
+        )
-            "\n"
+
-            "Removed          7\n"
+    def test_build_summary_plain_has_no_leading_indent(self):
-            "Orphaned         4\n"
+        _, body = build_summary(FakeQbt(), style="plain")
-            "Ignored          5\n"
+        for line in body.splitlines():
-            "Premature        3\n"
+            self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}")
-            "\n"
+
-            "Protected       10\n"
+    def test_build_summary_has_no_space_padding(self):
-            "Non-protected   20",
+        """Pushover uses a proportional font, so padded columns never align."""
-        )
+        for style in ("plain", "html"):
-
+            for emoji in (False, True):
-    def test_build_summary_plain_has_no_leading_indent(self):
+                _, body = build_summary(FakeQbt(), style=style, use_emoji=emoji)
-        _, body = build_summary(FakeQbt(), style="plain")
+                for line in body.splitlines():
-        for line in body.splitlines():
+                    stripped = line.replace("  \u00b7  ", "")
-            self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}")
+                    self.assertNotIn("   ", stripped,
-
+                                     f"padded column in {style}/{emoji}: {line!r}")
-    def test_build_summary_html(self):
+
-        title, body = build_summary(FakeQbt(), style="html")
+    def test_build_summary_groups_two_counters_per_line(self):
-        self.assertEqual(title, "qbit-maid: 7 torrents removed")
+        _, body = build_summary(FakeQbt(), style="plain")
-        self.assertIn("Scanned <b>42</b> torrents in <b>12s</b>", body)
+        rows = [ln for ln in body.splitlines() if "\u00b7" in ln]
-        self.assertIn("<b>Removed</b>", body)
+        self.assertEqual(len(rows), 3)
-        self.assertIn("<b>Non-protected</b>", body)
+        for row in rows:
-        self.assertNotIn("---", title)
+            self.assertEqual(row.count("\u00b7"), 1, row)
-
+
-    def test_build_summary_emoji(self):
+    def test_build_summary_value_leads_label(self):
-        _, plain = build_summary(FakeQbt(), style="plain", use_emoji=True)
+        """The number comes first so it is the first thing the eye lands on."""
-        self.assertIn("\U0001F5D1\uFE0F Removed: 7", plain)
+        _, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
-        self.assertIn("\U0001F47B Orphaned: 4", plain)
+        self.assertIn("7 removed", body)
-        _, html = build_summary(FakeQbt(), style="html", use_emoji=True)
+        self.assertIn("20 public", body)
-        self.assertIn("\U0001F5D1\uFE0F <b>Removed</b>: 7", html)
+        self.assertNotIn("Removed 7", body)
-
+        self.assertNotIn("Non-protected", body)
-    def test_build_summary_delete_disabled(self):
+
-        title, body = build_summary(FakeQbt(delete_torrents=False))
+    def test_build_summary_html(self):
-        self.assertEqual(title, "qbit-maid: 7 torrents would be removed")
+        title, body = build_summary(FakeQbt(), style="html")
-        self.assertIn("Would remove", body)
+        self.assertEqual(title, "qbit-maid: 7 torrents removed")
-        self.assertNotIn("Removed", body)
+        self.assertIn("Scanned <b>42</b> torrents in <b>12s</b>", body)
-
+        self.assertIn("<b>7</b> removed", body)
-    def test_build_summary_nothing_to_do(self):
+        self.assertIn("<b>20</b> public", body)
-        title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0))
+        self.assertNotIn("---", title)
-        self.assertEqual(title, "qbit-maid: nothing removed")
+
-        title, _ = build_summary(
+    def test_build_summary_emoji(self):
-            FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False)
+        _, plain = build_summary(FakeQbt(), style="plain", use_emoji=True)
-        )
+        self.assertIn("\U0001F5D1\uFE0F 7 removed", plain)
-        self.assertEqual(title, "qbit-maid: nothing to remove")
+        self.assertIn("\U0001F47B 4 orphaned", plain)
-
+        _, html = build_summary(FakeQbt(), style="html", use_emoji=True)
-    def test_build_summary_singular(self):
+        self.assertIn("\U0001F5D1\uFE0F <b>7</b> removed", html)
-        title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1))
+
-        self.assertEqual(title, "qbit-maid: 1 torrent removed")
+    def test_build_summary_emoji_off_has_no_emoji(self):
-
+        _, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
-    def test_build_summary_hostname(self):
+        for marker in ("\U0001F5D1", "\U0001F47B", "\u23ED", "\U0001F423",
-        _, body = build_summary(FakeQbt(), include_hostname=True)
+                       "\U0001F6E1", "\U0001F310"):
-        self.assertTrue(body.startswith("Host: "))
+            self.assertNotIn(marker, body)
-
+
-    def test_build_summary_unknown_style_falls_back(self):
+    def test_build_summary_delete_disabled(self):
-        _, body = build_summary(FakeQbt(), style="markdown")
+        title, body = build_summary(FakeQbt(delete_torrents=False))
-        self.assertNotIn("<b>", body)
+        self.assertEqual(title, "qbit-maid: 7 torrents would be removed")
-
+        self.assertIn("7 would be removed", body)
-    def test_get_notify_options_defaults(self):
+        self.assertNotIn("7 removed", body)
-        opts = get_notify_options(FakeQbt())
+
-        self.assertEqual(opts["style"], "plain")
+    def test_build_summary_body_does_not_repeat_title_phrase(self):
-        self.assertFalse(opts["use_emoji"])
+        """The title carries the outcome; the body must not restate it verbatim."""
-        self.assertFalse(opts["include_hostname"])
+        title, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
-        self.assertTrue(opts["notify_on_empty"])
+        self.assertNotIn("torrents removed", body)
-        self.assertIsNone(opts["priority"])
+        self.assertNotIn("torrents would be removed", body)
-        self.assertIsNone(opts["sound"])
+        self.assertIn("7 removed", body)
-
+        self.assertTrue(title.endswith("7 torrents removed"))
-    def test_get_notify_options_overrides(self):
+
-        fake = FakeQbt(config={"pushover": {
+    def test_build_summary_nothing_to_do(self):
-            "format": "HTML",
+        title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0))
-            "use_emoji": True,
+        self.assertEqual(title, "qbit-maid: nothing removed")
-            "include_hostname": True,
+        title, _ = build_summary(
-            "notify_on_empty": False,
+            FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False)
-            "priority": 2,
+        )
-            "sound": "siren",
+        self.assertEqual(title, "qbit-maid: nothing to remove")
-            "retry": 30,
+
-            "expire": 600,
+    def test_build_summary_singular(self):
-        }})
+        title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1))
-        opts = get_notify_options(fake)
+        self.assertEqual(title, "qbit-maid: 1 torrent removed")
-        self.assertEqual(opts["style"], "html")
+
-        self.assertTrue(opts["use_emoji"])
+    def test_build_summary_hostname(self):
-        self.assertTrue(opts["include_hostname"])
+        _, body = build_summary(FakeQbt(), include_hostname=True)
-        self.assertFalse(opts["notify_on_empty"])
+        self.assertTrue(body.startswith("Host: "))
-        self.assertEqual(opts["priority"], 2)
+
-        self.assertEqual(opts["sound"], "siren")
+    def test_build_summary_hostname_html(self):
-        self.assertEqual(opts["retry"], 30)
+        _, body = build_summary(FakeQbt(), style="html", include_hostname=True)
-        self.assertEqual(opts["expire"], 600)
+        self.assertTrue(body.startswith("<b>Host:</b> "))
-
+
-    def test_get_notify_options_invalid_style(self):
+    def test_build_summary_unknown_style_falls_back(self):
-        fake = FakeQbt(config={"pushover": {"format": "bogus"}})
+        _, body = build_summary(FakeQbt(), style="markdown")
-        self.assertEqual(get_notify_options(fake)["style"], "plain")
+        self.assertNotIn("<b>", body)
-
+
-    def test_tor_notify_summary_plain_payload(self):
+
-        fake = FakeQbt()
+class TestNotifyOptions(unittest.TestCase):
-        tor_notify_summary(fake)
+    def test_get_notify_options_defaults(self):
-        self.assertEqual(len(fake.poc.calls), 1)
+        opts = get_notify_options(FakeQbt())
-        call = fake.poc.calls[0]
+        self.assertEqual(opts["style"], "plain")
-        self.assertEqual(call["user"], "USERKEY")
+        self.assertTrue(opts["use_emoji"])
-        self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed")
+        self.assertFalse(opts["include_hostname"])
-        self.assertNotIn("html", call["kwargs"])
+        self.assertTrue(opts["notify_on_empty"])
-
+        self.assertIsNone(opts["priority"])
-    def test_tor_notify_summary_html_payload(self):
+        self.assertIsNone(opts["sound"])
-        fake = FakeQbt(config={"pushover": {"format": "html"}})
+
-        tor_notify_summary(fake)
+    def test_get_notify_options_emoji_can_be_disabled(self):
-        call = fake.poc.calls[0]
+        opts = get_notify_options(FakeQbt(config={"pushover": {"use_emoji": False}}))
-        self.assertEqual(call["kwargs"]["html"], 1)
+        self.assertFalse(opts["use_emoji"])
-        self.assertIn("<b>", call["body"])
+
-
+    def test_get_notify_options_overrides(self):
-    def test_tor_notify_summary_priority_two_adds_retry_expire(self):
+        fake = FakeQbt(config={"pushover": {
-        fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}})
+            "format": "HTML",
-        tor_notify_summary(fake)
+            "use_emoji": False,
-        kwargs = fake.poc.calls[0]["kwargs"]
+            "include_hostname": True,
-        self.assertEqual(kwargs["priority"], 2)
+            "notify_on_empty": False,
-        self.assertEqual(kwargs["retry"], 30)
+            "priority": 2,
-        self.assertEqual(kwargs["expire"], 600)
+            "sound": "siren",
-
+            "retry": 30,
-    def test_tor_notify_summary_priority_one_has_no_retry(self):
+            "expire": 600,
-        fake = FakeQbt(config={"pushover": {"priority": 1}})
+        }})
-        tor_notify_summary(fake)
+        opts = get_notify_options(fake)
-        kwargs = fake.poc.calls[0]["kwargs"]
+        self.assertEqual(opts["style"], "html")
-        self.assertEqual(kwargs["priority"], 1)
+        self.assertFalse(opts["use_emoji"])
-        self.assertNotIn("retry", kwargs)
+        self.assertTrue(opts["include_hostname"])
-
+        self.assertFalse(opts["notify_on_empty"])
-    def test_tor_notify_summary_skips_when_empty(self):
+        self.assertEqual(opts["priority"], 2)
-        fake = FakeQbt(
+        self.assertEqual(opts["sound"], "siren")
-            config={"pushover": {"notify_on_empty": False}},
+        self.assertEqual(opts["retry"], 30)
-            torrent_hash_delete_list=[],
+        self.assertEqual(opts["expire"], 600)
-            up_tor_counter=0,
+
-        )
+    def test_get_notify_options_invalid_style(self):
-        tor_notify_summary(fake)
+        fake = FakeQbt(config={"pushover": {"format": "bogus"}})
-        self.assertEqual(fake.poc.calls, [])
+        self.assertEqual(get_notify_options(fake)["style"], "plain")
-
+
-    def test_tor_notify_summary_sends_when_empty_by_default(self):
+
-        fake = FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)
+class TestTorNotifySummary(unittest.TestCase):
-        tor_notify_summary(fake)
+    def test_tor_notify_summary_plain_payload(self):
-        self.assertEqual(len(fake.poc.calls), 1)
+        fake = FakeQbt()
-
+        tor_notify_summary(fake)
-    def test_tor_notify_summary_swallows_send_errors(self):
+        self.assertEqual(len(fake.poc.calls), 1)
-        fake = FakeQbt()
+        call = fake.poc.calls[0]
-        fake.poc = FailingPushover()
+        self.assertEqual(call["user"], "USERKEY")
-        tor_notify_summary(fake)  # must not raise
+        self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed")
-
+        self.assertNotIn("html", call["kwargs"])
-    def test_tor_notify_apprise_uses_plain_body(self):
+
-        captured = {}
+    def test_tor_notify_summary_html_payload(self):
-
+        fake = FakeQbt(config={"pushover": {"format": "html"}})
-        def app_obj(req_obj, host, port, aurls, title, body):
+        tor_notify_summary(fake)
-            captured.update(host=host, port=port, aurls=aurls, title=title, body=body)
+        call = fake.poc.calls[0]
-
+        self.assertEqual(call["kwargs"]["html"], 1)
-        fake = FakeQbt(config={"pushover": {"format": "html"}})
+        self.assertIn("<b>", call["body"])
-        tor_notify_apprise(fake, None, app_obj)
+
-        self.assertEqual(captured["title"], "qbit-maid: 7 torrents removed")
+    def test_tor_notify_summary_priority_two_adds_retry_expire(self):
-        self.assertNotIn("<b>", captured["body"])
+        fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}})
-        self.assertEqual(captured["port"], 8088)
+        tor_notify_summary(fake)
-
+        kwargs = fake.poc.calls[0]["kwargs"]
-    def test_pushover_and_apprise_bodies_match(self):
+        self.assertEqual(kwargs["priority"], 2)
-        fake = FakeQbt()
+        self.assertEqual(kwargs["retry"], 30)
-        tor_notify_summary(fake)
+        self.assertEqual(kwargs["expire"], 600)
-        captured = {}
+
-        tor_notify_apprise(
+    def test_tor_notify_summary_priority_one_has_no_retry(self):
-            fake, None,
+        fake = FakeQbt(config={"pushover": {"priority": 1}})
-            lambda req, host, port, aurls, title, body: captured.update(title=title, body=body),
+        tor_notify_summary(fake)
-        )
+        kwargs = fake.poc.calls[0]["kwargs"]
-        self.assertEqual(fake.poc.calls[0]["body"], captured["body"])
+        self.assertEqual(kwargs["priority"], 1)
-        self.assertEqual(fake.poc.calls[0]["kwargs"]["title"], captured["title"])
+        self.assertNotIn("retry", kwargs)
-
+
-    # def test__sanity(self):
+    def test_tor_notify_summary_skips_when_empty(self):
-    #     pass
+        fake = FakeQbt(
-
+            config={"pushover": {"notify_on_empty": False}},
-    # def test_(self):
+            torrent_hash_delete_list=[],
-    #     pass
+            up_tor_counter=0,
-        
+        )
-if __name__ == '__main__':
+        tor_notify_summary(fake)
-    unittest.main()
+        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
+
+
+class TestTorNotifyApprise(unittest.TestCase):
+    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"])
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qbt.py

@@ -0,0 +1,350 @@
+"""End-to-end tests for the Qbt object in qbit-maid.py.
+
+The module name contains a hyphen, so it cannot be imported normally; it is
+loaded by path with ``qbittorrentapi`` stubbed out.
+"""
+import importlib.util
+import logging
+import os
+import shutil
+import sys
+import tempfile
+import types
+import unittest
+from unittest import mock
+
+from qbt_fakes import (
+    FailingPushover,
+    FakeAPIError,
+    FakeQbtClient,
+    StubPushover,
+    torrent,
+)
+
+REPO = os.path.dirname(os.path.abspath(__file__))
+
+CONFIG_TEMPLATE = """
+[qbittorrent]
+host = "127.0.0.1"
+port = 8080
+username = "user"
+password = "pass"
+
+[logging]
+use_log = true
+log_level = "{log_level}"
+log_path = "{log_path}"
+
+[app_tags]
+protected_tag = "ipt"
+non_protected_tag = "public"
+
+[torrent]
+max_age = 2419200
+min_age = 432000
+delete_torrents = {delete_torrents}
+
+[pushover]
+use_pushover = {use_pushover}
+po_key = "USERKEY"
+po_token = "TOKEN"
+
+[apprise]
+use_apprise = {use_apprise}
+host = "localhost"
+port = 8088
+aurls = "mailto://user:pass@example.com"
+
+[dragnet]
+enable_dragnet = false
+dragnet_outfile = "{dragnet_outfile}"
+
+[telemetry]
+enable_telemetry = false
+telemetry_outfile = "{telemetry_outfile}"
+
+[ignored_categories]
+tech = "tech"
+
+[ignored_domains]
+tracker-example = "tracker.example"
+
+[ignored_tags]
+save = "save"
+
+[healthcheck]
+use_healthcheck = {use_healthcheck}
+healthcheck_url = "https://example.com/ping/x"
+"""
+
+
+def load_qbit_maid(as_main=False):
+    """Import qbit-maid.py by path, with qbittorrentapi stubbed.
+
+    With ``as_main=True`` the module is executed as ``__main__``, which runs
+    the ``Qbt()`` entry point at import time.
+    """
+    # Install the stub once and reuse it, so tests can patch its Client.
+    # This keeps the suite off any real qBittorrent even when the real package
+    # is installed in the environment.
+    stub = sys.modules.get("qbittorrentapi")
+    if getattr(stub, "_qbit_maid_stub", False) is not True:
+        stub = types.ModuleType("qbittorrentapi")
+        stub.APIError = FakeAPIError
+        stub.Client = FakeQbtClient
+        stub._qbit_maid_stub = True
+        sys.modules["qbittorrentapi"] = stub
+    if REPO not in sys.path:
+        sys.path.insert(0, REPO)
+    spec = importlib.util.spec_from_file_location(
+        "__main__" if as_main else "qbit_maid", os.path.join(REPO, "qbit-maid.py")
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class QbtTestCase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self._cwd = os.getcwd()
+        os.chdir(self.tmp)
+        self.addCleanup(os.chdir, self._cwd)
+        self.addCleanup(self._close_log_handlers)
+        self.module = load_qbit_maid()
+
+    def _close_log_handlers(self):
+        logger = logging.getLogger("qbit-maid")
+        for handler in list(logger.handlers):
+            logger.removeHandler(handler)
+            handler.close()
+
+    def write_config(self, **overrides):
+        values = {
+            "log_level": "CRITICAL",  # keep the console handler quiet
+            "log_path": os.path.join(self.tmp, "qc.log"),
+            "delete_torrents": "true",
+            "use_pushover": "false",
+            "use_apprise": "false",
+            "use_healthcheck": "false",
+            "dragnet_outfile": os.path.join(self.tmp, "orphaned.csv"),
+            "telemetry_outfile": os.path.join(self.tmp, "telemetry.csv"),
+        }
+        values.update(overrides)
+        with open("config.toml", "w", encoding="utf-8") as fh:
+            fh.write(CONFIG_TEMPLATE.format(**values))
+
+    def build(self, torrents, client=None, **config_overrides):
+        self.write_config(**config_overrides)
+        client = client or FakeQbtClient(torrents=torrents)
+
+        def factory(**kwargs):
+            client.init_kwargs = kwargs
+            return client
+
+        with mock.patch.object(self.module.qbittorrentapi, "Client",
+                               side_effect=factory):
+            qbt = self.module.Qbt()
+        return qbt, client
+
+
+class TestQbtPipeline(QbtTestCase):
+    def sample_torrents(self):
+        return [
+            torrent(tags="ipt", ratio=0.5, time_active=9999999,
+                    infohash="del-protected"),
+            torrent(tags="public", tracker="https://other.example/a", ratio=0.1,
+                    infohash="del-public"),
+            torrent(tags="public", state="downloading", infohash="skip-downloading"),
+            torrent(tags="other", infohash="orphan-1"),
+            torrent(tracker="", infohash="ignored-blank-tracker"),
+            torrent(category="tech", infohash="ignored-category"),
+            torrent(tags="ipt", seeding_time=1, infohash="premature"),
+            torrent(tags="", infohash="new-blank-tag"),
+        ]
+
+    def test_full_run_counts_every_bucket(self):
+        qbt, client = self.build(self.sample_torrents())
+        self.assertEqual(qbt.total_torrents, 8)
+        self.assertEqual(qbt.ignored_counter, 2)
+        self.assertEqual(qbt.preme_tor_counter, 1)
+        self.assertEqual(qbt.up_tor_counter, 2)
+        self.assertCountEqual(qbt.torrent_hash_delete_list,
+                              ["del-protected", "del-public"])
+
+    def test_full_run_deletes_the_selected_hashes(self):
+        _, client = self.build(self.sample_torrents())
+        self.assertEqual(len(client.deleted), 1)
+        delete_files, hashes = client.deleted[0]
+        self.assertTrue(delete_files)
+        self.assertCountEqual(hashes, ["del-protected", "del-public"])
+
+    def test_full_run_tags_new_torrents(self):
+        _, client = self.build(self.sample_torrents())
+        self.assertIn(("ipt", "new-blank-tag"), client.added_tags)
+
+    def test_dry_run_does_not_delete(self):
+        qbt, client = self.build(self.sample_torrents(), delete_torrents="false")
+        self.assertFalse(qbt.delete_torrents)
+        self.assertEqual(client.deleted, [])
+        self.assertCountEqual(qbt.torrent_hash_delete_list,
+                              ["del-protected", "del-public"])
+
+    def test_client_is_constructed_from_config(self):
+        _, client = self.build([torrent(tags="ipt")])
+        self.assertEqual(client.init_kwargs["host"], "127.0.0.1")
+        self.assertEqual(client.init_kwargs["port"], 8080)
+        self.assertEqual(client.init_kwargs["username"], "user")
+
+    def test_client_logs_in(self):
+        _, client = self.build([torrent(tags="ipt")])
+        self.assertTrue(client.logged_in)
+
+    def test_runtime_is_recorded(self):
+        qbt, _ = self.build([torrent(tags="ipt")])
+        self.assertIsNotNone(qbt.et)
+        self.assertGreaterEqual(qbt.et, qbt.st)
+
+    def test_empty_torrent_list_is_handled(self):
+        """A qBittorrent instance with zero torrents must not abort the run."""
+        qbt, client = self.build([])
+        self.assertEqual(qbt.total_torrents, 0)
+        self.assertEqual(qbt.torrent_hash_delete_list, [])
+        self.assertEqual(client.deleted, [])
+
+    def test_empty_torrent_list_still_sends_a_summary(self):
+        stub = StubPushover()
+        with mock.patch.object(self.module.pushover, "Pushover",
+                               return_value=stub):
+            self.build([], use_pushover="true")
+        self.assertEqual(len(stub.calls), 1)
+        self.assertEqual(stub.calls[0]["kwargs"]["title"],
+                         "qbit-maid: nothing removed")
+
+
+class TestQbtConfigSources(QbtTestCase):
+    def test_config_is_read_from_toml_path_env(self):
+        config_path = os.path.join(self.tmp, "from-env.toml")
+        self.write_config()
+        os.replace("config.toml", config_path)
+        os.environ["toml_path"] = config_path
+        self.addCleanup(os.environ.pop, "toml_path", None)
+        with mock.patch.object(self.module.qbittorrentapi, "Client",
+                               return_value=FakeQbtClient(torrents=[torrent(tags="ipt")])):
+            qbt = self.module.Qbt()
+        self.assertEqual(qbt.tracker_protected_tag, "ipt")
+
+    def test_local_config_toml_wins_over_toml_path(self):
+        elsewhere = os.path.join(self.tmp, "elsewhere.toml")
+        with open(elsewhere, "w", encoding="utf-8") as fh:
+            fh.write(CONFIG_TEMPLATE.format(
+                log_level="CRITICAL",
+                log_path=os.path.join(self.tmp, "other.log"),
+                delete_torrents="true", use_pushover="false",
+                use_apprise="false", use_healthcheck="false",
+                dragnet_outfile="a.csv", telemetry_outfile="b.csv",
+            ).replace('protected_tag = "ipt"', 'protected_tag = "from-elsewhere"'))
+        os.environ["toml_path"] = elsewhere
+        self.addCleanup(os.environ.pop, "toml_path", None)
+        self.write_config()
+        with mock.patch.object(self.module.qbittorrentapi, "Client",
+                               return_value=FakeQbtClient(torrents=[torrent(tags="ipt")])):
+            qbt = self.module.Qbt()
+        self.assertEqual(qbt.tracker_protected_tag, "ipt")
+
+
+class TestQbtNotifiers(QbtTestCase):
+    def test_pushover_summary_is_sent_when_enabled(self):
+        stub = StubPushover()
+        with mock.patch.object(self.module.pushover, "Pushover",
+                               return_value=stub):
+            qbt, _ = self.build([torrent(tags="public", infohash="h1")],
+                                use_pushover="true")
+        self.assertEqual(len(stub.calls), 1)
+        self.assertEqual(stub.calls[0]["kwargs"]["title"],
+                         "qbit-maid: 1 torrent removed")
+
+    def test_apprise_is_sent_when_enabled(self):
+        captured = {}
+
+        def fake_apprise(req_obj, host, port, aurls, title, body):
+            captured.update(host=host, port=port, title=title, body=body)
+
+        with mock.patch.object(self.module, "apprise_notify", fake_apprise):
+            self.build([torrent(tags="public", infohash="h1")],
+                       use_apprise="true")
+        self.assertEqual(captured["host"], "localhost")
+        self.assertEqual(captured["port"], 8088)
+        self.assertIn("1 removed", captured["body"])
+
+    def test_healthcheck_is_pinged_twice_when_enabled(self):
+        calls = []
+
+        def fake_ping(obj, req_obj, url):
+            calls.append(url)
+
+        with mock.patch.object(self.module, "send_ping", fake_ping):
+            self.build([torrent(tags="ipt")], use_healthcheck="true")
+        self.assertEqual(len(calls), 2)
+        self.assertTrue(calls[0].endswith("/start"))
+        self.assertFalse(calls[1].endswith("/start"))
+
+
+class TestQbtLoginFailure(QbtTestCase):
+    def test_login_error_notifies_and_continues(self):
+        client = FakeQbtClient(torrents=[torrent(tags="ipt")],
+                               login_error=FakeAPIError("bad credentials"))
+        stub = StubPushover()
+        with mock.patch.object(self.module.pushover, "Pushover",
+                               return_value=stub):
+            qbt, _ = self.build([torrent(tags="ipt")], client=client,
+                                use_pushover="true")
+        # The error notification goes out first, then the run summary.
+        self.assertEqual(len(stub.calls), 2)
+        self.assertEqual(stub.calls[0]["kwargs"]["title"],
+                         "qbit-maid API ERROR")
+        self.assertIn("bad credentials", stub.calls[0]["body"])
+        self.assertEqual(qbt.total_torrents, 1)
+
+    def test_login_error_without_pushover_does_not_crash(self):
+        client = FakeQbtClient(torrents=[torrent(tags="ipt")],
+                               login_error=FakeAPIError("bad credentials"))
+        qbt, _ = self.build([torrent(tags="ipt")], client=client)
+        self.assertEqual(qbt.total_torrents, 1)
+
+    def test_login_error_survives_a_failing_pushover(self):
+        client = FakeQbtClient(torrents=[torrent(tags="ipt")],
+                               login_error=FakeAPIError("bad credentials"))
+        with mock.patch.object(self.module.pushover, "Pushover",
+                               return_value=FailingPushover()):
+            qbt, _ = self.build([torrent(tags="ipt")], client=client,
+                                use_pushover="true")
+        self.assertEqual(qbt.total_torrents, 1)
+
+
+class TestMainGuard(QbtTestCase):
+    def test_running_as_main_executes_the_pipeline(self):
+        self.write_config()
+        client = FakeQbtClient(torrents=[torrent(tags="ipt")])
+        stub = sys.modules["qbittorrentapi"]
+        with mock.patch.object(stub, "Client", return_value=client):
+            load_qbit_maid(as_main=True)
+        self.assertTrue(client.logged_in)
+        self.assertEqual(len(client.deleted), 1)
+
+
+class TestIgnoredTagsConfig(QbtTestCase):
+    def test_ignored_tags_reads_the_ignored_tags_section(self):
+        qbt, _ = self.build([torrent(tags="ipt")])
+        self.assertEqual(qbt.ignored_tags, {"save": "save"})
+        self.assertNotIn("tracker-example", qbt.ignored_tags)
+
+    def test_ignored_tag_from_config_is_skipped(self):
+        qbt, _ = self.build([torrent(tags="save", infohash="ignored-tag")])
+        self.assertEqual(qbt.ignored_counter, 1)
+        self.assertEqual(qbt.tracker_list, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qlist.py

@@ -0,0 +1,130 @@
+"""Tests for qlist.build_tor_list — the tagging and sorting pass."""
+import unittest
+
+from qbt_fakes import FakeQbt, torrent
+from qlist import build_tor_list, torrent_tag_for
+
+
+class TestTorrentTagFor(unittest.TestCase):
+    def test_protected_tracker_gets_the_protected_tag(self):
+        self.assertEqual(
+            torrent_tag_for("https://tracker.example/a", "ipt", "public",
+                            ["tracker.example"]),
+            "ipt",
+        )
+
+    def test_unknown_tracker_gets_the_public_tag(self):
+        self.assertEqual(
+            torrent_tag_for("https://other.example/a", "ipt", "public",
+                            ["tracker.example"]),
+            "public",
+        )
+
+    def test_blank_tracker_gets_no_tag(self):
+        self.assertIsNone(
+            torrent_tag_for("", "ipt", "public", ["tracker.example"])
+        )
+
+
+class TestBuildTorList(unittest.TestCase):
+    def run_build(self, torrents, **overrides):
+        qbt = FakeQbt(torrent_list=list(torrents), **overrides).reset_counters()
+        build_tor_list(qbt)
+        return qbt
+
+    def test_sets_total_torrents(self):
+        qbt = self.run_build([torrent(infohash="a"), torrent(infohash="b")])
+        self.assertEqual(qbt.total_torrents, 2)
+
+    def test_blank_tracker_is_ignored(self):
+        qbt = self.run_build([torrent(tracker="")])
+        self.assertEqual(qbt.ignored_counter, 1)
+        self.assertEqual(qbt.tracker_list, [])
+
+    def test_ignored_category_is_ignored(self):
+        qbt = self.run_build(
+            [torrent(category="tech")],
+            cat_whitelist={"tech": "tech"},
+        )
+        self.assertEqual(qbt.ignored_counter, 1)
+        self.assertEqual(qbt.tracker_list, [])
+
+    def test_ignored_tag_is_ignored(self):
+        qbt = self.run_build(
+            [torrent(tags="save,public")],
+            ignored_tags={"save": "save"},
+        )
+        self.assertEqual(qbt.ignored_counter, 1)
+        self.assertEqual(qbt.tracker_list, [])
+
+    def test_ignored_tag_does_not_match_a_longer_tag(self):
+        qbt = self.run_build(
+            [torrent(tags="iptorrents", tracker="https://other.example/a")],
+            ignored_tags={"ipt": "ipt"},
+        )
+        self.assertEqual(qbt.ignored_counter, 0)
+        self.assertEqual(len(qbt.tracker_list), 1)
+
+    def test_blank_tag_on_protected_tracker_gets_protected_tag(self):
+        qbt = self.run_build([torrent(tags="", infohash="h1")])
+        self.assertIn(("ipt", "h1"), qbt.qbt_client.added_tags)
+        self.assertEqual(len(qbt.tracker_list), 1)
+
+    def test_blank_tag_on_non_protected_tracker_gets_public_tag(self):
+        qbt = self.run_build(
+            [torrent(tags="", tracker="https://other.example/announce", infohash="h2")]
+        )
+        self.assertIn(("public", "h2"), qbt.qbt_client.added_tags)
+        self.assertEqual(len(qbt.tracker_list), 1)
+
+    def test_blank_tag_is_applied_once_for_non_premature_torrents(self):
+        """A blank-tag torrent is tagged once, not once per code path."""
+        qbt = self.run_build([torrent(tags="", infohash="h1")])
+        self.assertEqual(qbt.qbt_client.added_tags, [("ipt", "h1")])
+
+    def test_premature_blank_tag_torrent_is_tagged_once(self):
+        """Premature torrents are tagged by the first block, then skipped."""
+        qbt = self.run_build([torrent(tags="", seeding_time=1, infohash="h3")])
+        self.assertEqual(qbt.qbt_client.added_tags, [("ipt", "h3")])
+        self.assertEqual(qbt.preme_tor_counter, 1)
+        self.assertEqual(qbt.tracker_list, [])
+
+    def test_premature_torrent_is_counted_and_not_queued(self):
+        qbt = self.run_build([torrent(seeding_time=10, tags="ipt")])
+        self.assertEqual(qbt.preme_tor_counter, 1)
+        self.assertEqual(qbt.tracker_list, [])
+
+    def test_protected_torrent_with_tag_is_queued_without_retagging(self):
+        qbt = self.run_build([torrent(tags="ipt")])
+        self.assertEqual(qbt.qbt_client.added_tags, [])
+        self.assertEqual(len(qbt.tracker_list), 1)
+
+    def test_non_protected_torrent_with_tag_is_queued_without_retagging(self):
+        qbt = self.run_build(
+            [torrent(tags="public", tracker="https://other.example/announce")]
+        )
+        self.assertEqual(qbt.qbt_client.added_tags, [])
+        self.assertEqual(len(qbt.tracker_list), 1)
+
+    def test_mixed_batch_counts_each_bucket(self):
+        qbt = self.run_build([
+            torrent(tracker=""),                                   # ignored
+            torrent(category="tech"),                              # ignored
+            torrent(tags="save"),                                  # ignored
+            torrent(seeding_time=1, tags="ipt"),                   # premature
+            torrent(tags="ipt", infohash="p1"),                    # queued
+            torrent(tags="public", tracker="https://o.example/a",
+                    infohash="n1"),                                # queued
+        ], cat_whitelist={"tech": "tech"}, ignored_tags={"save": "save"})
+        self.assertEqual(qbt.ignored_counter, 3)
+        self.assertEqual(qbt.preme_tor_counter, 1)
+        self.assertEqual(len(qbt.tracker_list), 2)
+        self.assertEqual(qbt.total_torrents, 6)
+
+    def test_consumes_the_input_list(self):
+        qbt = self.run_build([torrent(tags="ipt")])
+        self.assertEqual(qbt.torrent_list, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qlogging_extra.py

@@ -0,0 +1,322 @@
+"""Tests for the qlogging helpers that the notifier tests do not reach."""
+import datetime
+import logging
+import os
+import shutil
+import tempfile
+import unittest
+from collections import Counter
+
+from qbt_fakes import FailingPushover, FakeQbt, StubPushover, torrent
+from qlogging import (
+    LOGGER_NAME,
+    debug_torrent_list,
+    format_runtime,
+    get_script_runtime,
+    list_first_tor,
+    list_qbit_api_info,
+    log_torrent,
+    print_processor,
+    process_counts,
+    send_ping,
+    setup_logging,
+    summary_runtime,
+    summary_stats,
+    tor_notify,
+    tor_notify_error,
+    tor_notify_summary,
+    tornotifytest,
+    torrent_count,
+)
+
+
+class FakePushoverModule:
+    """Stands in for the vendored ``pushover`` module."""
+
+    def __init__(self):
+        self.created = []
+
+    def Pushover(self, token):  # noqa: N802 - mirrors the library's name
+        self.created.append(token)
+        return StubPushover()
+
+
+class FailingRequests:
+    class RequestException(Exception):
+        pass
+
+    def __init__(self):
+        self.calls = []
+
+    def get(self, url, timeout=None):
+        self.calls.append(url)
+        raise self.RequestException("boom")
+
+
+class TestNotifySetup(unittest.TestCase):
+    def test_tor_notify_creates_client_when_enabled(self):
+        qbt = FakeQbt(use_pushover=True, po_token="TOKEN")
+        qbt.po = FakePushoverModule()
+        tor_notify(qbt)
+        self.assertEqual(qbt.po.created, ["TOKEN"])
+        self.assertIsInstance(qbt.poc, StubPushover)
+
+    def test_tor_notify_is_a_noop_when_disabled(self):
+        qbt = FakeQbt(use_pushover=False)
+        qbt.po = FakePushoverModule()
+        tor_notify(qbt)
+        self.assertEqual(qbt.po.created, [])
+
+    def test_tornotifytest_sends_a_test_message(self):
+        qbt = FakeQbt()
+        tornotifytest(qbt)
+        self.assertEqual(len(qbt.poc.calls), 1)
+        self.assertEqual(qbt.poc.calls[0]["user"], "USERKEY")
+        self.assertEqual(qbt.poc.calls[0]["body"], "Test Message")
+        self.assertEqual(qbt.poc.calls[0]["kwargs"]["title"], "qbit-maid")
+
+
+class TestTorNotifyError(unittest.TestCase):
+    def test_sends_through_the_pushover_client(self):
+        qbt = FakeQbt(use_pushover=True)
+        tor_notify_error(qbt, RuntimeError("boom"))
+        self.assertEqual(len(qbt.poc.calls), 1)
+        self.assertEqual(qbt.poc.calls[0]["user"], "USERKEY")
+        self.assertIn("boom", qbt.poc.calls[0]["body"])
+        self.assertEqual(qbt.poc.calls[0]["kwargs"]["title"],
+                         "qbit-maid API ERROR")
+
+    def test_skips_pushover_when_disabled(self):
+        qbt = FakeQbt(use_pushover=False)
+        tor_notify_error(qbt, RuntimeError("boom"))
+        self.assertEqual(qbt.poc.calls, [])
+
+    def test_skips_pushover_when_no_client_was_built(self):
+        qbt = FakeQbt(use_pushover=True, poc=None)
+        tor_notify_error(qbt, RuntimeError("boom"))  # must not raise
+
+    def test_survives_a_failing_pushover(self):
+        qbt = FakeQbt(use_pushover=True, poc=FailingPushover())
+        tor_notify_error(qbt, RuntimeError("boom"))  # must not raise
+
+    def test_accepts_a_custom_title(self):
+        qbt = FakeQbt(use_pushover=True)
+        tor_notify_error(qbt, RuntimeError("boom"), title="Custom")
+        self.assertEqual(qbt.poc.calls[0]["kwargs"]["title"], "Custom")
+
+
+class TestCounters(unittest.TestCase):
+    def test_process_counts_buckets_by_tag(self):
+        qbt = FakeQbt(
+            ct=Counter,
+            tracker_list=[torrent(tags="ipt"), torrent(tags="ipt"), torrent(tags="public")],
+        )
+        process_counts(qbt)
+        self.assertEqual(qbt.c["ipt"], 2)
+        self.assertEqual(qbt.c["public"], 1)
+
+    def test_print_processor_logs_every_counter(self):
+        qbt = FakeQbt()
+        with self.assertLogs("qbit-maid-test", level="INFO") as captured:
+            print_processor(qbt)
+        joined = "\n".join(captured.output)
+        for label in ("Total", "Premature", "Ignored", "Protected",
+                      "Non-protected", "Orphaned", "Marked for deletion"):
+            self.assertIn(label, joined)
+
+    def test_torrent_count_logs_both_tags(self):
+        qbt = FakeQbt()
+        with self.assertLogs("qbit-maid-test", level="DEBUG") as captured:
+            torrent_count(qbt)
+        joined = "\n".join(captured.output)
+        self.assertIn('tag["ipt"] 10', joined)
+        self.assertIn('tag["public"] 20', joined)
+
+
+class TestDebugHelpers(unittest.TestCase):
+    def test_list_first_tor_logs_each_field(self):
+        qbt = FakeQbt(torrent_list=[torrent(name="First")])
+        with self.assertLogs("qbit-maid-test", level="DEBUG") as captured:
+            list_first_tor(qbt)
+        self.assertIn("First", "\n".join(captured.output))
+
+    def test_list_first_tor_honours_the_index(self):
+        qbt = FakeQbt(torrent_list=[torrent(name="Zero"), torrent(name="One")])
+        with self.assertLogs("qbit-maid-test", level="DEBUG") as captured:
+            list_first_tor(qbt, index=1)
+        self.assertIn("One", "\n".join(captured.output))
+
+    def test_list_first_tor_handles_an_empty_list(self):
+        qbt = FakeQbt(torrent_list=[])
+        with self.assertLogs("qbit-maid-test", level="DEBUG") as captured:
+            list_first_tor(qbt)
+        self.assertIn("No torrents found", "\n".join(captured.output))
+
+    def test_list_qbit_api_info_logs_versions(self):
+        qbt = FakeQbt()
+        with self.assertLogs("qbit-maid-test", level="DEBUG") as captured:
+            list_qbit_api_info(qbt)
+        joined = "\n".join(captured.output)
+        self.assertIn("v4.6.2", joined)
+        self.assertIn("2.9.3", joined)
+
+    def test_debug_torrent_list_logs_the_list(self):
+        qbt = FakeQbt(torrent_list=[torrent(name="Logged")])
+        with self.assertLogs("qbit-maid-test", level="DEBUG") as captured:
+            debug_torrent_list(qbt)
+        self.assertIn("Logged", "\n".join(captured.output))
+
+
+class TestRuntime(unittest.TestCase):
+    def test_get_script_runtime_sets_extm_for_pushover(self):
+        qbt = FakeQbt(use_pushover=True, use_apprise=False)
+        get_script_runtime(qbt)
+        self.assertEqual(qbt.extm, "Execution time: [0:00:12]")
+
+    def test_get_script_runtime_sets_extm_for_apprise(self):
+        qbt = FakeQbt(use_pushover=False, use_apprise=True)
+        get_script_runtime(qbt)
+        self.assertEqual(qbt.extm, "Execution time: [0:00:12]")
+
+    def test_get_script_runtime_without_notifiers_leaves_extm_alone(self):
+        qbt = FakeQbt(use_pushover=False, use_apprise=False, extm="")
+        get_script_runtime(qbt)
+        self.assertEqual(qbt.extm, "")
+
+    def test_format_runtime_falls_back_to_str_for_non_timedelta(self):
+        self.assertEqual(format_runtime(12), "12")
+        self.assertEqual(format_runtime("n/a"), "n/a")
+
+    def test_summary_runtime_falls_back_to_extm(self):
+        class OnlyExtm:
+            extm = "Execution time: [5s]"
+
+        self.assertEqual(summary_runtime(OnlyExtm()), "5s")
+
+    def test_summary_runtime_returns_unknown_without_any_source(self):
+        class Nothing:
+            pass
+
+        self.assertEqual(summary_runtime(Nothing()), "unknown")
+
+
+class TestSummaryStatsEdgeCases(unittest.TestCase):
+    def test_missing_tag_returns_zero(self):
+        qbt = FakeQbt(c={})
+        stats = summary_stats(qbt)
+        self.assertEqual(stats["protected"], 0)
+        self.assertEqual(stats["non_protected"], 0)
+
+    def test_unindexable_counter_returns_zero(self):
+        qbt = FakeQbt(c=[])
+        stats = summary_stats(qbt)
+        self.assertEqual(stats["protected"], 0)
+
+    def test_blank_tag_returns_zero(self):
+        qbt = FakeQbt(tracker_protected_tag="", tracker_non_protected_tag="")
+        stats = summary_stats(qbt)
+        self.assertEqual(stats["protected"], 0)
+        self.assertEqual(stats["non_protected"], 0)
+
+
+class TestSendPing(unittest.TestCase):
+    def test_send_ping_swallows_request_errors(self):
+        qbt = FakeQbt()
+        failing = FailingRequests()
+        send_ping(qbt, failing, "https://example.com/ping/x")  # must not raise
+        self.assertEqual(failing.calls, ["https://example.com/ping/x"])
+
+    def test_send_ping_logs_the_failure(self):
+        qbt = FakeQbt()
+        with self.assertLogs("qbit-maid-test", level="INFO") as captured:
+            send_ping(qbt, FailingRequests(), "https://example.com/ping/x")
+        self.assertIn("Ping failed", "\n".join(captured.output))
+
+
+class TestSetupLogging(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self.log_path = os.path.join(self.tmp, "qc.log")
+        self.addCleanup(self._reset_logger)
+
+    def _reset_logger(self):
+        logger = logging.getLogger(LOGGER_NAME)
+        for handler in list(logger.handlers):
+            logger.removeHandler(handler)
+            handler.close()
+
+    def test_disabled_logging_uses_a_null_handler(self):
+        logger = setup_logging(False, self.log_path, "INFO")
+        self.assertEqual(logger.level, logging.CRITICAL)
+        self.assertEqual(len(logger.handlers), 1)
+        self.assertIsInstance(logger.handlers[0], logging.NullHandler)
+        self.assertFalse(os.path.exists(self.log_path))
+
+    def test_enabled_logging_writes_to_file_and_console(self):
+        logger = setup_logging(True, self.log_path, "DEBUG")
+        self.assertEqual(logger.level, logging.DEBUG)
+        self.assertEqual(len(logger.handlers), 2)
+        logger.info("hello from the test")
+        for handler in logger.handlers:
+            handler.flush()
+        with open(self.log_path, encoding="utf-8") as fh:
+            self.assertIn("hello from the test", fh.read())
+
+    def test_unknown_level_falls_back_to_info(self):
+        logger = setup_logging(True, self.log_path, "shouty")
+        self.assertEqual(logger.level, logging.INFO)
+
+    def test_setup_is_idempotent(self):
+        first = setup_logging(True, self.log_path, "INFO")
+        previous_file_handler = first.handlers[0]
+        self.assertIsNotNone(previous_file_handler.stream)
+        logger = setup_logging(True, self.log_path, "INFO")
+        self.assertEqual(len(logger.handlers), 2)
+        # The replaced handler is closed, so its file descriptor is released.
+        self.assertIsNone(previous_file_handler.stream)
+
+    def test_rotation_parameters_are_applied(self):
+        logger = setup_logging(True, self.log_path, "INFO",
+                               max_bytes=1234, backup_count=7)
+        file_handler = logger.handlers[0]
+        self.assertEqual(file_handler.maxBytes, 1234)
+        self.assertEqual(file_handler.backupCount, 7)
+
+
+class TestLogTorrent(unittest.TestCase):
+    def test_log_torrent_attaches_structured_fields(self):
+        logger = logging.getLogger("qbit-maid-test")
+        previous_level = logger.level
+        logger.setLevel(logging.DEBUG)
+        self.addCleanup(logger.setLevel, previous_level)
+        record = {}
+
+        class Capture(logging.Handler):
+            def emit(self, rec):
+                record.update(rec.__dict__)
+
+        handler = Capture()
+        logger.addHandler(handler)
+        self.addCleanup(logger.removeHandler, handler)
+
+        log_torrent(logger, logging.INFO, torrent(name="A" * 40, infohash="deadbeef"),
+                    message="processing")
+        self.assertEqual(record["torrent_hash"], "deadbeef")
+        self.assertEqual(record["torrent_name"], "A" * 20)
+        self.assertEqual(record["msg"], "processing")
+
+    def test_log_torrent_tolerates_missing_fields(self):
+        logger = logging.getLogger("qbit-maid-test")
+        log_torrent(logger, logging.INFO, {})  # must not raise
+
+
+class TestNotifySummarySound(unittest.TestCase):
+    def test_sound_is_passed_through(self):
+        qbt = FakeQbt(config={"pushover": {"sound": "siren"}})
+        tor_notify_summary(qbt)
+        self.assertEqual(qbt.poc.calls[0]["kwargs"]["sound"], "siren")
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qprocess.py

@@ -0,0 +1,103 @@
+"""Tests for qprocess — the deletion decision pass and the delete calls."""
+import csv
+import os
+import shutil
+import tempfile
+import unittest
+
+from qbt_fakes import FakeQbt, torrent
+from qprocess import tor_delete, tor_delete_tags, tor_processor
+
+
+class TestTorProcessor(unittest.TestCase):
+    def run_processor(self, candidates, **overrides):
+        qbt = FakeQbt(tracker_list=list(candidates), cv=csv, **overrides).reset_counters()
+        tor_processor(qbt)
+        return qbt
+
+    def test_downloading_torrent_is_skipped(self):
+        qbt = self.run_processor([torrent(state="downloading", tags="public")])
+        self.assertEqual(qbt.torrent_hash_delete_list, [])
+        self.assertEqual(qbt.up_tor_counter, 0)
+
+    def test_protected_under_ratio_and_old_is_deleted(self):
+        qbt = self.run_processor([
+            torrent(tags="ipt", ratio=0.5, time_active=9999999, infohash="old1"),
+        ])
+        self.assertEqual(qbt.torrent_hash_delete_list, ["old1"])
+
+    def test_protected_under_ratio_but_young_is_kept(self):
+        qbt = self.run_processor([
+            torrent(tags="ipt", ratio=0.5, time_active=1, infohash="young1"),
+        ])
+        self.assertEqual(qbt.torrent_hash_delete_list, [])
+        self.assertEqual(qbt.up_tor_counter, 0)
+
+    def test_protected_over_ratio_is_deleted(self):
+        qbt = self.run_processor([
+            torrent(tags="ipt", ratio=2.0, infohash="over1"),
+        ])
+        self.assertEqual(qbt.torrent_hash_delete_list, ["over1"])
+
+    def test_non_protected_is_deleted(self):
+        qbt = self.run_processor([
+            torrent(tags="public", ratio=0.1, infohash="np1"),
+        ])
+        self.assertEqual(qbt.torrent_hash_delete_list, ["np1"])
+
+    def test_unmatched_torrent_is_orphaned(self):
+        qbt = self.run_processor([torrent(tags="other", ratio=0.1)])
+        self.assertEqual(qbt.torrent_hash_delete_list, [])
+        self.assertEqual(qbt.up_tor_counter, 1)
+
+    def test_telemetry_writes_a_row_per_candidate(self):
+        tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, tmp, ignore_errors=True)
+        outfile = os.path.join(tmp, "telemetry.csv")
+        qbt = self.run_processor(
+            [torrent(tags="public", infohash="t1"), torrent(tags="public", infohash="t2")],
+            enable_telemetry=True,
+            telemetry_outfile=outfile,
+        )
+        with open(outfile, newline="", encoding="UTF8") as fh:
+            rows = list(csv.reader(fh))
+        self.assertEqual(rows[0], ["state", "ratio", "tags", "added", "hash", "name", "tracker"])
+        self.assertEqual(len(rows), 3)  # header + 2 candidates
+        self.assertEqual(qbt.torrent_hash_delete_list, ["t1", "t2"])
+
+    def test_dragnet_writes_orphans_only(self):
+        tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, tmp, ignore_errors=True)
+        outfile = os.path.join(tmp, "orphaned.csv")
+        qbt = self.run_processor(
+            [torrent(tags="other", infohash="orph1"), torrent(tags="public", infohash="del1")],
+            enable_dragnet=True,
+            dragnet_outfile=outfile,
+        )
+        with open(outfile, newline="", encoding="UTF8") as fh:
+            rows = list(csv.reader(fh))
+        self.assertEqual(rows[0], ["state", "ratio", "tags", "added", "thash", "tname", "trname"])
+        self.assertEqual(len(rows), 2)  # header + the single orphan
+        self.assertEqual(rows[1][4], "orph1")
+        self.assertEqual(qbt.up_tor_counter, 1)
+
+
+class TestDeleteCalls(unittest.TestCase):
+    def test_delete_tags_removes_both_tags(self):
+        qbt = FakeQbt()
+        tor_delete_tags(qbt)
+        self.assertEqual(qbt.qbt_client.deleted_tags, [["ipt", "public"]])
+
+    def test_delete_removes_files_for_each_hash(self):
+        qbt = FakeQbt(torrent_hash_delete_list=["h1", "h2"])
+        tor_delete(qbt)
+        self.assertEqual(qbt.qbt_client.deleted, [(True, ["h1", "h2"])])
+
+    def test_delete_is_a_noop_on_empty_list(self):
+        qbt = FakeQbt(torrent_hash_delete_list=[])
+        tor_delete(qbt)
+        self.assertEqual(qbt.qbt_client.deleted, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

test_write_csv.py

@@ -1,41 +1,50 @@
-from qprocess import write_csv
+"""Tests for qprocess.write_csv — header-once, append semantics."""
-import csv
+import csv
-import unittest
+import os
-import os
+import shutil
-# Torrent Items needed for testing
+import tempfile
-# canidate['state'] {canidate["ratio"]} {torrent["tags"]} torrent["seeding_time"] {torrent["category"]} {torrent["name"][0:20]} {canidate["time_active"]}
+import unittest
-class TestWriteCSV(unittest.TestCase):
+
-    def test_write_csv_dragnet(self):
+from qprocess import write_csv
-        self.cv = csv
+
-        outfile = './test_dragnet_outfile.csv'
+HEADER = ["state", "ratio", "tags", "added", "hash", "name", "tracker"]
-        state = 'downloading'
+ROW = ["downloading", 1.05, "ipt", 1, "asfasdf23412adfqwer",
-        ratio = 1.05
+       "thisismynamehahahah", "https://localhost.stackoverflow.tech/announce"]
-        tags = 'ipt'
+
-        added = 1
+
-        age = 240000
+class TestWriteCSV(unittest.TestCase):
-        time = 123456
+    def setUp(self):
-        thash = 'asfasdf23412adfqwer'
+        self.tmp = tempfile.mkdtemp()
-        tname = 'thisismynamehahahah'
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
-        trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
+        self.outfile = os.path.join(self.tmp, "out.csv")
-        header = ['state','ratio','tags','added','age','time','thash','tname','trname']
+
-        row = [state,ratio,tags,added,age,time,thash,tname,trname]
+    def read_rows(self):
-        write_csv(self.cv,outfile,header,row)
+        with open(self.outfile, newline="", encoding="UTF8") as fh:
-        self.assertTrue(os.path.exists(outfile))
+            return list(csv.reader(fh))
-
+
-    def test_write_csv_telemetry(self):
+    def test_writes_header_then_row(self):
-        self.cv = csv
+        write_csv(csv, self.outfile, HEADER, ROW)
-        outfile = './test_telemetry_outfile.csv'
+        rows = self.read_rows()
-        state = 'downloading'
+        self.assertEqual(rows[0], HEADER)
-        ratio = 1.05
+        self.assertEqual(len(rows), 2)
-        tags = 'ipt'
+        self.assertEqual(rows[1][4], "asfasdf23412adfqwer")
-        added = 1
+
-        thash = 'asfasdf23412adfqwer'
+    def test_header_is_written_only_once(self):
-        tname = 'thisismynamehahahah'
+        write_csv(csv, self.outfile, HEADER, ROW)
-        trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
+        write_csv(csv, self.outfile, HEADER, ROW)
-        header = ['state','ratio','tags','added','hash','name','tracker']
+        rows = self.read_rows()
-        row = [state,ratio,tags,added,thash,tname,trname]
+        self.assertEqual(rows.count(HEADER), 1)
-        write_csv(self.cv,outfile,header,row)
+        self.assertEqual(len(rows), 3)
-        self.assertTrue(os.path.exists(outfile))
+
-
+    def test_creates_the_file_when_missing(self):
-if __name__ == '__main__':
+        self.assertFalse(os.path.exists(self.outfile))
-    unittest.main()
+        write_csv(csv, self.outfile, HEADER, ROW)
+        self.assertTrue(os.path.exists(self.outfile))
+
+    def test_values_are_round_tripped(self):
+        write_csv(csv, self.outfile, HEADER, ROW)
+        self.assertEqual(self.read_rows()[1], [str(value) for value in ROW])
+
+
+if __name__ == "__main__":
+    unittest.main()