From 9d21cb476d07ab5583c99b1418d4ffb54fbb2a10 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Tue, 27 Oct 2026 01:16:37 +0100 Subject: Adding the actual tests to do aforementioned 'Massively increase coverage by pulling fullscrapes and sending udp packets' --- tests/integration/test_fullscrape.py | 114 +++++++++++++++++++++++++++++++++++ 1 file changed, 114 insertions(+) create mode 100644 tests/integration/test_fullscrape.py (limited to 'tests/integration/test_fullscrape.py') diff --git a/tests/integration/test_fullscrape.py b/tests/integration/test_fullscrape.py new file mode 100644 index 0000000..4059656 --- /dev/null +++ b/tests/integration/test_fullscrape.py @@ -0,0 +1,114 @@ +"""The full scrape, in every format the tracker offers. + +Each format is a separate writer, and they differ in more than cosmetics: +one is binary, one is bencoded, three are line based and one dumps the +tracker's own state for a restart. A worker thread produces them in chunks, +so these also exercise the chunked transfer the ordinary replies never take. +""" + +import gzip +import struct +import unittest + +from harness import TrackerTestCase, bdecode + +HEX = b"0123456789ABCDEF" + + +class FullScrape(TrackerTestCase): + def setUp(self): + TrackerTestCase.setUp(self) + self.seeding = self.hash_for("fsA") + self.leeching = self.hash_for("fsB") + self.tracker.announce(self.seeding, port=6001, left=0) + self.tracker.announce(self.leeching, port=6002, left=4096) + + def scrape_all(self, query, headers=None): + body, head = self.tracker.get_with_headers(query, headers) + if head.get("Content-Encoding") == "gzip": + body = gzip.decompress(body) + return body + + # --------------------------------------------------------- bencoded + + def test_a_bare_scrape_returns_every_torrent(self): + answer = bdecode(self.scrape_all("/scrape")) + self.assertIn(self.seeding, answer[b"files"]) + self.assertIn(self.leeching, answer[b"files"]) + self.assertEqual(answer[b"files"][self.seeding][b"complete"], 1) + self.assertEqual(answer[b"files"][self.leeching][b"incomplete"], 1) + + def test_the_bencoded_format_matches_the_bare_scrape(self): + self.assertEqual(bdecode(self.scrape_all("/stats?mode=tpbs&format=ben")), bdecode(self.scrape_all("/scrape"))) + + def test_a_full_scrape_can_be_gzipped(self): + plain = self.scrape_all("/scrape") + body, head = self.tracker.get_with_headers("/scrape", {"Accept-Encoding": "gzip"}) + self.assertEqual(head.get("Content-Encoding"), "gzip") + self.assertEqual(gzip.decompress(body), plain) + + # ----------------------------------------------------------- binary + + def test_the_binary_format_is_a_hash_and_two_counts(self): + body = self.scrape_all("/stats?mode=tpbs&format=bin") + self.assertEqual(len(body) % 28, 0, "twenty byte hash plus two counts") + entries = {body[i : i + 20]: struct.unpack(">II", body[i + 20 : i + 28]) for i in range(0, len(body), 28)} + self.assertEqual(entries[self.seeding], (1, 0), "one seeder, no leechers") + self.assertEqual(entries[self.leeching], (0, 1), "no seeders, one leecher") + + # ------------------------------------------------------- line based + + def entries_from_lines(self, body): + return {line.split(b":")[0]: line.split(b":")[1:] for line in body.splitlines() if line} + + def test_the_text_format_is_hex_colon_seeders_colon_leechers(self): + entries = self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txt")) + self.assertEqual(entries[self.seeding.hex().upper().encode()], [b"1", b"0"]) + self.assertEqual(entries[self.leeching.hex().upper().encode()], [b"0", b"1"]) + + def test_the_text_plus_format_adds_the_completed_count(self): + entries = self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txtp")) + self.assertEqual(entries[self.seeding.hex().upper().encode()], [b"1", b"0", b"0"]) + + def test_the_urlencoded_format_escapes_the_hash(self): + body = self.scrape_all("/stats?mode=tpbs&format=url") + lines = [line for line in body.splitlines() if line] + self.assertEqual(len(lines), 2) + for line in lines: + escaped = line.rsplit(b":", 2)[0] + self.assertIn(b"%", escaped, "these hashes are mostly unprintable bytes") + + def test_the_state_dump_carries_a_base_time_rather_than_a_peer_count(self): + """What a restart reads back, so its second column is the torrent's + own clock and not a count of anybody.""" + entries = self.entries_from_lines(self.scrape_all("/stats?mode=statedump")) + self.assertIn(self.seeding.hex().upper().encode(), entries) + base, completed = entries[self.seeding.hex().upper().encode()] + self.assertGreater(int(base), 0) + self.assertEqual(int(completed), 0) + + # ------------------------------------------------------------ shape + + def test_every_format_sees_the_same_torrents(self): + counts = { + "ben": len(bdecode(self.scrape_all("/stats?mode=tpbs&format=ben"))[b"files"]), + "bin": len(self.scrape_all("/stats?mode=tpbs&format=bin")) // 28, + "txt": len(self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txt"))), + "txtp": len(self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txtp"))), + "url": len([l for l in self.scrape_all("/stats?mode=tpbs&format=url").splitlines() if l]), + } + self.assertEqual(set(counts.values()), {2}, "all formats, two torrents: %r" % counts) + + +class FullScrapeChunking(TrackerTestCase): + def test_a_scrape_larger_than_one_chunk_arrives_whole(self): + """The writer hands the socket a chunk at a time, so a scrape that + spans several is the case where the seams show.""" + for i in range(400): + self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=7000 + i, left=0) + answer = bdecode(self.tracker.get("/scrape")) + self.assertEqual(len(answer[b"files"]), 400) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.3