From 687ef7fece04fb5dffa90b5d784fb1e34066c85a Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Mon, 26 Oct 2026 23:08:23 +0100 Subject: Add integration test and its runner --- tests/integration/test_announce.py | 129 +++++++++++++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 tests/integration/test_announce.py (limited to 'tests/integration/test_announce.py') diff --git a/tests/integration/test_announce.py b/tests/integration/test_announce.py new file mode 100644 index 0000000..ffbc07b --- /dev/null +++ b/tests/integration/test_announce.py @@ -0,0 +1,129 @@ +"""What a peer's announce does to the swarm it joins.""" + +import os +import socket +import sys +import unittest + +sys.path.insert(0, os.path.dirname(__file__)) +from harness import TrackerTestCase + + +def peers_of(answer, key=b"peers"): + """Compact peer list as a list of (address, port).""" + blob, out = answer.get(key, b""), [] + size = 6 if key == b"peers" else 18 + for offset in range(0, len(blob), size): + entry = blob[offset : offset + size] + address = socket.inet_ntop(socket.AF_INET if size == 6 else socket.AF_INET6, entry[:-2]) + out.append((address, entry[-2] << 8 | entry[-1])) + return out + + +class AnnounceLifetime(TrackerTestCase): + config = "" + + def test_a_leecher_is_counted_as_incomplete(self): + info_hash = self.hash_for("leecher") + answer = self.tracker.announce(info_hash, left=100) + self.assertEqual(answer[b"complete"], 0) + self.assertEqual(answer[b"incomplete"], 1) + + def test_no_bytes_left_means_seeding(self): + info_hash = self.hash_for("seeder") + answer = self.tracker.announce(info_hash, left=0) + self.assertEqual(answer[b"complete"], 1) + self.assertEqual(answer[b"incomplete"], 0) + + def test_a_peer_that_finishes_moves_from_one_column_to_the_other(self): + info_hash = self.hash_for("finishes") + peer_id = b"-OT0001-finisher0001" + self.tracker.announce(info_hash, peer_id=peer_id, left=100) + self.assertEqual(self.tracker.announce(info_hash, peer_id=peer_id, left=100)[b"incomplete"], 1) + + answer = self.tracker.announce(info_hash, peer_id=peer_id, left=0, event="completed") + self.assertEqual(answer[b"complete"], 1, "the peer should now count as a seeder") + self.assertEqual(answer[b"incomplete"], 0, "and no longer as a leecher") + + def test_stopped_removes_the_peer(self): + info_hash = self.hash_for("stopped") + peer_id = b"-OT0001-leaves000001" + self.tracker.announce(info_hash, peer_id=peer_id, left=100) + self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"incomplete"], 1) + + self.tracker.announce(info_hash, peer_id=peer_id, left=100, event="stopped") + self.assertEqual( + self.tracker.scrape(info_hash)[b"files"][info_hash][b"incomplete"], 0, "the peer should be gone" + ) + + def test_the_same_peer_announcing_twice_is_still_one_peer(self): + info_hash = self.hash_for("twice") + peer_id = b"-OT0001-repeats00001" + for _ in range(5): + answer = self.tracker.announce(info_hash, peer_id=peer_id, left=100) + self.assertEqual(answer[b"incomplete"], 1, "re-announcing must not duplicate the peer") + + def test_a_peer_is_its_address_and_port_not_its_peer_id(self): + """Two announces from one address and port are one peer, whatever + peer_id they carry, because that pair is the whole identity.""" + info_hash = self.hash_for("identity") + self.tracker.announce(info_hash, peer_id=b"-OT0001-a00000000001", port=6881, left=100) + answer = self.tracker.announce(info_hash, peer_id=b"-OT0001-b00000000002", port=6881, left=100) + self.assertEqual(answer[b"incomplete"], 1, "same address and port, so still one peer") + + answer = self.tracker.announce(info_hash, peer_id=b"-OT0001-c00000000003", port=6882, left=100) + self.assertEqual(answer[b"incomplete"], 2, "a different port makes a different peer") + + def test_a_downloaded_count_is_kept(self): + info_hash = self.hash_for("downloaded") + self.tracker.announce(info_hash, port=7001, left=0, event="completed") + self.tracker.announce(info_hash, port=7002, left=0, event="completed") + self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"downloaded"], 2) + + def test_a_peer_is_given_the_whole_swarm_including_itself(self): + """opentracker does not filter the announcing peer out of its own + answer. Clients are expected to ignore themselves.""" + info_hash = self.hash_for("sees") + self.tracker.announce(info_hash, port=1111, left=100) + answer = self.tracker.announce(info_hash, port=2222, left=100) + + ports = sorted(port for _, port in peers_of(answer)) + self.assertEqual(ports, [1111, 2222]) + + def test_an_unknown_torrent_scrapes_as_empty(self): + answer = self.tracker.scrape(self.hash_for("never-seen")) + self.assertEqual(answer[b"files"], {}, "a torrent nobody announced has no entry") + + +class AnnounceRejects(TrackerTestCase): + def assert_rejected(self, query): + import urllib.error + + with self.assertRaises(urllib.error.HTTPError) as caught: + self.tracker.get("/announce?" + query) + self.assertEqual(caught.exception.code, 400) + + def test_a_short_info_hash_is_rejected(self): + self.assert_rejected("info_hash=tooshort&peer_id=" + "b" * 20 + "&port=6881&left=0") + + def test_a_missing_info_hash_answers_a_failure_reason(self): + """Not a 400: the tracker answers 200 with a bencoded complaint, so + that a client shows the message to its user.""" + from harness import bdecode + + answer = bdecode(self.tracker.get("/announce?peer_id=" + "b" * 20 + "&port=6881&left=0")) + self.assertIn(b"failure reason", answer) + self.assertIn(b"info_hash", answer[b"failure reason"]) + + def test_a_port_out_of_range_is_rejected(self): + self.assert_rejected("info_hash=" + "a" * 20 + "&peer_id=" + "b" * 20 + "&port=65536&left=0") + + def test_a_port_that_is_not_a_number_is_rejected(self): + self.assert_rejected("info_hash=" + "a" * 20 + "&peer_id=" + "b" * 20 + "&port=abc&left=0") + + def test_two_info_hashes_are_rejected(self): + self.assert_rejected("info_hash=" + "a" * 20 + "&info_hash=" + "c" * 20 + "&peer_id=" + "b" * 20 + "&port=6881&left=0") + + +if __name__ == "__main__": + unittest.main() -- cgit v1.3