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_udp.py | 153 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 tests/integration/test_udp.py (limited to 'tests/integration/test_udp.py') diff --git a/tests/integration/test_udp.py b/tests/integration/test_udp.py new file mode 100644 index 0000000..1dfa21b --- /dev/null +++ b/tests/integration/test_udp.py @@ -0,0 +1,153 @@ +"""The UDP tracker protocol, BEP 15. + +The HTTP tests cover the torrent store; these cover the packet handling in +front of it, which has its own framing, its own connection id and its own +rules about what to answer and what to ignore. +""" + +import socket +import struct +import unittest + +from harness import TrackerTestCase + +MAGIC = 0x41727101980 + + +class UdpProtocol(TrackerTestCase): + def test_connect_echoes_the_transaction_id(self): + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(5) + try: + sock.sendto(struct.pack(">QII", MAGIC, 0, 0xDEADBEEF), ("127.0.0.1", self.tracker.port)) + action, transaction, connection_id = struct.unpack(">IIQ", sock.recv(2048)[:16]) + self.assertEqual(action, 0) + self.assertEqual(transaction, 0xDEADBEEF) + self.assertNotEqual(connection_id, 0) + finally: + sock.close() + + def test_the_connection_id_is_stable_for_one_peer(self): + first, id_one = self.tracker.udp_connect() + second, id_two = self.tracker.udp_connect() + try: + self.assertEqual(id_one, id_two) + finally: + first.close() + second.close() + + def test_a_wrong_connection_id_is_refused(self): + payload = struct.pack(">QII20s20sQQQIIIiH", 0, 1, 2, self.hash_for("udpbad"), b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881) + answer = self.tracker.udp_raw(payload) + self.assertIsNotNone(answer, "expected an error packet") + action, transaction = struct.unpack(">II", answer[:8]) + self.assertEqual(action, 3, "action 3 is the error packet") + self.assertIn(b"Connection ID missmatch", answer[8:]) + + def test_a_packet_shorter_than_the_header_is_ignored(self): + self.assertIsNone(self.tracker.udp_raw(b"\x00" * 15)) + + def test_an_unknown_action_is_ignored(self): + self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", MAGIC, 9, 1))) + + def test_connect_requires_the_protocol_magic(self): + self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", MAGIC + 1, 0, 1))) + + def test_an_announce_cut_short_is_ignored(self): + sock, connection_id = self.tracker.udp_connect() + try: + self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", connection_id, 1, 1) + b"\0" * 40)) + finally: + sock.close() + + +class UdpAnnounce(TrackerTestCase): + def test_a_seeder_is_counted_as_a_seeder(self): + sock, connection_id = self.tracker.udp_connect() + try: + answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpseed"), left=0) + self.assertEqual(answer["seeders"], 1) + self.assertEqual(answer["leechers"], 0) + self.assertGreater(answer["interval"], 0) + finally: + sock.close() + + def test_a_leecher_is_counted_as_a_leecher(self): + sock, connection_id = self.tracker.udp_connect() + try: + answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpleech"), left=1024) + self.assertEqual(answer["seeders"], 0) + self.assertEqual(answer["leechers"], 1) + finally: + sock.close() + + def test_stopped_removes_the_peer(self): + info_hash = self.hash_for("udpstop") + sock, connection_id = self.tracker.udp_connect() + try: + self.tracker.udp_announce(sock, connection_id, info_hash, left=0) + self.tracker.udp_announce(sock, connection_id, info_hash, left=0, event=3) + answer = self.tracker.udp_announce(sock, connection_id, info_hash, left=1024) + self.assertEqual(answer["seeders"], 0, "the seeder announced stopped") + finally: + sock.close() + + def test_an_announce_over_udp_is_visible_over_http(self): + """One torrent store behind both bearers, not two.""" + info_hash = self.hash_for("udpcross") + sock, connection_id = self.tracker.udp_connect() + try: + self.tracker.udp_announce(sock, connection_id, info_hash, left=0) + finally: + sock.close() + scrape = self.tracker.scrape(info_hash) + self.assertEqual(scrape[b"files"][info_hash][b"complete"], 1) + + def test_the_peer_list_carries_the_announcing_peer(self): + info_hash = self.hash_for("udppeers") + sock, connection_id = self.tracker.udp_connect() + try: + answer = self.tracker.udp_announce(sock, connection_id, info_hash, left=0, port=6881) + self.assertEqual(len(answer["peers"]) % 6, 0, "v4 peers are six bytes each") + self.assertGreaterEqual(len(answer["peers"]), 6) + self.assertEqual(struct.unpack(">H", answer["peers"][4:6])[0], 6881) + finally: + sock.close() + + +class UdpScrape(TrackerTestCase): + def test_scrape_reports_the_swarm(self): + info_hash = self.hash_for("udpscrape") + sock, connection_id = self.tracker.udp_connect() + try: + self.tracker.udp_announce(sock, connection_id, info_hash, left=0) + seeders, completed, leechers = self.tracker.udp_scrape(sock, connection_id, [info_hash])[0] + self.assertEqual(seeders, 1) + self.assertEqual(leechers, 0) + self.assertEqual(completed, 0) + finally: + sock.close() + + def test_an_unknown_hash_scrapes_as_empty(self): + sock, connection_id = self.tracker.udp_connect() + try: + self.assertEqual(self.tracker.udp_scrape(sock, connection_id, [self.hash_for("udpnone")])[0], (0, 0, 0)) + finally: + sock.close() + + def test_several_hashes_answer_in_order(self): + seeding = self.hash_for("udpmultiA") + leeching = self.hash_for("udpmultiB") + sock, connection_id = self.tracker.udp_connect() + try: + self.tracker.udp_announce(sock, connection_id, seeding, left=0) + self.tracker.udp_announce(sock, connection_id, leeching, left=1024) + answers = self.tracker.udp_scrape(sock, connection_id, [seeding, leeching]) + self.assertEqual(answers[0][0], 1, "first hash is the seeded one") + self.assertEqual(answers[1][2], 1, "second hash is the leeched one") + finally: + sock.close() + + +if __name__ == "__main__": + unittest.main() -- cgit v1.3