import hashlib import io import json import os import tempfile import unittest from pathlib import Path, PurePosixPath from unittest.mock import patch from urllib import error from archive_clients.bencode import encode from archive_clients.config import ServiceConfig from archive_clients.qbittorrent import QBittorrentReader class _Response: def __init__(self, value: bytes, status: int = 200): self.value = io.BytesIO(value) self.status = status def read(self, size=-1): return self.value.read(size) def __enter__(self): return self def __exit__(self, *_): return None class _Opener: def __init__(self, values): self.values = iter(values) self.calls = [] def open(self, call, timeout): self.calls.append(call) response = next(self.values) if isinstance(response, BaseException): raise response if isinstance(response, tuple): return _Response(*response) return _Response(response) class QBittorrentReaderTests(unittest.TestCase): def test_hash_scoped_lookup_exports_and_normalizes(self): info = { b"length": 3, b"name": b"a.txt", b"piece length": 16384, b"pieces": b"x" * 20, } torrent_bytes = encode({b"info": info}) torrent_hash = hashlib.sha1(encode(info)).hexdigest() responses = [ b"Ok.", json.dumps([{ "hash": torrent_hash, "name": "a.txt", "state": "uploading", }]).encode(), b'[{"index":0,"name":"a.txt","size":3,"progress":1,"priority":1}]', torrent_bytes, ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): resource = QBittorrentReader(config).get_resource(torrent_hash) self.assertIsNotNone(resource) self.assertEqual(resource.summary.resource_id.info_hash_v1_hex, torrent_hash) lookup_url = getattr( opener.calls[1], "full_url", opener.calls[1] ) self.assertTrue(lookup_url.endswith("/api/v2/torrents/info")) def test_full_v2_lookup_accepts_qbittorrent_truncated_hash(self): info = { b"file tree": { b"a.txt": { b"": { b"length": 3, b"pieces root": hashlib.sha256(b"abc").digest(), } } }, b"meta version": 2, b"name": b"a.txt", b"piece length": 16384, } torrent_bytes = encode({b"info": info}) full_hash = hashlib.sha256(encode(info)).hexdigest() qb_hash = full_hash[:40] responses = [ b"Ok.", json.dumps([{ "hash": qb_hash, "name": "a.txt", "state": "uploading", }]).encode(), b'[{"index":0,"name":"a.txt","size":3,"progress":1,"priority":1}]', torrent_bytes, ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): resource = QBittorrentReader(config).get_resource(full_hash) self.assertIsNotNone(resource) self.assertEqual(resource.summary.qb_torrent_id, qb_hash) self.assertEqual( resource.summary.resource_id.info_hash_v2_hex, full_hash ) self.assertNotIn("hashes=", opener.calls[1]) def test_hybrid_lookup_accepts_v1_alias_of_v2_primary_hash(self): info = { b"file tree": { b"a.txt": { b"": { b"length": 3, b"pieces root": hashlib.sha256(b"abc").digest(), } } }, b"length": 3, b"meta version": 2, b"name": b"a.txt", b"piece length": 16384, b"pieces": hashlib.sha1(b"abc").digest(), } torrent_bytes = encode({b"info": info}) encoded_info = encode(info) v1_hash = hashlib.sha1(encoded_info).hexdigest() v2_hash = hashlib.sha256(encoded_info).hexdigest() qb_hash = v2_hash[:40] responses = [ b"Ok.", json.dumps([{ "hash": qb_hash, "infohash_v1": v1_hash, "infohash_v2": v2_hash, "name": "a.txt", "state": "uploading", }]).encode(), b'[{"index":0,"name":"a.txt","size":3,"progress":1,"priority":1}]', torrent_bytes, ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): resource = QBittorrentReader(config).get_resource(v1_hash) self.assertIsNotNone(resource) self.assertEqual(resource.summary.qb_torrent_id, qb_hash) self.assertEqual( resource.summary.resource_id.info_hash_v1_hex, v1_hash ) self.assertEqual( resource.summary.resource_id.info_hash_v2_hex, v2_hash ) def test_current_empty_204_login_is_accepted(self): responses = [(b"", 204), b"[]"] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): resources = QBittorrentReader(config).list_resources() self.assertEqual(resources, []) self.assertEqual(len(opener.calls), 2) def test_stopped_add_selection_recheck_and_entry_only_delete(self): torrent_hash = "a" * 40 responses = [ b"Ok.", # login b"Ok.", # multipart add b"", # skip all files b"", # select requested files b"", # stop before recheck b'{"total_downloaded":0}', b"", # recheck json.dumps([{ "hash": torrent_hash, "state": "stoppedDL", }]).encode(), b'{"total_downloaded":0}', b'[{"index":0,"progress":0},{"index":1,"progress":0}]', json.dumps([{ "hash": torrent_hash, "state": "checkingUP", }]).encode(), b'{"total_downloaded":0}', b'[{"index":0,"progress":1},{"index":1,"progress":0}]', json.dumps([{ "hash": torrent_hash, "state": "stoppedUP", }]).encode(), b'{"total_downloaded":0}', b'[{"index":0,"progress":1},{"index":1,"progress":0}]', b"", # start after successful recheck b"", # entry-only delete ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): adapter = QBittorrentReader(config) adapter.add_stopped(b"torrent", "/downloads/archive") adapter.set_selection(torrent_hash, [0], 2) result = adapter.recheck_and_wait( torrent_hash, [0], timeout=1, poll_interval=0 ) adapter.start(torrent_hash) adapter.delete_entry(torrent_hash) self.assertEqual(result.final_state, "stoppedUP") self.assertEqual(result.selected_file_indices, (0,)) calls = [ call for call in opener.calls if not isinstance(call, str) ] add = calls[1] self.assertIn(b'name="stopped"\r\n\r\ntrue', add.data) form_bodies = [ call.data.decode() for call in calls[2:] if getattr(call, "data", None) ] self.assertIn("id=0%7C1", form_bodies[0]) self.assertIn("priority=0", form_bodies[0]) self.assertIn("priority=1", form_bodies[1]) self.assertIn("deleteFiles=false", form_bodies[-1]) def test_stop_falls_back_to_qbittorrent_4_pause_endpoint(self): missing = error.HTTPError( "http://qb/api/v2/torrents/stop", 404, "not found", {}, None ) responses = [b"Ok.", missing, b""] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): QBittorrentReader(config).stop("a" * 40) self.assertTrue(opener.calls[-1].full_url.endswith("/torrents/pause")) def test_start_falls_back_to_qbittorrent_4_resume_endpoint(self): missing = error.HTTPError( "http://qb/api/v2/torrents/start", 404, "not found", {}, None ) responses = [b"Ok.", missing, b""] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): QBittorrentReader(config).start("a" * 40) self.assertTrue(opener.calls[-1].full_url.endswith("/torrents/resume")) def test_wait_until_present_polls_until_added_torrent_is_visible(self): torrent_hash = "a" * 40 responses = [ b"Ok.", b"[]", json.dumps([{"hash": torrent_hash}]).encode(), ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): QBittorrentReader(config).wait_until_present( torrent_hash, timeout=1, poll_interval=0 ) self.assertEqual(len(opener.calls), 3) def test_stopped_add_retries_while_recently_deleted_hash_is_busy(self): torrent_hash = "a" * 40 responses = [ b"Ok.", b"Fails.", b"[]", b"Ok.", json.dumps([{"hash": torrent_hash}]).encode(), ] with tempfile.TemporaryDirectory() as directory: root = Path(directory) password = root / "password" password.write_text("secret", encoding="utf-8") os.chmod(password, 0o600) config = ServiceConfig( "http://qb", PurePosixPath("/downloads"), root, username="admin", password_file=password, ) opener = _Opener(responses) with patch( "archive_clients.qbittorrent.request.build_opener", return_value=opener, ): QBittorrentReader(config).add_stopped_with_retry( b"torrent", "/downloads", torrent_hash, max_attempts=2, initial_delay=0, ) self.assertEqual(len(opener.calls), 5) if __name__ == "__main__": unittest.main()