add ui

tests/test_anomaly.py

@@ -0,0 +1,42 @@
+import unittest
+
+from fgai.anomaly import anomaly_summary, detect_source_anomalies
+from fgai.logs import parse_log_line
+
+
+class AnomalyTests(unittest.TestCase):
+    def test_scores_repeated_utm_public_source_as_anomaly(self):
+        events = [
+            parse_log_line(f'type=traffic srcip=10.0.0.{i} dstip=1.1.1.1 service=https action=accept sentbyte=100 rcvdbyte=100')
+            for i in range(1, 8)
+        ]
+        events.extend(
+            parse_log_line(
+                'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https '
+                'action=blocked severity=critical sentbyte=0 rcvdbyte=0'
+            )
+            for _ in range(5)
+        )
+
+        findings = detect_source_anomalies(events)
+
+        self.assertEqual(findings[0].subject, "8.8.8.8")
+        self.assertGreaterEqual(findings[0].score, 60)
+        self.assertIn(findings[0].severity, {"high", "critical"})
+
+    def test_summary_counts_severities(self):
+        events = [
+            parse_log_line(
+                'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https '
+                'action=blocked severity=critical'
+            )
+            for _ in range(5)
+        ]
+
+        summary = anomaly_summary(detect_source_anomalies(events))
+
+        self.assertEqual(summary["total"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_monitor.py

@@ -0,0 +1,38 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from fgai.monitor import build_status, write_status
+
+
+class MonitorTests(unittest.TestCase):
+    def test_build_status_from_log_file(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = Path(tmp) / "fg.log"
+            log_path.write_text(
+                "\n".join(
+                    [
+                        'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https action=blocked severity=critical',
+                        'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https action=blocked severity=critical',
+                        'type=utm subtype=ips srcip=8.8.8.8 dstip=10.0.0.10 service=https action=blocked severity=critical',
+                    ]
+                ),
+                encoding="utf-8",
+            )
+
+            status = build_status(str(log_path))
+
+            self.assertEqual(status["summary"]["total"], 3)
+            self.assertGreaterEqual(len(status["anomalies"]), 1)
+
+    def test_write_status_creates_parent_directory(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            output = Path(tmp) / "state" / "status.json"
+
+            write_status({"ok": True}, str(output))
+
+            self.assertTrue(output.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()