102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
"""日志解析与模板化测试。"""
|
|
import unittest
|
|
|
|
from fault_log_analyzer.models import ParsedLog
|
|
from fault_log_analyzer.parser import LogParser
|
|
|
|
|
|
class TestParseJson(unittest.TestCase):
|
|
def setUp(self):
|
|
self.parser = LogParser()
|
|
|
|
def test_json_dict(self):
|
|
raw = {
|
|
"timestamp": "2025-01-01T10:18:00Z",
|
|
"level": "ERROR",
|
|
"message": "disk full",
|
|
"host_id": "h-1",
|
|
"service": "svc",
|
|
"trace_id": "tr-1",
|
|
}
|
|
parsed = self.parser.parse(raw)
|
|
self.assertEqual(parsed.level, "ERROR")
|
|
self.assertEqual(parsed.message, "disk full")
|
|
self.assertEqual(parsed.host_id, "h-1")
|
|
self.assertEqual(parsed.service, "svc")
|
|
self.assertEqual(parsed.trace_id, "tr-1")
|
|
|
|
def test_json_dict_uses_msg_alias(self):
|
|
parsed = self.parser.parse({"msg": "hello", "severity": "warn"})
|
|
self.assertEqual(parsed.message, "hello")
|
|
self.assertEqual(parsed.level, "WARN")
|
|
|
|
def test_json_line(self):
|
|
line = '{"timestamp":"2025-01-01T10:00:00Z","level":"FATAL","message":"oom"}'
|
|
parsed = self.parser.parse(line)
|
|
self.assertEqual(parsed.level, "FATAL")
|
|
self.assertEqual(parsed.message, "oom")
|
|
|
|
|
|
class TestParseLog4j(unittest.TestCase):
|
|
def setUp(self):
|
|
self.parser = LogParser()
|
|
|
|
def test_log4j(self):
|
|
line = "2025-01-01 10:18:00 ERROR [app] No space left on device"
|
|
parsed = self.parser.parse(line)
|
|
self.assertEqual(parsed.level, "ERROR")
|
|
self.assertEqual(parsed.service, "app")
|
|
self.assertEqual(parsed.message, "No space left on device")
|
|
|
|
def test_log4j_with_trace(self):
|
|
line = "2025-01-01 10:18:00 ERROR [svc] trace_id=abc-123-456 connection refused"
|
|
parsed = self.parser.parse(line)
|
|
self.assertEqual(parsed.trace_id, "abc-123-456")
|
|
|
|
|
|
class TestParseSyslog(unittest.TestCase):
|
|
def setUp(self):
|
|
self.parser = LogParser()
|
|
|
|
def test_syslog(self):
|
|
line = "<134>Jan 1 10:18:00 web-01 nginx[123]: connect timeout to upstream"
|
|
parsed = self.parser.parse(line)
|
|
self.assertEqual(parsed.host_id, "web-01")
|
|
self.assertEqual(parsed.service, "nginx")
|
|
self.assertIn("connect timeout", parsed.message)
|
|
|
|
|
|
class TestParseFallback(unittest.TestCase):
|
|
def setUp(self):
|
|
self.parser = LogParser()
|
|
|
|
def test_fallback_level_guess(self):
|
|
parsed = self.parser.parse("some random FATAL failure")
|
|
self.assertEqual(parsed.level, "FATAL")
|
|
self.assertEqual(parsed.message, "some random FATAL failure")
|
|
|
|
def test_passthrough_parsed_log(self):
|
|
pl = ParsedLog(level="ERROR", message="x")
|
|
self.assertIs(self.parser.parse(pl), pl)
|
|
|
|
|
|
class TestTemplate(unittest.TestCase):
|
|
def test_ip_uuid_number_replacement(self):
|
|
msg = "connect from 10.0.0.1 uuid 123e4567-e89b-12d3-a456-426614174000 retry 42"
|
|
templated = LogParser.template(msg)
|
|
self.assertIn("<IP>", templated)
|
|
self.assertIn("<UUID>", templated)
|
|
self.assertIn("<NUM>", templated)
|
|
self.assertNotIn("10.0.0.1", templated)
|
|
|
|
def test_tokenize_filters_short(self):
|
|
tokens = LogParser.tokenize("a b disk full error")
|
|
self.assertIn("disk", tokens)
|
|
self.assertIn("error", tokens)
|
|
self.assertNotIn("a", tokens)
|
|
self.assertNotIn("b", tokens)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|