1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2010-2014 Intel Corporation 3# Copyright(c) 2022 PANTHEON.tech s.r.o. 4# Copyright(c) 2022 University of New Hampshire 5 6""" 7DTS logger module with several log level. DTS framework and TestSuite logs 8are saved in different log files. 9""" 10 11import logging 12import os.path 13from typing import TypedDict 14 15from .settings import SETTINGS 16 17date_fmt = "%Y/%m/%d %H:%M:%S" 18stream_fmt = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" 19 20 21class LoggerDictType(TypedDict): 22 logger: "DTSLOG" 23 name: str 24 node: str 25 26 27# List for saving all using loggers 28Loggers: list[LoggerDictType] = [] 29 30 31class DTSLOG(logging.LoggerAdapter): 32 """ 33 DTS log class for framework and testsuite. 34 """ 35 36 logger: logging.Logger 37 node: str 38 sh: logging.StreamHandler 39 fh: logging.FileHandler 40 verbose_fh: logging.FileHandler 41 42 def __init__(self, logger: logging.Logger, node: str = "suite"): 43 self.logger = logger 44 # 1 means log everything, this will be used by file handlers if their level 45 # is not set 46 self.logger.setLevel(1) 47 48 self.node = node 49 50 # add handler to emit to stdout 51 sh = logging.StreamHandler() 52 sh.setFormatter(logging.Formatter(stream_fmt, date_fmt)) 53 sh.setLevel(logging.INFO) # console handler default level 54 55 if SETTINGS.verbose is True: 56 sh.setLevel(logging.DEBUG) 57 58 self.logger.addHandler(sh) 59 self.sh = sh 60 61 logging_path_prefix = os.path.join(SETTINGS.output_dir, node) 62 63 fh = logging.FileHandler(f"{logging_path_prefix}.log") 64 fh.setFormatter( 65 logging.Formatter( 66 fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s", 67 datefmt=date_fmt, 68 ) 69 ) 70 71 self.logger.addHandler(fh) 72 self.fh = fh 73 74 # This outputs EVERYTHING, intended for post-mortem debugging 75 # Also optimized for processing via AWK (awk -F '|' ...) 76 verbose_fh = logging.FileHandler(f"{logging_path_prefix}.verbose.log") 77 verbose_fh.setFormatter( 78 logging.Formatter( 79 fmt="%(asctime)s|%(name)s|%(levelname)s|%(pathname)s|%(lineno)d|" 80 "%(funcName)s|%(process)d|%(thread)d|%(threadName)s|%(message)s", 81 datefmt=date_fmt, 82 ) 83 ) 84 85 self.logger.addHandler(verbose_fh) 86 self.verbose_fh = verbose_fh 87 88 super(DTSLOG, self).__init__(self.logger, dict(node=self.node)) 89 90 def logger_exit(self) -> None: 91 """ 92 Remove stream handler and logfile handler. 93 """ 94 for handler in (self.sh, self.fh, self.verbose_fh): 95 handler.flush() 96 self.logger.removeHandler(handler) 97 98 99def getLogger(name: str, node: str = "suite") -> DTSLOG: 100 """ 101 Get logger handler and if there's no handler for specified Node will create one. 102 """ 103 global Loggers 104 # return saved logger 105 logger: LoggerDictType 106 for logger in Loggers: 107 if logger["name"] == name and logger["node"] == node: 108 return logger["logger"] 109 110 # return new logger 111 dts_logger: DTSLOG = DTSLOG(logging.getLogger(name), node) 112 Loggers.append({"logger": dts_logger, "name": name, "node": node}) 113 return dts_logger 114