1# SPDX-License-Identifier: MIT OR Apache-2.0
2# This file is dual licensed under the terms of the Apache License, Version
3# 2.0, and the MIT License. See the LICENSE file in the root of this
4# repository for complete details.
5
6"""
7Helpers to test your application's logging behavior.
8
9.. versionadded:: 20.1.0
10
11See :doc:`testing`.
12"""
13
14from __future__ import annotations
15
16from contextlib import contextmanager
17from typing import Any, Generator, NamedTuple, NoReturn
18
19from ._config import configure, get_config
20from ._log_levels import map_method_name
21from .exceptions import DropEvent
22from .typing import EventDict, WrappedLogger
23
24
25__all__ = [
26 "CapturedCall",
27 "CapturingLogger",
28 "CapturingLoggerFactory",
29 "LogCapture",
30 "ReturnLogger",
31 "ReturnLoggerFactory",
32 "capture_logs",
33]
34
35
36class LogCapture:
37 """
38 Class for capturing log messages in its entries list.
39 Generally you should use `structlog.testing.capture_logs`,
40 but you can use this class if you want to capture logs with other patterns.
41
42 :ivar List[structlog.typing.EventDict] entries: The captured log entries.
43
44 .. versionadded:: 20.1.0
45
46 .. versionchanged:: 24.3.0
47 Added mapping from "exception" to "error"
48 Added mapping from "warn" to "warning"
49 """
50
51 entries: list[EventDict]
52
53 def __init__(self) -> None:
54 self.entries = []
55
56 def __call__(
57 self, _: WrappedLogger, method_name: str, event_dict: EventDict
58 ) -> NoReturn:
59 event_dict["log_level"] = map_method_name(method_name)
60 self.entries.append(event_dict)
61
62 raise DropEvent
63
64
65@contextmanager
66def capture_logs() -> Generator[list[EventDict], None, None]:
67 """
68 Context manager that appends all logging statements to its yielded list
69 while it is active. Disables all configured processors for the duration
70 of the context manager.
71
72 Attention: this is **not** thread-safe!
73
74 .. versionadded:: 20.1.0
75 """
76 cap = LogCapture()
77 # Modify `_Configuration.default_processors` set via `configure` but always
78 # keep the list instance intact to not break references held by bound
79 # loggers.
80 processors = get_config()["processors"]
81 old_processors = processors.copy()
82 try:
83 # clear processors list and use LogCapture for testing
84 processors.clear()
85 processors.append(cap)
86 configure(processors=processors)
87 yield cap.entries
88 finally:
89 # remove LogCapture and restore original processors
90 processors.clear()
91 processors.extend(old_processors)
92 configure(processors=processors)
93
94
95class ReturnLogger:
96 """
97 Return the arguments that it's called with.
98
99 >>> from structlog import ReturnLogger
100 >>> ReturnLogger().info("hello")
101 'hello'
102 >>> ReturnLogger().info("hello", when="again")
103 (('hello',), {'when': 'again'})
104
105 .. versionchanged:: 0.3.0
106 Allow for arbitrary arguments and keyword arguments to be passed in.
107 """
108
109 def msg(self, *args: Any, **kw: Any) -> Any:
110 """
111 Return tuple of ``args, kw`` or just ``args[0]`` if only one arg passed
112 """
113 # Slightly convoluted for backwards compatibility.
114 if len(args) == 1 and not kw:
115 return args[0]
116
117 return args, kw
118
119 log = debug = info = warn = warning = msg
120 fatal = failure = err = error = critical = exception = msg
121
122
123class ReturnLoggerFactory:
124 r"""
125 Produce and cache `ReturnLogger`\ s.
126
127 To be used with `structlog.configure`\ 's *logger_factory*.
128
129 Positional arguments are silently ignored.
130
131 .. versionadded:: 0.4.0
132 """
133
134 def __init__(self) -> None:
135 self._logger = ReturnLogger()
136
137 def __call__(self, *args: Any) -> ReturnLogger:
138 return self._logger
139
140
141class CapturedCall(NamedTuple):
142 """
143 A call as captured by `CapturingLogger`.
144
145 Can also be unpacked like a tuple.
146
147 Args:
148 method_name: The method name that got called.
149
150 args: A tuple of the positional arguments.
151
152 kwargs: A dict of the keyword arguments.
153
154 .. versionadded:: 20.2.0
155 """
156
157 method_name: str
158 args: tuple[Any, ...]
159 kwargs: dict[str, Any]
160
161
162class CapturingLogger:
163 """
164 Store the method calls that it's been called with.
165
166 This is nicer than `ReturnLogger` for unit tests because the bound logger
167 doesn't have to cooperate.
168
169 **Any** method name is supported.
170
171 .. versionadded:: 20.2.0
172 """
173
174 calls: list[CapturedCall]
175
176 def __init__(self) -> None:
177 self.calls = []
178
179 def __repr__(self) -> str:
180 return f"<CapturingLogger with {len(self.calls)} call(s)>"
181
182 def __getattr__(self, name: str) -> Any:
183 """
184 Capture call to `calls`
185 """
186
187 def log(*args: Any, **kw: Any) -> None:
188 self.calls.append(CapturedCall(name, args, kw))
189
190 return log
191
192
193class CapturingLoggerFactory:
194 r"""
195 Produce and cache `CapturingLogger`\ s.
196
197 Each factory produces and reuses only **one** logger.
198
199 You can access it via the ``logger`` attribute.
200
201 To be used with `structlog.configure`\ 's *logger_factory*.
202
203 Positional arguments are silently ignored.
204
205 .. versionadded:: 20.2.0
206 """
207
208 logger: CapturingLogger
209
210 def __init__(self) -> None:
211 self.logger = CapturingLogger()
212
213 def __call__(self, *args: Any) -> CapturingLogger:
214 return self.logger