52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
import io
|
|
import json
|
|
import logging
|
|
import unittest
|
|
from app.common.logging import StructuredJsonFormatter, setup_logger
|
|
|
|
|
|
class TestStructuredJsonFormatter(unittest.TestCase):
|
|
def test_log_format_structure(self):
|
|
stream = io.StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
handler.setFormatter(StructuredJsonFormatter(service_name="test-service"))
|
|
|
|
logger = logging.getLogger("test_logger")
|
|
logger.setLevel(logging.INFO)
|
|
logger.addHandler(handler)
|
|
|
|
logger.info("Batch processed successfully", extra={"raw_id": "raw-12345"})
|
|
|
|
output = stream.getvalue().strip()
|
|
data = json.loads(output)
|
|
|
|
self.assertIn("timestamp", data)
|
|
self.assertEqual(data["level"], "INFO")
|
|
self.assertEqual(data["service"], "test-service")
|
|
self.assertEqual(data["raw_id"], "raw-12345")
|
|
self.assertEqual(data["event"], "Batch processed successfully")
|
|
|
|
def test_log_format_null_raw_id(self):
|
|
stream = io.StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
handler.setFormatter(StructuredJsonFormatter(service_name="test-service"))
|
|
|
|
logger = logging.getLogger("test_logger_null")
|
|
logger.setLevel(logging.INFO)
|
|
logger.addHandler(handler)
|
|
|
|
logger.info("Server started on port 8000")
|
|
|
|
output = stream.getvalue().strip()
|
|
data = json.loads(output)
|
|
|
|
self.assertIn("timestamp", data)
|
|
self.assertEqual(data["level"], "INFO")
|
|
self.assertIsNone(data["raw_id"])
|
|
self.assertEqual(data["event"], "Server started on port 8000")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|