"""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"))