Skip to content

Commit

Permalink
Fix tests fails (since pytest upgrade) due to standard logging misuse
Browse files Browse the repository at this point in the history
  • Loading branch information
Delgan committed Aug 4, 2020
1 parent 9a40e58 commit 084dffd
Show file tree
Hide file tree
Showing 4 changed files with 143 additions and 133 deletions.
27 changes: 13 additions & 14 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import asyncio
import contextlib
import loguru
import logging
import itertools
Expand Down Expand Up @@ -62,8 +63,6 @@ def reset():
loguru._logger.Core(), None, 0, False, False, False, False, True, None, {}
)
loguru._logger.context.set({})
logging.Logger.manager.loggerDict.clear()
logging.root = logging.RootLogger("WARNING")

reset()
yield
Expand Down Expand Up @@ -134,20 +133,20 @@ def patched_localtime(t):
return monkeypatch_date


@pytest.fixture
def make_logging_logger():
def make_logging_logger(name, handler, fmt="%(message)s", level="DEBUG"):
logging_logger = logging.getLogger(name)
logging_logger.setLevel(level)
formatter = logging.Formatter(fmt)

handler.setLevel(level)
handler.setFormatter(formatter)
logging_logger.addHandler(handler)
@contextlib.contextmanager
def make_logging_logger(name, handler, fmt="%(message)s", level="DEBUG"):
logging_logger = logging.getLogger(name)
logging_logger.setLevel(level)
formatter = logging.Formatter(fmt)

return logging_logger
handler.setLevel(level)
handler.setFormatter(formatter)
logging_logger.addHandler(handler)

yield make_logging_logger
try:
yield logging_logger
finally:
logging_logger.removeHandler(handler)


@pytest.fixture
Expand Down
42 changes: 29 additions & 13 deletions tests/test_coroutine_sink.py
Original file line number Diff line number Diff line change
Expand Up @@ -369,9 +369,9 @@ async def main():
assert lines[-1] == "--- End of logging error ---"


def test_exception_in_coroutine_not_caught(capsys):
def test_exception_in_coroutine_not_caught(capsys, caplog):
async def sink(msg):
raise Exception("Oh no")
raise ValueError("Oh no")

async def main():
logger.add(sink, catch=False)
Expand All @@ -382,10 +382,19 @@ async def main():
asyncio.run(main())

out, err = capsys.readouterr()
assert out == ""
assert "Logging error in Loguru Handler" not in err
assert "was never retrieved" not in err
assert err.strip().endswith("Exception: Oh no")
assert out == err == ""

records = caplog.records
assert len(records) == 1
record = records[0]

message = record.getMessage()
assert "Logging error in Loguru Handler" not in message
assert "was never retrieved" not in message

exc_type, exc_value, _ = record.exc_info
assert exc_type == ValueError
assert str(exc_value) == "Oh no"


def test_exception_in_coroutine_during_complete_caught(capsys):
Expand All @@ -410,10 +419,10 @@ async def main():
assert lines[-1] == "--- End of logging error ---"


def test_exception_in_coroutine_during_complete_not_caught(capsys):
def test_exception_in_coroutine_during_complete_not_caught(capsys, caplog):
async def sink(msg):
await asyncio.sleep(0.1)
raise Exception("Oh no")
raise ValueError("Oh no")

async def main():
logger.add(sink, catch=False)
Expand All @@ -423,12 +432,19 @@ async def main():
asyncio.run(main())

out, err = capsys.readouterr()
lines = err.strip().splitlines()
assert out == err == ""

assert out == ""
assert "Logging error in Loguru Handler" not in err
assert "was never retrieved" not in err
assert err.strip().endswith("Exception: Oh no")
records = caplog.records
assert len(records) == 1
record = records[0]

message = record.getMessage()
assert "Logging error in Loguru Handler" not in message
assert "was never retrieved" not in message

exc_type, exc_value, _ = record.exc_info
assert exc_type == ValueError
assert str(exc_value) == "Oh no"


@pytest.mark.skipif(sys.version_info < (3, 5, 3), reason="Coroutine can't access running loop")
Expand Down
127 changes: 61 additions & 66 deletions tests/test_interception.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import logging

from .conftest import make_logging_logger

from loguru import logger


Expand All @@ -20,139 +22,132 @@ def emit(self, record):
logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


def test_formatting(writer, make_logging_logger):
def test_formatting(writer):
fmt = "{name} - {file.name} - {function} - {level.name} - {level.no} - {line} - {module} - {message}"
expected = "tests.test_interception - test_interception.py - test_formatting - DEBUG - 10 - 31 - test_interception - This is the message\n"

logging_logger = make_logging_logger("tests", InterceptHandler())

logger.add(writer, format=fmt)

logging_logger.debug("This is the %s", "message")
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format=fmt)
logging_logger.debug("This is the %s", "message")

result = writer.read()
assert result == expected


def test_intercept(writer, make_logging_logger):
logging_logger = make_logging_logger(None, InterceptHandler())

logging_logger.info("Nope")

logger.add(writer, format="{message}")

logging_logger.info("Test")
def test_intercept(writer):
with make_logging_logger(None, InterceptHandler()) as logging_logger:
logging_logger.info("Nope")
logger.add(writer, format="{message}")
logging_logger.info("Test")

result = writer.read()
assert result == "Test\n"


def test_add_before_intercept(writer, make_logging_logger):
def test_add_before_intercept(writer):
logger.add(writer, format="{message}")
logging_logger = make_logging_logger(None, InterceptHandler())

logging_logger.info("Test")
with make_logging_logger(None, InterceptHandler()) as logging_logger:
logging_logger.info("Test")

result = writer.read()
assert result == "Test\n"


def test_remove_interception(writer, make_logging_logger):
def test_remove_interception(writer):
h = InterceptHandler()
logging_logger = make_logging_logger("foobar", h)
logger.add(writer, format="{message}")
logging_logger.debug("1")
logging_logger.removeHandler(h)
logging_logger.debug("2")

with make_logging_logger("foobar", h) as logging_logger:
logger.add(writer, format="{message}")
logging_logger.debug("1")
logging_logger.removeHandler(h)
logging_logger.debug("2")

result = writer.read()
assert result == "1\n"


def test_intercept_too_low(writer, make_logging_logger):
make_logging_logger("tests.test_interception", InterceptHandler())
logger.add(writer, format="{message}")
logging.getLogger("tests").error("Nope 1")
logging.getLogger("foobar").error("Nope 2")
def test_intercept_too_low(writer):
with make_logging_logger("tests.test_interception", InterceptHandler()):
logger.add(writer, format="{message}")
logging.getLogger("tests").error("Nope 1")
logging.getLogger("foobar").error("Nope 2")

result = writer.read()
assert result == ""


def test_multiple_intercept(writer, make_logging_logger):
logging_logger_1 = make_logging_logger("test_1", InterceptHandler())
logging_logger_2 = make_logging_logger("test_2", InterceptHandler())

logger.add(writer, format="{message}")

logging_logger_1.info("1")
logging_logger_2.info("2")
def test_multiple_intercept(writer):
with make_logging_logger("test_1", InterceptHandler()) as logging_logger_1:
with make_logging_logger("test_2", InterceptHandler()) as logging_logger_2:
logger.add(writer, format="{message}")
logging_logger_1.info("1")
logging_logger_2.info("2")

result = writer.read()
assert result == "1\n2\n"


def test_exception(writer, make_logging_logger):
logging_logger = make_logging_logger("tests.test_interception", InterceptHandler())
logger.add(writer, format="{message}")
def test_exception(writer):
with make_logging_logger("tests.test_interception", InterceptHandler()) as logging_logger:
logger.add(writer, format="{message}")

try:
1 / 0
except:
logging_logger.exception("Oops...")
try:
1 / 0
except:
logging_logger.exception("Oops...")

lines = writer.read().strip().splitlines()
assert lines[0] == "Oops..."
assert lines[-1] == "ZeroDivisionError: division by zero"
assert sum(line.startswith("> ") for line in lines) == 1


def test_level_is_no(writer, make_logging_logger):
logging_logger = make_logging_logger("tests", InterceptHandler())
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)

logging_logger.log(12, "Hop")
def test_level_is_no(writer):
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.log(12, "Hop")

result = writer.read()
assert result == "12 - Level 12 - Hop\x1b[0m\n"


def test_level_does_not_exist(writer, make_logging_logger):
def test_level_does_not_exist(writer):
logging.addLevelName(152, "FANCY_LEVEL")
logging_logger = make_logging_logger("tests", InterceptHandler())
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)

logging_logger.log(152, "Nop")
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.log(152, "Nop")

result = writer.read()
assert result == "152 - Level 152 - Nop\x1b[0m\n"


def test_level_exist_builtin(writer, make_logging_logger):
logging_logger = make_logging_logger("tests", InterceptHandler())
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)

logging_logger.error("Error...")
def test_level_exist_builtin(writer):
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.error("Error...")

result = writer.read()
assert result == "\x1b[31m\x1b[1m40 - ERROR - Error...\x1b[0m\n"


def test_level_exists_custom(writer, make_logging_logger):
def test_level_exists_custom(writer):
logging.addLevelName(99, "ANOTHER_FANCY_LEVEL")
logger.level("ANOTHER_FANCY_LEVEL", no=99, color="<green>", icon="")
logging_logger = make_logging_logger("tests", InterceptHandler())

logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)

logging_logger.log(99, "Yep!")
with make_logging_logger("tests", InterceptHandler()) as logging_logger:
logger.add(writer, format="<lvl>{level.no} - {level.name} - {message}</lvl>", colorize=True)
logging_logger.log(99, "Yep!")

result = writer.read()
assert result == "\x1b[32m99 - ANOTHER_FANCY_LEVEL - Yep!\x1b[0m\n"


def test_using_logging_function(writer):
logging.getLogger(None).addHandler(InterceptHandler())
logger.add(writer, format="{function} {line} {module} {file.name} {message}")
logging.warning("ABC")
with make_logging_logger(None, InterceptHandler()):
logger.add(writer, format="{function} {line} {module} {file.name} {message}")
logging.warning("ABC")

result = writer.read()
assert result == "test_using_logging_function 156 test_interception test_interception.py ABC\n"
assert result == "test_using_logging_function 150 test_interception test_interception.py ABC\n"
Loading

0 comments on commit 084dffd

Please sign in to comment.