29 lines
988 B
Python
29 lines
988 B
Python
import unittest
|
|
|
|
from fgai.entities import sample_timeline
|
|
from fgai.logs import parse_log_line
|
|
from fgai.query_details import event_query_details
|
|
|
|
|
|
class QueryDetailsTests(unittest.TestCase):
|
|
def test_builds_generic_graylog_query_from_aliases(self):
|
|
event = parse_log_line("src_addr=10.0.0.5 dst_addr=198.51.100.10 fw_action=allow")
|
|
|
|
details = event_query_details(event)
|
|
|
|
self.assertIn('srcip:"10.0.0.5"', details["query"])
|
|
self.assertIn('dstip:"198.51.100.10"', details["query"])
|
|
self.assertIn('action:"allow"', details["query"])
|
|
|
|
def test_timeline_contains_query_details(self):
|
|
event = parse_log_line('timestamp=2026-06-29T16:45:52Z fgai_stream=Firewall srcip=10.0.0.5 dstip=198.51.100.10 action=deny')
|
|
|
|
row = sample_timeline([event])[0]
|
|
|
|
self.assertEqual(row["query_details"]["stream"], "Firewall")
|
|
self.assertIn('srcip:"10.0.0.5"', row["graylog_query"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|