1# SPDX-License-Identifier: BSD-3-Clause 2# Copyright(c) 2010-2014 Intel Corporation 3# Copyright(c) 2022-2023 PANTHEON.tech s.r.o. 4# Copyright(c) 2022-2023 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 # prepare the output folder 62 if not os.path.exists(SETTINGS.output_dir): 63 os.mkdir(SETTINGS.output_dir) 64 65 logging_path_prefix = os.path.join(SETTINGS.output_dir, node) 66 67 fh = logging.FileHandler(f"{logging_path_prefix}.log") 68 fh.setFormatter( 69 logging.Formatter( 70 fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s", 71 datefmt=date_fmt, 72 ) 73 ) 74 75 self._logger.addHandler(fh) 76 self.fh = fh 77 78 # This outputs EVERYTHING, intended for post-mortem debugging 79 # Also optimized for processing via AWK (awk -F '|' ...) 80 verbose_fh = logging.FileHandler(f"{logging_path_prefix}.verbose.log") 81 verbose_fh.setFormatter( 82 logging.Formatter( 83 fmt="%(asctime)s|%(name)s|%(levelname)s|%(pathname)s|%(lineno)d|" 84 "%(funcName)s|%(process)d|%(thread)d|%(threadName)s|%(message)s", 85 datefmt=date_fmt, 86 ) 87 ) 88 89 self._logger.addHandler(verbose_fh) 90 self.verbose_fh = verbose_fh 91 92 super(DTSLOG, self).__init__(self._logger, dict(node=self.node)) 93 94 def logger_exit(self) -> None: 95 """ 96 Remove stream handler and logfile handler. 97 """ 98 for handler in (self.sh, self.fh, self.verbose_fh): 99 handler.flush() 100 self._logger.removeHandler(handler) 101 102 103def getLogger(name: str, node: str = "suite") -> DTSLOG: 104 """ 105 Get logger handler and if there's no handler for specified Node will create one. 106 """ 107 global Loggers 108 # return saved logger 109 logger: LoggerDictType 110 for logger in Loggers: 111 if logger["name"] == name and logger["node"] == node: 112 return logger["logger"] 113 114 # return new logger 115 dts_logger: DTSLOG = DTSLOG(logging.getLogger(name), node) 116 Loggers.append({"logger": dts_logger, "name": name, "node": node}) 117 return dts_logger 118