tests/test_multistructlog.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import multistructlog
import os
import logging
import logging.config
import shutil
import json
import unittest
import structlog

# test directory, cleared and left around after testing
test_scratch = "msl_test_scratch"


class TestMultiStructLog(unittest.TestCase):

    @classmethod
    def setUpClass(clsG):
        # delete and recreate test_scratch directory if it exists
        if os.path.isdir(test_scratch):
            shutil.rmtree(test_scratch)

        os.mkdir(test_scratch)

    def tearDown(self):
        structlog.reset_defaults()

    def test_reload(self):
        '''
        Test that creating creatinging multiple identical loggers will reuse
        existing loggers
        '''

        logger0 = multistructlog.create_logger({'version': 1, 'foo': 'bar'})
        logger1 = multistructlog.create_logger({'version': 1, 'foo': 'bar'})
        logger2 = multistructlog.create_logger()

        self.assertEqual(logger0, logger1)
        self.assertNotEqual(logger0, logger2)

    def test_different_formatters(self):
        '''
        Test different formattters and levels to different output streams
        NOTE: Only one test as logger has global state that is hard to reset
        between tests without breaking other things.
        '''

        f1 = os.path.join(test_scratch, 'different_formatters_test_file1')
        f2 = os.path.join(test_scratch, 'different_formatters_test_file2.json')

        logging_config = {
            'version': 1,
            'handlers': {
                'file1': {
                    'class': 'logging.handlers.RotatingFileHandler',
                    'level': 'WARNING',
                    'formatter': 'structured',
                    'filename': f1,
                },
                'file2': {
                    'class': 'logging.handlers.RotatingFileHandler',
                    'level': 'INFO',
                    'formatter': 'json',
                    'filename': f2,
                },
            },
            'formatters': {
                'json': {
                    '()': structlog.stdlib.ProcessorFormatter,
                    "processor": structlog.processors.JSONRenderer(),
                },
                'structured': {
                    '()': structlog.stdlib.ProcessorFormatter,
                    'processor': structlog.dev.ConsoleRenderer(colors=False),
                },
            },
            'loggers': {
                '': {
                    'handlers': ['file1', 'file2'],
                    'level': 'WARNING',
                    'propagate': True
                },
            }
        }

        # reset level to debug, overriding 'loggers' directive above
        logger = multistructlog.create_logger(logging_config,
                                              level=logging.DEBUG)

        extra_data = {'number': 42}

        logger.warning("should be in both files", extra=extra_data)

        # filtered by file1 handler
        logger.info("should only be in file2")

        # filtered by both handlers, but not by loggers
        logger.debug("should not be in either file")

        # test new trace level
        logger.trace("testing trace, shouldn't be in either file")

        # check contents of file1
        with open(f1) as f1fh:

            # regex at start should roughly match ISO8601 datetime
            f1_desired = r"[\dTZ\-\.\:]+ \[warning  \] should be in both files        extra={'number': 42}"
            self.assertRegexpMatches(f1fh.read(), f1_desired)

        # check contents of file2
        f2_read = []
        f2_desired = [
            {"event": "should be in both files", "extra": {"number": 42},
             "level": "warning", "timestamp": "removed"},
            {"event": "should only be in file2",
             "level": "info", "timestamp": "removed"},
        ]

        with open(f2) as f2fh:
            for line in f2fh:
                jl = json.loads(line)

                # assert there is a timestamp, and remove it as it changes
                self.assertTrue(True if "timestamp" in jl else False)
                jl['timestamp'] = "removed"

                f2_read.append(jl)

        self.assertEqual(f2_read, f2_desired)