6 files changed, 279 insertions, 22 deletions

diff --git a/tests/integration/harness.py b/tests/integration/harness.py
index 7e75574..03a0561 100644
--- a/tests/integration/harness.py
+++ b/tests/integration/harness.py
@@ -197,6 +197,10 @@ class Tracker:
         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."""
@@ -212,6 +216,73 @@ class Tracker:
             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."""
 
@@ -222,6 +293,13 @@ class TrackerTestCase(unittest.TestCase):
         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
index b4e134e..41ff654 100644
--- a/tests/integration/test_accesslist.py
+++ b/tests/integration/test_accesslist.py
@@ -37,12 +37,10 @@ class Blacklist(TrackerTestCase):
         self.assertNotIn(b"failure reason", answer)
         self.assertEqual(answer[b"complete"], 1)
 
-    def test_the_block_holds_over_udp_but_the_answer_is_malformed(self):
-        """The UDP path has no spelling for "not authorized": the announce
-        writes no body, so the tracker sends the eight byte header alone
-        where BEP 15 calls for at least twenty. Pinned rather than blessed —
-        an error packet or silence would both be defensible, a short announce
-        is neither."""
+    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(
@@ -50,10 +48,10 @@ class Blacklist(TrackerTestCase):
         )
         answer = self.tracker.udp_raw(request)
         self.assertIsNotNone(answer)
-        self.assertEqual(len(answer), 8, "header only, no interval or counts")
-        action, transaction = struct.unpack(">II", answer)
-        self.assertEqual(action, 1, "it still claims to be an announce")
+        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()
diff --git a/tests/integration/test_fullscrape.py b/tests/integration/test_fullscrape.py
index 4059656..be7b502 100644
--- a/tests/integration/test_fullscrape.py
+++ b/tests/integration/test_fullscrape.py
@@ -112,3 +112,35 @@ class FullScrapeChunking(TrackerTestCase):
 
 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
index 920d199..820b917 100644
--- a/tests/integration/test_stats.py
+++ b/tests/integration/test_stats.py
@@ -5,7 +5,9 @@ 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
@@ -14,8 +16,7 @@ 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; "busy" is advertised but unbuilt,
-# which the next case pins.
+# 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"]
 
@@ -34,14 +35,14 @@ class StatsShape(TrackerTestCase):
             for line in lines[:2]:
                 self.assertRegex(line, rb"^\d+$", "mode %s line %r" % (mode, line))
 
-    def test_the_busy_mode_is_advertised_but_not_built(self):
-        """TASK_STATS_BUSY_NETWORKS is in the enum and in the mode table, and
-        no switch anywhere handles it, so the worker frees the empty answer
-        and the request ends as a server error. Pinned rather than blessed:
-        either the report gets written or the mode leaves the table."""
+    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, 500)
+        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:
@@ -131,15 +132,37 @@ class StatsTop(TrackerTestCase):
 
 
 class StatsNetworks(TrackerTestCase):
-    def test_the_slash24_report_names_the_loopback_network(self):
+    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 masked out of a v4 peer's four address bytes and then
-        # printed through the v6 formatter, so 127.0.0.0/24 reads as "7f00::"
-        # rather than as an address anyone would recognise. Pinned as it is.
-        self.assertIn(b"7f00::", 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__":
diff --git a/tests/integration/test_udp.py b/tests/integration/test_udp.py
index 1dfa21b..1cb2389 100644
--- a/tests/integration/test_udp.py
+++ b/tests/integration/test_udp.py
@@ -149,5 +149,20 @@ class UdpScrape(TrackerTestCase):
             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()