feat(prune): add functionality to remove old single-host IP blocks while keeping CIDR ranges intact

tests/test_repositories.py

@@ -13,6 +13,7 @@ import os
 import sqlite3
 import tempfile
 import unittest
+from datetime import datetime, timedelta, timezone
 
 from app.db import connect
 from app.db import repositories as repo
@@ -110,7 +111,8 @@ class BlocklistTest(unittest.TestCase):
                                    added_by="scan:1")
         self.assertEqual(r1["active"], 2)
         # a later scan only sees 5.5.5.5; 6.6.6.6 stays (cumulative).
-        r2 = repo.record_blocklist(self.conn, ["5.5.5.5/32"], added_by="scan:2")
+        r2 = repo.record_blocklist(
+            self.conn, ["5.5.5.5/32"], added_by="scan:2")
         active = [e["cidr"] for e in repo.active_blocklist(self.conn)]
         self.assertIn("6.6.6.6/32", active)
         self.assertEqual(r2["active"], 2)
@@ -122,6 +124,38 @@ class BlocklistTest(unittest.TestCase):
                          [e["cidr"] for e in repo.active_blocklist(self.conn)])
         self.assertFalse(repo.unblock_entry(self.conn, "7.7.7.7/32"))
 
+    def test_prune_old_single_ip_blocks_keeps_ranges(self):
+        old = (datetime.now(timezone.utc) - timedelta(days=31)).isoformat(
+            timespec="seconds")
+        with repo.transaction(self.conn):
+            repo._block_entry_inner(self.conn, "1.2.3.4/32", last_seen=old)
+            repo._block_entry_inner(
+                self.conn, "2001:db8::1/128", last_seen=old)
+            repo._block_entry_inner(self.conn, "1.2.3.0/24", last_seen=old)
+            repo._block_entry_inner(self.conn, "2001:db8::/64", last_seen=old)
+
+        removed = repo.prune_old_single_ip_blocks(
+            self.conn, older_than_days=30)
+        self.assertCountEqual(removed, ["1.2.3.4/32", "2001:db8::1/128"])
+
+        active = {e["cidr"] for e in repo.active_blocklist(self.conn)}
+        self.assertIn("1.2.3.0/24", active)
+        self.assertIn("2001:db8::/64", active)
+        self.assertNotIn("1.2.3.4/32", active)
+        self.assertNotIn("2001:db8::1/128", active)
+
+    def test_prune_old_single_ip_blocks_respects_recency(self):
+        fresh = (datetime.now(timezone.utc) - timedelta(days=5)).isoformat(
+            timespec="seconds")
+        with repo.transaction(self.conn):
+            repo._block_entry_inner(self.conn, "9.9.9.9/32", last_seen=fresh)
+
+        removed = repo.prune_old_single_ip_blocks(
+            self.conn, older_than_days=30)
+        self.assertEqual(removed, [])
+        self.assertIn("9.9.9.9/32",
+                      {e["cidr"] for e in repo.active_blocklist(self.conn)})
+
 
 class WhitelistTest(unittest.TestCase):
     def setUp(self):