"""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 re 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; "busy" is advertised but unbuilt, # which the next case pins. 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_advertised_but_not_built(self): """TASK_STATS_BUSY_NETWORKS is in the enum and in the mode table, and no switch anywhere handles it, so the worker frees the empty answer and the request ends as a server error. Pinned rather than blessed: either the report gets written or the mode leaves the table.""" with self.assertRaises(urllib.error.HTTPError) as caught: self.tracker.stats("busy") self.assertEqual(caught.exception.code, 500) 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 test_the_slash24_report_names_the_loopback_network(self): 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 masked out of a v4 peer's four address bytes and then # printed through the v6 formatter, so 127.0.0.0/24 reads as "7f00::" # rather than as an address anyone would recognise. Pinned as it is. self.assertIn(b"7f00::", body) if __name__ == "__main__": unittest.main()