Select Git revision
test_configuration.py
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_configuration.py 8.73 KiB
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Tests for the ska_logging module."""
import logging
import logging.handlers
import time
import pytest
import ska_logging.configuration
from ska_logging import configure_logging, get_default_formatter, SkaLoggingTagsFormatError
@pytest.fixture
def reset_logging():
"""Cleanup logging module's state after each test (at least try to)."""
yield
manager = logging.root.manager
manager.disabled = logging.NOTSET
for logger in list(manager.loggerDict.values()) + [logging.root]:
if isinstance(logger, logging.Logger):
logger.setLevel(logging.NOTSET)
logger.propagate = True
logger.disabled = False
logger.filters.clear()
handlers = logger.handlers.copy()
for handler in handlers:
# Copied from `logging.shutdown`.
try:
handler.acquire()
handler.flush()
handler.close()
except (OSError, ValueError):
pass
finally:
handler.release()
logger.removeHandler(handler)
def get_named_handler(logger, name):
"""Search up through logger hierarchy to find a handler with the specified name."""
while logger:
for handler in logger.handlers:
if handler.name == name:
return handler
logger = logger.parent
class AppendHandler(logging.Handler):
"""Handler that keeps a history of the records and formatted log messages."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.logs = []
self.records = []
def emit(self, record):
self.records.append(record)
self.logs.append(self.format(record))
"""Override configuration for recording loggers"""
RECORDER_OVERRIDES = {
"handlers": {"recorder": {"()": AppendHandler, "formatter": "default"}},
"root": {"handlers": ["console", "recorder"]},
}
@pytest.mark.usefixtures("reset_logging")
class TestConfigureLogging:
"""Tests for :func:`~ska_logging.configuration.configure_logging`."""
@pytest.fixture
def default_logger(self):
"""Return user logger instance with default configuration."""
configure_logging()
yield logging.getLogger("ska.test.app")
@pytest.fixture
def recording_logger(self):
"""Return user logger, including a recording handler.
The additional handler has the name "recorder". It uses the default formatter,
and stores all formatted output strings as a list in its `logs` attribute.
It also keeps a list of the raw log records in its `records` attribute.
Note: we use this instead of pytest's `caplog` fixture because we cannot change
the formatter that it uses.
"""
configure_logging(overrides=RECORDER_OVERRIDES)
yield logging.getLogger("ska.logger")
@pytest.fixture
def recording_tags_logger(self):
"""Return user logger like :func:`recording_logger`, but including tags filter."""
class MyFilter(logging.Filter):
def filter(self, record):
record.tags = "key1:value1,key2:value2"
return True
configure_logging(tags_filter=MyFilter, overrides=RECORDER_OVERRIDES)
yield logging.getLogger("ska.logger")
def test_includes_console_handler(self, default_logger):
assert get_named_handler(default_logger, "console")
def test_multiple_calls_non_root_logger_still_enabled(self, default_logger):
logger = logging.getLogger("ska.logger.test")
assert not logger.disabled
configure_logging()
assert not logger.disabled
def test_default_log_level_info(self, default_logger):
logger = logging.getLogger("ska.logger.test")
assert logger.getEffectiveLevel() == logging.INFO
assert default_logger.getEffectiveLevel() == logging.INFO
def test_set_log_level_int(self):
configure_logging(level=logging.DEBUG)
logger = logging.getLogger("ska.logger.test")
assert logger.getEffectiveLevel() == logging.DEBUG
def test_set_log_level_string(self):
configure_logging(level="WARNING")
logger = logging.getLogger("ska.logger.test")
assert logger.getEffectiveLevel() == logging.WARNING
def test_default_uses_utc_time(self, recording_logger):
recording_logger.info("UTC message")
recorder = get_named_handler(recording_logger, "recorder")
record = recorder.records[0]
log_message = recorder.logs[0]
expected_time = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created))
assert "UTC message" in log_message
assert expected_time in log_message # just testing UTC, so ignore the milliseconds part
def test_default_no_tags(self, default_logger):
handler = get_named_handler(default_logger, "console")
formatter = handler.formatter
assert "%(tag)s" not in formatter._fmt
def test_tags_filter_adds_tags_field(self, recording_tags_logger):
handler = get_named_handler(recording_tags_logger, "console")
formatter = handler.formatter
assert "%(tags)s" in formatter._fmt
def test_tags_filter_emits_tags_value(self, recording_tags_logger):
recording_tags_logger.info("Tags message")
recorder = get_named_handler(recording_tags_logger, "recorder")
record = recorder.records[0]
log_message = recorder.logs[0]
assert record.tags == "key1:value1,key2:value2"
assert log_message.endswith("|key1:value1,key2:value2|Tags message")
def test_override(self):
overrides = {
"handlers": {"test": {"class": "logging.StreamHandler", "formatter": "default"}},
"root": {"handlers": ["console", "test"]},
}
configure_logging(overrides=overrides)
logger = logging.getLogger()
assert get_named_handler(logger, "console")
assert get_named_handler(logger, "test")
class TestGetDefaultFormatter:
"""Tests for :func:`~ska_logging.configuration.get_default_formatter`."""
def test_default_no_tags(self):
formatter = get_default_formatter()
assert isinstance(formatter, ska_logging.configuration._UTCFormatter)
assert formatter._fmt == ska_logging.configuration._FORMAT_STR_NO_TAGS
def test_get_tags_disabled(self):
formatter = get_default_formatter(tags=False)
assert isinstance(formatter, ska_logging.configuration._UTCFormatter)
assert formatter._fmt == ska_logging.configuration._FORMAT_STR_NO_TAGS
def test_get_tags_enabled(self):
formatter = get_default_formatter(tags=True)
assert isinstance(formatter, ska_logging.configuration._UTCFormatter)
assert formatter._fmt == ska_logging.configuration._FORMAT_STR_WITH_TAGS
def test_get_tags_static_string(self):
formatter = get_default_formatter(tags="test-key:test-value")
assert isinstance(formatter, ska_logging.configuration._UTCFormatter)
tags_format = ska_logging.configuration._FORMAT_STR_WITH_TAGS
expected_format = tags_format.replace("%(tags)s", "test-key:test-value")
assert formatter._fmt == expected_format
def test_get_tags_invalid_static_string(self):
with pytest.raises(SkaLoggingTagsFormatError):
get_default_formatter(tags="no|pipes|allowed")
with pytest.raises(SkaLoggingTagsFormatError):
get_default_formatter(tags="no%percentage%symbols%allowed")
class TestOverride:
"""Tests for :func:`~ska_logging.configuration._override`.
Code based on:
https://github.com/ska-sa/katsdpcontroller/blob/
40ec9ae57adf1d441fc42b5c6eac86aa57b6e3aa/katsdpcontroller/test/test_product_config.py#L42
Copyright (c) 2013-2019, National Research Foundation (SARAO)
All rights reserved.
(BSD-3-Clause license)
"""
def test_add(self):
out = ska_logging.configuration._override({"a": 1}, {"b": 2})
assert out == {"a": 1, "b": 2}
def test_remove(self):
out = ska_logging.configuration._override({"a": 1, "b": 2}, {"b": None})
assert out == {"a": 1}
out = ska_logging.configuration._override(out, {"b": None}) # Already absent
assert out == {"a": 1}
def test_replace(self):
out = ska_logging.configuration._override({"a": 1, "b": 2}, {"b": 3})
assert out == {"a": 1, "b": 3}
def test_recurse(self):
orig = {"a": {"aa": 1, "ab": 2}, "b": {"ba": {"c": 10}, "bb": [5]}}
override = {"a": {"aa": [], "ab": None, "ac": 3}, "b": {"bb": [1, 2]}}
out = ska_logging.configuration._override(orig, override)
assert out == {"a": {"aa": [], "ac": 3}, "b": {"ba": {"c": 10}, "bb": [1, 2]}}