386 lines
14 KiB
Python
386 lines
14 KiB
Python
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()
|