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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
"""What a peer's announce does to the swarm it joins."""
import os
import socket
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
from harness import TrackerTestCase
def peers_of(answer, key=b"peers"):
"""Compact peer list as a list of (address, port)."""
blob, out = answer.get(key, b""), []
size = 6 if key == b"peers" else 18
for offset in range(0, len(blob), size):
entry = blob[offset : offset + size]
address = socket.inet_ntop(socket.AF_INET if size == 6 else socket.AF_INET6, entry[:-2])
out.append((address, entry[-2] << 8 | entry[-1]))
return out
class AnnounceLifetime(TrackerTestCase):
config = ""
def test_a_leecher_is_counted_as_incomplete(self):
info_hash = self.hash_for("leecher")
answer = self.tracker.announce(info_hash, left=100)
self.assertEqual(answer[b"complete"], 0)
self.assertEqual(answer[b"incomplete"], 1)
def test_no_bytes_left_means_seeding(self):
info_hash = self.hash_for("seeder")
answer = self.tracker.announce(info_hash, left=0)
self.assertEqual(answer[b"complete"], 1)
self.assertEqual(answer[b"incomplete"], 0)
def test_a_peer_that_finishes_moves_from_one_column_to_the_other(self):
info_hash = self.hash_for("finishes")
peer_id = b"-OT0001-finisher0001"
self.tracker.announce(info_hash, peer_id=peer_id, left=100)
self.assertEqual(self.tracker.announce(info_hash, peer_id=peer_id, left=100)[b"incomplete"], 1)
answer = self.tracker.announce(info_hash, peer_id=peer_id, left=0, event="completed")
self.assertEqual(answer[b"complete"], 1, "the peer should now count as a seeder")
self.assertEqual(answer[b"incomplete"], 0, "and no longer as a leecher")
def test_stopped_removes_the_peer(self):
info_hash = self.hash_for("stopped")
peer_id = b"-OT0001-leaves000001"
self.tracker.announce(info_hash, peer_id=peer_id, left=100)
self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"incomplete"], 1)
self.tracker.announce(info_hash, peer_id=peer_id, left=100, event="stopped")
self.assertEqual(
self.tracker.scrape(info_hash)[b"files"][info_hash][b"incomplete"], 0, "the peer should be gone"
)
def test_the_same_peer_announcing_twice_is_still_one_peer(self):
info_hash = self.hash_for("twice")
peer_id = b"-OT0001-repeats00001"
for _ in range(5):
answer = self.tracker.announce(info_hash, peer_id=peer_id, left=100)
self.assertEqual(answer[b"incomplete"], 1, "re-announcing must not duplicate the peer")
def test_a_peer_is_its_address_and_port_not_its_peer_id(self):
"""Two announces from one address and port are one peer, whatever
peer_id they carry, because that pair is the whole identity."""
info_hash = self.hash_for("identity")
self.tracker.announce(info_hash, peer_id=b"-OT0001-a00000000001", port=6881, left=100)
answer = self.tracker.announce(info_hash, peer_id=b"-OT0001-b00000000002", port=6881, left=100)
self.assertEqual(answer[b"incomplete"], 1, "same address and port, so still one peer")
answer = self.tracker.announce(info_hash, peer_id=b"-OT0001-c00000000003", port=6882, left=100)
self.assertEqual(answer[b"incomplete"], 2, "a different port makes a different peer")
def test_a_downloaded_count_is_kept(self):
info_hash = self.hash_for("downloaded")
self.tracker.announce(info_hash, port=7001, left=0, event="completed")
self.tracker.announce(info_hash, port=7002, left=0, event="completed")
self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"downloaded"], 2)
def test_a_peer_is_given_the_whole_swarm_including_itself(self):
"""opentracker does not filter the announcing peer out of its own
answer. Clients are expected to ignore themselves."""
info_hash = self.hash_for("sees")
self.tracker.announce(info_hash, port=1111, left=100)
answer = self.tracker.announce(info_hash, port=2222, left=100)
ports = sorted(port for _, port in peers_of(answer))
self.assertEqual(ports, [1111, 2222])
def test_an_unknown_torrent_scrapes_as_empty(self):
answer = self.tracker.scrape(self.hash_for("never-seen"))
self.assertEqual(answer[b"files"], {}, "a torrent nobody announced has no entry")
class AnnounceRejects(TrackerTestCase):
def assert_rejected(self, query):
import urllib.error
with self.assertRaises(urllib.error.HTTPError) as caught:
self.tracker.get("/announce?" + query)
self.assertEqual(caught.exception.code, 400)
def test_a_short_info_hash_is_rejected(self):
self.assert_rejected("info_hash=tooshort&peer_id=" + "b" * 20 + "&port=6881&left=0")
def test_a_missing_info_hash_answers_a_failure_reason(self):
"""Not a 400: the tracker answers 200 with a bencoded complaint, so
that a client shows the message to its user."""
from harness import bdecode
answer = bdecode(self.tracker.get("/announce?peer_id=" + "b" * 20 + "&port=6881&left=0"))
self.assertIn(b"failure reason", answer)
self.assertIn(b"info_hash", answer[b"failure reason"])
def test_a_port_out_of_range_is_rejected(self):
self.assert_rejected("info_hash=" + "a" * 20 + "&peer_id=" + "b" * 20 + "&port=65536&left=0")
def test_a_port_that_is_not_a_number_is_rejected(self):
self.assert_rejected("info_hash=" + "a" * 20 + "&peer_id=" + "b" * 20 + "&port=abc&left=0")
def test_two_info_hashes_are_rejected(self):
self.assert_rejected("info_hash=" + "a" * 20 + "&info_hash=" + "c" * 20 + "&peer_id=" + "b" * 20 + "&port=6881&left=0")
if __name__ == "__main__":
unittest.main()
|
