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
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import io
import json
import logging
import sys
import types
import typing
import unittest
import warnings
import daiquiri
real_excepthook = sys.excepthook
class TestDaiquiri(unittest.TestCase):
def tearDown(self) -> None:
# Be sure to reset the warning capture
logging.captureWarnings(False)
# Reset exception hook
sys.excepthook = real_excepthook
super(TestDaiquiri, self).tearDown()
def test_setup(self) -> None:
daiquiri.setup()
daiquiri.setup(level=logging.DEBUG)
daiquiri.setup(program_name="foobar")
def test_excepthook(self) -> None:
hook_ran = False
def catcher(
exctype: type[BaseException],
value: BaseException,
traceback: types.TracebackType | None,
) -> None:
nonlocal hook_ran
hook_ran = True
real_excepthook(exctype, value, traceback)
sys.excepthook = catcher
sys.excepthook(Exception, Exception("boom"), None)
assert hook_ran
daiquiri.setup()
hook_ran = False
sys.excepthook(Exception, Exception("boom"), None)
assert hook_ran
def test_setup_json_formatter(self) -> None:
stream = io.StringIO()
daiquiri.setup(
outputs=(
daiquiri.output.Stream(
stream, formatter=daiquiri.formatter.JSON_FORMATTER
),
)
)
daiquiri.getLogger(__name__).warning("foobar")
expected: dict[str, typing.Any] = {"message": "foobar"}
self.assertEqual(expected, json.loads(stream.getvalue()))
def test_setup_json_formatter_with_extras(self) -> None:
stream = io.StringIO()
daiquiri.setup(
outputs=(
daiquiri.output.Stream(
stream, formatter=daiquiri.formatter.JSON_FORMATTER
),
)
)
daiquiri.getLogger(__name__).warning("foobar", foo="bar")
expected: dict[str, typing.Any] = {"message": "foobar", "foo": "bar"}
self.assertEqual(expected, json.loads(stream.getvalue()))
def test_get_logger_set_level(self) -> None:
logger = daiquiri.getLogger(__name__)
logger.setLevel(logging.DEBUG)
def test_capture_warnings(self) -> None:
stream = io.StringIO()
daiquiri.setup(outputs=(daiquiri.output.Stream(stream),))
warnings.warn("omg!")
line = stream.getvalue()
self.assertIn("WARNING py.warnings: ", line)
self.assertIn(
"daiquiri/tests/test_daiquiri.py:95: "
'UserWarning: omg!\n warnings.warn("omg!")\n',
line,
)
def test_no_capture_warnings(self) -> None:
stream = io.StringIO()
daiquiri.setup(
outputs=(daiquiri.output.Stream(stream),), capture_warnings=False
)
warnings.warn("omg!")
self.assertEqual("", stream.getvalue())
def test_set_default_log_levels(self) -> None:
daiquiri.set_default_log_levels((("amqp", "debug"), ("urllib3", "warn")))
def test_parse_and_set_default_log_levels(self) -> None:
daiquiri.parse_and_set_default_log_levels(("urllib3=warn", "foobar=debug"))
def test_string_as_setup_outputs_arg(self) -> None:
daiquiri.setup(outputs=("stderr", "stdout"))
if daiquiri.handlers.syslog is not None: # type: ignore[attr-defined]
daiquiri.setup(outputs=("syslog",))
if daiquiri.handlers.journal is not None: # type: ignore[attr-defined]
daiquiri.setup(outputs=("journal",))
def test_special_kwargs(self) -> None:
daiquiri.getLogger(__name__).info(
"foobar", foo="bar", exc_info=True, stack_info=True
)
def test_extra_with_two_loggers() -> None:
stream = io.StringIO()
daiquiri.setup(outputs=(daiquiri.output.Stream(stream),))
log1 = daiquiri.getLogger("foobar")
log1.error("argh")
log2 = daiquiri.getLogger("foobar", key="value")
log2.warning("boo")
lines = stream.getvalue().strip().split("\n")
assert lines[0].endswith("ERROR foobar: argh")
assert lines[1].endswith("WARNING foobar [key: value]: boo")
|
