From 9d21cb476d07ab5583c99b1418d4ffb54fbb2a10 Mon Sep 17 00:00:00 2001
From: Dirk Engling <erdgeist@erdgeist.org>
Date: Tue, 27 Oct 2026 01:16:37 +0100
Subject: Adding the actual tests to do aforementioned 'Massively increase
 coverage by pulling fullscrapes and sending udp packets'

---
 tests/integration/test_accesslist.py | 102 +++++++++++++++++++++++++++++++++++
 1 file changed, 102 insertions(+)
 create mode 100644 tests/integration/test_accesslist.py

(limited to 'tests/integration/test_accesslist.py')

diff --git a/tests/integration/test_accesslist.py b/tests/integration/test_accesslist.py
new file mode 100644
index 0000000..b4e134e
--- /dev/null
+++ b/tests/integration/test_accesslist.py
@@ -0,0 +1,102 @@
+"""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_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."""
+        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), 8, "header only, no interval or counts")
+        action, transaction = struct.unpack(">II", answer)
+        self.assertEqual(action, 1, "it still claims to be an announce")
+        self.assertEqual(transaction, 7)
+
+    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()
-- 
cgit v1.3