101 lines
3.2 KiB
Python
101 lines
3.2 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 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, 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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|