100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import _bootstrap # noqa: F401 (将 src 加入 sys.path)
|
|
|
|
import unittest
|
|
|
|
from fault_log_analyzer.parser import parse_log, parse_timestamp, templatize, tokenize
|
|
|
|
|
|
class TestParseTimestamp(unittest.TestCase):
|
|
def test_iso_with_z(self):
|
|
dt = parse_timestamp("2025-01-01T10:18:00Z")
|
|
self.assertEqual(dt.year, 2025)
|
|
self.assertEqual(dt.minute, 18)
|
|
self.assertIsNotNone(dt.tzinfo)
|
|
|
|
def test_epoch_seconds(self):
|
|
dt = parse_timestamp(1735726680) # 2025-01-01T10:18:00Z
|
|
self.assertEqual(dt.year, 2025)
|
|
self.assertIsNotNone(dt.tzinfo)
|
|
|
|
def test_epoch_millis(self):
|
|
dt = parse_timestamp(1735726680000)
|
|
self.assertEqual(dt.year, 2025)
|
|
self.assertEqual(dt.hour, 10)
|
|
|
|
def test_epoch_nanos(self):
|
|
dt = parse_timestamp(1735726680000000000)
|
|
self.assertEqual(dt.year, 2025)
|
|
|
|
def test_common_format(self):
|
|
dt = parse_timestamp("2025-01-01 10:18:00")
|
|
self.assertEqual(dt.year, 2025)
|
|
self.assertEqual(dt.hour, 10)
|
|
|
|
def test_invalid_returns_none(self):
|
|
self.assertIsNone(parse_timestamp("not-a-date"))
|
|
|
|
def test_datetime_passthrough(self):
|
|
from datetime import datetime, timezone
|
|
|
|
dt = parse_timestamp(datetime(2025, 1, 1, tzinfo=timezone.utc))
|
|
self.assertEqual(dt.year, 2025)
|
|
|
|
|
|
class TestParseLog(unittest.TestCase):
|
|
def test_structured_json(self):
|
|
raw = {
|
|
"timestamp": "2025-01-01T10:18:00Z",
|
|
"level": "ERROR",
|
|
"message": "No space left on device",
|
|
"host_id": "h-001",
|
|
"service": "app",
|
|
"trace_id": "tr-1",
|
|
}
|
|
entry = parse_log(raw)
|
|
self.assertEqual(entry.level, "ERROR")
|
|
self.assertEqual(entry.host_id, "h-001")
|
|
self.assertEqual(entry.trace_id, "tr-1")
|
|
|
|
def test_level_extracted_from_message(self):
|
|
entry = parse_log({"message": "ERROR: something bad"})
|
|
self.assertEqual(entry.level, "ERROR")
|
|
|
|
def test_json_message_expands_fields(self):
|
|
raw = {"message": '{"message": "disk full", "level": "FATAL", "host_id": "h-2"}'}
|
|
entry = parse_log(raw)
|
|
self.assertEqual(entry.message, "disk full")
|
|
self.assertEqual(entry.level, "FATAL")
|
|
self.assertEqual(entry.host_id, "h-2")
|
|
|
|
def test_fields_fallback(self):
|
|
raw = {
|
|
"message": "boom",
|
|
"fields": {"hostname": "web-01", "service_name": "api", "request_id": "r-9"},
|
|
}
|
|
entry = parse_log(raw)
|
|
self.assertEqual(entry.host_id, "web-01")
|
|
self.assertEqual(entry.service, "api")
|
|
self.assertEqual(entry.trace_id, "r-9")
|
|
|
|
|
|
class TestTemplatize(unittest.TestCase):
|
|
def test_replace_variables(self):
|
|
msg = "Connection from 10.0.0.11 port 8080 failed after 3 retries"
|
|
tpl = templatize(msg)
|
|
self.assertIn("<IP>", tpl)
|
|
self.assertIn("<NUM>", tpl)
|
|
|
|
def test_replace_uuid(self):
|
|
tpl = templatize("request 123e4567-e89b-12d3-a456-426614174000 failed")
|
|
self.assertIn("<UUID>", tpl)
|
|
|
|
def test_tokenize(self):
|
|
tokens = tokenize("No space left on device")
|
|
self.assertIn("space", tokens)
|
|
self.assertIn("device", tokens)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|