81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
import json
|
|
import logging
|
|
import io
|
|
|
|
from agent.json_logging import JsonFormatter, setup_logging
|
|
|
|
|
|
def test_json_formatter_basic():
|
|
formatter = JsonFormatter()
|
|
record = logging.LogRecord(
|
|
name="test", level=logging.INFO, pathname="test.py",
|
|
lineno=1, msg="테스트 메시지", args=(), exc_info=None,
|
|
)
|
|
output = formatter.format(record)
|
|
parsed = json.loads(output)
|
|
assert parsed["message"] == "테스트 메시지"
|
|
assert parsed["level"] == "INFO"
|
|
assert "timestamp" in parsed
|
|
|
|
|
|
def test_json_formatter_with_extra():
|
|
formatter = JsonFormatter()
|
|
record = logging.LogRecord(
|
|
name="test", level=logging.INFO, pathname="test.py",
|
|
lineno=1, msg="작업 시작", args=(), exc_info=None,
|
|
)
|
|
record.thread_id = "uuid-123"
|
|
record.issue = 42
|
|
output = formatter.format(record)
|
|
parsed = json.loads(output)
|
|
assert parsed["thread_id"] == "uuid-123"
|
|
assert parsed["issue"] == 42
|
|
|
|
|
|
def test_json_formatter_with_exception():
|
|
formatter = JsonFormatter()
|
|
try:
|
|
raise ValueError("test error")
|
|
except ValueError:
|
|
import sys
|
|
record = logging.LogRecord(
|
|
name="test", level=logging.ERROR, pathname="test.py",
|
|
lineno=1, msg="에러 발생", args=(), exc_info=sys.exc_info(),
|
|
)
|
|
output = formatter.format(record)
|
|
parsed = json.loads(output)
|
|
assert "exception" in parsed
|
|
assert "ValueError" in parsed["exception"]
|
|
|
|
|
|
def test_setup_logging_json():
|
|
test_logger = logging.getLogger("test_json_setup")
|
|
test_logger.handlers.clear()
|
|
test_logger.setLevel(logging.DEBUG)
|
|
stream = io.StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
test_logger.addHandler(handler)
|
|
setup_logging(log_format="json", logger=test_logger)
|
|
test_logger.info("hello")
|
|
output = stream.getvalue().strip()
|
|
parsed = json.loads(output)
|
|
assert parsed["message"] == "hello"
|
|
|
|
|
|
def test_setup_logging_text():
|
|
test_logger = logging.getLogger("test_text_setup")
|
|
test_logger.handlers.clear()
|
|
test_logger.setLevel(logging.DEBUG)
|
|
stream = io.StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
test_logger.addHandler(handler)
|
|
setup_logging(log_format="text", logger=test_logger)
|
|
test_logger.info("hello")
|
|
output = stream.getvalue().strip()
|
|
assert "hello" in output
|
|
try:
|
|
json.loads(output)
|
|
assert False, "Should not be valid JSON"
|
|
except json.JSONDecodeError:
|
|
pass
|