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) self.assertIn("hashes=", opener.calls[1]) 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()