64 lines
2.7 KiB
Python
64 lines
2.7 KiB
Python
import unittest
|
|
|
|
from fgai.logs import parse_log_line
|
|
from fgai.profile_suggestions import suggest_stream_profiles
|
|
|
|
|
|
class ProfileSuggestionTests(unittest.TestCase):
|
|
def test_suggests_common_profile_fields_per_stream(self):
|
|
events = [
|
|
parse_log_line(f"fgai_stream_id=firewall fgai_stream=Firewall srcip=10.0.0.{index} dstip=8.8.8.8 dstport=443 action=deny timestamp=2026-06-29T10:00:{index:02d}Z hitcount={index}")
|
|
for index in range(1, 25)
|
|
]
|
|
|
|
suggestions = suggest_stream_profiles(events)
|
|
profile = suggestions[0]["profile"]
|
|
|
|
self.assertEqual(profile["stream_id"], "firewall")
|
|
self.assertIn("srcip", profile["entity_fields"])
|
|
self.assertEqual(profile["timestamp_field"], "eventtime")
|
|
self.assertIn("action", profile["categorical_fields"])
|
|
self.assertIn("dstport", profile["categorical_fields"])
|
|
self.assertIn("hitcount", profile["numeric_fields"])
|
|
self.assertIn("deny_action", profile["detectors"])
|
|
|
|
def test_suggests_windows_event_fields(self):
|
|
events = [
|
|
parse_log_line(
|
|
f"fgai_stream_id=windows fgai_stream=Windows winlog_event_id=4625 "
|
|
f"winlog_event_data_targetusername=alice winlog_computer_name=host{index % 3} "
|
|
f"event_provider=Microsoft-Windows-Security-Auditing logon_type=3 "
|
|
f"eventtime=2026-06-29T10:00:{index:02d}Z action=failure"
|
|
)
|
|
for index in range(1, 30)
|
|
]
|
|
|
|
profile = suggest_stream_profiles(events)[0]["profile"]
|
|
|
|
self.assertIn("username", profile["entity_fields"])
|
|
self.assertIn("hostname", profile["entity_fields"])
|
|
self.assertIn("eventid", profile["categorical_fields"])
|
|
self.assertIn("logon_type", profile["categorical_fields"])
|
|
self.assertIn("auth_failure", profile["detectors"])
|
|
|
|
def test_discovers_non_priority_application_fields(self):
|
|
events = [
|
|
parse_log_line(
|
|
f"fgai_stream_id=app fgai_stream=BillingApp tenant_id=t{index % 4} actor_id=user{index % 7} "
|
|
f"workflow_state={'approved' if index % 2 else 'rejected'} payment_provider=stripe risk_points={index % 9} "
|
|
f"created_at=2026-06-29T10:00:{index:02d}Z"
|
|
)
|
|
for index in range(1, 40)
|
|
]
|
|
|
|
profile = suggest_stream_profiles(events)[0]["profile"]
|
|
|
|
self.assertIn("username", profile["entity_fields"])
|
|
self.assertIn("tenant_id", profile["categorical_fields"])
|
|
self.assertIn("workflow_state", profile["categorical_fields"])
|
|
self.assertIn("risk_points", profile["numeric_fields"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|