Skip to content
Snippets Groups Projects
Select Git revision
  • 8339bf17dc2d768d98a033814aaf77a9d67c47af
  • SAR-150_add_transaction_id_in_all_logs default protected
  • master
  • sar-293_fix_lint_issue
  • sar-259/Testing-skallop-docs
  • sar-255/Publish-package-to-CAR
  • test-package
  • SAR-189-use-single-namespace-for-package-naming
  • revert-52d118c0
  • 0.4.1
  • 0.4.0
  • 0.3.0
  • 0.2.1
  • 0.2.0
  • 0.1.0
15 results

test_configuration.py

Blame
  • 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]}}