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

Added the optional Ollama profile advisor

Browse Source
This commit is contained in:
larssand
2026-06-30 09:09:04 +02:00
parent f2385a00a9
commit d7b8c494e9
9 changed files with 178 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
tests/test_monitor.py
View File

@@ -95,6 +95,20 @@ class MonitorTests(unittest.TestCase):
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"])
if __name__ == "__main__":
unittest.main()

23
tests/test_profile_suggestions.py
View File

@@ -1,7 +1,7 @@
import unittest
from fgai.logs import parse_log_line
from fgai.profile_suggestions import suggest_stream_profiles
from fgai.profile_suggestions import apply_profile_advice, suggest_stream_profiles
class ProfileSuggestionTests(unittest.TestCase):
@@ -58,6 +58,27 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertIn("workflow_state", profile["categorical_fields"])
self.assertIn("risk_points", profile["numeric_fields"])
def test_applies_valid_llm_advice_and_rejects_unknown_fields(self):
suggestion = suggest_stream_profiles([
parse_log_line("fgai_stream_id=windows fgai_stream=Windows username=alice hostname=host01 eventid=4625 action=failure")
])[0]
advised = apply_profile_advice([suggestion], [{
"stream_id": "windows",
"entity_fields": ["username", "not_a_field"],
"timestamp_field": "eventtime",
"categorical_fields": ["eventid", "full_message"],
"numeric_fields": ["missing_number"],
"detectors": {"auth_failure": {"enabled": True, "minimum": 3, "z_threshold": 2.5}, "made_up": {"enabled": True}},
"reason": "Windows auth fields",
}])[0]
self.assertEqual(advised["profile_advisor"]["status"], "ok")
self.assertEqual(advised["profile"]["entity_fields"], ["username"])
self.assertIn("eventid", advised["profile"]["categorical_fields"])
self.assertNotIn("full_message", advised["profile"]["categorical_fields"])
self.assertEqual(set(advised["profile"]["detectors"]), {"auth_failure"})
if __name__ == "__main__":
unittest.main()