1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
"""The full scrape, in every format the tracker offers.
Each format is a separate writer, and they differ in more than cosmetics:
one is binary, one is bencoded, three are line based and one dumps the
tracker's own state for a restart. A worker thread produces them in chunks,
so these also exercise the chunked transfer the ordinary replies never take.
"""
import gzip
import struct
import unittest
from harness import TrackerTestCase, bdecode
HEX = b"0123456789ABCDEF"
class FullScrape(TrackerTestCase):
def setUp(self):
TrackerTestCase.setUp(self)
self.seeding = self.hash_for("fsA")
self.leeching = self.hash_for("fsB")
self.tracker.announce(self.seeding, port=6001, left=0)
self.tracker.announce(self.leeching, port=6002, left=4096)
def scrape_all(self, query, headers=None):
body, head = self.tracker.get_with_headers(query, headers)
if head.get("Content-Encoding") == "gzip":
body = gzip.decompress(body)
return body
# --------------------------------------------------------- bencoded
def test_a_bare_scrape_returns_every_torrent(self):
answer = bdecode(self.scrape_all("/scrape"))
self.assertIn(self.seeding, answer[b"files"])
self.assertIn(self.leeching, answer[b"files"])
self.assertEqual(answer[b"files"][self.seeding][b"complete"], 1)
self.assertEqual(answer[b"files"][self.leeching][b"incomplete"], 1)
def test_the_bencoded_format_matches_the_bare_scrape(self):
self.assertEqual(bdecode(self.scrape_all("/stats?mode=tpbs&format=ben")), bdecode(self.scrape_all("/scrape")))
def test_a_full_scrape_can_be_gzipped(self):
plain = self.scrape_all("/scrape")
body, head = self.tracker.get_with_headers("/scrape", {"Accept-Encoding": "gzip"})
self.assertEqual(head.get("Content-Encoding"), "gzip")
self.assertEqual(gzip.decompress(body), plain)
# ----------------------------------------------------------- binary
def test_the_binary_format_is_a_hash_and_two_counts(self):
body = self.scrape_all("/stats?mode=tpbs&format=bin")
self.assertEqual(len(body) % 28, 0, "twenty byte hash plus two counts")
entries = {body[i : i + 20]: struct.unpack(">II", body[i + 20 : i + 28]) for i in range(0, len(body), 28)}
self.assertEqual(entries[self.seeding], (1, 0), "one seeder, no leechers")
self.assertEqual(entries[self.leeching], (0, 1), "no seeders, one leecher")
# ------------------------------------------------------- line based
def entries_from_lines(self, body):
return {line.split(b":")[0]: line.split(b":")[1:] for line in body.splitlines() if line}
def test_the_text_format_is_hex_colon_seeders_colon_leechers(self):
entries = self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txt"))
self.assertEqual(entries[self.seeding.hex().upper().encode()], [b"1", b"0"])
self.assertEqual(entries[self.leeching.hex().upper().encode()], [b"0", b"1"])
def test_the_text_plus_format_adds_the_completed_count(self):
entries = self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txtp"))
self.assertEqual(entries[self.seeding.hex().upper().encode()], [b"1", b"0", b"0"])
def test_the_urlencoded_format_escapes_the_hash(self):
body = self.scrape_all("/stats?mode=tpbs&format=url")
lines = [line for line in body.splitlines() if line]
self.assertEqual(len(lines), 2)
for line in lines:
escaped = line.rsplit(b":", 2)[0]
self.assertIn(b"%", escaped, "these hashes are mostly unprintable bytes")
def test_the_state_dump_carries_a_base_time_rather_than_a_peer_count(self):
"""What a restart reads back, so its second column is the torrent's
own clock and not a count of anybody."""
entries = self.entries_from_lines(self.scrape_all("/stats?mode=statedump"))
self.assertIn(self.seeding.hex().upper().encode(), entries)
base, completed = entries[self.seeding.hex().upper().encode()]
self.assertGreater(int(base), 0)
self.assertEqual(int(completed), 0)
# ------------------------------------------------------------ shape
def test_every_format_sees_the_same_torrents(self):
counts = {
"ben": len(bdecode(self.scrape_all("/stats?mode=tpbs&format=ben"))[b"files"]),
"bin": len(self.scrape_all("/stats?mode=tpbs&format=bin")) // 28,
"txt": len(self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txt"))),
"txtp": len(self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txtp"))),
"url": len([l for l in self.scrape_all("/stats?mode=tpbs&format=url").splitlines() if l]),
}
self.assertEqual(set(counts.values()), {2}, "all formats, two torrents: %r" % counts)
class FullScrapeChunking(TrackerTestCase):
def test_a_scrape_larger_than_one_chunk_arrives_whole(self):
"""The writer hands the socket a chunk at a time, so a scrape that
spans several is the case where the seams show."""
for i in range(400):
self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=7000 + i, left=0)
answer = bdecode(self.tracker.get("/scrape"))
self.assertEqual(len(answer[b"files"]), 400)
if __name__ == "__main__":
unittest.main()
|
