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
|
import sys
import time
import pytest
from loguru import logger
class StopSinkError:
def write(self, message):
print(message, end="")
def stop(self):
raise OSError("Stop error")
def test_remove_all(tmp_path, writer, capsys):
file = tmp_path / "test.log"
logger.debug("This shouldn't be printed.")
logger.add(file, format="{message}")
logger.add(sys.stdout, format="{message}")
logger.add(sys.stderr, format="{message}")
logger.add(writer, format="{message}")
message = "some message"
expected = message + "\n"
logger.debug(message)
logger.remove()
logger.debug("This shouldn't be printed neither.")
out, err = capsys.readouterr()
assert file.read_text() == expected
assert out == expected
assert err == expected
assert writer.read() == expected
def test_remove_simple(writer):
i = logger.add(writer, format="{message}")
logger.debug("1")
logger.remove(i)
logger.debug("2")
assert writer.read() == "1\n"
def test_remove_enqueue(writer):
i = logger.add(writer, format="{message}", enqueue=True)
logger.debug("1")
time.sleep(0.1)
logger.remove(i)
logger.debug("2")
assert writer.read() == "1\n"
def test_remove_enqueue_filesink(tmp_path):
file = tmp_path / "test.log"
i = logger.add(file, format="{message}", enqueue=True)
logger.debug("1")
logger.remove(i)
assert file.read_text() == "1\n"
def test_exception_in_stop_during_remove_one(capsys):
i = logger.add(StopSinkError(), catch=False, format="{message}")
logger.info("A")
with pytest.raises(OSError, match=r"Stop error"):
logger.remove(i)
logger.info("Nope")
out, err = capsys.readouterr()
assert out == "A\n"
assert err == ""
def test_exception_in_stop_not_caught_during_remove_all(capsys):
logger.add(StopSinkError(), catch=False, format="{message}")
logger.add(StopSinkError(), catch=False, format="{message}")
with pytest.raises(OSError, match=r"Stop error"):
logger.remove()
logger.info("A")
with pytest.raises(OSError, match=r"Stop error"):
logger.remove()
logger.info("Nope")
out, err = capsys.readouterr()
assert out == "A\n"
assert err == ""
def test_invalid_handler_id_value(writer):
logger.add(writer)
with pytest.raises(ValueError, match=r"^There is no existing handler.*"):
logger.remove(42)
@pytest.mark.parametrize("handler_id", [sys.stderr, sys, object(), int])
def test_invalid_handler_id_type(handler_id):
with pytest.raises(TypeError, match=r"^Invalid handler id.*"):
logger.remove(handler_id)
|
