Produits Entr'ouvert » Hobo

DEBUG_LOG_PATH = '/var/log/%s/debug' % PROJECT_NAME

DEBUG_LOG_FORMAT = (

    '%(asctime)s \x1f%(tenant)s \x1f%(ip)s \x1f%(user)r \x1f%(request_id)s \x1f'

    '%(levelname)s \x1f%(name)s \x1f%(message)s'

)

        'debug_log': {

            '()': 'hobo.logger.DebugLogFilter',

        },

        'debug': {

            'format': DEBUG_LOG_FORMAT,

        },

            'level': 'INFO',

            'level': 'INFO',

        'debug': {

            'level': 'DEBUG',

            'class': 'hobo.logger.TimedRotatingFileHandler',

            'formatter': 'debug',

            'filename': DEBUG_LOG_PATH,

            'when': 'midnight',

            'backupCount': 1,

            'interval': 1,

            'filters': ['request_context', 'debug_log'],

        }

            'level': 'DEBUG',

            'handlers': ['syslog', 'mail_admins', 'debug'],

# Copyright (C) 2015-2020  Entr'ouvert

import ast

import datetime

import logging.handlers

import os

class LogRecord(logging.LogRecord):

    def getMessage(self):

        return super(LogRecord, self).getMessage().replace('\n', '\n ')

class TimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler):

    def format(self, record):

        old_class = record.__class__

        record.__class__ = LogRecord

        try:

            return super(TimedRotatingFileHandler, self).format(record)

        finally:

            record.__class__ = old_class

class DebugLogFilter(object):

    '''Filter debug log records based on the DEBUG_LOG setting'''

    def filter(self, record):

        debug_log = getattr(settings, 'DEBUG_LOG', False)

        # add space after newlines to reconstruct message when reading

        # debug.log

        if not debug_log:

            return False

        # change class to add space after newlines in message

        record.__class__ = LogRecord

        if debug_log is True:

            return True

        elif hasattr(debug_log, 'encode'):

            # debug_log is a string

            domains = [domain.strip() for domain in debug_log.split(',')]

            return any(record.name == domain or (record.name.startswith(domain) and record.name[len(domain)] == '.')

                       for domain in domains)

        else:

            return bool(debug_log)

class DebugLog(object):

    def __init__(self, path):

        self.path = path

    def _pre_lines(self, cursor=0):

        if not os.path.exists(self.path):

            return

        with open(self.path, 'rb') as fd:

            accum = ''

            fd.seek(cursor)

            for line in fd:

                size = len(line)

                cursor += size

                line = line.decode('utf-8')

                if not accum:

                    accum = line

                elif not line.startswith(' '):

                    yield cursor - size, accum

                    accum = line

                else:

                    accum += line[1:]

            if accum:

                yield cursor, accum

    keys = ['tenant', 'ip', 'user', 'level', 'request_id', 'name']

    def _parse(self, cursor=0):

        for cursor, line in self._pre_lines(cursor=cursor):

            if line.endswith('\n'):

                line = line[:-1]

            parts = line.split(' \x1f', settings.DEBUG_LOG_FORMAT.count(' \x1f'))

            try:

                timestamp = datetime.datetime.strptime(parts[0], '%Y-%m-%d %H:%M:%S,%f')

            except ValueError:

                continue

            message = parts[-1]

            d = {key: value for key, value in zip(self.keys, parts[1:-1])}

            if 'user' in d:

                try:

                    d['user'] = ast.literal_eval(d['user'])

                except SyntaxError:

                    pass

            d.update({

                'cursor': cursor,

                'timestamp': timestamp,

                'message': message,

            })

            yield d

    @classmethod

    def lines(cls, cursor=0):

        debug_log_path = getattr(settings, 'DEBUG_LOG_PATH', None)

        if not debug_log_path:

            return None

        if not os.path.exists(debug_log_path):

            return None

        for record in cls(debug_log_path)._parse(cursor=cursor):

            yield record

LOGGING['handlers']['debug']['filename'] = 'debug.log'

# hobo - portal to configure and deploy applications

# Copyright (C) 2019  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import datetime

import logging

import logging.config

import pytest

from hobo.logger import DebugLog

@pytest.fixture

def debug_log(settings, tmpdir):

    debug_log_path = str(tmpdir / 'debug.log')

    settings.DEBUG_LOG_PATH = debug_log_path

    LOGGING = {

        'version': 1,

        'disable_existing_loggers': True,

        'filters': {

            'debug_log': {

                '()': 'hobo.logger.DebugLogFilter',

            },

            'request_context': {

                '()': 'hobo.logger.RequestContextFilter',

            },

        },

        'formatters': {

            'debug': {

                'format': settings.DEBUG_LOG_FORMAT,

            },

        },

        'handlers': {

            'debug': {

                'level': 'DEBUG',

                'class': 'hobo.logger.TimedRotatingFileHandler',

                'formatter': 'debug',

                'filename': debug_log_path,

                'when': 'midnight',

                'backupCount': 1,

                'interval': 1,

                'filters': ['request_context', 'debug_log'],

            }

        },

        'loggers': {

            'multitenant': {

                'level': 'DEBUG',

                'handlers': ['debug'],

            },

        },

    }

    logging.config.dictConfig(LOGGING)

    logger = logging.getLogger('multitenant')

    yield logger

    logger.removeHandler(logger.handlers[0])

def test_debug_log(settings, rf, debug_log, freezer):

    freezer.move_to('2020-4-20')

    request = rf.get('/path/')

    debug_log.info('test %s is ok', 1, extra={'request': request, 'tenant': 'yes'})

    lines = list(DebugLog.lines())

    assert len(lines) == 0

    settings.DEBUG_LOG = True

    debug_log.info('log %s is \nok', 2, extra={'request': request, 'tenant': 'tenant1', 'user': 'jean darmette'})

    debug_log.debug('log %s is \nok', 3, extra={'request': request})

    lines = list(DebugLog.lines())

    assert len(lines) == 2

    assert lines[0] == {

        'cursor': 93,

        'ip': '127.0.0.1',

        'level': 'r:' + hex(id(request))[2:].upper(),

        'message': 'log 2 is \nok',

        'request_id': 'INFO',

        'tenant': '-',

        'timestamp': datetime.datetime(2020, 4, 20, 2, 0),

        'user': "-",

        'name': 'multitenant',

    }

    # check that seeking by cursor gives the same lines

    lines2 = list(DebugLog.lines(cursor=lines[0]['cursor']))

    assert len(lines2) == 1

    assert lines[1] == lines2[0]

from hobo.logger import RequestContextFilter, DebugLogFilter

def test_debug_log_filter(caplog, settings):

    # default caplog log level is INFO

    caplog.set_level(logging.DEBUG)

    root_logger = logging.getLogger()

    assert len(root_logger.handlers) == 1

    root_logger.handlers[0].addFilter(DebugLogFilter())

    root_logger.debug('l1')

    assert 'l1' not in caplog.text

    settings.DEBUG_LOG = True

    root_logger.debug('l2')

    assert 'l2' in caplog.text

    settings.DEBUG_LOG = False

    root_logger.debug('l3')

    assert 'l3' not in caplog.text

    settings.DEBUG_LOG = 'app1,app2'

    root_logger.debug('l4')

    assert 'l4' not in caplog.text

    logging.getLogger('app3').debug('l5')

    assert 'l5' not in caplog.text

    logging.getLogger('app1').debug('l6')

    assert 'l6' in caplog.text

    logging.getLogger('app2').debug('l7')

    assert 'l7' in caplog.text

LOGGING['handlers']['debug']['filename'] = 'debug.log'