.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:

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

test_apprise.py

@@ -0,0 +1,173 @@
+"""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):
+        # The DOCKER branch reads five env vars before toml_path is reached, so
+        # all of them must be present for the toml_path branch to run at all.
+        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_ignored_without_docker(self):
+        """KNOWN BUG: the bare ``except`` swallows the DOCKER KeyError and
+        skips the rest of the try block, so ``toml_path`` is never read unless
+        ``DOCKER`` is also set. With no ./config.toml either, the client then
+        crashes on the missing ``host`` attribute.
+        """
+        os.environ["toml_path"] = self._write_elsewhere_config()
+        with self.assertRaises(AttributeError):
+            AppriseClient()
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qbitmaid.py

@@ -1,167 +1,142 @@
 import datetime
-import logging
 import unittest
-from collections import Counter
 
-import requests as r
-from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag
-from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor
+from qbt_fakes import FailingPushover, FakeQbt, StubPushover
+from qlist import (
+    is_cat_ignored,
+    is_ignored_tag,
+    is_not_protected_tracker,
+    is_preme,
+    is_protected_tracker,
+    is_tag_blank,
+    is_tracker_blank,
+)
 from qlogging import (
-    send_ping,
     build_summary,
     format_runtime,
     get_notify_options,
+    send_ping,
     summary_stats,
     tor_notify_apprise,
     tor_notify_summary,
 )
+from qprocess import (
+    is_downloading,
+    is_not_protected_tor,
+    is_old_tor,
+    is_protected_over_ratio,
+    is_protected_under_ratio,
+)
 
 
-class FakeQbt:
-    """Minimal stand-in for the Qbt object, holding only what notifiers read."""
+class RecordingRequests:
+    """Stands in for the requests module without touching the network."""
 
-    def __init__(self, **overrides):
-        self.config = {"pushover": {}}
-        self.total_torrents = 42
-        self.torrent_hash_delete_list = ["a"] * 7
-        self.up_tor_counter = 4
-        self.ignored_counter = 5
-        self.preme_tor_counter = 3
-        self.tracker_protected_tag = "ipt"
-        self.tracker_non_protected_tag = "public"
-        self.c = Counter({"ipt": 10, "public": 20})
-        self.delete_torrents = True
-        self.st = datetime.datetime(2026, 10, 27, 9, 54, 0)
-        self.et = datetime.datetime(2026, 10, 27, 9, 54, 12)
-        self.po_key = "USERKEY"
-        self.poc = StubPushover()
-        self.apprise_host = "localhost"
-        self.apprise_port = 8088
-        self.apprise_aurls = "mailto://user:pass@example.com"
-        self.tl = logging.getLogger("qbit-maid-test")
-        for key, value in overrides.items():
-            setattr(self, key, value)
-
-
-class StubPushover:
-    """Records message() calls instead of hitting the Pushover API."""
+    class RequestException(Exception):
+        pass
 
     def __init__(self):
         self.calls = []
+        self.timeouts = []
 
-    def message(self, user, body, **kwargs):
-        self.calls.append({"user": user, "body": body, "kwargs": kwargs})
+    def get(self, url, timeout=None):
+        self.calls.append(url)
+        self.timeouts.append(timeout)
+        return "ok"
 
 
-class FailingPushover:
-    def message(self, *args, **kwargs):
-        raise RuntimeError("pushover is down")
+class TestPredicates(unittest.TestCase):
+    def test_is_preme(self):
+        self.assertTrue(is_preme(1, 1))  # equal to min_age counts as premature
+        self.assertTrue(is_preme(1, 2))
+        self.assertFalse(is_preme(2, 1))
 
-class TestQbitmaid(unittest.TestCase):
-    def test_ispreme_sanity(self):
-        self.assertTrue(is_preme(1,1))
-        self.assertTrue(is_preme(1,2))
-        self.assertFalse(is_preme(2,1))
+    def test_is_cat_ignored(self):
+        cats = ["a", "b", "c"]
+        for cat in cats:
+            self.assertTrue(is_cat_ignored(cat, cats))
+        self.assertFalse(is_cat_ignored("d", cats))
+        self.assertFalse(is_cat_ignored(1, cats))
+        self.assertFalse(is_cat_ignored(1.0, cats))
 
-    def test_ispreme(self):
-        pass
-
-    def test_iscatignored_sanity(self):
-        self.assertTrue(is_cat_ignored('a', ['a','b','c']))
-        self.assertTrue(is_cat_ignored('b', ['a','b','c']))
-        self.assertTrue(is_cat_ignored('c', ['a','b','c']))
-        self.assertFalse(is_cat_ignored('d', ['a','b','c']))
-        self.assertFalse(is_cat_ignored(1, ['a','b','d']))
-        self.assertFalse(is_cat_ignored(1.0000000, ['a','b','c']))
-
-    def test_iscatignored(self):
-        pass
-
-    def test_istrackerblank_sanity(self):
-        self.assertTrue(is_tracker_blank(''))
-        self.assertFalse(is_tracker_blank('a'))
+    def test_is_tracker_blank(self):
+        self.assertTrue(is_tracker_blank(""))
+        self.assertFalse(is_tracker_blank("a"))
         self.assertFalse(is_tracker_blank(1))
-        self.assertFalse(is_tracker_blank(1.00000000))
+        self.assertFalse(is_tracker_blank(1.0))
 
-    def test_istrackerblank(self):
-        pass
+    def test_is_protected_tracker(self):
+        whitelist = ["a.com", "b.me", "c.io"]
+        self.assertTrue(is_protected_tracker("https://a.com/", whitelist))
+        self.assertFalse(is_protected_tracker("https://google.com/", whitelist))
+        self.assertFalse(is_protected_tracker("https://d.com", whitelist))
+        self.assertFalse(is_protected_tracker("", whitelist))
 
-    def test_isprotectedtracker_sanity(self):
-        self.assertTrue(is_protected_tracker('https://a.com/',['a.com','b.me','c.io']))
-        self.assertFalse(is_protected_tracker('https://google.com/',['a.com','b.me','c.io']))
-        self.assertFalse(is_protected_tracker('https://d.com',['a.com','b.me','c.io']))
+    def test_is_not_protected_tracker(self):
+        whitelist = ["a.com", "b.me", "c.io"]
+        self.assertFalse(is_not_protected_tracker("https://a.com/", whitelist))
+        self.assertTrue(is_not_protected_tracker("https://google.com/", whitelist))
+        self.assertTrue(is_not_protected_tracker("https://d.com", whitelist))
+        self.assertFalse(is_not_protected_tracker("", whitelist))
 
-    def test_isprotectedtracker(self):
-        pass
+    def test_is_tag_blank(self):
+        self.assertTrue(is_tag_blank(""))
+        for value in ("a", 1, 1.0001, False, True):
+            self.assertFalse(is_tag_blank(value))
 
-    def test_isnotprotectedtracker_sanity(self):
-        self.assertFalse(is_not_protected_tracker('https://a.com/',['a.com','b.me','c.io']))
-        self.assertTrue(is_not_protected_tracker('https://google.com/',['a.com','b.me','c.io']))
-        self.assertTrue(is_not_protected_tracker('https://d.com',['a.com','b.me','c.io']))
+    def test_is_downloading(self):
+        self.assertTrue(is_downloading("downloading"))
+        self.assertFalse(is_downloading("DOWNLOADING"))
+        self.assertFalse(is_downloading("dOwNlOaDiNg"))
+        self.assertFalse(is_downloading("uploading"))
 
-    def test_isnotprotectedtracker(self):
-        pass
+    def test_is_protected_under_ratio(self):
+        self.assertTrue(is_protected_under_ratio(0.5, 1, "a", "a,b,c"))
+        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_istagblank(self):
-        self.assertTrue(is_tag_blank(''))
-        self.assertFalse(is_tag_blank('a'))
-        self.assertFalse(is_tag_blank(1))
-        self.assertFalse(is_tag_blank(1.0001))
-        self.assertFalse(is_tag_blank(False))
-        self.assertFalse(is_tag_blank(True))
+    def test_is_old_tor(self):
+        self.assertTrue(is_old_tor(1, 1))
+        self.assertTrue(is_old_tor(2, 1))
+        self.assertFalse(is_old_tor(1, 2))
 
-    def test_isdownloading_sanity(self):
-        self.assertTrue(is_downloading('downloading'))
+    def test_is_protected_over_ratio(self):
+        self.assertTrue(is_protected_over_ratio(2, 1, "a", "a,b,c"))
+        self.assertFalse(is_protected_over_ratio(0.5, 1, "a", "a,b,c"))
+        self.assertFalse(is_protected_over_ratio(2, 1, "z", "a,b,c"))
 
-    def test_isdownloading(self):
-        self.assertFalse(is_downloading('DOWNLOADING'))
-        self.assertFalse(is_downloading('dOwNlOaDiNg'))
+    def test_is_not_protected_tor(self):
+        self.assertTrue(is_not_protected_tor("a", "a,b,c"))
+        self.assertFalse(is_not_protected_tor("z", "a,b,c"))
 
-    def test_isprotectedunderratio_sanity(self):
-        self.assertTrue(is_protected_under_ratio(0.5,1,'a','a,b,c'))
+    def test_is_ignored_tag(self):
+        self.assertTrue(is_ignored_tag(["save"], "save,public,ipt"))
+        self.assertFalse(is_ignored_tag(["save"], "public,ipt"))
 
-    def test_isprotectedunderratio(self):
-        pass
-
-    def test_isoldtor_sanity(self):
-        self.assertFalse(is_old_tor(1,2))
-
-    def test_isoldtor(self):
-        self.assertTrue(is_old_tor(1,1))
-        self.assertTrue(is_old_tor(2,1))
-        self.assertFalse(is_old_tor(1,2))
-
-    def test_isprotectedoverratio_sanity(self):
-        self.assertTrue(is_protected_over_ratio(2,1,'a','a,b,c'))
-
-    def test_isprotectedoverratio(self):
-        pass
-
-    def test_isnonprotectedtor_sanity(self):
-        self.assertTrue(is_not_protected_tor('a','a,b,c'))
-
-    def test_isnonprotectedtor(self):
-        pass
-
-    def test_isignoredtag_sanity(self):
-        self.assertTrue(is_ignored_tag(['a','b','c'], 'first,second,third,a'))
-    
-    def test_isignoredtag(self):
-        self.assertTrue(is_ignored_tag(['save'], 'save,public,ipt'))
-
-    def test_sendpingstart_sanity(self):
-        send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb" + "/start")
-        url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb"
-        send_ping(self, r, url)
-
-    def test_sendping_start(self):
-        url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb/"
-        send_ping(self, r, url.strip("/") + "/start")
-        send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb")
+    def test_is_ignored_tag_matches_substrings_not_tags(self):
+        """KNOWN BUG: matching is substring-based, so a short tag matches any
+        longer tag that contains it. 'ipt' matches 'iptorrents' and 'save'
+        matches 'unsaved', which silently ignores unrelated torrents.
+        """
+        self.assertTrue(is_ignored_tag(["ipt"], "iptorrents"))
+        self.assertTrue(is_ignored_tag(["save"], "unsaved"))
 
 
-    # --- notification formatting -------------------------------------------
+class TestSendPing(unittest.TestCase):
+    def test_send_ping_requests_the_url(self):
+        req = RecordingRequests()
+        send_ping(FakeQbt(), req, "https://example.com/ping/x")
+        self.assertEqual(req.calls, ["https://example.com/ping/x"])
 
+    def test_send_ping_uses_a_five_second_timeout(self):
+        req = RecordingRequests()
+        send_ping(FakeQbt(), req, "https://example.com/ping/x")
+        self.assertEqual(req.timeouts, [5])
+
+
+# --- notification formatting -----------------------------------------------
+
+class TestFormatRuntime(unittest.TestCase):
     def test_format_runtime(self):
         self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s")
         self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s")
@@ -169,6 +144,8 @@ class TestQbitmaid(unittest.TestCase):
         self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s")
         self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s")
 
+
+class TestSummaryStats(unittest.TestCase):
     def test_summary_stats(self):
         stats = summary_stats(FakeQbt())
         self.assertEqual(stats["total"], 42)
@@ -189,6 +166,8 @@ class TestQbitmaid(unittest.TestCase):
         self.assertEqual(stats["removed"], 0)
         self.assertEqual(stats["runtime"], "unknown")
 
+
+class TestBuildSummary(unittest.TestCase):
     def test_build_summary_plain(self):
         title, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
         self.assertEqual(title, "qbit-maid: 7 torrents removed")
@@ -295,10 +274,16 @@ class TestQbitmaid(unittest.TestCase):
         _, body = build_summary(FakeQbt(), include_hostname=True)
         self.assertTrue(body.startswith("Host: "))
 
+    def test_build_summary_hostname_html(self):
+        _, body = build_summary(FakeQbt(), style="html", include_hostname=True)
+        self.assertTrue(body.startswith("<b>Host:</b> "))
+
     def test_build_summary_unknown_style_falls_back(self):
         _, body = build_summary(FakeQbt(), style="markdown")
         self.assertNotIn("<b>", body)
 
+
+class TestNotifyOptions(unittest.TestCase):
     def test_get_notify_options_defaults(self):
         opts = get_notify_options(FakeQbt())
         self.assertEqual(opts["style"], "plain")
@@ -337,6 +322,8 @@ class TestQbitmaid(unittest.TestCase):
         fake = FakeQbt(config={"pushover": {"format": "bogus"}})
         self.assertEqual(get_notify_options(fake)["style"], "plain")
 
+
+class TestTorNotifySummary(unittest.TestCase):
     def test_tor_notify_summary_plain_payload(self):
         fake = FakeQbt()
         tor_notify_summary(fake)
@@ -387,6 +374,8 @@ class TestQbitmaid(unittest.TestCase):
         fake.poc = FailingPushover()
         tor_notify_summary(fake)  # must not raise
 
+
+class TestTorNotifyApprise(unittest.TestCase):
     def test_tor_notify_apprise_uses_plain_body(self):
         captured = {}
 
@@ -410,11 +399,6 @@ class TestQbitmaid(unittest.TestCase):
         self.assertEqual(fake.poc.calls[0]["body"], captured["body"])
         self.assertEqual(fake.poc.calls[0]["kwargs"]["title"], captured["title"])
 
-    # def test__sanity(self):
-    #     pass
 
-    # def test_(self):
-    #     pass
-        
-if __name__ == '__main__':
+if __name__ == "__main__":
     unittest.main()

test_qbt.py

@@ -0,0 +1,326 @@
+"""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 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_crashes(self):
+        """KNOWN BUG: list_first_tor() indexes torrent_list[0] unconditionally,
+        so a qBittorrent instance with zero torrents raises IndexError and the
+        whole run aborts before any summary is sent.
+        """
+        with self.assertRaises(IndexError):
+            self.build([])
+
+
+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"))
+        with mock.patch.object(self.module.pushover, "send_message",
+                               create=True) as send_message:
+            qbt, _ = self.build([torrent(tags="ipt")], client=client)
+        send_message.assert_called_once()
+        self.assertEqual(send_message.call_args.kwargs["title"],
+                         "qbit-maid API ERROR")
+        self.assertEqual(qbt.total_torrents, 1)
+
+    def test_login_error_crashes_without_a_send_message_helper(self):
+        """KNOWN BUG: the vendored pushover module has no module-level
+        ``send_message``, so the error path raises AttributeError instead of
+        notifying anyone.
+        """
+        client = FakeQbtClient(torrents=[torrent(tags="ipt")],
+                               login_error=FakeAPIError("bad credentials"))
+        with self.assertRaises(AttributeError):
+            self.build([torrent(tags="ipt")], client=client)
+
+
+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_domains_section(self):
+        """KNOWN BUG: qbit-maid.py assigns ``ignored_tags`` from
+        ``config["ignored_domains"]``, so the ``[ignored_tags]`` section is
+        never read and tag-based ignoring silently does nothing.
+        """
+        qbt, _ = self.build([torrent(tags="ipt")])
+        self.assertEqual(qbt.ignored_tags, {"tracker-example": "tracker.example"})
+        self.assertNotIn("save", qbt.ignored_tags)
+
+
+if __name__ == "__main__":
+    unittest.main()

test_qlist.py

@@ -0,0 +1,107 @@
+"""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
+
+
+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_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_twice_for_non_premature_torrents(self):
+        """KNOWN REDUNDANCY: the tagging block runs twice for the same torrent.
+
+        Lines 24-29 tag any blank-tag torrent; lines 34-43 tag it again inside
+        the tracker branch. qBittorrent treats the second call as a no-op, so
+        the outcome is correct but one API call per new torrent is wasted.
+        Pinned here so the duplication is visible rather than silent.
+        """
+        qbt = self.run_build([torrent(tags="", infohash="h1")])
+        self.assertEqual(qbt.qbt_client.added_tags, [("ipt", "h1"), ("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,288 @@
+"""Tests for the qlogging helpers that the notifier tests do not reach."""
+import datetime
+import logging
+import os
+import shutil
+import tempfile
+import unittest
+import warnings
+from collections import Counter
+
+from qbt_fakes import 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_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 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_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):
+        setup_logging(True, self.log_path, "INFO")
+        # setup_logging drops the previous handlers without closing them, so a
+        # second call leaks the old file descriptor. Known issue; silenced here
+        # so the suite output stays readable.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", ResourceWarning)
+            logger = setup_logging(True, self.log_path, "INFO")
+        self.assertEqual(len(logger.handlers), 2)
+
+    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 unittest
 import os
-# Torrent Items needed for testing
-# canidate['state'] {canidate["ratio"]} {torrent["tags"]} torrent["seeding_time"] {torrent["category"]} {torrent["name"][0:20]} {canidate["time_active"]}
+import shutil
+import tempfile
+import unittest
+
+from qprocess import write_csv
+
+HEADER = ["state", "ratio", "tags", "added", "hash", "name", "tracker"]
+ROW = ["downloading", 1.05, "ipt", 1, "asfasdf23412adfqwer",
+       "thisismynamehahahah", "https://localhost.stackoverflow.tech/announce"]
+
+
 class TestWriteCSV(unittest.TestCase):
-    def test_write_csv_dragnet(self):
-        self.cv = csv
-        outfile = './test_dragnet_outfile.csv'
-        state = 'downloading'
-        ratio = 1.05
-        tags = 'ipt'
-        added = 1
-        age = 240000
-        time = 123456
-        thash = 'asfasdf23412adfqwer'
-        tname = 'thisismynamehahahah'
-        trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
-        header = ['state','ratio','tags','added','age','time','thash','tname','trname']
-        row = [state,ratio,tags,added,age,time,thash,tname,trname]
-        write_csv(self.cv,outfile,header,row)
-        self.assertTrue(os.path.exists(outfile))
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
+        self.outfile = os.path.join(self.tmp, "out.csv")
 
-    def test_write_csv_telemetry(self):
-        self.cv = csv
-        outfile = './test_telemetry_outfile.csv'
-        state = 'downloading'
-        ratio = 1.05
-        tags = 'ipt'
-        added = 1
-        thash = 'asfasdf23412adfqwer'
-        tname = 'thisismynamehahahah'
-        trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce'
-        header = ['state','ratio','tags','added','hash','name','tracker']
-        row = [state,ratio,tags,added,thash,tname,trname]
-        write_csv(self.cv,outfile,header,row)
-        self.assertTrue(os.path.exists(outfile))
+    def read_rows(self):
+        with open(self.outfile, newline="", encoding="UTF8") as fh:
+            return list(csv.reader(fh))
 
-if __name__ == '__main__':
+    def test_writes_header_then_row(self):
+        write_csv(csv, self.outfile, HEADER, ROW)
+        rows = self.read_rows()
+        self.assertEqual(rows[0], HEADER)
+        self.assertEqual(len(rows), 2)
+        self.assertEqual(rows[1][4], "asfasdf23412adfqwer")
+
+    def test_header_is_written_only_once(self):
+        write_csv(csv, self.outfile, HEADER, ROW)
+        write_csv(csv, self.outfile, HEADER, ROW)
+        rows = self.read_rows()
+        self.assertEqual(rows.count(HEADER), 1)
+        self.assertEqual(len(rows), 3)
+
+    def test_creates_the_file_when_missing(self):
+        self.assertFalse(os.path.exists(self.outfile))
+        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()