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

stream_coverage i status-API.

Browse Source
This commit is contained in:
larssand
2026-06-29 18:36:50 +02:00
parent c95634f976
commit 0b31e6c55e
5 changed files with 78 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
tests/test_monitor.py
View File

@@ -54,6 +54,29 @@ class MonitorTests(unittest.TestCase):
self.assertEqual(status["profile_readiness"][0]["profile_name"], "Fortigate profile")
self.assertEqual(status["stream_profiles"][0]["name"], "Fortigate profile")
def test_stream_coverage_marks_missing_profiles(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = Path(tmp) / "config.json"
config_path.write_text(
json.dumps(
{
"graylog_streams": [
{"id": "firewall", "title": "Firewall", "enabled": True},
{"id": "dns", "title": "DNS", "enabled": True},
],
"graylog_stream_profiles": [{"stream_id": "firewall", "name": "Firewall profile", "entity_field": "srcip", "categorical_fields": ["action"]}],
}
),
encoding="utf-8",
)
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))
coverage = {item["stream_id"]: item for item in status["stream_coverage"]}
self.assertEqual(coverage["firewall"]["health"], "no_events")
self.assertEqual(coverage["firewall"]["readiness"], "1/1")
self.assertEqual(coverage["dns"]["health"], "missing_profile")
def test_add_llm_assessment_records_error_without_ollama(self):
status = {"summary": {}, "anomalies": []}