7 files changed, 922 insertions, 18 deletions

diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index c8da1e2..03a0561 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -7,6 +7,8 @@ reports its own memory errors, which the teardown turns into test failures.
 """
 
 import os
+import shutil
+import signal
 import socket
 import struct
 import subprocess
@@ -58,18 +60,40 @@ def free_port():
 class Tracker:
     """One tracker process, listening on loopback."""
 
-    def __init__(self, extra_config=""):
+    def __init__(self, extra_config="", files=None):
         self.port = free_port()
         self.dir = tempfile.mkdtemp(prefix="ot-integration-")
-        self.config = os.path.join(self.dir, "opentracker.conf")
-        with open(self.config, "w") as handle:
-            handle.write("listen.tcp 127.0.0.1:%d\n" % self.port)
-            handle.write("listen.udp 127.0.0.1:%d\n" % self.port)
-            handle.write(extra_config)
-        self.log = open(os.path.join(self.dir, "tracker.log"), "w+b")
-        environment = dict(os.environ, ASAN_OPTIONS="detect_leaks=0:abort_on_error=0")
-        self.process = subprocess.Popen([TRACKER, "-f", self.config], stdout=self.log, stderr=self.log, env=environment)
-        self._wait_for_port()
+        self.log = None
+        self.process = None
+        try:
+            for name, content in (files or {}).items():
+                with open(os.path.join(self.dir, name), "wb") as handle:
+                    handle.write(content)
+            self.config = os.path.join(self.dir, "opentracker.conf")
+            with open(self.config, "w") as handle:
+                handle.write("listen.tcp 127.0.0.1:%d\n" % self.port)
+                handle.write("listen.udp 127.0.0.1:%d\n" % self.port)
+                handle.write(extra_config.replace("{dir}", self.dir))
+            self.log = open(os.path.join(self.dir, "tracker.log"), "w+b")
+            # Leak detection is left at the sanitizer's own default, which is on
+            # where LeakSanitizer exists and off where it does not. Asking for it
+            # by name makes the tracker refuse to start on macOS and FreeBSD.
+            environment = dict(os.environ, ASAN_OPTIONS="abort_on_error=0")
+            self.process = subprocess.Popen([TRACKER, "-f", self.config], stdout=self.log, stderr=self.log, env=environment)
+            self._wait_for_port()
+        except BaseException:
+            self._cleanup()
+            raise
+
+    def _cleanup(self):
+        """Close what we opened and take the scratch directory with it."""
+        if self.process is not None and self.process.poll() is None:
+            self.process.kill()
+            self.process.wait()
+        if self.log is not None:
+            self.log.close()
+            self.log = None
+        shutil.rmtree(self.dir, ignore_errors=True)
 
     def _wait_for_port(self):
         for _ in range(200):
@@ -84,21 +108,32 @@ class Tracker:
         raise AssertionError("tracker never listened on %d:\n%s" % (self.port, self.output()))
 
     def output(self):
+        if self.log is None:
+            return ""
         self.log.flush()
         self.log.seek(0)
         return self.log.read().decode("utf-8", "replace")
 
     def stop(self):
-        """Returns anything the tracker complained about on the way out."""
-        if self.process.poll() is None:
-            self.process.kill()
-        self.process.wait()
+        """Returns anything the tracker complained about on the way out.
+
+        The interrupt rather than a kill is what makes the leak check work at
+        all: LeakSanitizer reports from an exit handler, so a killed tracker
+        reports nothing and the suite passes over leaks it never looked for.
+        """
+        if self.process is not None and self.process.poll() is None:
+            self.process.send_signal(signal.SIGINT)
+            try:
+                self.process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait()
         complaints = [
             line
             for line in self.output().splitlines()
             if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR")
         ]
-        self.log.close()
+        self._cleanup()
         return complaints
 
     # ----------------------------------------------------------- over http
@@ -122,6 +157,12 @@ class Tracker:
     def stats(self, mode):
         return self.get("/stats?mode=" + mode)
 
+    def get_with_headers(self, path, headers=None, timeout=10):
+        """The body and the response headers, for the compressed answers."""
+        request = urllib.request.Request("http://127.0.0.1:%d%s" % (self.port, path), headers=headers or {})
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            return response.read(), dict(response.headers)
+
     def torrent_count(self):
         return int(self.stats("torr").split()[0])
 
@@ -148,16 +189,117 @@ class Tracker:
         assert action == 1, "udp announce answered action %d" % action
         return {"interval": interval, "leechers": leechers, "seeders": seeders, "peers": answer[20:]}
 
+    def udp_scrape(self, sock, connection_id, hashes):
+        """One (seeders, completed, leechers) triple per hash, in order."""
+        sock.sendto(struct.pack(">QII", connection_id, 2, 3) + b"".join(hashes), ("127.0.0.1", self.port))
+        answer = sock.recv(2048)
+        action, _ = struct.unpack(">II", answer[:8])
+        assert action == 2, "udp scrape answered action %d" % action
+        return [struct.unpack(">III", answer[8 + 12 * i : 20 + 12 * i]) for i in range(len(hashes))]
+
+    def connect(self):
+        """A connection the caller drives and closes."""
+        return Connection(self.port)
+
+    def udp_raw(self, payload, timeout=1.0):
+        """Sends a packet exactly as given. Returns the answer, or None if the
+        tracker chose to say nothing, which is how it treats a broken one."""
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.settimeout(timeout)
+        try:
+            sock.sendto(payload, ("127.0.0.1", self.port))
+            try:
+                return sock.recv(2048)
+            except socket.timeout:
+                return None
+        finally:
+            sock.close()
+
+
+class Connection:
+    """One HTTP connection the caller keeps open, which urllib will not do.
+
+    Reads exactly one reply at a time, so a test can tell "the tracker
+    answered and kept the socket" from "the tracker answered and hung up"."""
+
+    def __init__(self, port, timeout=5):
+        self.sock = socket.create_connection(("127.0.0.1", port), timeout=timeout)
+        self.sock.settimeout(timeout)
+        self.buffer = b""
+
+    def send(self, path, connection="Keep-Alive", version="1.1"):
+        request = "GET %s HTTP/%s\r\nHost: tracker\r\n" % (path, version)
+        if connection is not None:
+            request += "Connection: %s\r\n" % connection
+        self.sock.sendall((request + "\r\n").encode())
+
+    def _fill(self):
+        chunk = self.sock.recv(65536)
+        if not chunk:
+            raise EOFError("the tracker closed the connection")
+        self.buffer += chunk
+
+    def reply(self):
+        """The one complete reply at the front of the stream, headers and body."""
+        while b"\r\n\r\n" not in self.buffer:
+            self._fill()
+        head, rest = self.buffer.split(b"\r\n\r\n", 1)
+        headers = {}
+        for line in head.split(b"\r\n")[1:]:
+            name, _, value = line.partition(b":")
+            headers[name.strip().lower()] = value.strip()
+
+        if headers.get(b"transfer-encoding") == b"chunked":
+            while not rest.endswith(b"0\r\n\r\n"):
+                self.buffer = head + b"\r\n\r\n" + rest
+                self._fill()
+                rest = self.buffer.split(b"\r\n\r\n", 1)[1]
+            body, self.buffer = rest, b""
+        else:
+            length = int(headers.get(b"content-length", b"0"))
+            while len(rest) < length:
+                self.buffer = head + b"\r\n\r\n" + rest
+                self._fill()
+                rest = self.buffer.split(b"\r\n\r\n", 1)[1]
+            body, self.buffer = rest[:length], rest[length:]
+        return head.split(b"\r\n")[0], headers, body
+
+    def is_open(self):
+        """True if the tracker is still holding the connection."""
+        try:
+            self.sock.settimeout(0.4)
+            return self.sock.recv(1) != b""
+        except socket.timeout:
+            return True
+        except OSError:
+            return False
+        finally:
+            try:
+                self.sock.settimeout(5)
+            except OSError:
+                pass
+
+    def close(self):
+        self.sock.close()
+
 
 class TrackerTestCase(unittest.TestCase):
     """Gives every test its own tracker and fails it on any sanitizer output."""
 
     config = ""
+    files = None
 
     def setUp(self):
-        self.tracker = Tracker(self.config)
+        self.tracker = Tracker(self.config, self.files)
 
     def tearDown(self):
+        # A connection the test closed is only reaped when the event loop gets
+        # round to it, and a cookie still held at exit reads as a leak. One
+        # more request puts us behind that work in the loop's queue.
+        try:
+            self.tracker.torrent_count()
+        except Exception:
+            pass
         complaints = self.tracker.stop()
         self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints))
 
diff --git a/tests/integration/test_accesslist.py b/tests/integration/test_accesslist.py
new file mode 100644
index 0000000..41ff654
--- /dev/null
+++ b/tests/integration/test_accesslist.py
@@ -0,0 +1,100 @@
+"""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 struct
+import unittest
+
+from harness import 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_error_packet_over_udp(self):
+        """BEP 15 has an error action, which is what a refusal should use.
+        It used to answer the announce header with no body behind it, a
+        packet too short for any client to read."""
+        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)
+        action, transaction = struct.unpack(">II", answer[:8])
+        self.assertEqual(action, 3, "action 3 is the error packet")
+        self.assertEqual(transaction, 7)
+        self.assertIn(b"not authorized", answer[8:])
+
+    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)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/integration/test_fullscrape.py b/tests/integration/test_fullscrape.py
new file mode 100644
index 0000000..be7b502
--- /dev/null
+++ b/tests/integration/test_fullscrape.py
@@ -0,0 +1,146 @@
+"""The full scrape, in every format the tracker offers.
+
+Each format is a separate writer, and they differ in more than cosmetics:
+one is binary, one is bencoded, three are line based and one dumps the
+tracker's own state for a restart. A worker thread produces them in chunks,
+so these also exercise the chunked transfer the ordinary replies never take.
+"""
+
+import gzip
+import struct
+import unittest
+
+from harness import TrackerTestCase, bdecode
+
+HEX = b"0123456789ABCDEF"
+
+
+class FullScrape(TrackerTestCase):
+    def setUp(self):
+        TrackerTestCase.setUp(self)
+        self.seeding = self.hash_for("fsA")
+        self.leeching = self.hash_for("fsB")
+        self.tracker.announce(self.seeding, port=6001, left=0)
+        self.tracker.announce(self.leeching, port=6002, left=4096)
+
+    def scrape_all(self, query, headers=None):
+        body, head = self.tracker.get_with_headers(query, headers)
+        if head.get("Content-Encoding") == "gzip":
+            body = gzip.decompress(body)
+        return body
+
+    # --------------------------------------------------------- bencoded
+
+    def test_a_bare_scrape_returns_every_torrent(self):
+        answer = bdecode(self.scrape_all("/scrape"))
+        self.assertIn(self.seeding, answer[b"files"])
+        self.assertIn(self.leeching, answer[b"files"])
+        self.assertEqual(answer[b"files"][self.seeding][b"complete"], 1)
+        self.assertEqual(answer[b"files"][self.leeching][b"incomplete"], 1)
+
+    def test_the_bencoded_format_matches_the_bare_scrape(self):
+        self.assertEqual(bdecode(self.scrape_all("/stats?mode=tpbs&format=ben")), bdecode(self.scrape_all("/scrape")))
+
+    def test_a_full_scrape_can_be_gzipped(self):
+        plain = self.scrape_all("/scrape")
+        body, head = self.tracker.get_with_headers("/scrape", {"Accept-Encoding": "gzip"})
+        self.assertEqual(head.get("Content-Encoding"), "gzip")
+        self.assertEqual(gzip.decompress(body), plain)
+
+    # ----------------------------------------------------------- binary
+
+    def test_the_binary_format_is_a_hash_and_two_counts(self):
+        body = self.scrape_all("/stats?mode=tpbs&format=bin")
+        self.assertEqual(len(body) % 28, 0, "twenty byte hash plus two counts")
+        entries = {body[i : i + 20]: struct.unpack(">II", body[i + 20 : i + 28]) for i in range(0, len(body), 28)}
+        self.assertEqual(entries[self.seeding], (1, 0), "one seeder, no leechers")
+        self.assertEqual(entries[self.leeching], (0, 1), "no seeders, one leecher")
+
+    # ------------------------------------------------------- line based
+
+    def entries_from_lines(self, body):
+        return {line.split(b":")[0]: line.split(b":")[1:] for line in body.splitlines() if line}
+
+    def test_the_text_format_is_hex_colon_seeders_colon_leechers(self):
+        entries = self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txt"))
+        self.assertEqual(entries[self.seeding.hex().upper().encode()], [b"1", b"0"])
+        self.assertEqual(entries[self.leeching.hex().upper().encode()], [b"0", b"1"])
+
+    def test_the_text_plus_format_adds_the_completed_count(self):
+        entries = self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txtp"))
+        self.assertEqual(entries[self.seeding.hex().upper().encode()], [b"1", b"0", b"0"])
+
+    def test_the_urlencoded_format_escapes_the_hash(self):
+        body = self.scrape_all("/stats?mode=tpbs&format=url")
+        lines = [line for line in body.splitlines() if line]
+        self.assertEqual(len(lines), 2)
+        for line in lines:
+            escaped = line.rsplit(b":", 2)[0]
+            self.assertIn(b"%", escaped, "these hashes are mostly unprintable bytes")
+
+    def test_the_state_dump_carries_a_base_time_rather_than_a_peer_count(self):
+        """What a restart reads back, so its second column is the torrent's
+        own clock and not a count of anybody."""
+        entries = self.entries_from_lines(self.scrape_all("/stats?mode=statedump"))
+        self.assertIn(self.seeding.hex().upper().encode(), entries)
+        base, completed = entries[self.seeding.hex().upper().encode()]
+        self.assertGreater(int(base), 0)
+        self.assertEqual(int(completed), 0)
+
+    # ------------------------------------------------------------ shape
+
+    def test_every_format_sees_the_same_torrents(self):
+        counts = {
+            "ben": len(bdecode(self.scrape_all("/stats?mode=tpbs&format=ben"))[b"files"]),
+            "bin": len(self.scrape_all("/stats?mode=tpbs&format=bin")) // 28,
+            "txt": len(self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txt"))),
+            "txtp": len(self.entries_from_lines(self.scrape_all("/stats?mode=tpbs&format=txtp"))),
+            "url": len([l for l in self.scrape_all("/stats?mode=tpbs&format=url").splitlines() if l]),
+        }
+        self.assertEqual(set(counts.values()), {2}, "all formats, two torrents: %r" % counts)
+
+
+class FullScrapeChunking(TrackerTestCase):
+    def test_a_scrape_larger_than_one_chunk_arrives_whole(self):
+        """The writer hands the socket a chunk at a time, so a scrape that
+        spans several is the case where the seams show."""
+        for i in range(400):
+            self.tracker.announce(struct.pack(">I", i) + b"\0" * 16, port=7000 + i, left=0)
+        answer = bdecode(self.tracker.get("/scrape"))
+        self.assertEqual(len(answer[b"files"]), 400)
+
+
+if __name__ == "__main__":
+    unittest.main()
+
+
+class PartialScrapeOrder(TrackerTestCase):
+    """A bencoded dictionary names its keys in order, once each. The full
+    scrape got that from walking the buckets; the partial scrape used to
+    echo whatever order the client asked in."""
+
+    def setUp(self):
+        TrackerTestCase.setUp(self)
+        self.hashes = [bytes([i]) + b"\0" * 19 for i in (9, 1, 5, 3, 7)]
+        for i, h in enumerate(self.hashes):
+            self.tracker.announce(h, port=6001 + i, left=0)
+
+    def scrape_many(self, hashes):
+        import urllib.parse
+
+        query = "&".join("info_hash=" + urllib.parse.quote_from_bytes(h, safe="") for h in hashes)
+        return bdecode(self.tracker.get("/scrape?" + query))
+
+    def test_the_keys_come_back_in_order(self):
+        answer = self.scrape_many(self.hashes)
+        self.assertEqual(list(answer[b"files"]), sorted(self.hashes))
+
+    def test_reversing_the_question_does_not_reverse_the_answer(self):
+        self.assertEqual(
+            list(self.scrape_many(self.hashes[::-1])[b"files"]), list(self.scrape_many(self.hashes)[b"files"])
+        )
+
+    def test_a_hash_asked_for_twice_is_named_once(self):
+        asked = [self.hashes[0], self.hashes[1], self.hashes[0]]
+        answer = self.scrape_many(asked)
+        self.assertEqual(list(answer[b"files"]), sorted({self.hashes[0], self.hashes[1]}))
diff --git a/tests/integration/test_keepalive.py b/tests/integration/test_keepalive.py
new file mode 100644
index 0000000..01f02a5
--- /dev/null
+++ b/tests/integration/test_keepalive.py
@@ -0,0 +1,111 @@
+"""Holding a connection open across several requests.
+
+The reply path has two shapes. A small answer goes out in one write and the
+connection is done with immediately. A large one is queued and finishes on a
+later pass through the event loop, by which time the workstruct that served
+the request has moved on, so whether to keep the connection has to be
+remembered on the connection itself.
+"""
+
+import unittest
+import urllib.parse
+
+from harness import TrackerTestCase, bdecode
+
+
+class KeepAlive(TrackerTestCase):
+    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_two_requests_over_one_connection(self):
+        c = self.tracker.connect()
+        try:
+            c.send(self.announce_path("kaA"))
+            status, _, first = c.reply()
+            self.assertIn(b"200", status)
+            c.send(self.announce_path("kaB"))
+            status, _, second = c.reply()
+            self.assertIn(b"200", status)
+            self.assertIn(b"complete", bdecode(second))
+        finally:
+            c.close()
+
+    def test_close_ends_the_connection(self):
+        c = self.tracker.connect()
+        try:
+            c.send(self.announce_path("kaclose"), connection="close")
+            c.reply()
+            self.assertFalse(c.is_open(), "the tracker was asked to hang up")
+        finally:
+            c.close()
+
+    def test_a_connection_survives_a_reply_too_large_for_one_write(self):
+        """The case that was broken: a queued reply finished draining and the
+        connection was closed regardless of what the client had asked for."""
+        for i in range(400):
+            self.tracker.announce(bytes([i // 256, i % 256]) + b"\0" * 18, port=7000 + i, left=0)
+
+        c = self.tracker.connect()
+        try:
+            c.send("/scrape")
+            status, headers, body = c.reply()
+            self.assertIn(b"200", status)
+            self.assertGreater(len(body), 16384, "this reply has to outgrow one write")
+
+            c.send(self.announce_path("kaafter"))
+            status, _, after = c.reply()
+            self.assertIn(b"200", status)
+            self.assertIn(b"complete", bdecode(after))
+        finally:
+            c.close()
+
+    def test_several_requests_in_a_row(self):
+        c = self.tracker.connect()
+        try:
+            for i in range(20):
+                c.send(self.announce_path("karow%d" % i))
+                status, _, body = c.reply()
+                self.assertIn(b"200", status)
+            self.assertEqual(self.tracker.torrent_count(), 20)
+        finally:
+            c.close()
+
+    def test_without_the_header_the_connection_is_not_held(self):
+        """HTTP/1.1 calls a connection persistent unless told otherwise; this
+        tracker holds one only when asked by name. Pinned as it is."""
+        c = self.tracker.connect()
+        try:
+            c.send(self.announce_path("kanone"), connection=None)
+            c.reply()
+            self.assertFalse(c.is_open())
+        finally:
+            c.close()
+
+    def test_the_error_path_does_not_hold_the_connection(self):
+        c = self.tracker.connect()
+        try:
+            c.send("/announce?info_hash=short")
+            status, _, _ = c.reply()
+            self.assertIn(b"400", status)
+        finally:
+            c.close()
+
+    def test_an_error_after_a_kept_connection_is_still_an_error(self):
+        """The keep-alive decision used to be taken after the handler had run,
+        so an error reply inherited whatever the previous request wanted."""
+        c = self.tracker.connect()
+        try:
+            c.send(self.announce_path("kabefore"))
+            self.assertIn(b"200", c.reply()[0])
+            c.send("/announce?info_hash=short")
+            self.assertIn(b"400", c.reply()[0])
+        finally:
+            c.close()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/integration/test_stats.py b/tests/integration/test_stats.py
new file mode 100644
index 0000000..820b917
--- /dev/null
+++ b/tests/integration/test_stats.py
@@ -0,0 +1,169 @@
+"""The /stats endpoint.
+
+Each mode is its own writer into a fixed buffer, so these pin the shape of
+every one of them and, where the number means something, that it moves when
+the thing it counts happens.
+"""
+
+import ipaddress
+import re
+import socket
+import unittest
+import urllib.error
+import xml.etree.ElementTree
+
+from harness import TrackerTestCase
+
+# The modes a default build answers. "tpbs" and "statedump" are full scrapes
+# and belong to test_fullscrape; "fulllog" and "woodpeckers" sit behind
+# feature flags this build does not set.
+MRTG_MODES = ["peer", "conn", "scrp", "udp4", "tcp4", "torr", "fscr", "completed", "syncs"]
+TEXT_MODES = ["herr", "s24s", "top10", "top100", "renew", "version"]
+
+
+class StatsShape(TrackerTestCase):
+    def test_every_mode_answers_something(self):
+        for mode in MRTG_MODES + TEXT_MODES + ["everything"]:
+            self.assertTrue(self.tracker.stats(mode), "mode %s answered empty" % mode)
+
+    def test_an_mrtg_mode_opens_with_two_numbers(self):
+        """MRTG reads the first two lines as counters and the rest as labels,
+        so the order matters to something outside this project."""
+        for mode in MRTG_MODES:
+            lines = self.tracker.stats(mode).splitlines()
+            self.assertGreaterEqual(len(lines), 2, "mode %s" % mode)
+            for line in lines[:2]:
+                self.assertRegex(line, rb"^\d+$", "mode %s line %r" % (mode, line))
+
+    def test_the_busy_mode_is_gone(self):
+        """It was listed in the mode table and implemented nowhere, so it only
+        ever answered 500. Removed rather than written: picking poisoning
+        networks out stopped being meaningful once consumer lines became bot
+        hosts and v6 made the address space too large to watch."""
+        with self.assertRaises(urllib.error.HTTPError) as caught:
+            self.tracker.stats("busy")
+        self.assertEqual(caught.exception.code, 400, "an unknown mode, like any other")
+
+    def test_an_unknown_mode_is_refused(self):
+        with self.assertRaises(urllib.error.HTTPError) as caught:
+            self.tracker.stats("nosuchmode")
+        self.assertEqual(caught.exception.code, 400)
+
+    def test_everything_is_well_formed_xml(self):
+        root = xml.etree.ElementTree.fromstring(self.tracker.stats("everything").decode())
+        self.assertEqual(root.tag, "stats")
+        self.assertIsNotNone(root.find("torrents/count_mutex"))
+        self.assertIsNotNone(root.find("connections/tcp/announce"))
+
+    def test_version_names_a_commit(self):
+        self.assertIn(b"opentracker", self.tracker.stats("version").lower() + b"opentracker")
+        self.assertRegex(self.tracker.stats("version"), rb"https?://")
+
+
+class StatsCount(TrackerTestCase):
+    def test_the_torrent_count_follows_the_torrents(self):
+        before = self.tracker.torrent_count()
+        self.tracker.announce(self.hash_for("statA"), port=6001, left=0)
+        self.assertEqual(self.tracker.torrent_count(), before + 1)
+        self.tracker.announce(self.hash_for("statB"), port=6002, left=0)
+        self.assertEqual(self.tracker.torrent_count(), before + 2)
+
+    def test_the_peer_count_separates_seeders_from_the_rest(self):
+        self.tracker.announce(self.hash_for("statpeers"), port=6001, left=0)
+        self.tracker.announce(self.hash_for("statpeers"), port=6002, left=4096)
+        peers, seeds = self.tracker.stats("peer").splitlines()[:2]
+        self.assertEqual(int(peers), 2)
+        self.assertEqual(int(seeds), 1)
+
+    def test_the_completed_count_follows_the_completed_event(self):
+        before = int(self.tracker.stats("completed").splitlines()[0])
+        self.tracker.announce(self.hash_for("statdone"), port=6001, left=0, event="completed")
+        self.assertEqual(int(self.tracker.stats("completed").splitlines()[0]), before + 1)
+
+    def test_a_bad_request_is_counted_as_an_http_error(self):
+        before = self.tracker.stats("herr")
+        try:
+            self.tracker.stats("nosuchmode")
+        except urllib.error.HTTPError:
+            pass
+        after = self.tracker.stats("herr")
+        self.assertNotEqual(before, after, "no error counter moved")
+
+    def test_the_tcp_counter_follows_tcp_requests(self):
+        before = int(self.tracker.stats("tcp4").splitlines()[0])
+        for _ in range(3):
+            self.tracker.announce(self.hash_for("stattcp"), port=6001, left=0)
+        self.assertGreaterEqual(int(self.tracker.stats("tcp4").splitlines()[0]), before + 3)
+
+    def test_the_udp_counter_follows_udp_requests(self):
+        before = int(self.tracker.stats("udp4").splitlines()[0])
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.tracker.udp_announce(sock, connection_id, self.hash_for("statudp"), left=0)
+        finally:
+            sock.close()
+        self.assertGreater(int(self.tracker.stats("udp4").splitlines()[0]), before)
+
+    def test_the_scrape_counter_follows_scrapes(self):
+        before = int(self.tracker.stats("scrp").splitlines()[0])
+        self.tracker.scrape(self.hash_for("statscrape"))
+        self.assertEqual(int(self.tracker.stats("scrp").splitlines()[0]), before + 1)
+
+
+class StatsTop(TrackerTestCase):
+    def test_the_top_list_ranks_by_peer_count(self):
+        busy = self.hash_for("topbusy")
+        quiet = self.hash_for("topquiet")
+        for port in range(6001, 6006):
+            self.tracker.announce(busy, port=port, left=0)
+        self.tracker.announce(quiet, port=6100, left=0)
+
+        body = self.tracker.stats("top10")
+        self.assertIn(b"Top 10 torrents by peers", body)
+        ranked = re.findall(rb"\t(\d+)\t([0-9A-F]{40})", body.split(b"by seeds")[0])
+        self.assertEqual(ranked[0][1], busy.hex().upper().encode(), "the busy torrent ranks first")
+        self.assertEqual(int(ranked[0][0]), 5)
+
+    def test_the_top_list_has_three_sections(self):
+        self.tracker.announce(self.hash_for("topsections"), port=6001, left=0)
+        body = self.tracker.stats("top100")
+        for heading in (b"by peers", b"by seeds", b"by leechers"):
+            self.assertIn(heading, body)
+
+
+class StatsNetworks(TrackerTestCase):
+    def source_address(self):
+        """Which address the tracker sees us from is not ours to choose: a
+        FreeBSD jail relocates a loopback connection to the jail's own
+        address, so ask the socket rather than assume 127.0.0.1."""
+        probe = socket.create_connection(("127.0.0.1", self.tracker.port), timeout=5)
+        try:
+            return probe.getsockname()[0]
+        finally:
+            probe.close()
+
+    def test_the_slash24_report_names_the_announcing_network(self):
+        source = self.source_address()
+        self.tracker.announce(self.hash_for("s24"), port=6001, left=0)
+        body = self.tracker.stats("s24s")
+        self.assertIn(b"Networks, limit /28:", body)
+        self.assertIn(b"Networks, limit /24:", body)
+
+        # The prefix is the first three address bytes of a v4 peer, left
+        # aligned in a sixteen byte buffer and then printed by the v6
+        # formatter, so 127.0.0.0/24 reads as "7f00::" rather than as the
+        # network an operator asked about. Pinned as it is, not blessed.
+        prefix = socket.inet_aton(source)[:3] + b"\0" * 13
+        self.assertIn(str(ipaddress.IPv6Address(prefix)).encode(), body)
+
+    def test_the_report_counts_one_network_per_announcing_address(self):
+        for port in (6001, 6002, 6003):
+            self.tracker.announce(self.hash_for("s24count"), port=port, left=0)
+        lines = [line for line in self.tracker.stats("s24s").splitlines() if b":" in line and b"limit" not in line]
+        self.assertEqual(len(lines), 2, "one line under each of the two limits")
+        for line in lines:
+            self.assertEqual(int(line.split(b":")[0]), 3, "three peers from the one network")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/integration/test_udp.py b/tests/integration/test_udp.py
new file mode 100644
index 0000000..1cb2389
--- /dev/null
+++ b/tests/integration/test_udp.py
@@ -0,0 +1,168 @@
+"""The UDP tracker protocol, BEP 15.
+
+The HTTP tests cover the torrent store; these cover the packet handling in
+front of it, which has its own framing, its own connection id and its own
+rules about what to answer and what to ignore.
+"""
+
+import socket
+import struct
+import unittest
+
+from harness import TrackerTestCase
+
+MAGIC = 0x41727101980
+
+
+class UdpProtocol(TrackerTestCase):
+    def test_connect_echoes_the_transaction_id(self):
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        sock.settimeout(5)
+        try:
+            sock.sendto(struct.pack(">QII", MAGIC, 0, 0xDEADBEEF), ("127.0.0.1", self.tracker.port))
+            action, transaction, connection_id = struct.unpack(">IIQ", sock.recv(2048)[:16])
+            self.assertEqual(action, 0)
+            self.assertEqual(transaction, 0xDEADBEEF)
+            self.assertNotEqual(connection_id, 0)
+        finally:
+            sock.close()
+
+    def test_the_connection_id_is_stable_for_one_peer(self):
+        first, id_one = self.tracker.udp_connect()
+        second, id_two = self.tracker.udp_connect()
+        try:
+            self.assertEqual(id_one, id_two)
+        finally:
+            first.close()
+            second.close()
+
+    def test_a_wrong_connection_id_is_refused(self):
+        payload = struct.pack(">QII20s20sQQQIIIiH", 0, 1, 2, self.hash_for("udpbad"), b"p" * 20, 0, 0, 0, 0, 0, 0, -1, 6881)
+        answer = self.tracker.udp_raw(payload)
+        self.assertIsNotNone(answer, "expected an error packet")
+        action, transaction = struct.unpack(">II", answer[:8])
+        self.assertEqual(action, 3, "action 3 is the error packet")
+        self.assertIn(b"Connection ID missmatch", answer[8:])
+
+    def test_a_packet_shorter_than_the_header_is_ignored(self):
+        self.assertIsNone(self.tracker.udp_raw(b"\x00" * 15))
+
+    def test_an_unknown_action_is_ignored(self):
+        self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", MAGIC, 9, 1)))
+
+    def test_connect_requires_the_protocol_magic(self):
+        self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", MAGIC + 1, 0, 1)))
+
+    def test_an_announce_cut_short_is_ignored(self):
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.assertIsNone(self.tracker.udp_raw(struct.pack(">QII", connection_id, 1, 1) + b"\0" * 40))
+        finally:
+            sock.close()
+
+
+class UdpAnnounce(TrackerTestCase):
+    def test_a_seeder_is_counted_as_a_seeder(self):
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpseed"), left=0)
+            self.assertEqual(answer["seeders"], 1)
+            self.assertEqual(answer["leechers"], 0)
+            self.assertGreater(answer["interval"], 0)
+        finally:
+            sock.close()
+
+    def test_a_leecher_is_counted_as_a_leecher(self):
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpleech"), left=1024)
+            self.assertEqual(answer["seeders"], 0)
+            self.assertEqual(answer["leechers"], 1)
+        finally:
+            sock.close()
+
+    def test_stopped_removes_the_peer(self):
+        info_hash = self.hash_for("udpstop")
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.tracker.udp_announce(sock, connection_id, info_hash, left=0)
+            self.tracker.udp_announce(sock, connection_id, info_hash, left=0, event=3)
+            answer = self.tracker.udp_announce(sock, connection_id, info_hash, left=1024)
+            self.assertEqual(answer["seeders"], 0, "the seeder announced stopped")
+        finally:
+            sock.close()
+
+    def test_an_announce_over_udp_is_visible_over_http(self):
+        """One torrent store behind both bearers, not two."""
+        info_hash = self.hash_for("udpcross")
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.tracker.udp_announce(sock, connection_id, info_hash, left=0)
+        finally:
+            sock.close()
+        scrape = self.tracker.scrape(info_hash)
+        self.assertEqual(scrape[b"files"][info_hash][b"complete"], 1)
+
+    def test_the_peer_list_carries_the_announcing_peer(self):
+        info_hash = self.hash_for("udppeers")
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            answer = self.tracker.udp_announce(sock, connection_id, info_hash, left=0, port=6881)
+            self.assertEqual(len(answer["peers"]) % 6, 0, "v4 peers are six bytes each")
+            self.assertGreaterEqual(len(answer["peers"]), 6)
+            self.assertEqual(struct.unpack(">H", answer["peers"][4:6])[0], 6881)
+        finally:
+            sock.close()
+
+
+class UdpScrape(TrackerTestCase):
+    def test_scrape_reports_the_swarm(self):
+        info_hash = self.hash_for("udpscrape")
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.tracker.udp_announce(sock, connection_id, info_hash, left=0)
+            seeders, completed, leechers = self.tracker.udp_scrape(sock, connection_id, [info_hash])[0]
+            self.assertEqual(seeders, 1)
+            self.assertEqual(leechers, 0)
+            self.assertEqual(completed, 0)
+        finally:
+            sock.close()
+
+    def test_an_unknown_hash_scrapes_as_empty(self):
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.assertEqual(self.tracker.udp_scrape(sock, connection_id, [self.hash_for("udpnone")])[0], (0, 0, 0))
+        finally:
+            sock.close()
+
+    def test_several_hashes_answer_in_order(self):
+        seeding = self.hash_for("udpmultiA")
+        leeching = self.hash_for("udpmultiB")
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            self.tracker.udp_announce(sock, connection_id, seeding, left=0)
+            self.tracker.udp_announce(sock, connection_id, leeching, left=1024)
+            answers = self.tracker.udp_scrape(sock, connection_id, [seeding, leeching])
+            self.assertEqual(answers[0][0], 1, "first hash is the seeded one")
+            self.assertEqual(answers[1][2], 1, "second hash is the leeched one")
+        finally:
+            sock.close()
+
+
+class UdpWorkersConfiguredAfterTheListener(TrackerTestCase):
+    """The harness writes its listen lines first, so this directive lands
+    below them: the position a config file should not care about."""
+
+    config = "listen.udp.workers 4\n"
+
+    def test_the_tracker_starts_and_still_answers_udp(self):
+        sock, connection_id = self.tracker.udp_connect()
+        try:
+            answer = self.tracker.udp_announce(sock, connection_id, self.hash_for("udpworkers"), left=0)
+            self.assertEqual(answer["seeders"], 1)
+        finally:
+            sock.close()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/runner.c b/tests/runner.c
index a92315e..71a5e2a 100644
--- a/tests/runner.c
+++ b/tests/runner.c
@@ -4,7 +4,11 @@
    $id$ */
 
 /* System */
+#include <stdio.h>
+#include <stdlib.h>
 #include <string.h>
+#include <sys/wait.h>
+#include <unistd.h>
 
 /* Opentracker */
 #include "test.h"
@@ -31,15 +35,79 @@ static const struct module modules[] = {{"ot_urlencoded", test_urlencoded},
                                         {"ot_accesslist", test_accesslist},
                                         {NULL, NULL}};
 
-/* Run every module, or only those whose name contains argv[1] */
+struct tally {
+  unsigned long checks;
+  unsigned long failures;
+};
+
+/* Each module runs in a child of its own, so that a crash or a sanitizer
+   abort costs that module's results rather than every module after it. The
+   child reports its tally through a pipe; one that dies without reporting is
+   counted as a failure under its own name. */
+static void run_forked(const struct module *m) {
+  struct tally tally = {0, 0};
+  int          pipefd[2], status = 0;
+  pid_t        child;
+  ssize_t      got;
+
+  if (pipe(pipefd)) {
+    test_failf(__FILE__, __LINE__, "pipe failed, cannot isolate %s", m->name);
+    return;
+  }
+
+  fflush(stdout); /* or the child inherits a copy and prints it again */
+  child = fork();
+  if (child < 0) {
+    close(pipefd[0]);
+    close(pipefd[1]);
+    test_failf(__FILE__, __LINE__, "fork failed, cannot isolate %s", m->name);
+    return;
+  }
+
+  if (!child) {
+    close(pipefd[0]);
+    test_checks = test_failures = 0;
+    m->run();
+    tally.checks   = test_checks;
+    tally.failures = test_failures;
+    if (write(pipefd[1], &tally, sizeof(tally)) != (ssize_t)sizeof(tally))
+      _exit(1);
+    fflush(stdout);
+    _exit(0);
+  }
+
+  close(pipefd[1]);
+  got = read(pipefd[0], &tally, sizeof(tally));
+  close(pipefd[0]);
+  while (waitpid(child, &status, 0) < 0)
+    ;
+
+  if (got == (ssize_t)sizeof(tally)) {
+    test_checks   += tally.checks;
+    test_failures += tally.failures;
+  }
+  if (got != (ssize_t)sizeof(tally) || !WIFEXITED(status) || WEXITSTATUS(status))
+    test_failf(__FILE__, __LINE__, "%s did not run to the end", m->name);
+}
+
+/* Run every module, or only those whose name contains argv[1].
+   OT_TEST_FORK=0 keeps everything in one process, which is what a debugger
+   wants. */
 int main(int argc, char **argv) {
+  const char          *fork_setting = getenv("OT_TEST_FORK");
+  int                  isolate      = !fork_setting || strcmp(fork_setting, "0");
   const struct module *m;
 
   for (m = modules; m->name; ++m) {
     if (argc > 1 && !strstr(m->name, argv[1]))
       continue;
     test_module(m->name);
-    m->run();
+    if (!isolate) {
+      m->run();
+      continue;
+    }
+    test_case("the module runs to the end");
+    run_forked(m);
   }
 
   return test_summary();