blob: 2ffddcd2c2758109e4bbd7d31c56ff3d16b9464b (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
"""Tests for shared structured logging module."""
import io
import json
import logging
import pytest
import structlog
@pytest.fixture(autouse=True)
def _reset_structlog():
"""Reset structlog configuration between tests."""
structlog.reset_defaults()
# Remove all handlers from root logger to avoid cross-test leakage
root = logging.getLogger()
root.handlers.clear()
yield
structlog.reset_defaults()
root.handlers.clear()
def test_setup_logging_returns_logger():
"""setup_logging should return a bound structlog logger."""
from shared.logging import setup_logging
logger = setup_logging(service_name="test-svc")
assert logger is not None
# Should be usable as a logger (has info, debug, etc.)
assert callable(getattr(logger, "info", None))
assert callable(getattr(logger, "error", None))
def test_service_name_bound_in_context():
"""The returned logger should have service=service_name bound."""
from shared.logging import setup_logging
logger = setup_logging(service_name="my-service", log_format="json")
# Capture output via a stream handler on the root logger
stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.setLevel(logging.DEBUG)
# Use the same formatter structlog sets up
root = logging.getLogger()
handler.setFormatter(root.handlers[0].formatter)
root.addHandler(handler)
logger.info("hello")
output = stream.getvalue()
parsed = json.loads(output)
assert parsed["service"] == "my-service"
def test_log_level_set_correctly():
"""setup_logging should set the stdlib root logger level."""
from shared.logging import setup_logging
setup_logging(service_name="test", log_level="WARNING")
root = logging.getLogger()
assert root.level == logging.WARNING
def test_json_format_produces_json():
"""log_format='json' should produce JSON output."""
from shared.logging import setup_logging
logger = setup_logging(service_name="json-svc", log_format="json")
stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.setLevel(logging.DEBUG)
root = logging.getLogger()
handler.setFormatter(root.handlers[0].formatter)
root.addHandler(handler)
logger.info("test-json")
output = stream.getvalue().strip()
parsed = json.loads(output)
assert "event" in parsed
assert parsed["event"] == "test-json"
assert "timestamp" in parsed
assert "level" in parsed
def test_console_format_produces_output():
"""log_format='console' should produce human-readable output (not JSON)."""
from shared.logging import setup_logging
logger = setup_logging(service_name="console-svc", log_format="console")
stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.setLevel(logging.DEBUG)
root = logging.getLogger()
handler.setFormatter(root.handlers[0].formatter)
root.addHandler(handler)
logger.info("test-console")
output = stream.getvalue().strip()
assert len(output) > 0
# Console output should NOT be valid JSON
with pytest.raises(json.JSONDecodeError):
json.loads(output)
|
