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_dualstack.py | 89 +++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 tests/integration/test_dualstack.py (limited to 'tests/integration/test_dualstack.py') diff --git a/tests/integration/test_dualstack.py b/tests/integration/test_dualstack.py new file mode 100644 index 0000000..cd5ba7d --- /dev/null +++ b/tests/integration/test_dualstack.py @@ -0,0 +1,89 @@ +"""A tracker serving both families from one socket. + +These need WANT_IP_FROM_QUERY_STRING so a peer's address can be dictated +rather than being whatever loopback the test happened to come from. +""" + +import os +import socket +import sys +import unittest + +sys.path.insert(0, os.path.dirname(__file__)) +from harness import TrackerTestCase + + +def addresses(answer, key): + 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] + out.append(socket.inet_ntop(socket.AF_INET if size == 6 else socket.AF_INET6, entry[:-2])) + return out + + +class DualStack(TrackerTestCase): + def test_an_ipv4_peer_lands_in_the_v4_list(self): + info_hash = self.hash_for("v4only") + self.tracker.announce(info_hash, ip="10.1.2.3", port=5001, left=100) + answer = self.tracker.announce(info_hash, ip="10.1.2.4", port=5002, left=100) + + self.assertEqual(sorted(addresses(answer, b"peers")), ["10.1.2.3", "10.1.2.4"]) + self.assertEqual(addresses(answer, b"peers6"), [], "no v6 peer was announced") + + def test_an_ipv6_peer_lands_in_the_v6_list(self): + info_hash = self.hash_for("v6only") + self.tracker.announce(info_hash, ip="2001:db8::1", port=5001, left=100) + answer = self.tracker.announce(info_hash, ip="2001:db8::2", port=5002, left=100) + + self.assertEqual(sorted(addresses(answer, b"peers6")), ["2001:db8::1", "2001:db8::2"]) + self.assertEqual(addresses(answer, b"peers"), [], "no v4 peer was announced") + + def test_the_two_families_are_kept_apart_but_both_returned(self): + info_hash = self.hash_for("both") + self.tracker.announce(info_hash, ip="10.9.9.9", port=6001, left=100) + answer = self.tracker.announce(info_hash, ip="2001:db8::9", port=6002, left=100) + + self.assertEqual(addresses(answer, b"peers"), ["10.9.9.9"]) + self.assertEqual(addresses(answer, b"peers6"), ["2001:db8::9"]) + + def test_a_v4_mapped_address_is_the_same_peer_as_the_plain_one(self): + info_hash = self.hash_for("v4mapped") + self.tracker.announce(info_hash, ip="10.5.5.5", port=7001, left=100) + answer = self.tracker.announce(info_hash, ip="::ffff:10.5.5.5", port=7001, left=100) + + self.assertEqual(answer[b"incomplete"], 1, "the same address written two ways is one peer") + self.assertEqual(addresses(answer, b"peers"), ["10.5.5.5"], "and it is held as v4") + + def test_the_two_families_count_into_one_swarm(self): + info_hash = self.hash_for("counts") + self.tracker.announce(info_hash, ip="10.4.4.4", port=8001, left=0) + answer = self.tracker.announce(info_hash, ip="2001:db8::4", port=8002, left=100) + + self.assertEqual(answer[b"complete"], 1) + self.assertEqual(answer[b"incomplete"], 1) + self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 1) + + def test_stopped_removes_only_the_family_it_came_from(self): + info_hash = self.hash_for("stopfamily") + self.tracker.announce(info_hash, ip="10.3.3.3", port=9001, left=100) + self.tracker.announce(info_hash, ip="2001:db8::3", port=9002, left=100) + self.tracker.announce(info_hash, ip="10.3.3.3", port=9001, left=100, event="stopped") + + answer = self.tracker.announce(info_hash, ip="2001:db8::33", port=9003, left=100) + self.assertEqual(addresses(answer, b"peers"), [], "the v4 peer left") + self.assertIn("2001:db8::3", addresses(answer, b"peers6"), "the v6 one stayed") + + def test_a_malformed_address_is_rejected(self): + import urllib.error + + # The whole field has to be the address: a prefix that happens to + # parse would quietly record the peer somewhere else + for bad in ("999.1.2.3", "nonsense", "2001:db8:::1", "10.1.2.3.4", "10.1.2.3junk", "2001:db8::1%eth0"): + with self.assertRaises(urllib.error.HTTPError, msg="ip=%s should be refused" % bad) as caught: + self.tracker.announce(self.hash_for("badip"), ip=bad, port=6881, left=0) + self.assertEqual(caught.exception.code, 400) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.3