mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-09-30 11:05:04 +01:00
Alex
a560089ce3
- Add new booklore API file formats - Renamed cookie for better login persistence with reverse proxy - Updated fs.py to try hardlink before atomic move from tmp dir - Fix transmission URL parsing - Fix scenario where file processing of huge files starves the healthcheck - Large enhancements to custom scripting, including passing JSON download info, more consistent activation across output types, decoupling from staging behavior, and added full documentation.
432 lines
15 KiB
Python
432 lines
15 KiB
Python
"""
|
|
Tests for torrent utility functions.
|
|
|
|
Tests:
|
|
- parse_transmission_url
|
|
- bencode_encode/decode
|
|
- extract_info_hash_from_torrent
|
|
- extract_hash_from_magnet
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
|
|
import pytest
|
|
|
|
from shelfmark.release_sources.prowlarr.clients.torrent_utils import (
|
|
parse_transmission_url,
|
|
bencode_decode,
|
|
bencode_encode,
|
|
extract_info_hash_from_torrent,
|
|
extract_hash_from_magnet,
|
|
)
|
|
|
|
|
|
class TestParseTransmissionUrl:
|
|
"""Tests for parse_transmission_url function."""
|
|
|
|
def test_parse_simple_url(self):
|
|
"""Test parsing a simple URL with host and port."""
|
|
protocol, host, port, path = parse_transmission_url("http://localhost:9091")
|
|
assert protocol == "http"
|
|
assert host == "localhost"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_url_with_custom_port(self):
|
|
"""Test parsing URL with custom port."""
|
|
protocol, host, port, path = parse_transmission_url("http://myserver:8080")
|
|
assert protocol == "http"
|
|
assert host == "myserver"
|
|
assert port == 8080
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_url_with_path(self):
|
|
"""Test parsing URL with existing path."""
|
|
protocol, host, port, path = parse_transmission_url("http://localhost:9091/transmission/rpc")
|
|
assert protocol == "http"
|
|
assert host == "localhost"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_url_with_partial_path(self):
|
|
"""Test parsing URL with partial path appends /rpc."""
|
|
protocol, host, port, path = parse_transmission_url("http://localhost:9091/custom")
|
|
assert protocol == "http"
|
|
assert host == "localhost"
|
|
assert port == 9091
|
|
assert path == "/custom/transmission/rpc"
|
|
|
|
def test_parse_url_with_trailing_slash(self):
|
|
"""Test parsing URL with trailing slash."""
|
|
protocol, host, port, path = parse_transmission_url("http://localhost:9091/")
|
|
assert protocol == "http"
|
|
assert host == "localhost"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_url_without_port(self):
|
|
"""Test parsing URL without port uses default 9091."""
|
|
protocol, host, port, path = parse_transmission_url("http://transmission")
|
|
assert protocol == "http"
|
|
assert host == "transmission"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_https_url(self):
|
|
"""Test parsing HTTPS URL."""
|
|
protocol, host, port, path = parse_transmission_url("https://secure.transmission.local:9091")
|
|
assert protocol == "https"
|
|
assert host == "secure.transmission.local"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_url_with_ip_address(self):
|
|
"""Test parsing URL with IP address."""
|
|
protocol, host, port, path = parse_transmission_url("http://192.168.1.100:9091")
|
|
assert protocol == "http"
|
|
assert host == "192.168.1.100"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
def test_parse_empty_url_uses_defaults(self):
|
|
"""Test parsing empty URL uses localhost defaults."""
|
|
protocol, host, port, path = parse_transmission_url("")
|
|
assert protocol == "http"
|
|
assert host == "localhost"
|
|
assert port == 9091
|
|
assert path == "/transmission/rpc"
|
|
|
|
|
|
class TestBencodeDecode:
|
|
"""Tests for bencode decoding."""
|
|
|
|
def test_decode_integer(self):
|
|
"""Test decoding integers."""
|
|
result, remaining = bencode_decode(b"i42e")
|
|
assert result == 42
|
|
assert remaining == b""
|
|
|
|
def test_decode_negative_integer(self):
|
|
"""Test decoding negative integers."""
|
|
result, remaining = bencode_decode(b"i-42e")
|
|
assert result == -42
|
|
|
|
def test_decode_zero(self):
|
|
"""Test decoding zero."""
|
|
result, remaining = bencode_decode(b"i0e")
|
|
assert result == 0
|
|
|
|
def test_decode_large_integer(self):
|
|
"""Test decoding large integers."""
|
|
result, remaining = bencode_decode(b"i999999999999e")
|
|
assert result == 999999999999
|
|
|
|
def test_decode_string(self):
|
|
"""Test decoding byte strings."""
|
|
result, remaining = bencode_decode(b"5:hello")
|
|
assert result == b"hello"
|
|
assert remaining == b""
|
|
|
|
def test_decode_empty_string(self):
|
|
"""Test decoding empty string."""
|
|
result, remaining = bencode_decode(b"0:")
|
|
assert result == b""
|
|
|
|
def test_decode_unicode_string(self):
|
|
"""Test decoding unicode bytes."""
|
|
data = "tëst".encode("utf-8")
|
|
encoded = f"{len(data)}:".encode() + data
|
|
result, remaining = bencode_decode(encoded)
|
|
assert result == data
|
|
|
|
def test_decode_list(self):
|
|
"""Test decoding lists."""
|
|
result, remaining = bencode_decode(b"li1ei2ei3ee")
|
|
assert result == [1, 2, 3]
|
|
assert remaining == b""
|
|
|
|
def test_decode_empty_list(self):
|
|
"""Test decoding empty list."""
|
|
result, remaining = bencode_decode(b"le")
|
|
assert result == []
|
|
|
|
def test_decode_nested_list(self):
|
|
"""Test decoding nested lists."""
|
|
result, remaining = bencode_decode(b"lli1eeli2eee")
|
|
assert result == [[1], [2]]
|
|
|
|
def test_decode_mixed_list(self):
|
|
"""Test decoding list with mixed types."""
|
|
result, remaining = bencode_decode(b"l5:helloi42ee")
|
|
assert result == [b"hello", 42]
|
|
|
|
def test_decode_dict(self):
|
|
"""Test decoding dictionaries."""
|
|
result, remaining = bencode_decode(b"d3:key5:valuee")
|
|
assert result == {b"key": b"value"}
|
|
assert remaining == b""
|
|
|
|
def test_decode_empty_dict(self):
|
|
"""Test decoding empty dictionary."""
|
|
result, remaining = bencode_decode(b"de")
|
|
assert result == {}
|
|
|
|
def test_decode_complex_structure(self):
|
|
"""Test decoding complex nested structures."""
|
|
# Dict with string, int, and list values
|
|
data = b"d3:agei25e4:name4:John5:itemsli1ei2ei3eee"
|
|
result, remaining = bencode_decode(data)
|
|
assert result == {
|
|
b"age": 25,
|
|
b"name": b"John",
|
|
b"items": [1, 2, 3],
|
|
}
|
|
|
|
def test_decode_invalid_data_raises(self):
|
|
"""Test that invalid data raises ValueError."""
|
|
with pytest.raises(ValueError):
|
|
bencode_decode(b"x")
|
|
|
|
|
|
class TestBencodeEncode:
|
|
"""Tests for bencode encoding."""
|
|
|
|
def test_encode_integer(self):
|
|
"""Test encoding integers."""
|
|
assert bencode_encode(42) == b"i42e"
|
|
assert bencode_encode(-42) == b"i-42e"
|
|
assert bencode_encode(0) == b"i0e"
|
|
|
|
def test_encode_bytes(self):
|
|
"""Test encoding byte strings."""
|
|
assert bencode_encode(b"hello") == b"5:hello"
|
|
assert bencode_encode(b"") == b"0:"
|
|
|
|
def test_encode_string(self):
|
|
"""Test encoding regular strings (UTF-8 encoded)."""
|
|
assert bencode_encode("hello") == b"5:hello"
|
|
assert bencode_encode("") == b"0:"
|
|
|
|
def test_encode_list(self):
|
|
"""Test encoding lists."""
|
|
assert bencode_encode([1, 2, 3]) == b"li1ei2ei3ee"
|
|
assert bencode_encode([]) == b"le"
|
|
|
|
def test_encode_dict(self):
|
|
"""Test encoding dictionaries."""
|
|
result = bencode_encode({b"key": b"value"})
|
|
assert result == b"d3:key5:valuee"
|
|
|
|
def test_encode_dict_keys_sorted(self):
|
|
"""Test that dictionary keys are sorted."""
|
|
# Keys should be sorted: a < m < z
|
|
result = bencode_encode({b"z": 1, b"a": 2, b"m": 3})
|
|
assert result == b"d1:ai2e1:mi3e1:zi1ee"
|
|
|
|
def test_encode_nested_structure(self):
|
|
"""Test encoding nested structures."""
|
|
data = {b"list": [1, 2, 3], b"num": 42}
|
|
result = bencode_encode(data)
|
|
assert result == b"d4:listli1ei2ei3ee3:numi42ee"
|
|
|
|
|
|
|
|
def test_encode_invalid_type_raises(self):
|
|
"""Test that invalid types raise ValueError."""
|
|
with pytest.raises(ValueError):
|
|
bencode_encode(3.14) # floats not supported
|
|
|
|
|
|
class TestBencodeRoundTrip:
|
|
"""Tests for encoding then decoding (roundtrip)."""
|
|
|
|
def test_roundtrip_integer(self):
|
|
"""Test roundtrip for integers."""
|
|
original = 12345
|
|
encoded = bencode_encode(original)
|
|
decoded, _ = bencode_decode(encoded)
|
|
assert decoded == original
|
|
|
|
def test_roundtrip_bytes(self):
|
|
"""Test roundtrip for byte strings."""
|
|
original = b"hello world"
|
|
encoded = bencode_encode(original)
|
|
decoded, _ = bencode_decode(encoded)
|
|
assert decoded == original
|
|
|
|
def test_roundtrip_list(self):
|
|
"""Test roundtrip for lists."""
|
|
original = [1, 2, b"three", [4, 5]]
|
|
encoded = bencode_encode(original)
|
|
decoded, _ = bencode_decode(encoded)
|
|
assert decoded == original
|
|
|
|
def test_roundtrip_dict(self):
|
|
"""Test roundtrip for dictionaries."""
|
|
original = {b"name": b"test", b"value": 123}
|
|
encoded = bencode_encode(original)
|
|
decoded, _ = bencode_decode(encoded)
|
|
assert decoded == original
|
|
|
|
def test_roundtrip_complex_torrent_like_structure(self):
|
|
"""Test roundtrip for a structure similar to a torrent file."""
|
|
original = {
|
|
b"announce": b"http://tracker.example.com/announce",
|
|
b"info": {
|
|
b"name": b"TestFile.txt",
|
|
b"length": 1024,
|
|
b"piece length": 16384,
|
|
b"pieces": b"\x00" * 20, # SHA1 hashes
|
|
},
|
|
}
|
|
encoded = bencode_encode(original)
|
|
decoded, _ = bencode_decode(encoded)
|
|
assert decoded == original
|
|
|
|
|
|
class TestExtractInfoHash:
|
|
"""Tests for extracting info hash from torrent files."""
|
|
|
|
def test_extract_hash_from_simple_torrent(self):
|
|
"""Test extracting hash from a simple torrent structure."""
|
|
info_dict = {
|
|
b"name": b"test.txt",
|
|
b"length": 100,
|
|
b"piece length": 16384,
|
|
b"pieces": b"\x00" * 20,
|
|
}
|
|
torrent = {b"info": info_dict}
|
|
torrent_bytes = bencode_encode(torrent)
|
|
|
|
result = extract_info_hash_from_torrent(torrent_bytes)
|
|
|
|
# Should return a 40-character hex string
|
|
assert result is not None
|
|
assert len(result) == 40
|
|
assert all(c in "0123456789abcdef" for c in result)
|
|
|
|
def test_extract_hash_returns_none_for_invalid(self):
|
|
"""Test that invalid data returns None."""
|
|
assert extract_info_hash_from_torrent(b"not a torrent") is None
|
|
assert extract_info_hash_from_torrent(b"") is None
|
|
|
|
def test_extract_hash_returns_none_without_info(self):
|
|
"""Test that torrent without info dict returns None."""
|
|
torrent = {b"announce": b"http://tracker.example.com"}
|
|
torrent_bytes = bencode_encode(torrent)
|
|
|
|
result = extract_info_hash_from_torrent(torrent_bytes)
|
|
assert result is None
|
|
|
|
def test_extract_hash_is_consistent(self):
|
|
"""Test that same torrent always produces same hash."""
|
|
info_dict = {b"name": b"consistent.txt", b"length": 500}
|
|
torrent = {b"info": info_dict}
|
|
torrent_bytes = bencode_encode(torrent)
|
|
|
|
hash1 = extract_info_hash_from_torrent(torrent_bytes)
|
|
hash2 = extract_info_hash_from_torrent(torrent_bytes)
|
|
|
|
assert hash1 == hash2
|
|
|
|
def test_extract_hash_different_for_different_torrents(self):
|
|
"""Test that different torrents produce different hashes."""
|
|
torrent1 = {b"info": {b"name": b"file1.txt", b"length": 100}}
|
|
torrent2 = {b"info": {b"name": b"file2.txt", b"length": 100}}
|
|
|
|
hash1 = extract_info_hash_from_torrent(bencode_encode(torrent1))
|
|
hash2 = extract_info_hash_from_torrent(bencode_encode(torrent2))
|
|
|
|
assert hash1 != hash2
|
|
|
|
def test_extract_hash_v2_without_pieces(self):
|
|
"""Use SHA-256 when torrent lacks v1 pieces."""
|
|
info_dict = {
|
|
b"meta version": 2,
|
|
b"file tree": {b"test.txt": {b"": {b"length": 123}}},
|
|
b"piece length": 16384,
|
|
}
|
|
torrent = {b"info": info_dict}
|
|
torrent_bytes = bencode_encode(torrent)
|
|
|
|
expected = hashlib.sha256(bencode_encode(info_dict)).hexdigest().lower()
|
|
assert extract_info_hash_from_torrent(torrent_bytes) == expected
|
|
|
|
|
|
class TestExtractHashFromMagnet:
|
|
"""Tests for extracting hash from magnet links."""
|
|
|
|
def test_extract_hash_from_hex_magnet(self):
|
|
"""Test extracting 40-char hex hash from magnet link."""
|
|
magnet = "magnet:?xt=urn:btih:3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0&dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result == "3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0"
|
|
|
|
def test_extract_hash_from_base32_magnet(self):
|
|
"""Test extracting 32-char base32 hash from magnet link."""
|
|
# Base32 encoded hash: "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"
|
|
magnet = "magnet:?xt=urn:btih:ABCDEFGHIJKLMNOPQRSTUVWXYZ234567&dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
# Should be converted to hex (lowercase)
|
|
assert result is not None
|
|
assert len(result) == 40
|
|
assert all(c in "0123456789abcdef" for c in result)
|
|
|
|
def test_extract_hash_uppercase_hex(self):
|
|
"""Test that uppercase hex is converted to lowercase."""
|
|
magnet = "magnet:?xt=urn:btih:3B245504CF5F11BBDBE1201CEA6A6BF45AEE1BC0&dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result == "3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0"
|
|
|
|
def test_extract_hash_no_btih(self):
|
|
"""Test that magnets without btih return None."""
|
|
magnet = "magnet:?dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result is None
|
|
|
|
def test_extract_hash_invalid_format(self):
|
|
"""Test that invalid hash format returns None."""
|
|
magnet = "magnet:?xt=urn:btih:invalid&dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result is None
|
|
|
|
def test_extract_hash_not_magnet(self):
|
|
"""Test that non-magnet URLs return None."""
|
|
result = extract_hash_from_magnet("https://example.com/file.torrent")
|
|
assert result is None
|
|
|
|
def test_extract_hash_empty_string(self):
|
|
"""Test that empty string returns None."""
|
|
result = extract_hash_from_magnet("")
|
|
assert result is None
|
|
|
|
def test_extract_hash_complex_magnet(self):
|
|
"""Test extracting from complex magnet with many parameters."""
|
|
magnet = (
|
|
"magnet:?xt=urn:btih:3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0"
|
|
"&dn=Ubuntu+22.04"
|
|
"&tr=udp://tracker.example.com:80"
|
|
"&tr=udp://tracker2.example.com:6969"
|
|
"&xl=12345"
|
|
)
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result == "3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0"
|
|
|
|
def test_extract_hash_from_btmh_hex(self):
|
|
"""Test extracting v2 hash from btmh (hex multihash)."""
|
|
digest = bytes(range(1, 33))
|
|
multihash = b"\x12\x20" + digest
|
|
magnet = f"magnet:?xt=urn:btmh:{multihash.hex()}&dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result == digest.hex()
|
|
|
|
def test_extract_hash_from_btmh_base32(self):
|
|
"""Test extracting v2 hash from btmh (base32 multihash)."""
|
|
digest = bytes(range(1, 33))
|
|
multihash = b"\x12\x20" + digest
|
|
b32 = base64.b32encode(multihash).decode("ascii").rstrip("=")
|
|
magnet = f"magnet:?xt=urn:btmh:{b32}&dn=test"
|
|
result = extract_hash_from_magnet(magnet)
|
|
assert result == digest.hex()
|