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