28 lines
1.1 KiB
Python
28 lines
1.1 KiB
Python
import unittest
|
|
|
|
from fgai.triage import build_triage_queue
|
|
|
|
|
|
class TriageTests(unittest.TestCase):
|
|
def test_builds_compact_queue_from_raw_deviations(self):
|
|
queue = build_triage_queue(
|
|
[{"id": "abc", "entity": "alice", "score": 42, "severity": "medium", "correlated_streams": ["Windows"], "evidence": ["login burst"]}],
|
|
{
|
|
"alice": [
|
|
{"score": 60, "detector": "auth_failure_burst", "stream_name": "Windows", "reason": "auth failure burst"},
|
|
{"score": 0, "detector": "rare_value", "stream_name": "DNS", "reason": "known", "feedback": "expected"},
|
|
]
|
|
},
|
|
[{"entity": "alice", "streams": ["Windows", "DNS"], "security_events": 2}],
|
|
[],
|
|
)
|
|
|
|
self.assertEqual(queue[0]["entity"], "alice")
|
|
self.assertEqual(queue[0]["active_deviation_count"], 1)
|
|
self.assertIn("auth_failure_burst", queue[0]["detectors"])
|
|
self.assertEqual(queue[0]["next_action"], "investigate cross-source activity")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|