2 files changed, 117 insertions, 0 deletions

diff --git a/tests/integration/test_dynamic_accesslist.py b/tests/integration/test_dynamic_accesslist.py
new file mode 100644
index 0000000..7fa5fde
--- /dev/null
+++ b/tests/integration/test_dynamic_accesslist.py
@@ -0,0 +1,50 @@
+"""Hashes added to and removed from the blacklist while the tracker runs.
+
+The two FIFOs are read by threads of their own, which is also what makes
+this the test that drives those threads through a shutdown.
+"""
+
+import os
+import time
+import unittest
+
+from harness import TrackerTestCase
+
+BLOCKED = bytes(range(1, 21))
+
+
+def as_hex(info_hash):
+    return "".join("%02x" % b for b in info_hash)
+
+
+class DynamicAccesslist(TrackerTestCase):
+    config = ("access.blacklist {dir}/blacklist\n"
+              "access.fifo_add {dir}/add.fifo\n"
+              "access.fifo_delete {dir}/delete.fifo\n")
+    files = {"blacklist": b""}
+
+    def feed(self, name, info_hash):
+        """Writing to the FIFO blocks until the reader has it, so this also
+           tells us the thread is alive."""
+        with open(os.path.join(self.tracker.dir, name), "w") as fifo:
+            fifo.write("%s added by a test\n" % as_hex(info_hash))
+
+    def wait_for(self, info_hash, blocked, patience=5.0):
+        """The reader thread picks the hash up on its own schedule, so this
+           waits for the answer to change rather than assuming it already has."""
+        deadline = time.time() + patience
+        while time.time() < deadline:
+            answer = self.tracker.announce(info_hash)
+            if (b"failure reason" in answer) == blocked:
+                return True
+            time.sleep(0.05)
+        return False
+
+    def test_a_hash_can_be_blocked_and_released_while_running(self):
+        self.assertNotIn(b"failure reason", self.tracker.announce(BLOCKED))
+
+        self.feed("add.fifo", BLOCKED)
+        self.assertTrue(self.wait_for(BLOCKED, blocked=True), "the hash was never blocked")
+
+        self.feed("delete.fifo", BLOCKED)
+        self.assertTrue(self.wait_for(BLOCKED, blocked=False), "the hash was never released")
diff --git a/tests/integration/test_tcpworkers.py b/tests/integration/test_tcpworkers.py
new file mode 100644
index 0000000..e43612a
--- /dev/null
+++ b/tests/integration/test_tcpworkers.py
@@ -0,0 +1,67 @@
+"""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)