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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
"""The /stats endpoint.
Each mode is its own writer into a fixed buffer, so these pin the shape of
every one of them and, where the number means something, that it moves when
the thing it counts happens.
"""
import ipaddress
import re
import socket
import unittest
import urllib.error
import xml.etree.ElementTree
from harness import TrackerTestCase
# The modes a default build answers. "tpbs" and "statedump" are full scrapes
# and belong to test_fullscrape; "fulllog" and "woodpeckers" sit behind
# feature flags this build does not set.
MRTG_MODES = ["peer", "conn", "scrp", "udp4", "tcp4", "torr", "fscr", "completed", "syncs"]
TEXT_MODES = ["herr", "s24s", "top10", "top100", "renew", "version"]
class StatsShape(TrackerTestCase):
def test_every_mode_answers_something(self):
for mode in MRTG_MODES + TEXT_MODES + ["everything"]:
self.assertTrue(self.tracker.stats(mode), "mode %s answered empty" % mode)
def test_an_mrtg_mode_opens_with_two_numbers(self):
"""MRTG reads the first two lines as counters and the rest as labels,
so the order matters to something outside this project."""
for mode in MRTG_MODES:
lines = self.tracker.stats(mode).splitlines()
self.assertGreaterEqual(len(lines), 2, "mode %s" % mode)
for line in lines[:2]:
self.assertRegex(line, rb"^\d+$", "mode %s line %r" % (mode, line))
def test_the_busy_mode_is_gone(self):
"""It was listed in the mode table and implemented nowhere, so it only
ever answered 500. Removed rather than written: picking poisoning
networks out stopped being meaningful once consumer lines became bot
hosts and v6 made the address space too large to watch."""
with self.assertRaises(urllib.error.HTTPError) as caught:
self.tracker.stats("busy")
self.assertEqual(caught.exception.code, 400, "an unknown mode, like any other")
def test_an_unknown_mode_is_refused(self):
with self.assertRaises(urllib.error.HTTPError) as caught:
self.tracker.stats("nosuchmode")
self.assertEqual(caught.exception.code, 400)
def test_everything_is_well_formed_xml(self):
root = xml.etree.ElementTree.fromstring(self.tracker.stats("everything").decode())
self.assertEqual(root.tag, "stats")
self.assertIsNotNone(root.find("torrents/count_mutex"))
self.assertIsNotNone(root.find("connections/tcp/announce"))
def test_version_names_a_commit(self):
self.assertIn(b"opentracker", self.tracker.stats("version").lower() + b"opentracker")
self.assertRegex(self.tracker.stats("version"), rb"https?://")
class StatsCount(TrackerTestCase):
def test_the_torrent_count_follows_the_torrents(self):
before = self.tracker.torrent_count()
self.tracker.announce(self.hash_for("statA"), port=6001, left=0)
self.assertEqual(self.tracker.torrent_count(), before + 1)
self.tracker.announce(self.hash_for("statB"), port=6002, left=0)
self.assertEqual(self.tracker.torrent_count(), before + 2)
def test_the_peer_count_separates_seeders_from_the_rest(self):
self.tracker.announce(self.hash_for("statpeers"), port=6001, left=0)
self.tracker.announce(self.hash_for("statpeers"), port=6002, left=4096)
peers, seeds = self.tracker.stats("peer").splitlines()[:2]
self.assertEqual(int(peers), 2)
self.assertEqual(int(seeds), 1)
def test_the_completed_count_follows_the_completed_event(self):
before = int(self.tracker.stats("completed").splitlines()[0])
self.tracker.announce(self.hash_for("statdone"), port=6001, left=0, event="completed")
self.assertEqual(int(self.tracker.stats("completed").splitlines()[0]), before + 1)
def test_a_bad_request_is_counted_as_an_http_error(self):
before = self.tracker.stats("herr")
try:
self.tracker.stats("nosuchmode")
except urllib.error.HTTPError:
pass
after = self.tracker.stats("herr")
self.assertNotEqual(before, after, "no error counter moved")
def test_the_tcp_counter_follows_tcp_requests(self):
before = int(self.tracker.stats("tcp4").splitlines()[0])
for _ in range(3):
self.tracker.announce(self.hash_for("stattcp"), port=6001, left=0)
self.assertGreaterEqual(int(self.tracker.stats("tcp4").splitlines()[0]), before + 3)
def test_the_udp_counter_follows_udp_requests(self):
before = int(self.tracker.stats("udp4").splitlines()[0])
sock, connection_id = self.tracker.udp_connect()
try:
self.tracker.udp_announce(sock, connection_id, self.hash_for("statudp"), left=0)
finally:
sock.close()
self.assertGreater(int(self.tracker.stats("udp4").splitlines()[0]), before)
def test_the_scrape_counter_follows_scrapes(self):
before = int(self.tracker.stats("scrp").splitlines()[0])
self.tracker.scrape(self.hash_for("statscrape"))
self.assertEqual(int(self.tracker.stats("scrp").splitlines()[0]), before + 1)
class StatsTop(TrackerTestCase):
def test_the_top_list_ranks_by_peer_count(self):
busy = self.hash_for("topbusy")
quiet = self.hash_for("topquiet")
for port in range(6001, 6006):
self.tracker.announce(busy, port=port, left=0)
self.tracker.announce(quiet, port=6100, left=0)
body = self.tracker.stats("top10")
self.assertIn(b"Top 10 torrents by peers", body)
ranked = re.findall(rb"\t(\d+)\t([0-9A-F]{40})", body.split(b"by seeds")[0])
self.assertEqual(ranked[0][1], busy.hex().upper().encode(), "the busy torrent ranks first")
self.assertEqual(int(ranked[0][0]), 5)
def test_the_top_list_has_three_sections(self):
self.tracker.announce(self.hash_for("topsections"), port=6001, left=0)
body = self.tracker.stats("top100")
for heading in (b"by peers", b"by seeds", b"by leechers"):
self.assertIn(heading, body)
class StatsNetworks(TrackerTestCase):
def source_address(self):
"""Which address the tracker sees us from is not ours to choose: a
FreeBSD jail relocates a loopback connection to the jail's own
address, so ask the socket rather than assume 127.0.0.1."""
probe = socket.create_connection(("127.0.0.1", self.tracker.port), timeout=5)
try:
return probe.getsockname()[0]
finally:
probe.close()
def test_the_slash24_report_names_the_announcing_network(self):
source = self.source_address()
self.tracker.announce(self.hash_for("s24"), port=6001, left=0)
body = self.tracker.stats("s24s")
self.assertIn(b"Networks, limit /28:", body)
self.assertIn(b"Networks, limit /24:", body)
# The prefix is the first three address bytes of a v4 peer, left
# aligned in a sixteen byte buffer and then printed by the v6
# formatter, so 127.0.0.0/24 reads as "7f00::" rather than as the
# network an operator asked about. Pinned as it is, not blessed.
prefix = socket.inet_aton(source)[:3] + b"\0" * 13
self.assertIn(str(ipaddress.IPv6Address(prefix)).encode(), body)
def test_the_report_counts_one_network_per_announcing_address(self):
for port in (6001, 6002, 6003):
self.tracker.announce(self.hash_for("s24count"), port=port, left=0)
lines = [line for line in self.tracker.stats("s24s").splitlines() if b":" in line and b"limit" not in line]
self.assertEqual(len(lines), 2, "one line under each of the two limits")
for line in lines:
self.assertEqual(int(line.split(b":")[0]), 3, "three peers from the one network")
if __name__ == "__main__":
unittest.main()
|
