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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
import inspect
import logging
from loguru import logger
from .conftest import make_logging_logger
class InterceptHandler(logging.Handler):
def emit(self, record):
# Get corresponding Loguru level if it exists.
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message.
frame, depth = inspect.currentframe(), 0
while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
def test_formatting(writer):
fmt = (
"{name} - {file.name} - {function} - {level.name} - "
"{level.no} - {line} - {module} - {message}"
)
expected = (
"tests.test_interception - test_interception.py - test_formatting - DEBUG - "
"10 - 39 - test_interception - This is the message\n"
)
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format=fmt)
logging_logger.debug("This is the %s", "message")
result = writer.read()
assert result == expected
def test_intercept(writer):
with make_logging_logger(None, InterceptHandler()) as logging_logger:
logging_logger.info("Nope")
logger.add(writer, format="{message}")
logging_logger.info("Test")
result = writer.read()
assert result == "Test\n"
def test_add_before_intercept(writer):
logger.add(writer, format="{message}")
with make_logging_logger(None, InterceptHandler()) as logging_logger:
logging_logger.info("Test")
result = writer.read()
assert result == "Test\n"
def test_remove_interception(writer):
h = InterceptHandler()
with make_logging_logger("foobar", h) as logging_logger:
logger.add(writer, format="{message}")
logging_logger.debug("1")
logging_logger.removeHandler(h)
logging_logger.debug("2")
result = writer.read()
assert result == "1\n"
def test_intercept_too_low(writer):
with make_logging_logger("tests.test_interception", InterceptHandler()):
logger.add(writer, format="{message}")
logging.getLogger("tests").error("Nope 1")
logging.getLogger("foobar").error("Nope 2")
result = writer.read()
assert result == ""
def test_multiple_intercept(writer):
with make_logging_logger("test_1", InterceptHandler()) as logging_logger_1:
with make_logging_logger("test_2", InterceptHandler()) as logging_logger_2:
logger.add(writer, format="{message}")
logging_logger_1.info("1")
logging_logger_2.info("2")
result = writer.read()
assert result == "1\n2\n"
def test_exception(writer):
with make_logging_logger("tests.test_interception", InterceptHandler()) as logging_logger:
logger.add(writer, format="{message}")
try:
1 / 0 # noqa: B018
except Exception:
logging_logger.exception("Oops...")
lines = writer.read().strip().splitlines()
assert lines[0] == "Oops..."
assert lines[-1] == "ZeroDivisionError: division by zero"
assert sum(line.startswith("> ") for line in lines) == 1
def test_level_is_no(writer):
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.log(12, "Hop")
result = writer.read()
assert result == "12 - Level 12 - Hop\x1b[0m\n"
def test_level_does_not_exist(writer):
logging.addLevelName(152, "FANCY_LEVEL")
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.log(152, "Nop")
result = writer.read()
assert result == "152 - Level 152 - Nop\x1b[0m\n"
def test_level_exist_builtin(writer):
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.error("Error...")
result = writer.read()
assert result == "\x1b[31m\x1b[1m40 - ERROR - Error...\x1b[0m\n"
def test_level_exists_custom(writer):
logging.addLevelName(99, "ANOTHER_FANCY_LEVEL")
logger.level("ANOTHER_FANCY_LEVEL", no=99, color="<green>", icon="")
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.log(99, "Yep!")
result = writer.read()
assert result == "\x1b[32m99 - ANOTHER_FANCY_LEVEL - Yep!\x1b[0m\n"
def test_using_logging_function(writer):
with make_logging_logger(None, InterceptHandler()):
logger.add(writer, format="{function} {line} {module} {file.name} {message}")
logging.warning("ABC")
result = writer.read()
assert result == "test_using_logging_function 158 test_interception test_interception.py ABC\n"
|
