"""The blacklist, loaded from a file at startup.

A blacklisted info hash must be refused at both the announce and the scrape
entry, and the file parser has to cope with the shapes an operator actually
produces: an empty file, and a last line with no newline after it.
"""

import os
import struct
import unittest

from harness import Tracker, TrackerTestCase

BLOCKED = b"\xb1" * 20
ALLOWED = b"\xa1" * 20


def listing(*hashes):
    return b"".join(h.hex().upper().encode() + b"\n" for h in hashes)


class Blacklist(TrackerTestCase):
    config = "access.blacklist {dir}/blacklist\n"
    files = {"blacklist": listing(BLOCKED)}

    def test_a_blacklisted_hash_is_refused(self):
        answer = self.tracker.announce(BLOCKED, port=6881, left=0)
        self.assertIn(b"failure reason", answer)
        self.assertIn(b"not authorized", answer[b"failure reason"])

    def test_a_blacklisted_hash_never_becomes_a_torrent(self):
        before = self.tracker.torrent_count()
        self.tracker.announce(BLOCKED, port=6881, left=0)
        self.assertEqual(self.tracker.torrent_count(), before, "the refused hash was stored anyway")

    def test_any_other_hash_is_served(self):
        answer = self.tracker.announce(ALLOWED, port=6881, left=0)
        self.assertNotIn(b"failure reason", answer)
        self.assertEqual(answer[b"complete"], 1)

    def test_the_block_is_an_empty_announce_over_udp(self):
        """An empty swarm with a long interval, not an error. BEP 15 binds a
        client to the interval and says nothing about what to do with an
        error, so an error puts the client on its own retry schedule, which
        starts in the tens of seconds."""
        sock, connection_id = self.tracker.udp_connect()
        sock.close()
        request = struct.pack(
            ">QII20s20sQQQIIIiH", connection_id, 1, 7, BLOCKED, b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881
        )
        answer = self.tracker.udp_raw(request)
        self.assertIsNotNone(answer)
        self.assertEqual(len(answer), 20, "header and counts, no peers behind them")
        action, transaction, interval, leechers, seeders = struct.unpack(">IIIII", answer)
        self.assertEqual(action, 1, "action 1 is the announce packet")
        self.assertEqual(transaction, 7)
        self.assertEqual((leechers, seeders), (0, 0))
        self.assertGreaterEqual(interval, 60 * 60, "far enough out to be worth the trouble")
        self.assertLessEqual(interval, 60 * 60 * 6, "not so far out that a client clamps it")

    def test_a_blocked_hash_is_not_stored_when_it_arrives_over_udp(self):
        sock, connection_id = self.tracker.udp_connect()
        sock.close()
        self.tracker.udp_raw(
            struct.pack(">QII20s20sQQQIIIiH", connection_id, 1, 8, BLOCKED, b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881)
        )
        self.assertEqual(self.tracker.torrent_count(), 0)


class BlacklistWithoutTrailingNewline(TrackerTestCase):
    """A single hash with no newline after it used to size the list at zero
    and then write into it, which took the tracker down before it listened."""

    config = "access.blacklist {dir}/blacklist\n"
    files = {"blacklist": BLOCKED.hex().upper().encode()}

    def test_the_last_line_counts_even_without_its_newline(self):
        answer = self.tracker.announce(BLOCKED, port=6881, left=0)
        self.assertIn(b"failure reason", answer)

    def test_and_the_tracker_still_serves_everything_else(self):
        self.assertEqual(self.tracker.announce(ALLOWED, port=6881, left=0)[b"complete"], 1)


class EmptyBlacklist(TrackerTestCase):
    config = "access.blacklist {dir}/blacklist\n"
    files = {"blacklist": b""}

    def test_an_empty_file_blocks_nothing_and_starts_cleanly(self):
        self.assertEqual(self.tracker.announce(BLOCKED, port=6881, left=0)[b"complete"], 1)
        self.assertEqual(self.tracker.announce(ALLOWED, port=6882, left=0)[b"complete"], 1)


class MissingBlacklist(TrackerTestCase):
    """The file is named but absent: the tracker warns and serves everything,
    because it may be created later."""

    config = "access.blacklist {dir}/absent\n"

    def test_a_missing_file_blocks_nothing(self):
        self.assertEqual(self.tracker.announce(BLOCKED, port=6881, left=0)[b"complete"], 1)


class ListIsInPlaceBeforeTheStateFile(TrackerTestCase):
    """The loader used to run in a thread of its own, so a state file read at
    startup was checked against a list that was still empty."""

    config = "access.blacklist {dir}/blacklist\n"
    files = {"blacklist": listing(BLOCKED)}

    def test_a_blacklisted_torrent_in_a_state_file_is_not_restored(self):
        path = os.path.join(self.tracker.dir, "state")
        with open(path, "w") as handle:
            handle.write("%s:5:7\n%s:5:7\n" % (BLOCKED.hex(), ALLOWED.hex()))

        restarted = Tracker(self.config, self.files, extra_args=["-l", path])
        try:
            self.assertNotIn(BLOCKED, restarted.scrape(BLOCKED)[b"files"], "the blacklist was not in place when the state file was read")
            self.assertIn(ALLOWED, restarted.scrape(ALLOWED)[b"files"], "the state file was not read at all")
            restarted.stop()
        finally:
            restarted._cleanup()


if __name__ == "__main__":
    unittest.main()