"""The tracker with its tcp side handed to worker threads.

With workers configured, a connection is accepted and answered on a worker's
own stack. Anything it cannot finish there is copied to the heap and handed
to the event loop, which is the one place where a worker and the main thread
touch the same structures, so these drive all three outcomes: answered
inline, held open, and parked waiting for another thread entirely.
"""

import threading
import unittest
import urllib.parse

from harness import TrackerTestCase, bdecode

WORKERS = 4


class TcpWorkers(TrackerTestCase):
    config = "listen.tcp.workers %d\n" % WORKERS

    def announce_path(self, name, port=6881):
        return "/announce?info_hash=%s&peer_id=%s&port=%d&left=0" % (
            urllib.parse.quote_from_bytes(self.hash_for(name), safe=""),
            "x" * 20,
            port,
        )

    def test_many_at_once_are_all_answered(self):
        """Every worker gets something to do, and the swarm adds up."""
        info_hash = self.hash_for("workers-swarm")
        errors = []

        def announce(index):
            try:
                self.tracker.announce(info_hash, peer_id=b"-OT0001-%012d" % index, port=7000 + index, left=0)
            except Exception as problem:  # noqa: BLE001 - reported, not swallowed
                errors.append("%d: %r" % (index, problem))

        threads = [threading.Thread(target=announce, args=(i,)) for i in range(32)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()

        self.assertEqual(errors, [], "\n".join(errors))
        self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 32)

    def test_a_kept_connection_outlives_the_worker_that_took_it(self):
        """Keep-alive is the plain case of a connection a worker hands on."""
        connection = self.tracker.connect()
        try:
            for i in range(4):
                connection.send(self.announce_path("workers-kept"))
                status, _, body = connection.reply()
                self.assertIn(b"200", status)
                self.assertIn(b"complete", bdecode(body))
        finally:
            connection.close()

    def test_a_full_scrape_asked_for_by_a_worker_still_arrives(self):
        """The reply is built by another thread, so the worker parks the
           connection waiting for nothing at all before handing it over."""
        for i in range(20):
            self.tracker.announce(bytes([i]) + b"\0" * 19, port=7100 + i, left=0)
        body = self.tracker.get("/scrape")
        self.assertEqual(len(bdecode(body)[b"files"]), 20)