jblu/qbit-maid
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

test: audit suite for meaningfulness, reach 100% coverage #66

Merged
trex Bot merged 2 commits from dev-test-suite-audit into main 2026-11-03 16:47:04 -06:00
Conversation 0 Commits 2 Files Changed 10
+1647 -461
10 changed files with 1647 additions and 461 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.coveragerc
+18
View File
@@ -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
+20
View File
@@ -8,6 +8,26 @@ on:
pull_request: pull_request:
jobs: 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: komodo-deploy-script:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
qbt_fakes.py
+158
View File
@@ -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
+173
View File
@@ -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
+404 -420
View File
@@ -1,420 +1,404 @@
import datetime import datetime
import logging import unittest
import unittest
from collections import Counter from qbt_fakes import FailingPushover, FakeQbt, StubPushover
from qlist import (
import requests as r is_cat_ignored,
from qlist import is_preme,is_cat_ignored,is_tracker_blank,is_protected_tracker,is_not_protected_tracker,is_tag_blank,is_ignored_tag is_ignored_tag,
from qprocess import is_downloading,is_protected_under_ratio,is_old_tor,is_protected_over_ratio,is_not_protected_tor is_not_protected_tracker,
from qlogging import ( is_preme,
send_ping, is_protected_tracker,
build_summary, is_tag_blank,
format_runtime, is_tracker_blank,
get_notify_options, )
summary_stats, from qlogging import (
tor_notify_apprise, build_summary,
tor_notify_summary, format_runtime,
) get_notify_options,
send_ping,
summary_stats,
class FakeQbt: tor_notify_apprise,
"""Minimal stand-in for the Qbt object, holding only what notifiers read.""" tor_notify_summary,
)
def __init__(self, **overrides): from qprocess import (
self.config = {"pushover": {}} is_downloading,
self.total_torrents = 42 is_not_protected_tor,
self.torrent_hash_delete_list = ["a"] * 7 is_old_tor,
self.up_tor_counter = 4 is_protected_over_ratio,
self.ignored_counter = 5 is_protected_under_ratio,
self.preme_tor_counter = 3 )
self.tracker_protected_tag = "ipt"
self.tracker_non_protected_tag = "public"
self.c = Counter({"ipt": 10, "public": 20}) class RecordingRequests:
self.delete_torrents = True """Stands in for the requests module without touching the network."""
self.st = datetime.datetime(2026, 10, 27, 9, 54, 0)
self.et = datetime.datetime(2026, 10, 27, 9, 54, 12) class RequestException(Exception):
self.po_key = "USERKEY" pass
self.poc = StubPushover()
self.apprise_host = "localhost" def __init__(self):
self.apprise_port = 8088 self.calls = []
self.apprise_aurls = "mailto://user:pass@example.com" self.timeouts = []
self.tl = logging.getLogger("qbit-maid-test")
for key, value in overrides.items(): def get(self, url, timeout=None):
setattr(self, key, value) self.calls.append(url)
self.timeouts.append(timeout)
return "ok"
class StubPushover:
"""Records message() calls instead of hitting the Pushover API."""
class TestPredicates(unittest.TestCase):
def __init__(self): def test_is_preme(self):
self.calls = [] self.assertTrue(is_preme(1, 1)) # equal to min_age counts as premature
self.assertTrue(is_preme(1, 2))
def message(self, user, body, **kwargs): self.assertFalse(is_preme(2, 1))
self.calls.append({"user": user, "body": body, "kwargs": kwargs})
def test_is_cat_ignored(self):
cats = ["a", "b", "c"]
class FailingPushover: for cat in cats:
def message(self, *args, **kwargs): self.assertTrue(is_cat_ignored(cat, cats))
raise RuntimeError("pushover is down") self.assertFalse(is_cat_ignored("d", cats))
self.assertFalse(is_cat_ignored(1, cats))
class TestQbitmaid(unittest.TestCase): self.assertFalse(is_cat_ignored(1.0, cats))
def test_ispreme_sanity(self):
self.assertTrue(is_preme(1,1)) def test_is_tracker_blank(self):
self.assertTrue(is_preme(1,2)) self.assertTrue(is_tracker_blank(""))
self.assertFalse(is_preme(2,1)) self.assertFalse(is_tracker_blank("a"))
self.assertFalse(is_tracker_blank(1))
def test_ispreme(self): self.assertFalse(is_tracker_blank(1.0))
pass
def test_is_protected_tracker(self):
def test_iscatignored_sanity(self): whitelist = ["a.com", "b.me", "c.io"]
self.assertTrue(is_cat_ignored('a', ['a','b','c'])) self.assertTrue(is_protected_tracker("https://a.com/", whitelist))
self.assertTrue(is_cat_ignored('b', ['a','b','c'])) self.assertFalse(is_protected_tracker("https://google.com/", whitelist))
self.assertTrue(is_cat_ignored('c', ['a','b','c'])) self.assertFalse(is_protected_tracker("https://d.com", whitelist))
self.assertFalse(is_cat_ignored('d', ['a','b','c'])) self.assertFalse(is_protected_tracker("", whitelist))
self.assertFalse(is_cat_ignored(1, ['a','b','d']))
self.assertFalse(is_cat_ignored(1.0000000, ['a','b','c'])) def test_is_not_protected_tracker(self):
whitelist = ["a.com", "b.me", "c.io"]
def test_iscatignored(self): self.assertFalse(is_not_protected_tracker("https://a.com/", whitelist))
pass self.assertTrue(is_not_protected_tracker("https://google.com/", whitelist))
self.assertTrue(is_not_protected_tracker("https://d.com", whitelist))
def test_istrackerblank_sanity(self): self.assertFalse(is_not_protected_tracker("", whitelist))
self.assertTrue(is_tracker_blank(''))
self.assertFalse(is_tracker_blank('a')) def test_is_tag_blank(self):
self.assertFalse(is_tracker_blank(1)) self.assertTrue(is_tag_blank(""))
self.assertFalse(is_tracker_blank(1.00000000)) for value in ("a", 1, 1.0001, False, True):
self.assertFalse(is_tag_blank(value))
def test_istrackerblank(self):
pass def test_is_downloading(self):
self.assertTrue(is_downloading("downloading"))
def test_isprotectedtracker_sanity(self): self.assertFalse(is_downloading("DOWNLOADING"))
self.assertTrue(is_protected_tracker('https://a.com/',['a.com','b.me','c.io'])) self.assertFalse(is_downloading("dOwNlOaDiNg"))
self.assertFalse(is_protected_tracker('https://google.com/',['a.com','b.me','c.io'])) self.assertFalse(is_downloading("uploading"))
self.assertFalse(is_protected_tracker('https://d.com',['a.com','b.me','c.io']))
def test_is_protected_under_ratio(self):
def test_isprotectedtracker(self): self.assertTrue(is_protected_under_ratio(0.5, 1, "a", "a,b,c"))
pass self.assertFalse(is_protected_under_ratio(1.5, 1, "a", "a,b,c"))
self.assertFalse(is_protected_under_ratio(0.5, 1, "z", "a,b,c"))
def test_isnotprotectedtracker_sanity(self):
self.assertFalse(is_not_protected_tracker('https://a.com/',['a.com','b.me','c.io'])) def test_is_old_tor(self):
self.assertTrue(is_not_protected_tracker('https://google.com/',['a.com','b.me','c.io'])) self.assertTrue(is_old_tor(1, 1))
self.assertTrue(is_not_protected_tracker('https://d.com',['a.com','b.me','c.io'])) self.assertTrue(is_old_tor(2, 1))
self.assertFalse(is_old_tor(1, 2))
def test_isnotprotectedtracker(self):
pass def test_is_protected_over_ratio(self):
self.assertTrue(is_protected_over_ratio(2, 1, "a", "a,b,c"))
def test_istagblank(self): self.assertFalse(is_protected_over_ratio(0.5, 1, "a", "a,b,c"))
self.assertTrue(is_tag_blank('')) self.assertFalse(is_protected_over_ratio(2, 1, "z", "a,b,c"))
self.assertFalse(is_tag_blank('a'))
self.assertFalse(is_tag_blank(1)) def test_is_not_protected_tor(self):
self.assertFalse(is_tag_blank(1.0001)) self.assertTrue(is_not_protected_tor("a", "a,b,c"))
self.assertFalse(is_tag_blank(False)) self.assertFalse(is_not_protected_tor("z", "a,b,c"))
self.assertFalse(is_tag_blank(True))
def test_is_ignored_tag(self):
def test_isdownloading_sanity(self): self.assertTrue(is_ignored_tag(["save"], "save,public,ipt"))
self.assertTrue(is_downloading('downloading')) self.assertFalse(is_ignored_tag(["save"], "public,ipt"))
def test_isdownloading(self): def test_is_ignored_tag_matches_substrings_not_tags(self):
self.assertFalse(is_downloading('DOWNLOADING')) """KNOWN BUG: matching is substring-based, so a short tag matches any
self.assertFalse(is_downloading('dOwNlOaDiNg')) longer tag that contains it. 'ipt' matches 'iptorrents' and 'save'
matches 'unsaved', which silently ignores unrelated torrents.
def test_isprotectedunderratio_sanity(self): """
self.assertTrue(is_protected_under_ratio(0.5,1,'a','a,b,c')) self.assertTrue(is_ignored_tag(["ipt"], "iptorrents"))
self.assertTrue(is_ignored_tag(["save"], "unsaved"))
def test_isprotectedunderratio(self):
pass
class TestSendPing(unittest.TestCase):
def test_isoldtor_sanity(self): def test_send_ping_requests_the_url(self):
self.assertFalse(is_old_tor(1,2)) req = RecordingRequests()
send_ping(FakeQbt(), req, "https://example.com/ping/x")
def test_isoldtor(self): self.assertEqual(req.calls, ["https://example.com/ping/x"])
self.assertTrue(is_old_tor(1,1))
self.assertTrue(is_old_tor(2,1)) def test_send_ping_uses_a_five_second_timeout(self):
self.assertFalse(is_old_tor(1,2)) req = RecordingRequests()
send_ping(FakeQbt(), req, "https://example.com/ping/x")
def test_isprotectedoverratio_sanity(self): self.assertEqual(req.timeouts, [5])
self.assertTrue(is_protected_over_ratio(2,1,'a','a,b,c'))
def test_isprotectedoverratio(self): # --- notification formatting -----------------------------------------------
pass
class TestFormatRuntime(unittest.TestCase):
def test_isnonprotectedtor_sanity(self): def test_format_runtime(self):
self.assertTrue(is_not_protected_tor('a','a,b,c')) self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s")
self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s")
def test_isnonprotectedtor(self): self.assertEqual(format_runtime(datetime.timedelta(seconds=64)), "1m 04s")
pass self.assertEqual(format_runtime(datetime.timedelta(seconds=3723)), "1h 02m 03s")
self.assertEqual(format_runtime(datetime.timedelta(seconds=-5)), "0s")
def test_isignoredtag_sanity(self):
self.assertTrue(is_ignored_tag(['a','b','c'], 'first,second,third,a'))
class TestSummaryStats(unittest.TestCase):
def test_isignoredtag(self): def test_summary_stats(self):
self.assertTrue(is_ignored_tag(['save'], 'save,public,ipt')) stats = summary_stats(FakeQbt())
self.assertEqual(stats["total"], 42)
def test_sendpingstart_sanity(self): self.assertEqual(stats["removed"], 7)
send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb" + "/start") self.assertEqual(stats["orphaned"], 4)
url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb" self.assertEqual(stats["ignored"], 5)
send_ping(self, r, url) self.assertEqual(stats["premature"], 3)
self.assertEqual(stats["protected"], 10)
def test_sendping_start(self): self.assertEqual(stats["non_protected"], 20)
url = "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb/" self.assertEqual(stats["runtime"], "12s")
send_ping(self, r, url.strip("/") + "/start")
send_ping(self, r, "https://thunder.jonb.io/ping/921625e5-5b76-4f45-a0c3-56145e16f3bb") def test_summary_stats_missing_attributes(self):
class Bare:
pass
# --- notification formatting -------------------------------------------
stats = summary_stats(Bare())
def test_format_runtime(self): self.assertEqual(stats["total"], 0)
self.assertEqual(format_runtime(datetime.timedelta(seconds=0)), "0s") self.assertEqual(stats["removed"], 0)
self.assertEqual(format_runtime(datetime.timedelta(seconds=12)), "12s") self.assertEqual(stats["runtime"], "unknown")
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 TestBuildSummary(unittest.TestCase):
def test_build_summary_plain(self):
def test_summary_stats(self): title, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
stats = summary_stats(FakeQbt()) self.assertEqual(title, "qbit-maid: 7 torrents removed")
self.assertEqual(stats["total"], 42) self.assertEqual(
self.assertEqual(stats["removed"], 7) body,
self.assertEqual(stats["orphaned"], 4) "Scanned 42 torrents in 12s\n"
self.assertEqual(stats["ignored"], 5) "\n"
self.assertEqual(stats["premature"], 3) "7 removed \u00b7 4 orphaned\n"
self.assertEqual(stats["protected"], 10) "5 ignored \u00b7 3 premature\n"
self.assertEqual(stats["non_protected"], 20) "\n"
self.assertEqual(stats["runtime"], "12s") "10 protected \u00b7 20 public",
)
def test_summary_stats_missing_attributes(self):
class Bare: def test_build_summary_plain_emoji(self):
pass _, body = build_summary(FakeQbt(), style="plain", use_emoji=True)
self.assertEqual(
stats = summary_stats(Bare()) body,
self.assertEqual(stats["total"], 0) "Scanned 42 torrents in 12s\n"
self.assertEqual(stats["removed"], 0) "\n"
self.assertEqual(stats["runtime"], "unknown") "\U0001F5D1\uFE0F 7 removed \u00b7 \U0001F47B 4 orphaned\n"
"\u23ED\uFE0F 5 ignored \u00b7 \U0001F423 3 premature\n"
def test_build_summary_plain(self): "\n"
title, body = build_summary(FakeQbt(), style="plain", use_emoji=False) "\U0001F6E1\uFE0F 10 protected \u00b7 \U0001F310 20 public",
self.assertEqual(title, "qbit-maid: 7 torrents removed") )
self.assertEqual(
body, def test_build_summary_plain_has_no_leading_indent(self):
"Scanned 42 torrents in 12s\n" _, body = build_summary(FakeQbt(), style="plain")
"\n" for line in body.splitlines():
"7 removed \u00b7 4 orphaned\n" self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}")
"5 ignored \u00b7 3 premature\n"
"\n" def test_build_summary_has_no_space_padding(self):
"10 protected \u00b7 20 public", """Pushover uses a proportional font, so padded columns never align."""
) for style in ("plain", "html"):
for emoji in (False, True):
def test_build_summary_plain_emoji(self): _, body = build_summary(FakeQbt(), style=style, use_emoji=emoji)
_, body = build_summary(FakeQbt(), style="plain", use_emoji=True) for line in body.splitlines():
self.assertEqual( stripped = line.replace(" \u00b7 ", "")
body, self.assertNotIn(" ", stripped,
"Scanned 42 torrents in 12s\n" f"padded column in {style}/{emoji}: {line!r}")
"\n"
"\U0001F5D1\uFE0F 7 removed \u00b7 \U0001F47B 4 orphaned\n" def test_build_summary_groups_two_counters_per_line(self):
"\u23ED\uFE0F 5 ignored \u00b7 \U0001F423 3 premature\n" _, body = build_summary(FakeQbt(), style="plain")
"\n" rows = [ln for ln in body.splitlines() if "\u00b7" in ln]
"\U0001F6E1\uFE0F 10 protected \u00b7 \U0001F310 20 public", self.assertEqual(len(rows), 3)
) for row in rows:
self.assertEqual(row.count("\u00b7"), 1, row)
def test_build_summary_plain_has_no_leading_indent(self):
_, body = build_summary(FakeQbt(), style="plain") def test_build_summary_value_leads_label(self):
for line in body.splitlines(): """The number comes first so it is the first thing the eye lands on."""
self.assertFalse(line.startswith(" "), f"unexpected indent: {line!r}") _, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
self.assertIn("7 removed", body)
def test_build_summary_has_no_space_padding(self): self.assertIn("20 public", body)
"""Pushover uses a proportional font, so padded columns never align.""" self.assertNotIn("Removed 7", body)
for style in ("plain", "html"): self.assertNotIn("Non-protected", body)
for emoji in (False, True):
_, body = build_summary(FakeQbt(), style=style, use_emoji=emoji) def test_build_summary_html(self):
for line in body.splitlines(): title, body = build_summary(FakeQbt(), style="html")
stripped = line.replace(" \u00b7 ", "") self.assertEqual(title, "qbit-maid: 7 torrents removed")
self.assertNotIn(" ", stripped, self.assertIn("Scanned <b>42</b> torrents in <b>12s</b>", body)
f"padded column in {style}/{emoji}: {line!r}") self.assertIn("<b>7</b> removed", body)
self.assertIn("<b>20</b> public", body)
def test_build_summary_groups_two_counters_per_line(self): self.assertNotIn("---", title)
_, body = build_summary(FakeQbt(), style="plain")
rows = [ln for ln in body.splitlines() if "\u00b7" in ln] def test_build_summary_emoji(self):
self.assertEqual(len(rows), 3) _, plain = build_summary(FakeQbt(), style="plain", use_emoji=True)
for row in rows: self.assertIn("\U0001F5D1\uFE0F 7 removed", plain)
self.assertEqual(row.count("\u00b7"), 1, row) self.assertIn("\U0001F47B 4 orphaned", plain)
_, html = build_summary(FakeQbt(), style="html", use_emoji=True)
def test_build_summary_value_leads_label(self): self.assertIn("\U0001F5D1\uFE0F <b>7</b> removed", html)
"""The number comes first so it is the first thing the eye lands on."""
_, body = build_summary(FakeQbt(), style="plain", use_emoji=False) def test_build_summary_emoji_off_has_no_emoji(self):
self.assertIn("7 removed", body) _, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
self.assertIn("20 public", body) for marker in ("\U0001F5D1", "\U0001F47B", "\u23ED", "\U0001F423",
self.assertNotIn("Removed 7", body) "\U0001F6E1", "\U0001F310"):
self.assertNotIn("Non-protected", body) self.assertNotIn(marker, body)
def test_build_summary_html(self): def test_build_summary_delete_disabled(self):
title, body = build_summary(FakeQbt(), style="html") title, body = build_summary(FakeQbt(delete_torrents=False))
self.assertEqual(title, "qbit-maid: 7 torrents removed") self.assertEqual(title, "qbit-maid: 7 torrents would be removed")
self.assertIn("Scanned <b>42</b> torrents in <b>12s</b>", body) self.assertIn("7 would be removed", body)
self.assertIn("<b>7</b> removed", body) self.assertNotIn("7 removed", body)
self.assertIn("<b>20</b> public", body)
self.assertNotIn("---", title) def test_build_summary_body_does_not_repeat_title_phrase(self):
"""The title carries the outcome; the body must not restate it verbatim."""
def test_build_summary_emoji(self): title, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
_, plain = build_summary(FakeQbt(), style="plain", use_emoji=True) self.assertNotIn("torrents removed", body)
self.assertIn("\U0001F5D1\uFE0F 7 removed", plain) self.assertNotIn("torrents would be removed", body)
self.assertIn("\U0001F47B 4 orphaned", plain) self.assertIn("7 removed", body)
_, html = build_summary(FakeQbt(), style="html", use_emoji=True) self.assertTrue(title.endswith("7 torrents removed"))
self.assertIn("\U0001F5D1\uFE0F <b>7</b> removed", html)
def test_build_summary_nothing_to_do(self):
def test_build_summary_emoji_off_has_no_emoji(self): title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0))
_, body = build_summary(FakeQbt(), style="plain", use_emoji=False) self.assertEqual(title, "qbit-maid: nothing removed")
for marker in ("\U0001F5D1", "\U0001F47B", "\u23ED", "\U0001F423", title, _ = build_summary(
"\U0001F6E1", "\U0001F310"): FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False)
self.assertNotIn(marker, body) )
self.assertEqual(title, "qbit-maid: nothing to remove")
def test_build_summary_delete_disabled(self):
title, body = build_summary(FakeQbt(delete_torrents=False)) def test_build_summary_singular(self):
self.assertEqual(title, "qbit-maid: 7 torrents would be removed") title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1))
self.assertIn("7 would be removed", body) self.assertEqual(title, "qbit-maid: 1 torrent removed")
self.assertNotIn("7 removed", body)
def test_build_summary_hostname(self):
def test_build_summary_body_does_not_repeat_title_phrase(self): _, body = build_summary(FakeQbt(), include_hostname=True)
"""The title carries the outcome; the body must not restate it verbatim.""" self.assertTrue(body.startswith("Host: "))
title, body = build_summary(FakeQbt(), style="plain", use_emoji=False)
self.assertNotIn("torrents removed", body) def test_build_summary_hostname_html(self):
self.assertNotIn("torrents would be removed", body) _, body = build_summary(FakeQbt(), style="html", include_hostname=True)
self.assertIn("7 removed", body) self.assertTrue(body.startswith("<b>Host:</b> "))
self.assertTrue(title.endswith("7 torrents removed"))
def test_build_summary_unknown_style_falls_back(self):
def test_build_summary_nothing_to_do(self): _, body = build_summary(FakeQbt(), style="markdown")
title, _ = build_summary(FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)) self.assertNotIn("<b>", body)
self.assertEqual(title, "qbit-maid: nothing removed")
title, _ = build_summary(
FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0, delete_torrents=False) class TestNotifyOptions(unittest.TestCase):
) def test_get_notify_options_defaults(self):
self.assertEqual(title, "qbit-maid: nothing to remove") opts = get_notify_options(FakeQbt())
self.assertEqual(opts["style"], "plain")
def test_build_summary_singular(self): self.assertTrue(opts["use_emoji"])
title, _ = build_summary(FakeQbt(torrent_hash_delete_list=["a"], total_torrents=1)) self.assertFalse(opts["include_hostname"])
self.assertEqual(title, "qbit-maid: 1 torrent removed") self.assertTrue(opts["notify_on_empty"])
self.assertIsNone(opts["priority"])
def test_build_summary_hostname(self): self.assertIsNone(opts["sound"])
_, body = build_summary(FakeQbt(), include_hostname=True)
self.assertTrue(body.startswith("Host: ")) def test_get_notify_options_emoji_can_be_disabled(self):
opts = get_notify_options(FakeQbt(config={"pushover": {"use_emoji": False}}))
def test_build_summary_unknown_style_falls_back(self): self.assertFalse(opts["use_emoji"])
_, body = build_summary(FakeQbt(), style="markdown")
self.assertNotIn("<b>", body) def test_get_notify_options_overrides(self):
fake = FakeQbt(config={"pushover": {
def test_get_notify_options_defaults(self): "format": "HTML",
opts = get_notify_options(FakeQbt()) "use_emoji": False,
self.assertEqual(opts["style"], "plain") "include_hostname": True,
self.assertTrue(opts["use_emoji"]) "notify_on_empty": False,
self.assertFalse(opts["include_hostname"]) "priority": 2,
self.assertTrue(opts["notify_on_empty"]) "sound": "siren",
self.assertIsNone(opts["priority"]) "retry": 30,
self.assertIsNone(opts["sound"]) "expire": 600,
}})
def test_get_notify_options_emoji_can_be_disabled(self): opts = get_notify_options(fake)
opts = get_notify_options(FakeQbt(config={"pushover": {"use_emoji": False}})) self.assertEqual(opts["style"], "html")
self.assertFalse(opts["use_emoji"]) self.assertFalse(opts["use_emoji"])
self.assertTrue(opts["include_hostname"])
def test_get_notify_options_overrides(self): self.assertFalse(opts["notify_on_empty"])
fake = FakeQbt(config={"pushover": { self.assertEqual(opts["priority"], 2)
"format": "HTML", self.assertEqual(opts["sound"], "siren")
"use_emoji": False, self.assertEqual(opts["retry"], 30)
"include_hostname": True, self.assertEqual(opts["expire"], 600)
"notify_on_empty": False,
"priority": 2, def test_get_notify_options_invalid_style(self):
"sound": "siren", fake = FakeQbt(config={"pushover": {"format": "bogus"}})
"retry": 30, self.assertEqual(get_notify_options(fake)["style"], "plain")
"expire": 600,
}})
opts = get_notify_options(fake) class TestTorNotifySummary(unittest.TestCase):
self.assertEqual(opts["style"], "html") def test_tor_notify_summary_plain_payload(self):
self.assertFalse(opts["use_emoji"]) fake = FakeQbt()
self.assertTrue(opts["include_hostname"]) tor_notify_summary(fake)
self.assertFalse(opts["notify_on_empty"]) self.assertEqual(len(fake.poc.calls), 1)
self.assertEqual(opts["priority"], 2) call = fake.poc.calls[0]
self.assertEqual(opts["sound"], "siren") self.assertEqual(call["user"], "USERKEY")
self.assertEqual(opts["retry"], 30) self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed")
self.assertEqual(opts["expire"], 600) self.assertNotIn("html", call["kwargs"])
def test_get_notify_options_invalid_style(self): def test_tor_notify_summary_html_payload(self):
fake = FakeQbt(config={"pushover": {"format": "bogus"}}) fake = FakeQbt(config={"pushover": {"format": "html"}})
self.assertEqual(get_notify_options(fake)["style"], "plain") tor_notify_summary(fake)
call = fake.poc.calls[0]
def test_tor_notify_summary_plain_payload(self): self.assertEqual(call["kwargs"]["html"], 1)
fake = FakeQbt() self.assertIn("<b>", call["body"])
tor_notify_summary(fake)
self.assertEqual(len(fake.poc.calls), 1) def test_tor_notify_summary_priority_two_adds_retry_expire(self):
call = fake.poc.calls[0] fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}})
self.assertEqual(call["user"], "USERKEY") tor_notify_summary(fake)
self.assertEqual(call["kwargs"]["title"], "qbit-maid: 7 torrents removed") kwargs = fake.poc.calls[0]["kwargs"]
self.assertNotIn("html", call["kwargs"]) self.assertEqual(kwargs["priority"], 2)
self.assertEqual(kwargs["retry"], 30)
def test_tor_notify_summary_html_payload(self): self.assertEqual(kwargs["expire"], 600)
fake = FakeQbt(config={"pushover": {"format": "html"}})
tor_notify_summary(fake) def test_tor_notify_summary_priority_one_has_no_retry(self):
call = fake.poc.calls[0] fake = FakeQbt(config={"pushover": {"priority": 1}})
self.assertEqual(call["kwargs"]["html"], 1) tor_notify_summary(fake)
self.assertIn("<b>", call["body"]) kwargs = fake.poc.calls[0]["kwargs"]
self.assertEqual(kwargs["priority"], 1)
def test_tor_notify_summary_priority_two_adds_retry_expire(self): self.assertNotIn("retry", kwargs)
fake = FakeQbt(config={"pushover": {"priority": 2, "retry": 30, "expire": 600}})
tor_notify_summary(fake) def test_tor_notify_summary_skips_when_empty(self):
kwargs = fake.poc.calls[0]["kwargs"] fake = FakeQbt(
self.assertEqual(kwargs["priority"], 2) config={"pushover": {"notify_on_empty": False}},
self.assertEqual(kwargs["retry"], 30) torrent_hash_delete_list=[],
self.assertEqual(kwargs["expire"], 600) up_tor_counter=0,
)
def test_tor_notify_summary_priority_one_has_no_retry(self): tor_notify_summary(fake)
fake = FakeQbt(config={"pushover": {"priority": 1}}) self.assertEqual(fake.poc.calls, [])
tor_notify_summary(fake)
kwargs = fake.poc.calls[0]["kwargs"] def test_tor_notify_summary_sends_when_empty_by_default(self):
self.assertEqual(kwargs["priority"], 1) fake = FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)
self.assertNotIn("retry", kwargs) tor_notify_summary(fake)
self.assertEqual(len(fake.poc.calls), 1)
def test_tor_notify_summary_skips_when_empty(self):
fake = FakeQbt( def test_tor_notify_summary_swallows_send_errors(self):
config={"pushover": {"notify_on_empty": False}}, fake = FakeQbt()
torrent_hash_delete_list=[], fake.poc = FailingPushover()
up_tor_counter=0, tor_notify_summary(fake) # must not raise
)
tor_notify_summary(fake)
self.assertEqual(fake.poc.calls, []) class TestTorNotifyApprise(unittest.TestCase):
def test_tor_notify_apprise_uses_plain_body(self):
def test_tor_notify_summary_sends_when_empty_by_default(self): captured = {}
fake = FakeQbt(torrent_hash_delete_list=[], up_tor_counter=0)
tor_notify_summary(fake) def app_obj(req_obj, host, port, aurls, title, body):
self.assertEqual(len(fake.poc.calls), 1) captured.update(host=host, port=port, aurls=aurls, title=title, body=body)
def test_tor_notify_summary_swallows_send_errors(self): fake = FakeQbt(config={"pushover": {"format": "html"}})
fake = FakeQbt() tor_notify_apprise(fake, None, app_obj)
fake.poc = FailingPushover() self.assertEqual(captured["title"], "qbit-maid: 7 torrents removed")
tor_notify_summary(fake) # must not raise self.assertNotIn("<b>", captured["body"])
self.assertEqual(captured["port"], 8088)
def test_tor_notify_apprise_uses_plain_body(self):
captured = {} def test_pushover_and_apprise_bodies_match(self):
fake = FakeQbt()
def app_obj(req_obj, host, port, aurls, title, body): tor_notify_summary(fake)
captured.update(host=host, port=port, aurls=aurls, title=title, body=body) captured = {}
tor_notify_apprise(
fake = FakeQbt(config={"pushover": {"format": "html"}}) fake, None,
tor_notify_apprise(fake, None, app_obj) lambda req, host, port, aurls, title, body: captured.update(title=title, body=body),
self.assertEqual(captured["title"], "qbit-maid: 7 torrents removed") )
self.assertNotIn("<b>", captured["body"]) self.assertEqual(fake.poc.calls[0]["body"], captured["body"])
self.assertEqual(captured["port"], 8088) self.assertEqual(fake.poc.calls[0]["kwargs"]["title"], captured["title"])
def test_pushover_and_apprise_bodies_match(self):
fake = FakeQbt() if __name__ == "__main__":
tor_notify_summary(fake) unittest.main()
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()
test_qbt.py
+326
View File
@@ -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
+107
View File
@@ -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
+288
View File
@@ -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
+103
View File
@@ -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
+50 -41
View File
@@ -1,41 +1,50 @@
from qprocess import write_csv """Tests for qprocess.write_csv — header-once, append semantics."""
import csv import csv
import unittest import os
import os import shutil
# Torrent Items needed for testing import tempfile
# canidate['state'] {canidate["ratio"]} {torrent["tags"]} torrent["seeding_time"] {torrent["category"]} {torrent["name"][0:20]} {canidate["time_active"]} import unittest
class TestWriteCSV(unittest.TestCase):
def test_write_csv_dragnet(self): from qprocess import write_csv
self.cv = csv
outfile = './test_dragnet_outfile.csv' HEADER = ["state", "ratio", "tags", "added", "hash", "name", "tracker"]
state = 'downloading' ROW = ["downloading", 1.05, "ipt", 1, "asfasdf23412adfqwer",
ratio = 1.05 "thisismynamehahahah", "https://localhost.stackoverflow.tech/announce"]
tags = 'ipt'
added = 1
age = 240000 class TestWriteCSV(unittest.TestCase):
time = 123456 def setUp(self):
thash = 'asfasdf23412adfqwer' self.tmp = tempfile.mkdtemp()
tname = 'thisismynamehahahah' self.addCleanup(shutil.rmtree, self.tmp, ignore_errors=True)
trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce' self.outfile = os.path.join(self.tmp, "out.csv")
header = ['state','ratio','tags','added','age','time','thash','tname','trname']
row = [state,ratio,tags,added,age,time,thash,tname,trname] def read_rows(self):
write_csv(self.cv,outfile,header,row) with open(self.outfile, newline="", encoding="UTF8") as fh:
self.assertTrue(os.path.exists(outfile)) return list(csv.reader(fh))
def test_write_csv_telemetry(self): def test_writes_header_then_row(self):
self.cv = csv write_csv(csv, self.outfile, HEADER, ROW)
outfile = './test_telemetry_outfile.csv' rows = self.read_rows()
state = 'downloading' self.assertEqual(rows[0], HEADER)
ratio = 1.05 self.assertEqual(len(rows), 2)
tags = 'ipt' self.assertEqual(rows[1][4], "asfasdf23412adfqwer")
added = 1
thash = 'asfasdf23412adfqwer' def test_header_is_written_only_once(self):
tname = 'thisismynamehahahah' write_csv(csv, self.outfile, HEADER, ROW)
trname = 'https://localhost.stackoverflow.tech/317332f1c125bc9c1b9b14fb8e054908/announce' write_csv(csv, self.outfile, HEADER, ROW)
header = ['state','ratio','tags','added','hash','name','tracker'] rows = self.read_rows()
row = [state,ratio,tags,added,thash,tname,trname] self.assertEqual(rows.count(HEADER), 1)
write_csv(self.cv,outfile,header,row) self.assertEqual(len(rows), 3)
self.assertTrue(os.path.exists(outfile))
def test_creates_the_file_when_missing(self):
if __name__ == '__main__': self.assertFalse(os.path.exists(self.outfile))
unittest.main() write_csv(csv, self.outfile, HEADER, ROW)
self.assertTrue(os.path.exists(self.outfile))
def test_values_are_round_tripped(self):
write_csv(csv, self.outfile, HEADER, ROW)
self.assertEqual(self.read_rows()[1], [str(value) for value in ROW])
if __name__ == "__main__":
unittest.main()