mirror of
https://github.com/beetbox/beets.git
synced 2026-07-17 14:47:04 -04:00
230bf19c7f
- Removing the coverage omit configuration exposed unused branches, helpers, and fixtures across the test suite. - Delete that redundant logic and simplify affected tests while preserving their behavior.
367 lines
12 KiB
Python
367 lines
12 KiB
Python
"""Stupid tests that ensure logging works as expected"""
|
||
|
||
import logging as log
|
||
import sys
|
||
import threading
|
||
from types import ModuleType
|
||
|
||
import pytest
|
||
|
||
import beets.logging as blog
|
||
from beets import plugins, ui
|
||
from beets.test.helper import AsIsImporterMixin, ImportHelper, PluginMixin
|
||
|
||
|
||
class TestStrFormatLogger:
|
||
"""Tests for the custom str-formatting logger."""
|
||
|
||
def test_logger_creation(self):
|
||
l1 = log.getLogger("foo123")
|
||
l2 = blog.getLogger("foo123")
|
||
assert l1 == l2
|
||
assert l1.__class__ == log.Logger
|
||
|
||
l3 = blog.getLogger("bar123")
|
||
l4 = log.getLogger("bar123")
|
||
assert l3 == l4
|
||
assert l3.__class__ == blog.BeetsLogger
|
||
assert isinstance(
|
||
l3, (blog.StrFormatLogger, blog.ThreadLocalLevelLogger)
|
||
)
|
||
|
||
l5 = l3.getChild("shalala")
|
||
assert l5.__class__ == blog.BeetsLogger
|
||
|
||
l6 = blog.getLogger()
|
||
assert l1 != l6
|
||
|
||
@pytest.mark.parametrize(
|
||
"level", [log.DEBUG, log.INFO, log.WARNING, log.ERROR]
|
||
)
|
||
@pytest.mark.parametrize(
|
||
"msg, args, kwargs, expected",
|
||
[
|
||
("foo {} bar {}", ("oof", "baz"), {}, "foo oof bar baz"),
|
||
(
|
||
"foo {bar} baz {foo}",
|
||
(),
|
||
{"foo": "oof", "bar": "baz"},
|
||
"foo baz baz oof",
|
||
),
|
||
("no args", (), {}, "no args"),
|
||
("foo {} bar {baz}", ("oof",), {"baz": "baz"}, "foo oof bar baz"),
|
||
],
|
||
)
|
||
def test_str_format_logging(
|
||
self, level, msg, args, kwargs, expected, caplog
|
||
):
|
||
logger = blog.getLogger("test_logger")
|
||
logger.setLevel(level)
|
||
|
||
with caplog.at_level(level, logger="test_logger"):
|
||
logger.log(level, msg, *args, **kwargs)
|
||
|
||
assert caplog.records, "No log records were captured"
|
||
assert str(caplog.records[0].msg) == expected
|
||
|
||
|
||
class TestLogSanitization:
|
||
"""Log messages should have control characters removed from:
|
||
- String arguments
|
||
- Keyword argument values
|
||
- Bytes arguments (which get decoded first)
|
||
"""
|
||
|
||
@pytest.mark.parametrize(
|
||
"msg, args, kwargs, expected",
|
||
[
|
||
# Valid UTF-8 bytes are decoded and preserved
|
||
(
|
||
"foo {} bar {bar}",
|
||
(b"oof \xc3\xa9",),
|
||
{"bar": b"baz \xc3\xa9"},
|
||
"foo oof é bar baz é",
|
||
),
|
||
# Invalid UTF-8 bytes are decoded with replacement characters
|
||
(
|
||
"foo {} bar {bar}",
|
||
(b"oof \xff",),
|
||
{"bar": b"baz \xff"},
|
||
"foo oof <20> bar baz <20>",
|
||
),
|
||
# Control characters should be removed
|
||
(
|
||
"foo {} bar {bar}",
|
||
("oof \x9e",),
|
||
{"bar": "baz \x9e"},
|
||
"foo oof <20> bar baz <20>",
|
||
),
|
||
# Whitespace control characters should be preserved
|
||
(
|
||
"foo {} bar {bar}",
|
||
("foo\t\n",),
|
||
{"bar": "bar\r"},
|
||
"foo foo\t\n bar bar\r",
|
||
),
|
||
],
|
||
)
|
||
def test_sanitization(self, msg, args, kwargs, expected, caplog):
|
||
level = log.INFO
|
||
logger = blog.getLogger("test_logger")
|
||
logger.setLevel(level)
|
||
|
||
with caplog.at_level(level, logger="test_logger"):
|
||
logger.log(level, msg, *args, **kwargs)
|
||
|
||
assert caplog.records, "No log records were captured"
|
||
assert str(caplog.records[0].msg) == expected
|
||
|
||
|
||
class DummyModule(ModuleType):
|
||
class DummyPlugin(plugins.BeetsPlugin):
|
||
def __init__(self):
|
||
plugins.BeetsPlugin.__init__(self, "dummy")
|
||
self.import_stages = [self.import_stage]
|
||
self.register_listener("dummy_event", self.listener)
|
||
|
||
def log_all(self, name):
|
||
self._log.debug("debug {}", name)
|
||
self._log.info("info {}", name)
|
||
self._log.warning("warning {}", name)
|
||
|
||
def commands(self):
|
||
cmd = ui.Subcommand("dummy")
|
||
cmd.func = lambda _, __, ___: self.log_all("cmd")
|
||
return (cmd,)
|
||
|
||
def import_stage(self, session, task):
|
||
self.log_all("import_stage")
|
||
|
||
def listener(self):
|
||
self.log_all("listener")
|
||
|
||
def __init__(self, *_, **__):
|
||
module_name = "beetsplug.dummy"
|
||
super().__init__(module_name)
|
||
self.DummyPlugin.__module__ = module_name
|
||
self.DummyPlugin = self.DummyPlugin
|
||
|
||
|
||
class TestLoggingLevel(AsIsImporterMixin, PluginMixin, ImportHelper):
|
||
plugin = "dummy"
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _patch_dummy_module(self, monkeypatch):
|
||
monkeypatch.setitem(sys.modules, "beetsplug.dummy", DummyModule())
|
||
|
||
def test_command_level0(self, caplog):
|
||
self.config["verbose"] = 0
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_command("dummy")
|
||
assert "warning cmd" in caplog.messages
|
||
assert "info cmd" in caplog.messages
|
||
assert "debug cmd" not in caplog.messages
|
||
|
||
def test_command_level1(self, caplog):
|
||
self.config["verbose"] = 1
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_command("dummy")
|
||
assert "warning cmd" in caplog.messages
|
||
assert "info cmd" in caplog.messages
|
||
assert "debug cmd" in caplog.messages
|
||
|
||
def test_command_level2(self, caplog):
|
||
self.config["verbose"] = 2
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_command("dummy")
|
||
assert "warning cmd" in caplog.messages
|
||
assert "info cmd" in caplog.messages
|
||
assert "debug cmd" in caplog.messages
|
||
|
||
def test_listener_level0(self, caplog):
|
||
self.config["verbose"] = 0
|
||
with caplog.at_level("DEBUG"):
|
||
plugins.send("dummy_event")
|
||
assert "warning listener" in caplog.messages
|
||
assert "info listener" not in caplog.messages
|
||
assert "debug listener" not in caplog.messages
|
||
|
||
def test_listener_level1(self, caplog):
|
||
self.config["verbose"] = 1
|
||
with caplog.at_level("DEBUG"):
|
||
plugins.send("dummy_event")
|
||
assert "warning listener" in caplog.messages
|
||
assert "info listener" in caplog.messages
|
||
assert "debug listener" not in caplog.messages
|
||
|
||
def test_listener_level2(self, caplog):
|
||
self.config["verbose"] = 2
|
||
with caplog.at_level("DEBUG"):
|
||
plugins.send("dummy_event")
|
||
assert "warning listener" in caplog.messages
|
||
assert "info listener" in caplog.messages
|
||
assert "debug listener" in caplog.messages
|
||
|
||
def test_import_stage_level0(self, caplog):
|
||
self.config["verbose"] = 0
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_asis_importer()
|
||
assert "warning import_stage" in caplog.messages
|
||
assert "info import_stage" not in caplog.messages
|
||
assert "debug import_stage" not in caplog.messages
|
||
|
||
def test_import_stage_level1(self, caplog):
|
||
self.config["verbose"] = 1
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_asis_importer()
|
||
assert "warning import_stage" in caplog.messages
|
||
assert "info import_stage" in caplog.messages
|
||
assert "debug import_stage" not in caplog.messages
|
||
|
||
def test_import_stage_level2(self, caplog):
|
||
self.config["verbose"] = 2
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_asis_importer()
|
||
assert "warning import_stage" in caplog.messages
|
||
assert "info import_stage" in caplog.messages
|
||
assert "debug import_stage" in caplog.messages
|
||
|
||
|
||
class TestConcurrentEvents(AsIsImporterMixin, ImportHelper):
|
||
"""Similar to LoggingLevelTest but lower-level and focused on multiple
|
||
events interaction. Since this is a bit heavy we don't do it in
|
||
LoggingLevelTest.
|
||
"""
|
||
|
||
db_on_disk = True
|
||
|
||
class DummyPlugin(plugins.BeetsPlugin):
|
||
def __init__(self, test_case):
|
||
plugins.BeetsPlugin.__init__(self, "dummy")
|
||
self.register_listener("dummy_event1", self.listener1)
|
||
self.register_listener("dummy_event2", self.listener2)
|
||
self.lock1 = threading.Lock()
|
||
self.lock2 = threading.Lock()
|
||
self.test_case = test_case
|
||
self.exc = None
|
||
self.t1_step = self.t2_step = 0
|
||
|
||
def listener1(self):
|
||
try:
|
||
assert self._log.level == log.INFO
|
||
self.t1_step = 1
|
||
self.lock1.acquire()
|
||
assert self._log.level == log.INFO
|
||
self.t1_step = 2
|
||
except Exception as e:
|
||
self.exc = e
|
||
|
||
def listener2(self):
|
||
try:
|
||
assert self._log.level == log.DEBUG
|
||
self.t2_step = 1
|
||
self.lock2.acquire()
|
||
assert self._log.level == log.DEBUG
|
||
self.t2_step = 2
|
||
except Exception as e:
|
||
self.exc = e
|
||
|
||
def test_concurrent_events(self):
|
||
dp = self.DummyPlugin(self)
|
||
|
||
def check_dp_exc():
|
||
if dp.exc:
|
||
raise dp.exc
|
||
|
||
try:
|
||
dp.lock1.acquire()
|
||
dp.lock2.acquire()
|
||
assert dp._log.level == log.NOTSET
|
||
|
||
self.config["verbose"] = 1
|
||
t1 = threading.Thread(target=dp.listeners["dummy_event1"][0])
|
||
t1.start() # blocked. t1 tested its log level
|
||
while dp.t1_step != 1:
|
||
check_dp_exc()
|
||
assert t1.is_alive()
|
||
assert dp._log.level == log.NOTSET
|
||
|
||
self.config["verbose"] = 2
|
||
t2 = threading.Thread(target=dp.listeners["dummy_event2"][0])
|
||
t2.start() # blocked. t2 tested its log level
|
||
while dp.t2_step != 1:
|
||
check_dp_exc()
|
||
assert t2.is_alive()
|
||
assert dp._log.level == log.NOTSET
|
||
|
||
dp.lock1.release() # dummy_event1 tests its log level + finishes
|
||
while dp.t1_step != 2:
|
||
check_dp_exc()
|
||
t1.join(0.1)
|
||
assert not t1.is_alive()
|
||
assert t2.is_alive()
|
||
assert dp._log.level == log.NOTSET
|
||
|
||
dp.lock2.release() # dummy_event2 tests its log level + finishes
|
||
while dp.t2_step != 2:
|
||
check_dp_exc()
|
||
t2.join(0.1)
|
||
assert not t2.is_alive()
|
||
|
||
except Exception:
|
||
print("Alive threads:", threading.enumerate())
|
||
if dp.lock1.locked():
|
||
print("Releasing lock1 after exception in test")
|
||
dp.lock1.release()
|
||
if dp.lock2.locked():
|
||
print("Releasing lock2 after exception in test")
|
||
dp.lock2.release()
|
||
print("Alive threads:", threading.enumerate())
|
||
raise
|
||
|
||
def test_root_logger_levels(self, caplog):
|
||
"""Root logger level should be shared between threads."""
|
||
self.config["threaded"] = True
|
||
|
||
blog.getLogger("beets").set_global_level(blog.WARNING)
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_asis_importer()
|
||
assert caplog.messages == []
|
||
|
||
blog.getLogger("beets").set_global_level(blog.INFO)
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_asis_importer()
|
||
for line in caplog.messages:
|
||
assert "import" in line
|
||
assert "album" in line
|
||
|
||
blog.getLogger("beets").set_global_level(blog.DEBUG)
|
||
with caplog.at_level("DEBUG"):
|
||
self.run_asis_importer()
|
||
assert "Sending event: database_change" in caplog.messages
|
||
|
||
|
||
class TestLegacyFormatter:
|
||
"""Tests for ``LegacyFormatter``, which strips the ``beets.`` prefix
|
||
from logger names and prepends the remainder to the message.
|
||
"""
|
||
|
||
@pytest.mark.parametrize(
|
||
"logger_name, expected",
|
||
[
|
||
("beets", "hello world"),
|
||
("beets.musicbrainz", "musicbrainz: hello world"),
|
||
("beets.musicbrainz.sub", "musicbrainz.sub: hello world"),
|
||
("root.foobar", "foobar: hello world"),
|
||
],
|
||
)
|
||
def test_format(self, logger_name, expected):
|
||
"""Root logger passes message unchanged; child loggers get
|
||
``prefix: msg`` format after the ``beets.`` prefix is stripped.
|
||
"""
|
||
formatter = blog.LegacyFormatter("%(legacy_prefix)s%(message)s")
|
||
record = log.LogRecord(
|
||
logger_name, log.INFO, __file__, 0, "hello world", (), None
|
||
)
|
||
assert formatter.format(record) == expected
|