diff --git a/.coveragerc b/.coveragerc new file mode 100644 index 0000000..19fa0e9 --- /dev/null +++ b/.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 diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index 6e456d5..7763df3 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -8,6 +8,24 @@ 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. + run: | + python3 -m pip install --upgrade pip + python3 -m pip install 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: diff --git a/qbt_fakes.py b/qbt_fakes.py new file mode 100644 index 0000000..25aa8ba --- /dev/null +++ b/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 diff --git a/test_apprise.py b/test_apprise.py new file mode 100644 index 0000000..c404162 --- /dev/null +++ b/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() diff --git a/test_qbitmaid.py b/test_qbitmaid.py index 92a2c6f..ce91b13 100755 --- a/test_qbitmaid.py +++ b/test_qbitmaid.py @@ -1,420 +1,404 @@ -import datetime -import logging -import unittest -from collections import Counter - -import requests as r -from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag -from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor -from qlogging import ( - send_ping, - build_summary, - format_runtime, - get_notify_options, - summary_stats, - tor_notify_apprise, - tor_notify_summary, -) - - -class FakeQbt: - """Minimal stand-in for the Qbt object, holding only what notifiers read.""" - - def __init__(self, **overrides): - self.config = {"pushover": {}} - self.total_torrents = 42 - self.torrent_hash_delete_list = ["a"] * 7 - self.up_tor_counter = 4 - self.ignored_counter = 5 - self.preme_tor_counter = 3 - self.tracker_protected_tag = "ipt" - self.tracker_non_protected_tag = "public" - self.c = Counter({"ipt": 10, "public": 20}) - self.delete_torrents = True - self.st = datetime.datetime(2026, 10, 27, 9, 54, 0) - self.et = datetime.datetime(2026, 10, 27, 9, 54, 12) - self.po_key = "USERKEY" - self.poc = StubPushover() - self.apprise_host = "localhost" - self.apprise_port = 8088 - self.apprise_aurls = "mailto://user:pass@example.com" - self.tl = logging.getLogger("qbit-maid-test") - for key, value in overrides.items(): - setattr(self, key, value) - - -class StubPushover: - """Records message() calls instead of hitting the Pushover API.""" - - def __init__(self): - self.calls = [] - - def message(self, user, body, **kwargs): - self.calls.append({"user": user, "body": body, "kwargs": kwargs}) - - -class FailingPushover: - def message(self, *args, **kwargs): - raise RuntimeError("pushover is down") - -class TestQbitmaid(unittest.TestCase): - def test_ispreme_sanity(self): - self.assertTrue(is_preme(1,1)) - self.assertTrue(is_preme(1,2)) - self.assertFalse(is_preme(2,1)) - - 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')) - self.assertFalse(is_tracker_blank(1)) - self.assertFalse(is_tracker_blank(1.00000000)) - - def test_istrackerblank(self): - pass - - 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_isprotectedtracker(self): - pass - - 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_isnotprotectedtracker(self): - pass - - 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_isdownloading_sanity(self): - self.assertTrue(is_downloading('downloading')) - - def test_isdownloading(self): - self.assertFalse(is_downloading('DOWNLOADING')) - self.assertFalse(is_downloading('dOwNlOaDiNg')) - - def test_isprotectedunderratio_sanity(self): - self.assertTrue(is_protected_under_ratio(0.5,1,'a','a,b,c')) - - 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") - - - # --- notification formatting ------------------------------------------- - - def test_format_runtime(self): - self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s") - self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s") - self.assertEqual(format_runtime(datetime.timedelta(seconds=64)), "1m 04s") - self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s") - self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s") - - def test_summary_stats(self): - stats = summary_stats(FakeQbt()) - self.assertEqual(stats["total"], 42) - self.assertEqual(stats["removed"], 7) - self.assertEqual(stats["orphaned"], 4) - self.assertEqual(stats["ignored"], 5) - self.assertEqual(stats["premature"], 3) - self.assertEqual(stats["protected"], 10) - self.assertEqual(stats["non_protected"], 20) - self.assertEqual(stats["runtime"], "12s") - - def test_summary_stats_missing_attributes(self): - class Bare: - pass - - stats = summary_stats(Bare()) - self.assertEqual(stats["total"], 0) - self.assertEqual(stats["removed"], 0) - self.assertEqual(stats["runtime"], "unknown") - - def test_build_summary_plain(self): - title, body = build_summary(FakeQbt(), style="plain", use_emoji=False) - self.assertEqual(title, "qbit-maid: 7 torrents removed") - self.assertEqual( - body, - "Scanned 42 torrents in 12s\n" - "\n" - "7 removed \u00b7 4 orphaned\n" - "5 ignored \u00b7 3 premature\n" - "\n" - "10 protected \u00b7 20 public", - ) - - def test_build_summary_plain_emoji(self): - _, body = build_summary(FakeQbt(), style="plain", use_emoji=True) - self.assertEqual( - body, - "Scanned 42 torrents in 12s\n" - "\n" - "\U0001F5D1\uFE0F 7 removed \u00b7 \U0001F47B 4 orphaned\n" - "\u23ED\uFE0F 5 ignored \u00b7 \U0001F423 3 premature\n" - "\n" - "\U0001F6E1\uFE0F 10 protected \u00b7 \U0001F310 20 public", - ) - - def test_build_summary_plain_has_no_leading_indent(self): - _, body = build_summary(FakeQbt(), style="plain") - for line in body.splitlines(): - self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}") - - def test_build_summary_has_no_space_padding(self): - """Pushover uses a proportional font, so padded columns never align.""" - for style in ("plain", "html"): - for emoji in (False, True): - _, body = build_summary(FakeQbt(), style=style, use_emoji=emoji) - for line in body.splitlines(): - stripped = line.replace(" \u00b7 ", "") - self.assertNotIn(" ", stripped, - f"padded column in {style}/{emoji}: {line!r}") - - def test_build_summary_groups_two_counters_per_line(self): - _, body = build_summary(FakeQbt(), style="plain") - rows = [ln for ln in body.splitlines() if "\u00b7" in ln] - self.assertEqual(len(rows), 3) - for row in rows: - self.assertEqual(row.count("\u00b7"), 1, row) - - def test_build_summary_value_leads_label(self): - """The number comes first so it is the first thing the eye lands on.""" - _, body = build_summary(FakeQbt(), style="plain", use_emoji=False) - self.assertIn("7 removed", body) - self.assertIn("20 public", body) - self.assertNotIn("Removed 7", body) - self.assertNotIn("Non-protected", body) - - def test_build_summary_html(self): - title, body = build_summary(FakeQbt(), style="html") - self.assertEqual(title, "qbit-maid: 7 torrents removed") - self.assertIn("Scanned 42 torrents in 12s", body) - self.assertIn("7 removed", body) - self.assertIn("20 public", body) - self.assertNotIn("---", title) - - def test_build_summary_emoji(self): - _, plain = build_summary(FakeQbt(), style="plain", use_emoji=True) - self.assertIn("\U0001F5D1\uFE0F 7 removed", plain) - self.assertIn("\U0001F47B 4 orphaned", plain) - _, html = build_summary(FakeQbt(), style="html", use_emoji=True) - self.assertIn("\U0001F5D1\uFE0F 7 removed", html) - - def test_build_summary_emoji_off_has_no_emoji(self): - _, body = build_summary(FakeQbt(), style="plain", use_emoji=False) - for marker in ("\U0001F5D1", "\U0001F47B", "\u23ED", "\U0001F423", - "\U0001F6E1", "\U0001F310"): - self.assertNotIn(marker, body) - - def test_build_summary_delete_disabled(self): - title, body = build_summary(FakeQbt(delete_torrents=False)) - self.assertEqual(title, "qbit-maid: 7 torrents would be removed") - self.assertIn("7 would be removed", body) - self.assertNotIn("7 removed", body) - - def test_build_summary_body_does_not_repeat_title_phrase(self): - """The title carries the outcome; the body must not restate it verbatim.""" - title, body = build_summary(FakeQbt(), style="plain", use_emoji=False) - self.assertNotIn("torrents removed", body) - self.assertNotIn("torrents would be removed", body) - self.assertIn("7 removed", body) - self.assertTrue(title.endswith("7 torrents removed")) - - def test_build_summary_nothing_to_do(self): - title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)) - self.assertEqual(title, "qbit-maid: nothing removed") - title, _ = build_summary( - FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False) - ) - self.assertEqual(title, "qbit-maid: nothing to remove") - - def test_build_summary_singular(self): - title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1)) - self.assertEqual(title, "qbit-maid: 1 torrent removed") - - def test_build_summary_hostname(self): - _, body = build_summary(FakeQbt(), include_hostname=True) - self.assertTrue(body.startswith("Host: ")) - - def test_build_summary_unknown_style_falls_back(self): - _, body = build_summary(FakeQbt(), style="markdown") - self.assertNotIn("", body) - - def test_get_notify_options_defaults(self): - opts = get_notify_options(FakeQbt()) - self.assertEqual(opts["style"], "plain") - self.assertTrue(opts["use_emoji"]) - self.assertFalse(opts["include_hostname"]) - self.assertTrue(opts["notify_on_empty"]) - self.assertIsNone(opts["priority"]) - self.assertIsNone(opts["sound"]) - - def test_get_notify_options_emoji_can_be_disabled(self): - opts = get_notify_options(FakeQbt(config={"pushover": {"use_emoji": False}})) - self.assertFalse(opts["use_emoji"]) - - def test_get_notify_options_overrides(self): - fake = FakeQbt(config={"pushover": { - "format": "HTML", - "use_emoji": False, - "include_hostname": True, - "notify_on_empty": False, - "priority": 2, - "sound": "siren", - "retry": 30, - "expire": 600, - }}) - opts = get_notify_options(fake) - self.assertEqual(opts["style"], "html") - self.assertFalse(opts["use_emoji"]) - self.assertTrue(opts["include_hostname"]) - self.assertFalse(opts["notify_on_empty"]) - self.assertEqual(opts["priority"], 2) - self.assertEqual(opts["sound"], "siren") - self.assertEqual(opts["retry"], 30) - self.assertEqual(opts["expire"], 600) - - def test_get_notify_options_invalid_style(self): - fake = FakeQbt(config={"pushover": {"format": "bogus"}}) - self.assertEqual(get_notify_options(fake)["style"], "plain") - - def test_tor_notify_summary_plain_payload(self): - fake = FakeQbt() - tor_notify_summary(fake) - self.assertEqual(len(fake.poc.calls), 1) - call = fake.poc.calls[0] - self.assertEqual(call["user"], "USERKEY") - self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed") - self.assertNotIn("html", call["kwargs"]) - - def test_tor_notify_summary_html_payload(self): - fake = FakeQbt(config={"pushover": {"format": "html"}}) - tor_notify_summary(fake) - call = fake.poc.calls[0] - self.assertEqual(call["kwargs"]["html"], 1) - self.assertIn("", call["body"]) - - def test_tor_notify_summary_priority_two_adds_retry_expire(self): - fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}}) - tor_notify_summary(fake) - kwargs = fake.poc.calls[0]["kwargs"] - self.assertEqual(kwargs["priority"], 2) - self.assertEqual(kwargs["retry"], 30) - self.assertEqual(kwargs["expire"], 600) - - def test_tor_notify_summary_priority_one_has_no_retry(self): - fake = FakeQbt(config={"pushover": {"priority": 1}}) - tor_notify_summary(fake) - kwargs = fake.poc.calls[0]["kwargs"] - self.assertEqual(kwargs["priority"], 1) - self.assertNotIn("retry", kwargs) - - def test_tor_notify_summary_skips_when_empty(self): - fake = FakeQbt( - config={"pushover": {"notify_on_empty": False}}, - torrent_hash_delete_list=[], - up_tor_counter=0, - ) - tor_notify_summary(fake) - self.assertEqual(fake.poc.calls, []) - - def test_tor_notify_summary_sends_when_empty_by_default(self): - fake = FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0) - tor_notify_summary(fake) - self.assertEqual(len(fake.poc.calls), 1) - - def test_tor_notify_summary_swallows_send_errors(self): - fake = FakeQbt() - fake.poc = FailingPushover() - tor_notify_summary(fake) # must not raise - - def test_tor_notify_apprise_uses_plain_body(self): - captured = {} - - def app_obj(req_obj, host, port, aurls, title, body): - captured.update(host=host, port=port, aurls=aurls, title=title, body=body) - - fake = FakeQbt(config={"pushover": {"format": "html"}}) - tor_notify_apprise(fake, None, app_obj) - self.assertEqual(captured["title"], "qbit-maid: 7 torrents removed") - self.assertNotIn("", captured["body"]) - self.assertEqual(captured["port"], 8088) - - def test_pushover_and_apprise_bodies_match(self): - fake = FakeQbt() - tor_notify_summary(fake) - captured = {} - tor_notify_apprise( - fake, None, - lambda req, host, port, aurls, title, body: captured.update(title=title, body=body), - ) - self.assertEqual(fake.poc.calls[0]["body"], captured["body"]) - self.assertEqual(fake.poc.calls[0]["kwargs"]["title"], captured["title"]) - - # def test__sanity(self): - # pass - - # def test_(self): - # pass - -if __name__ == '__main__': - unittest.main() +import datetime +import unittest + +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 ( + 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 RecordingRequests: + """Stands in for the requests module without touching the network.""" + + class RequestException(Exception): + pass + + def __init__(self): + self.calls = [] + self.timeouts = [] + + def get(self, url, timeout=None): + self.calls.append(url) + self.timeouts.append(timeout) + return "ok" + + +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)) + + 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_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.0)) + + 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_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_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_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_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_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_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_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_is_ignored_tag(self): + self.assertTrue(is_ignored_tag(["save"], "save,public,ipt")) + self.assertFalse(is_ignored_tag(["save"], "public,ipt")) + + 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")) + + +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") + self.assertEqual(format_runtime(datetime.timedelta(seconds=64)), "1m 04s") + self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s") + self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s") + + +class TestSummaryStats(unittest.TestCase): + def test_summary_stats(self): + stats = summary_stats(FakeQbt()) + self.assertEqual(stats["total"], 42) + self.assertEqual(stats["removed"], 7) + self.assertEqual(stats["orphaned"], 4) + self.assertEqual(stats["ignored"], 5) + self.assertEqual(stats["premature"], 3) + self.assertEqual(stats["protected"], 10) + self.assertEqual(stats["non_protected"], 20) + self.assertEqual(stats["runtime"], "12s") + + def test_summary_stats_missing_attributes(self): + class Bare: + pass + + stats = summary_stats(Bare()) + self.assertEqual(stats["total"], 0) + self.assertEqual(stats["removed"], 0) + self.assertEqual(stats["runtime"], "unknown") + + +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") + self.assertEqual( + body, + "Scanned 42 torrents in 12s\n" + "\n" + "7 removed \u00b7 4 orphaned\n" + "5 ignored \u00b7 3 premature\n" + "\n" + "10 protected \u00b7 20 public", + ) + + def test_build_summary_plain_emoji(self): + _, body = build_summary(FakeQbt(), style="plain", use_emoji=True) + self.assertEqual( + body, + "Scanned 42 torrents in 12s\n" + "\n" + "\U0001F5D1\uFE0F 7 removed \u00b7 \U0001F47B 4 orphaned\n" + "\u23ED\uFE0F 5 ignored \u00b7 \U0001F423 3 premature\n" + "\n" + "\U0001F6E1\uFE0F 10 protected \u00b7 \U0001F310 20 public", + ) + + def test_build_summary_plain_has_no_leading_indent(self): + _, body = build_summary(FakeQbt(), style="plain") + for line in body.splitlines(): + self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}") + + def test_build_summary_has_no_space_padding(self): + """Pushover uses a proportional font, so padded columns never align.""" + for style in ("plain", "html"): + for emoji in (False, True): + _, body = build_summary(FakeQbt(), style=style, use_emoji=emoji) + for line in body.splitlines(): + stripped = line.replace(" \u00b7 ", "") + self.assertNotIn(" ", stripped, + f"padded column in {style}/{emoji}: {line!r}") + + def test_build_summary_groups_two_counters_per_line(self): + _, body = build_summary(FakeQbt(), style="plain") + rows = [ln for ln in body.splitlines() if "\u00b7" in ln] + self.assertEqual(len(rows), 3) + for row in rows: + self.assertEqual(row.count("\u00b7"), 1, row) + + def test_build_summary_value_leads_label(self): + """The number comes first so it is the first thing the eye lands on.""" + _, body = build_summary(FakeQbt(), style="plain", use_emoji=False) + self.assertIn("7 removed", body) + self.assertIn("20 public", body) + self.assertNotIn("Removed 7", body) + self.assertNotIn("Non-protected", body) + + def test_build_summary_html(self): + title, body = build_summary(FakeQbt(), style="html") + self.assertEqual(title, "qbit-maid: 7 torrents removed") + self.assertIn("Scanned 42 torrents in 12s", body) + self.assertIn("7 removed", body) + self.assertIn("20 public", body) + self.assertNotIn("---", title) + + def test_build_summary_emoji(self): + _, plain = build_summary(FakeQbt(), style="plain", use_emoji=True) + self.assertIn("\U0001F5D1\uFE0F 7 removed", plain) + self.assertIn("\U0001F47B 4 orphaned", plain) + _, html = build_summary(FakeQbt(), style="html", use_emoji=True) + self.assertIn("\U0001F5D1\uFE0F 7 removed", html) + + def test_build_summary_emoji_off_has_no_emoji(self): + _, body = build_summary(FakeQbt(), style="plain", use_emoji=False) + for marker in ("\U0001F5D1", "\U0001F47B", "\u23ED", "\U0001F423", + "\U0001F6E1", "\U0001F310"): + self.assertNotIn(marker, body) + + def test_build_summary_delete_disabled(self): + title, body = build_summary(FakeQbt(delete_torrents=False)) + self.assertEqual(title, "qbit-maid: 7 torrents would be removed") + self.assertIn("7 would be removed", body) + self.assertNotIn("7 removed", body) + + def test_build_summary_body_does_not_repeat_title_phrase(self): + """The title carries the outcome; the body must not restate it verbatim.""" + title, body = build_summary(FakeQbt(), style="plain", use_emoji=False) + self.assertNotIn("torrents removed", body) + self.assertNotIn("torrents would be removed", body) + self.assertIn("7 removed", body) + self.assertTrue(title.endswith("7 torrents removed")) + + def test_build_summary_nothing_to_do(self): + title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)) + self.assertEqual(title, "qbit-maid: nothing removed") + title, _ = build_summary( + FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False) + ) + self.assertEqual(title, "qbit-maid: nothing to remove") + + def test_build_summary_singular(self): + title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1)) + self.assertEqual(title, "qbit-maid: 1 torrent removed") + + def test_build_summary_hostname(self): + _, body = build_summary(FakeQbt(), include_hostname=True) + self.assertTrue(body.startswith("Host: ")) + + def test_build_summary_hostname_html(self): + _, body = build_summary(FakeQbt(), style="html", include_hostname=True) + self.assertTrue(body.startswith("Host: ")) + + def test_build_summary_unknown_style_falls_back(self): + _, body = build_summary(FakeQbt(), style="markdown") + self.assertNotIn("", body) + + +class TestNotifyOptions(unittest.TestCase): + def test_get_notify_options_defaults(self): + opts = get_notify_options(FakeQbt()) + self.assertEqual(opts["style"], "plain") + self.assertTrue(opts["use_emoji"]) + self.assertFalse(opts["include_hostname"]) + self.assertTrue(opts["notify_on_empty"]) + self.assertIsNone(opts["priority"]) + self.assertIsNone(opts["sound"]) + + def test_get_notify_options_emoji_can_be_disabled(self): + opts = get_notify_options(FakeQbt(config={"pushover": {"use_emoji": False}})) + self.assertFalse(opts["use_emoji"]) + + def test_get_notify_options_overrides(self): + fake = FakeQbt(config={"pushover": { + "format": "HTML", + "use_emoji": False, + "include_hostname": True, + "notify_on_empty": False, + "priority": 2, + "sound": "siren", + "retry": 30, + "expire": 600, + }}) + opts = get_notify_options(fake) + self.assertEqual(opts["style"], "html") + self.assertFalse(opts["use_emoji"]) + self.assertTrue(opts["include_hostname"]) + self.assertFalse(opts["notify_on_empty"]) + self.assertEqual(opts["priority"], 2) + self.assertEqual(opts["sound"], "siren") + self.assertEqual(opts["retry"], 30) + self.assertEqual(opts["expire"], 600) + + def test_get_notify_options_invalid_style(self): + fake = FakeQbt(config={"pushover": {"format": "bogus"}}) + self.assertEqual(get_notify_options(fake)["style"], "plain") + + +class TestTorNotifySummary(unittest.TestCase): + def test_tor_notify_summary_plain_payload(self): + fake = FakeQbt() + tor_notify_summary(fake) + self.assertEqual(len(fake.poc.calls), 1) + call = fake.poc.calls[0] + self.assertEqual(call["user"], "USERKEY") + self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed") + self.assertNotIn("html", call["kwargs"]) + + def test_tor_notify_summary_html_payload(self): + fake = FakeQbt(config={"pushover": {"format": "html"}}) + tor_notify_summary(fake) + call = fake.poc.calls[0] + self.assertEqual(call["kwargs"]["html"], 1) + self.assertIn("", call["body"]) + + def test_tor_notify_summary_priority_two_adds_retry_expire(self): + fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}}) + tor_notify_summary(fake) + kwargs = fake.poc.calls[0]["kwargs"] + self.assertEqual(kwargs["priority"], 2) + self.assertEqual(kwargs["retry"], 30) + self.assertEqual(kwargs["expire"], 600) + + def test_tor_notify_summary_priority_one_has_no_retry(self): + fake = FakeQbt(config={"pushover": {"priority": 1}}) + tor_notify_summary(fake) + kwargs = fake.poc.calls[0]["kwargs"] + self.assertEqual(kwargs["priority"], 1) + self.assertNotIn("retry", kwargs) + + def test_tor_notify_summary_skips_when_empty(self): + fake = FakeQbt( + config={"pushover": {"notify_on_empty": False}}, + torrent_hash_delete_list=[], + up_tor_counter=0, + ) + tor_notify_summary(fake) + self.assertEqual(fake.poc.calls, []) + + def test_tor_notify_summary_sends_when_empty_by_default(self): + fake = FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0) + tor_notify_summary(fake) + self.assertEqual(len(fake.poc.calls), 1) + + def test_tor_notify_summary_swallows_send_errors(self): + fake = FakeQbt() + fake.poc = FailingPushover() + tor_notify_summary(fake) # must not raise + + +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("", 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() diff --git a/test_qbt.py b/test_qbt.py new file mode 100644 index 0000000..c8c94e1 --- /dev/null +++ b/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() diff --git a/test_qlist.py b/test_qlist.py new file mode 100644 index 0000000..f2e4dfe --- /dev/null +++ b/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() diff --git a/test_qlogging_extra.py b/test_qlogging_extra.py new file mode 100644 index 0000000..2024e5c --- /dev/null +++ b/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() diff --git a/test_qprocess.py b/test_qprocess.py new file mode 100644 index 0000000..3202e49 --- /dev/null +++ b/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() diff --git a/test_write_csv.py b/test_write_csv.py index 380fa58..acac514 100644 --- a/test_write_csv.py +++ b/test_write_csv.py @@ -1,41 +1,50 @@ -from qprocess import write_csv -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"]} -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 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)) - -if __name__ == '__main__': - unittest.main() \ No newline at end of file +"""Tests for qprocess.write_csv — header-once, append semantics.""" +import csv +import os +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 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 read_rows(self): + with open(self.outfile, newline="", encoding="UTF8") as fh: + return list(csv.reader(fh)) + + 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()