40 lines
2.2 KiB
Python
40 lines
2.2 KiB
Python
import unittest
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from fgai.incidents import IncidentStore, build_incidents
|
|
from fgai.models import AnomalyFinding
|
|
|
|
class IncidentTests(unittest.TestCase):
|
|
def test_merges_anomaly_and_correlation(self):
|
|
anomaly = AnomalyFinding("10.0.0.1", 60, "high", "high", ["burst"], {})
|
|
result = build_incidents([anomaly], {"10.0.0.1": [{"score": 15, "reason": "new domain"}]}, [{"source_ip": "10.0.0.1", "streams": ["DNS", "Firewall"]}])
|
|
self.assertEqual(result[0]["score"], 85)
|
|
|
|
def test_creates_incident_for_profile_entity_without_network_anomaly(self):
|
|
result = build_incidents([], {"alice": [{"score": 15, "reason": "new login country", "stream_id": "windows"}]}, [])
|
|
self.assertEqual(result[0]["entity"], "alice")
|
|
self.assertEqual(result[0]["field_deviations"], 1)
|
|
|
|
def test_incident_uses_stream_name_for_field_deviation(self):
|
|
result = build_incidents([], {"alice": [{"score": 15, "reason": "new login country", "stream_id": "6a3993", "stream_name": "Windows"}]}, [])
|
|
self.assertEqual(result[0]["correlated_streams"], ["Windows"])
|
|
|
|
def test_incident_store_persists_lifecycle_state(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
store = IncidentStore(str(Path(directory) / "incidents.json"))
|
|
incident = build_incidents([], {"alice": [{"score": 15, "reason": "new login country", "stream_id": "windows"}]}, [])[0]
|
|
applied = store.apply([incident])[0]
|
|
self.assertEqual(applied["lifecycle_status"], "open")
|
|
|
|
store.update(str(applied["id"]), "acknowledged", "checking vpn logs")
|
|
applied_again = store.apply([incident])[0]
|
|
|
|
self.assertEqual(applied_again["lifecycle_status"], "acknowledged")
|
|
self.assertEqual(applied_again["note"], "checking vpn logs")
|
|
|
|
def test_incident_id_is_stable_when_evidence_changes(self):
|
|
first = build_incidents([], {"alice": [{"score": 15, "reason": "new login country", "stream_id": "windows"}]}, [])[0]
|
|
second = build_incidents([], {"alice": [{"score": 25, "reason": "new source ip", "stream_id": "windows"}]}, [])[0]
|
|
self.assertEqual(first["id"], second["id"])
|