Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

multi-entity stream profiles.

Browse Source
This commit is contained in:
larssand
2026-06-29 19:20:08 +02:00
parent aab9f15c6d
commit 4a85e53869
1 changed files with 3 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/test_monitor.py
View File

@@ -22,7 +22,7 @@ class MonitorTests(unittest.TestCase):
encoding="utf-8", encoding="utf-8",
) )
status = build_status(str(log_path)) status = build_status(str(log_path), incident_path=str(Path(tmp) / "incidents.json"))
self.assertEqual(status["summary"]["total"], 3) self.assertEqual(status["summary"]["total"], 3)
self.assertGreaterEqual(len(status["anomalies"]), 1) self.assertGreaterEqual(len(status["anomalies"]), 1)
@@ -48,7 +48,7 @@ class MonitorTests(unittest.TestCase):
encoding="utf-8", encoding="utf-8",
) )
with patch("fgai.monitor.BaselineStore.profile_readiness", return_value=[{"stream_id": "66fe", "field": "action", "buckets": 12, "ready": True}]): with patch("fgai.monitor.BaselineStore.profile_readiness", return_value=[{"stream_id": "66fe", "field": "action", "buckets": 12, "ready": True}]):
status = build_status(str(Path(tmp) / "missing.log"), baseline_path=str(Path(tmp) / "baseline.sqlite3"), config_path=str(config_path)) status = build_status(str(Path(tmp) / "missing.log"), baseline_path=str(Path(tmp) / "baseline.sqlite3"), config_path=str(config_path), incident_path=str(Path(tmp) / "incidents.json"))
self.assertEqual(status["profile_readiness"][0]["stream_name"], "Fortigate") self.assertEqual(status["profile_readiness"][0]["stream_name"], "Fortigate")
self.assertEqual(status["profile_readiness"][0]["profile_name"], "Fortigate profile") self.assertEqual(status["profile_readiness"][0]["profile_name"], "Fortigate profile")
@@ -70,7 +70,7 @@ class MonitorTests(unittest.TestCase):
encoding="utf-8", encoding="utf-8",
) )
with patch("fgai.monitor.BaselineStore.profile_readiness", return_value=[{"stream_id": "firewall", "field": "action", "buckets": 12, "ready": True}]): with patch("fgai.monitor.BaselineStore.profile_readiness", return_value=[{"stream_id": "firewall", "field": "action", "buckets": 12, "ready": True}]):
status = build_status(str(Path(tmp) / "missing.log"), baseline_path=str(Path(tmp) / "baseline.sqlite3"), config_path=str(config_path)) status = build_status(str(Path(tmp) / "missing.log"), baseline_path=str(Path(tmp) / "baseline.sqlite3"), config_path=str(config_path), incident_path=str(Path(tmp) / "incidents.json"))
coverage = {item["stream_id"]: item for item in status["stream_coverage"]} coverage = {item["stream_id"]: item for item in status["stream_coverage"]}
self.assertEqual(coverage["firewall"]["health"], "no_events") self.assertEqual(coverage["firewall"]["health"], "no_events")