58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from fgai.monitor import add_llm_assessment, 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())
|
|
|
|
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")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|