139 lines
6.9 KiB
Python
139 lines
6.9 KiB
Python
import tempfile
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from fgai.history import StatusSnapshotStore
|
|
from fgai.monitor import add_llm_assessment, build_status, cached_status_with_error, 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), incident_path=str(Path(tmp) / "incidents.json"))
|
|
|
|
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())
|
|
|
|
def test_profile_readiness_includes_stream_and_profile_names(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_path = Path(tmp) / "config.json"
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"graylog_streams": [{"id": "66fe", "title": "Fortigate", "enabled": True}],
|
|
"graylog_stream_profiles": [{"stream_id": "66fe", "name": "66fe", "entity_field": "srcip", "categorical_fields": ["action"]}],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
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), incident_path=str(Path(tmp) / "incidents.json"))
|
|
|
|
self.assertEqual(status["profile_readiness"][0]["stream_name"], "Fortigate")
|
|
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), incident_path=str(Path(tmp) / "incidents.json"))
|
|
|
|
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": []}
|
|
|
|
with patch("fgai.monitor.ollama_dashboard_assessment", side_effect=TimeoutError("timeout")):
|
|
add_llm_assessment(status, previous="old text")
|
|
|
|
self.assertEqual(status["llm_assessment"]["status"], "error")
|
|
self.assertEqual(status["llm_assessment"]["text"], "old text")
|
|
|
|
def test_add_llm_assessment_records_text(self):
|
|
status = {"summary": {}, "anomalies": []}
|
|
|
|
with patch("fgai.monitor.ollama_dashboard_assessment", return_value="looks noisy"):
|
|
add_llm_assessment(status)
|
|
|
|
self.assertEqual(status["llm_assessment"]["status"], "ok")
|
|
self.assertEqual(status["llm_assessment"]["text"], "looks noisy")
|
|
|
|
def test_profile_advisor_status_records_error(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
log_path = Path(tmp) / "events.log"
|
|
log_path.write_text("fgai_stream_id=windows fgai_stream=Windows username=alice eventid=4625 action=failure\n", encoding="utf-8")
|
|
config_path = Path(tmp) / "config.json"
|
|
config_path.write_text(json.dumps({"profile_advisor_enabled": True, "profile_advisor_model": "qwen3:8b"}), encoding="utf-8")
|
|
|
|
with patch("fgai.monitor.ollama_profile_advice", side_effect=TimeoutError("timeout")):
|
|
status = build_status(str(log_path), config_path=str(config_path), incident_path=str(Path(tmp) / "incidents.json"))
|
|
|
|
advisor = status["capabilities"]["profile_advisor"]
|
|
self.assertEqual(advisor["status"], "error")
|
|
self.assertIn("timeout", advisor["error"])
|
|
|
|
def test_cached_status_with_error_keeps_last_good_dashboard_data(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
cache_path = str(Path(tmp) / "status-cache.sqlite3")
|
|
StatusSnapshotStore(cache_path).save(
|
|
"last_good",
|
|
{
|
|
"summary": {"total": 42},
|
|
"capabilities": {"graylog_mcp": {"status": "connected"}},
|
|
"stream_coverage": [{"stream_name": "Firewall", "aggregate_status": "error"}],
|
|
"cross_source_correlations": [{"entity": "10.0.0.1"}],
|
|
},
|
|
)
|
|
|
|
status = cached_status_with_error(cache_path, {"status": "error", "error": "mcp down"})
|
|
|
|
self.assertIsNotNone(status)
|
|
self.assertEqual(status["summary"]["total"], 42)
|
|
self.assertTrue(status["stale"])
|
|
self.assertEqual(status["capabilities"]["graylog_mcp"]["status"], "error")
|
|
self.assertTrue(status["status_cache"]["served_from_cache"])
|
|
self.assertEqual(status["stream_coverage"][0]["aggregate_error"], "")
|
|
self.assertEqual(status["stream_coverage"][0]["raw_error"], "")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|