enodia-sentinal/tests/test_event_envelope.py

61 lines
2.4 KiB
Python

# SPDX-License-Identifier: GPL-3.0-or-later
"""Compatibility tests for the enodia.event.v1 envelope."""
import os
import unittest
from datetime import datetime
from enodia_sentinel import event, schemas
class TestEventEnvelope(unittest.TestCase):
def test_schema_id_is_stable(self):
self.assertEqual(schemas.EVENT_V1, "enodia.event.v1")
def test_build_event_shape_and_types(self):
env = event.build_event("alert", {"sid": 100010})
self.assertEqual(env["schema"], schemas.EVENT_V1)
self.assertEqual(env["event_type"], "alert")
self.assertIsInstance(env["timestamp"], str)
self.assertIsInstance(env["host"], str)
# EVE-style: payload lives under a key named by event_type.
self.assertEqual(env["alert"], {"sid": 100010})
def test_timestamp_defaults_to_iso8601(self):
env = event.build_event("status", {"running": True})
# Parses as ISO-8601 without raising.
datetime.fromisoformat(env["timestamp"])
def test_host_defaults_to_nodename(self):
env = event.build_event("status", {"running": True})
self.assertEqual(env["host"], os.uname().nodename)
def test_overrides_are_respected(self):
env = event.build_event(
"incident", {"id": "x"}, host="host-a", timestamp="2026-07-10T00:00:00-07:00")
self.assertEqual(env["host"], "host-a")
self.assertEqual(env["timestamp"], "2026-07-10T00:00:00-07:00")
def test_unknown_event_type_rejected(self):
with self.assertRaises(ValueError):
event.build_event("nope", {})
def test_event_types_is_closed_set(self):
self.assertEqual(event.EVENT_TYPES, frozenset({"alert", "incident", "status"}))
def test_from_alert_wraps_alert_payload(self):
from enodia_sentinel.alert import Alert, Severity
alert = Alert(
Severity.CRITICAL, "reverse_shell", "rsh:4242",
"pid=4242 peer=[8.8.8.8:4444]", (4242,),
sid=100010, classtype="command-and-control")
env = event.from_alert(alert, host="host-a")
self.assertEqual(env["event_type"], "alert")
self.assertEqual(env["host"], "host-a")
# Payload is the untouched enodia.alert.v1 object.
self.assertEqual(env["alert"], alert.to_dict())
self.assertEqual(env["alert"]["signature"], "reverse_shell")
self.assertEqual(env["alert"]["sid"], 100010)
if __name__ == "__main__":
unittest.main()