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
|
"""Settings the tracker has to accept, and still work with.
An unrecognised key only warns, so coming up proves nothing on its own and a
test that cares has to look at what the tracker said. A key that is
recognised but cannot be parsed is another matter, and for the listen
statements it stops the tracker.
"""
import unittest
from harness import Tracker, TrackerTestCase, free_port
class TcpWorkers(TrackerTestCase):
config = "listen.tcp.workers 2\n"
def test_the_tracker_serves_with_tcp_workers_asked_for(self):
self.assertEqual(self.tracker.announce(self.hash_for("tcpworkers"))[b"complete"], 1)
self.assertNotIn("Unhandled line", self.tracker.output())
class WorkersOnAListenStatement(unittest.TestCase):
def test_a_listen_statement_may_carry_a_worker_count(self):
"""The option is read off the statement rather than inherited."""
second = free_port()
tracker = Tracker("listen.tcp 127.0.0.1:%d workers=2\n" % second)
try:
self.assertEqual(tracker.announce(b"listenopts\0\0\0\0\0\0\0\0\0\0")[b"complete"], 1)
complaints = tracker.stop()
self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints))
finally:
tracker._cleanup()
class AListenLineThatDoesNotParse(unittest.TestCase):
"""Warning and carrying on left a tracker that had been told where to
listen coming up on its default instead, which is every interface."""
def refuses(self, line):
with self.assertRaises(AssertionError) as caught:
Tracker(line)
return str(caught.exception)
def test_an_address_that_is_not_one_stops_the_tracker(self):
self.assertIn("want an address", self.refuses("listen.tcp 10.0.0.300:7999\n"))
def test_so_does_a_port_out_of_range(self):
self.assertIn("want an address", self.refuses("listen.udp 127.0.0.1:99999\n"))
def test_so_does_an_option_it_does_not_know(self):
self.assertIn("want an address", self.refuses("listen.tcp 127.0.0.1:7999 wokers=2\n"))
class EmptyTorrentTimeout(TrackerTestCase):
config = "tracker.empty_torrent_timeout 60\n"
def test_the_tracker_accepts_the_option_and_serves(self):
self.assertEqual(self.tracker.announce(self.hash_for("emptyto"))[b"complete"], 1)
self.assertNotIn("Unhandled line", self.tracker.output())
self.assertNotIn("Parse error", self.tracker.output())
class EmptyTorrentTimeoutThatDoesNotParse(unittest.TestCase):
def test_a_non_numeric_value_warns_but_the_tracker_serves(self):
"""A scalar that will not parse warns and is ignored, unlike a listen
line, which stops the tracker."""
tracker = Tracker("tracker.empty_torrent_timeout later\n")
try:
self.assertEqual(tracker.announce(b"emptytobad\0\0\0\0\0\0\0\0\0\0")[b"complete"], 1)
self.assertIn("Parse error", tracker.output())
finally:
tracker._cleanup()
|
