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