28 lines
1.0 KiB
Python
28 lines
1.0 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"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|