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