poc3

tests/test_config.py

@@ -0,0 +1,23 @@
+import os
+import unittest
+from unittest.mock import patch
+
+from app.config import Config
+
+
+class ConfigTests(unittest.TestCase):
+    def test_rule_update_interval_defaults_to_24_hours(self):
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(Config.from_env().rule_update_interval_hours, 24)
+
+    def test_rule_update_interval_can_be_changed_or_disabled(self):
+        with patch.dict(os.environ, {"RULE_UPDATE_INTERVAL_HOURS": "6"}, clear=True):
+            self.assertEqual(Config.from_env().rule_update_interval_hours, 6)
+        with patch.dict(os.environ, {"RULE_UPDATE_INTERVAL_HOURS": "0"}, clear=True):
+            self.assertEqual(Config.from_env().rule_update_interval_hours, 0)
+        with patch.dict(os.environ, {"RULE_UPDATE_INTERVAL_HOURS": "-5"}, clear=True):
+            self.assertEqual(Config.from_env().rule_update_interval_hours, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_forensics.py

@@ -0,0 +1,48 @@
+import ipaddress
+import os
+import struct
+import tempfile
+import unittest
+
+from app.forensics import ForensicPcapRing
+
+
+def ipv4_frame(src: str, dst: str, payload: bytes = b"evidence") -> bytes:
+    ethernet = b"\x00\x11\x22\x33\x44\x55\x66\x77\x88\x99\xaa\xbb\x08\x00"
+    total = 20 + len(payload)
+    header = bytearray(20)
+    header[0] = 0x45
+    header[2:4] = total.to_bytes(2, "big")
+    header[8] = 64
+    header[9] = 6
+    header[12:16] = ipaddress.IPv4Address(src).packed
+    header[16:20] = ipaddress.IPv4Address(dst).packed
+    return ethernet + bytes(header) + payload
+
+
+class ForensicPcapTests(unittest.TestCase):
+    def test_blocks_mode_persists_only_when_capture_is_requested(self):
+        with tempfile.TemporaryDirectory() as td:
+            ring = ForensicPcapRing("blocks", td, window_seconds=60, memory_mb=1)
+            ring.observe(ipv4_frame("192.0.2.10", "198.51.100.4"))
+            self.assertEqual(os.listdir(td), [])
+
+            result = ring.capture_target("198.51.100.4", label="sid-42")
+            self.assertIsNotNone(result)
+            path = os.path.join(td, result["name"])
+            self.assertTrue(os.path.isfile(path))
+            self.assertEqual(result["packet_count"], 1)
+            with open(path, "rb") as handle:
+                magic = struct.unpack("<I", handle.read(4))[0]
+            self.assertEqual(magic, 0xA1B2C3D4)
+
+    def test_alerts_mode_leaves_persistence_to_suricata(self):
+        with tempfile.TemporaryDirectory() as td:
+            ring = ForensicPcapRing("alerts", td)
+            ring.observe(ipv4_frame("192.0.2.10", "198.51.100.4"))
+            self.assertIsNone(ring.capture_target("198.51.100.4"))
+            self.assertEqual(os.listdir(td), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_redis_service.py

@@ -20,3 +20,15 @@ def test_redis_uses_aof_everysec_and_rdb_snapshot():
         assert cmd[cmd.index("--dir") + 1] == td
         assert cmd[cmd.index("--maxmemory") + 1] == "0"
         assert cmd[cmd.index("--maxmemory-policy") + 1] == "noeviction"
+
+
+def test_redis_status_exposes_runtime_details_for_system_ui():
+    with tempfile.TemporaryDirectory() as td:
+        supervisor = RedisSupervisor(True, td, port=6381, snapshot_seconds=1200, aof=True)
+        supervisor.executable = "/usr/bin/redis-server"
+        status = supervisor.status()
+        assert status["managed"] is True
+        assert status["port"] == 6381
+        assert status["data_dir"] == td
+        assert status["snapshot_seconds"] == 1200
+        assert status["persistence"] == "AOF everysec + RDB"

tests/test_rules.py

@@ -170,6 +170,45 @@ Enabled sources:
                 ],
             )
 
+
+    def test_manual_url_source_is_added_and_rules_rebuilt(self):
+        with tempfile.TemporaryDirectory() as td:
+            manager = self.make_manager(td)
+            manager.suricata_available = True
+            calls = []
+            manager._run_suricata_update = lambda args, timeout: (calls.append(list(args)) or subprocess.CompletedProcess(args, 0, stdout="added"))
+            manager._run_vendor_update_unlocked = lambda: RuleActionResult(True, "rebuilt")
+
+            result = manager.add_manual_source("vendor/community", "https://rules.example.invalid/feed.rules")
+            self.assertTrue(result.ok)
+            self.assertEqual(calls, [["add-source", "vendor/community", "https://rules.example.invalid/feed.rules", "--no-checksum"]])
+
+    def test_manual_url_source_rejects_non_http_url(self):
+        with tempfile.TemporaryDirectory() as td:
+            manager = self.make_manager(td)
+            manager.suricata_available = True
+            result = manager.add_manual_source("vendor/community", "file:///tmp/feed.rules")
+            self.assertFalse(result.ok)
+
+    def test_merged_rules_are_searchable_and_paginated(self):
+        with tempfile.TemporaryDirectory() as td:
+            manager = self.make_manager(td)
+            rules = os.path.join(td, "lib", "suricata", "rules")
+            os.makedirs(rules, exist_ok=True)
+            with open(os.path.join(rules, "suricata.rules"), "w", encoding="utf-8") as handle:
+                handle.write('# generated\n')
+                handle.write('alert tcp any any -> any 80 (msg:"one"; sid:1;)\n')
+                handle.write('alert dns any any -> any any (msg:"two"; sid:2;)\n')
+                handle.write('alert tcp any any -> any 443 (msg:"three"; sid:3;)\n')
+
+            result = manager.merged_rules("tcp", offset=0, limit=1)
+            self.assertTrue(result["ok"])
+            self.assertEqual(result["total_rules"], 3)
+            self.assertEqual(result["matched"], 2)
+            self.assertEqual(result["next_offset"], 1)
+            self.assertIn('sid:1', result["content"])
+            self.assertEqual(manager.status()["vendor_rule_count"], 3)
+
     def test_adaptive_threshold_uses_global_limit_and_snapshot_is_persistent(self):
         with tempfile.TemporaryDirectory() as td:
             manager = self.make_manager(td)

tests/test_webui.py

@@ -54,6 +54,24 @@ class WebUITests(unittest.TestCase):
     def test_intelligence_stages_column_has_dedicated_width_hook(self):
         self.assertIn('<th class="stages-col">Stages</th>', DASHBOARD)
 
+    def test_security_uses_compact_internal_tabs(self):
+        for tab in ("incidents", "analytics", "telemetry"):
+            self.assertIn(f'data-subtab-group="security" data-subtab="{tab}"', DASHBOARD)
+            self.assertIn(f'data-subtab-panel="security:{tab}"', DASHBOARD)
+
+    def test_intelligence_uses_compact_internal_tabs(self):
+        for tab in ("incidents", "assets", "threat-intel", "forensics"):
+            self.assertIn(f'data-subtab-group="intelligence" data-subtab="{tab}"', DASHBOARD)
+            self.assertIn(f'data-subtab-panel="intelligence:{tab}"', DASHBOARD)
+        self.assertIn("Asset intelligence", DASHBOARD)
+        self.assertIn("Threat intelligence repository", DASHBOARD)
+        self.assertIn("Forensic PCAP ring", DASHBOARD)
+
+    def test_system_has_dedicated_redis_status_panel(self):
+        self.assertIn('<h2>Redis</h2>', DASHBOARD)
+        self.assertIn('id="redisStateBadge"', DASHBOARD)
+        self.assertIn('id="redisStatus"', DASHBOARD)
+
     def test_live_stream_is_opt_in_and_bounded(self):
         self.assertIn('Continuous streaming is off by default', DASHBOARD)
         self.assertIn('id="toggleLive"', DASHBOARD)
@@ -76,9 +94,19 @@ class WebUITests(unittest.TestCase):
             "clearSourceSelection",
             "queueSelectedSources",
             "sourceQueueStatus",
+            "feedRuleCount",
+            "ruleUpdateSchedule",
         ):
             self.assertIn(f'id="{element_id}"', DASHBOARD)
 
+    def test_merged_rules_ui_has_explicit_counters(self):
+        self.assertIn('id="mergedRuleCount"', DASHBOARD)
+        self.assertIn('id="mergedRuleMatchCount"', DASHBOARD)
+
+    def test_merged_rules_panel_is_collapsed_by_default(self):
+        self.assertIn('<details id="mergedRulesPanel" class="panel mt-4 collapsible-panel">', DASHBOARD)
+        self.assertNotIn('<details id="mergedRulesPanel" class="panel mt-4 collapsible-panel" open', DASHBOARD)
+
     def test_autonomous_ids_operations_are_exposed_in_ui(self):
         for element_id in (
             "ruleIntelRows", "ruleSnapshotRows", "createRuleSnapshot",