1import sys
2import pytest
3from loguru import logger
4import loguru
5import re
6import time
7
8
9def broken_sink(m):
10    raise Exception("Error!")
11
12
13def test_catch_is_true(capsys):
14    logger.add(broken_sink, catch=True)
15    logger.debug("Fail")
16    out, err = capsys.readouterr()
17    assert out == ""
18    assert err != ""
19
20
21def test_catch_is_false(capsys):
22    logger.add(broken_sink, catch=False)
23    with pytest.raises(Exception):
24        logger.debug("Fail")
25    out, err = capsys.readouterr()
26    assert out == err == ""
27
28
29def test_no_sys_stderr(capsys, monkeypatch):
30    monkeypatch.setattr(sys, "stderr", None)
31    logger.add(broken_sink, catch=True)
32    logger.debug("a")
33
34    out, err = capsys.readouterr()
35    assert out == err == ""
36
37
38def test_broken_sys_stderr(capsys, monkeypatch):
39    def broken_write(*args, **kwargs):
40        raise OSError
41
42    monkeypatch.setattr(sys.stderr, "write", broken_write)
43    logger.add(broken_sink, catch=True)
44    logger.debug("a")
45
46    out, err = capsys.readouterr()
47    assert out == err == ""
48
49
50def test_encoding_error(capsys):
51    def sink(m):
52        raise UnicodeEncodeError("utf8", "", 10, 11, "too bad")
53
54    logger.add(sink, catch=True)
55    logger.debug("test")
56
57    out, err = capsys.readouterr()
58    lines = err.strip().splitlines()
59
60    assert out == ""
61    assert lines[0] == "--- Logging error in Loguru Handler #0 ---"
62    assert lines[1].startswith("Record was: {")
63    assert lines[1].endswith("}")
64    assert lines[-2].startswith("UnicodeEncodeError:")
65    assert lines[-1] == "--- End of logging error ---"
66
67
68def test_unprintable_record(writer, capsys):
69    class Unprintable:
70        def __repr__(self):
71            raise ValueError("Failed")
72
73    logger.add(writer, format="{message} {extra[unprintable]}", catch=True)
74    logger.bind(unprintable=1).debug("a")
75    logger.bind(unprintable=Unprintable()).debug("b")
76    logger.bind(unprintable=2).debug("c")
77
78    out, err = capsys.readouterr()
79    lines = err.strip().splitlines()
80
81    assert writer.read() == "a 1\nc 2\n"
82    assert out == ""
83    assert lines[0] == "--- Logging error in Loguru Handler #0 ---"
84    assert lines[1] == "Record was: /!\\ Unprintable record /!\\"
85    assert lines[-2] == "ValueError: Failed"
86    assert lines[-1] == "--- End of logging error ---"
87
88
89@pytest.mark.parametrize("enqueue", [False, True])
90def test_broken_sink_message(capsys, enqueue):
91    logger.add(broken_sink, catch=True, enqueue=enqueue)
92    logger.debug("Oops")
93    time.sleep(0.1)
94
95    out, err = capsys.readouterr()
96    lines = err.strip().splitlines()
97
98    assert out == ""
99    assert lines[0] == "--- Logging error in Loguru Handler #0 ---"
100    assert re.match(r"Record was: \{.*Oops.*\}", lines[1])
101    assert lines[-2].startswith("Exception: Error!")
102    assert lines[-1] == "--- End of logging error ---"
103
104
105@pytest.mark.parametrize("enqueue", [False, True])
106def test_broken_sink_caught_keep_working(enqueue):
107    output = ""
108
109    def half_broken_sink(m):
110        nonlocal output
111        if m.startswith("NOK"):
112            raise ValueError("Broken!")
113        else:
114            output += m
115
116    logger.add(half_broken_sink, format="{message}", enqueue=enqueue, catch=True)
117    logger.info("A")
118    logger.info("NOK")
119    logger.info("B")
120
121    time.sleep(0.1)
122    assert output == "A\nB\n"
123
124
125def test_broken_sink_not_caught_enqueue():
126    called = 0
127
128    def broken_sink(m):
129        nonlocal called
130        called += 1
131        raise ValueError("Nop")
132
133    logger.add(broken_sink, format="{message}", enqueue=True, catch=False)
134
135    logger.info("A")
136    logger.info("B")
137    time.sleep(0.1)
138    assert called == 1
139