147 lines
5.0 KiB
Python
147 lines
5.0 KiB
Python
import hashlib
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from archive_clients.bencode import BencodeError, decode_metainfo, encode
|
|
from archive_clients.resources import (
|
|
ResourceError,
|
|
build_content_tree,
|
|
normalize_resource,
|
|
)
|
|
from archive_control.v1 import resource_pb2
|
|
|
|
|
|
class ResourceTests(unittest.TestCase):
|
|
def test_pure_v2_identity_and_single_file_tree_normalization(self):
|
|
info = {
|
|
b"file tree": {
|
|
b"": {
|
|
b"length": 3,
|
|
b"pieces root": b"x" * 32,
|
|
},
|
|
},
|
|
b"meta version": 2,
|
|
b"name": b"v2.bin",
|
|
b"piece length": 16384,
|
|
}
|
|
metainfo_bytes = encode({b"info": info})
|
|
decoded = decode_metainfo(metainfo_bytes)
|
|
self.assertEqual(decoded.info_hash_v1_hex, "")
|
|
self.assertEqual(
|
|
decoded.info_hash_v2_hex,
|
|
hashlib.sha256(encode(info)).hexdigest(),
|
|
)
|
|
self.assertEqual(decoded.files[0].path, "v2.bin")
|
|
|
|
normalized = normalize_resource(
|
|
{
|
|
"hash": decoded.info_hash_v2_hex,
|
|
"name": "v2.bin",
|
|
"state": "stoppedUP",
|
|
},
|
|
[{
|
|
"index": 0,
|
|
"name": "v2.bin",
|
|
"size": 3,
|
|
"completed": 3,
|
|
"priority": 1,
|
|
}],
|
|
metainfo_bytes,
|
|
)
|
|
self.assertEqual(
|
|
normalized.summary.resource_id.info_hash_v2_hex,
|
|
decoded.info_hash_v2_hex,
|
|
)
|
|
self.assertFalse(normalized.summary.resource_id.info_hash_v1_hex)
|
|
|
|
def test_hybrid_identity_and_selection_normalization(self):
|
|
info = {
|
|
b"file tree": {
|
|
b"a.txt": {b"": {b"length": 3}},
|
|
b"b.bin": {b"": {b"attr": b"p", b"length": 5}},
|
|
},
|
|
b"files": [
|
|
{b"length": 3, b"path": [b"a.txt"]},
|
|
{b"attr": b"p", b"length": 5, b"path": [b"b.bin"]},
|
|
],
|
|
b"meta version": 2,
|
|
b"name": b"resource",
|
|
b"piece length": 16384,
|
|
b"pieces": b"x" * 20,
|
|
}
|
|
metainfo_bytes = encode({b"announce": b"http://tracker", b"info": info})
|
|
decoded = decode_metainfo(metainfo_bytes)
|
|
encoded_info = encode(info)
|
|
self.assertEqual(
|
|
decoded.info_hash_v1_hex, hashlib.sha1(encoded_info).hexdigest()
|
|
)
|
|
self.assertEqual(
|
|
decoded.info_hash_v2_hex, hashlib.sha256(encoded_info).hexdigest()
|
|
)
|
|
self.assertEqual(decoded.files[1].path, "resource/b.bin")
|
|
self.assertTrue(decoded.files[1].padding)
|
|
|
|
observed = datetime(2026, 1, 2, tzinfo=timezone.utc)
|
|
normalized = normalize_resource(
|
|
{
|
|
"hash": decoded.info_hash_v1_hex,
|
|
"name": "resource",
|
|
"state": "stoppedUP",
|
|
},
|
|
[
|
|
{
|
|
"index": 0, "name": "resource/a.txt", "size": 3,
|
|
"progress": 1.0, "priority": 1,
|
|
},
|
|
{
|
|
"index": 1, "name": "resource/b.bin", "size": 5,
|
|
"progress": 0.5, "priority": 0,
|
|
},
|
|
],
|
|
metainfo_bytes,
|
|
observed,
|
|
)
|
|
summary = normalized.summary
|
|
self.assertEqual(
|
|
summary.runtime_state, resource_pb2.TORRENT_RUNTIME_STATE_STOPPED
|
|
)
|
|
self.assertEqual(summary.selected_files.ranges[0].first, 0)
|
|
self.assertEqual(summary.selected_complete_files.ranges[0].last, 0)
|
|
self.assertEqual(summary.selected_logical_bytes, 3)
|
|
self.assertTrue(summary.canonical_paths)
|
|
tree = build_content_tree(normalized.files, {0})
|
|
root = next(entry for entry in tree if entry.canonical_path == "resource")
|
|
self.assertEqual(root.available_file_count, 1)
|
|
self.assertEqual(root.available_logical_bytes, 3)
|
|
|
|
def test_noncanonical_and_unsafe_paths_are_distinct(self):
|
|
info = {
|
|
b"length": 3,
|
|
b"name": b"a.txt",
|
|
b"piece length": 16384,
|
|
b"pieces": b"x" * 20,
|
|
}
|
|
metainfo = encode({b"info": info})
|
|
torrent = {
|
|
"hash": hashlib.sha1(encode(info)).hexdigest(),
|
|
"name": "renamed", "state": "uploading",
|
|
}
|
|
renamed = normalize_resource(torrent, [{
|
|
"index": 0, "name": "renamed.txt", "size": 3,
|
|
"progress": 1.0, "priority": 1,
|
|
}], metainfo)
|
|
self.assertFalse(renamed.summary.canonical_paths)
|
|
with self.assertRaisesRegex(ResourceError, "unsafe"):
|
|
normalize_resource(torrent, [{
|
|
"index": 0, "name": "../escape", "size": 3,
|
|
"progress": 1.0, "priority": 1,
|
|
}], metainfo)
|
|
|
|
def test_noncanonical_bencode_is_rejected(self):
|
|
with self.assertRaisesRegex(BencodeError, "unsorted"):
|
|
decode_metainfo(b"d4:infod1:b1:x1:a1:yee")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|