mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-10-05 22:05:50 +01:00
601 lines
22 KiB
Python
601 lines
22 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
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from shelfmark.download.clients.torrent_utils import (
|
|
bencode_decode,
|
|
bencode_encode,
|
|
extract_hash_from_magnet,
|
|
extract_info_hash_from_torrent,
|
|
extract_torrent_info,
|
|
parse_transmission_url,
|
|
)
|
|
|
|
|
|
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()
|
|
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 TestExtractTorrentInfo:
|
|
"""Tests for extracting torrent info from user-supplied URLs."""
|
|
|
|
def test_fetches_untrusted_torrent_url_to_recover_missing_hash(self, monkeypatch):
|
|
"""Regression for #1012.
|
|
|
|
A download URL on a non-Prowlarr origin (e.g. a direct tracker link, or
|
|
Prowlarr reached through a separate proxy) must still be prefetched so
|
|
the info_hash can be recovered when the feed did not provide one.
|
|
Otherwise qBittorrent fails with "Could not determine torrent hash from
|
|
URL".
|
|
"""
|
|
info_dict = {
|
|
b"name": b"book.txt",
|
|
b"length": 100,
|
|
b"piece length": 16384,
|
|
b"pieces": b"\x00" * 20,
|
|
}
|
|
torrent_data = bencode_encode({b"info": info_dict})
|
|
expected_hash = hashlib.sha1(bencode_encode(info_dict)).hexdigest().lower()
|
|
|
|
config_values = {"PROWLARR_URL": "https://prowlarr.example"}
|
|
monkeypatch.setattr(
|
|
"shelfmark.download.clients.torrent_utils.config.get",
|
|
lambda key, default="": config_values.get(key, default),
|
|
)
|
|
response = MagicMock(status_code=200, content=torrent_data)
|
|
response.raise_for_status = MagicMock()
|
|
mock_get = MagicMock(return_value=response)
|
|
monkeypatch.setattr("shelfmark.download.clients.torrent_utils.requests.get", mock_get)
|
|
|
|
# No expected_hash supplied: the hash can only come from the prefetch.
|
|
result = extract_torrent_info(
|
|
"https://tracker.example/download/book.torrent",
|
|
fetch_torrent=True,
|
|
)
|
|
|
|
assert result.info_hash == expected_hash
|
|
assert result.torrent_data == torrent_data
|
|
assert result.is_magnet is False
|
|
mock_get.assert_called_once()
|
|
|
|
def test_does_not_send_api_key_to_untrusted_torrent_url(self, monkeypatch):
|
|
"""The Prowlarr API key is never sent to a download URL on an untrusted origin."""
|
|
info_dict = {
|
|
b"name": b"book.txt",
|
|
b"length": 100,
|
|
b"piece length": 16384,
|
|
b"pieces": b"\x00" * 20,
|
|
}
|
|
torrent_data = bencode_encode({b"info": info_dict})
|
|
|
|
config_values = {
|
|
"PROWLARR_URL": "https://prowlarr.example",
|
|
"PROWLARR_API_KEY": "secret",
|
|
}
|
|
monkeypatch.setattr(
|
|
"shelfmark.download.clients.torrent_utils.config.get",
|
|
lambda key, default="": config_values.get(key, default),
|
|
)
|
|
response = MagicMock(status_code=200, content=torrent_data)
|
|
response.raise_for_status = MagicMock()
|
|
mock_get = MagicMock(return_value=response)
|
|
monkeypatch.setattr("shelfmark.download.clients.torrent_utils.requests.get", mock_get)
|
|
|
|
extract_torrent_info(
|
|
"https://attacker.example/book.torrent",
|
|
fetch_torrent=True,
|
|
)
|
|
|
|
mock_get.assert_called_once()
|
|
sent_headers = mock_get.call_args.kwargs.get("headers", {})
|
|
assert "X-Api-Key" not in sent_headers
|
|
|
|
def test_fetches_configured_prowlarr_torrent_url(self, monkeypatch):
|
|
"""Configured Prowlarr download URLs can still be prefetched and parsed."""
|
|
info_dict = {
|
|
b"name": b"trusted.txt",
|
|
b"length": 100,
|
|
b"piece length": 16384,
|
|
b"pieces": b"\x00" * 20,
|
|
}
|
|
torrent_data = bencode_encode({b"info": info_dict})
|
|
expected_hash = hashlib.sha1(bencode_encode(info_dict)).hexdigest().lower()
|
|
|
|
config_values = {
|
|
"PROWLARR_URL": "https://prowlarr.example",
|
|
"PROWLARR_API_KEY": "secret",
|
|
}
|
|
monkeypatch.setattr(
|
|
"shelfmark.download.clients.torrent_utils.config.get",
|
|
lambda key, default="": config_values.get(key, default),
|
|
)
|
|
response = MagicMock(status_code=200, content=torrent_data)
|
|
response.raise_for_status = MagicMock()
|
|
mock_get = MagicMock(return_value=response)
|
|
monkeypatch.setattr("shelfmark.download.clients.torrent_utils.requests.get", mock_get)
|
|
|
|
result = extract_torrent_info(
|
|
"https://prowlarr.example/1/download?apikey=secret&indexer=7",
|
|
fetch_torrent=True,
|
|
)
|
|
|
|
assert result.info_hash == expected_hash
|
|
assert result.torrent_data == torrent_data
|
|
assert result.is_magnet is False
|
|
mock_get.assert_called_once()
|
|
|
|
def test_normalizes_configured_origin_before_trusting_torrent_url(self, monkeypatch):
|
|
"""Configured Prowlarr URLs match the same normalization used by the source."""
|
|
info_dict = {
|
|
b"name": b"trusted.txt",
|
|
b"length": 100,
|
|
b"piece length": 16384,
|
|
b"pieces": b"\x00" * 20,
|
|
}
|
|
torrent_data = bencode_encode({b"info": info_dict})
|
|
expected_hash = hashlib.sha1(bencode_encode(info_dict)).hexdigest().lower()
|
|
|
|
config_values = {
|
|
"PROWLARR_URL": "prowlarr.example:9696/",
|
|
"PROWLARR_API_KEY": "secret",
|
|
}
|
|
monkeypatch.setattr(
|
|
"shelfmark.download.clients.torrent_utils.config.get",
|
|
lambda key, default="": config_values.get(key, default),
|
|
)
|
|
response = MagicMock(status_code=200, content=torrent_data)
|
|
response.raise_for_status = MagicMock()
|
|
mock_get = MagicMock(return_value=response)
|
|
monkeypatch.setattr("shelfmark.download.clients.torrent_utils.requests.get", mock_get)
|
|
|
|
result = extract_torrent_info(
|
|
"http://prowlarr.example:9696/1/download?apikey=secret&indexer=7",
|
|
fetch_torrent=True,
|
|
)
|
|
|
|
assert result.info_hash == expected_hash
|
|
assert result.torrent_data == torrent_data
|
|
mock_get.assert_called_once()
|
|
|
|
def test_does_not_follow_trusted_torrent_url_redirect_to_untrusted_host(self, monkeypatch):
|
|
"""Trusted HTTP prefetch does not continue through arbitrary redirects."""
|
|
expected_hash = "3b245504cf5f11bbdbe1201cea6a6bf45aee1bc0"
|
|
monkeypatch.setattr(
|
|
"shelfmark.download.clients.torrent_utils.config.get",
|
|
lambda key, default="": "https://prowlarr.example" if key == "PROWLARR_URL" else "",
|
|
)
|
|
response = MagicMock(status_code=302)
|
|
response.headers = {"Location": "https://attacker.example/book.torrent"}
|
|
mock_get = MagicMock(return_value=response)
|
|
monkeypatch.setattr("shelfmark.download.clients.torrent_utils.requests.get", mock_get)
|
|
|
|
result = extract_torrent_info(
|
|
"https://prowlarr.example/1/download?apikey=secret&indexer=7",
|
|
fetch_torrent=True,
|
|
expected_hash=expected_hash,
|
|
)
|
|
|
|
assert result.info_hash == expected_hash
|
|
assert result.torrent_data is None
|
|
assert result.is_magnet is False
|
|
mock_get.assert_called_once()
|
|
|
|
|
|
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()
|