Skip to content
Snippets Groups Projects
Select Git revision
  • 16c9928d133a86ac049dfcd81a12e59535d80f99
  • master default protected
  • gec-84-c-compiler-dependency
  • gec-50-only-test-plot-with-gtkmm
  • make-baseline-aoqplot-nicer
  • use-system-pybind11
  • fix-filterbank-imagesets
  • fix-drawing-large-images
  • allow-writing-to-filterbanksets
  • implement-nroamlize-bandpass-for-non-complex-data
  • support-multiple-pols-in-filterbank
  • solve-flipped-axis-crashes
  • improve-statistics-speed
  • avoid-deprecated-gtkmm
  • ast-1621-deprecated-deletetable
  • ast-1613-update-hdf5-url
  • rename-use-input-flags-variable
  • dockerfile-fix
  • open-correct-data-column
  • add-target-cpu
  • add-git-to-format-conatiner
  • v3.4.0
  • v3.3.0
  • v3.2.0
  • test-vx.y
  • v3.1.0
  • v3.0.0
  • v2.15.0
  • v2.14.0
  • v2.13.0
  • v2.12.1
  • v2.12.0
  • v2.11.0
  • v2.10.0
  • v2.9.0
  • v2.8.0
  • v2.7.0
  • v2.6.0
  • v2.5.0
  • v2.4.0
  • v2.3.0
41 results

testconfig.py.in

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    test_configuration.py 8.03 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
    
    
    @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
    
    
    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]}}