1 files changed, 89 insertions, 0 deletions

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()