1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2022 Huawei Device Co., Ltd. 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17# 18 19import datetime 20import logging 21import os 22import sys 23import traceback 24from xdevice import get_cst_time 25from xdevice import platform_logger 26 27 28class DeviceTestLog: 29 _log = platform_logger("DeviceTestLog") 30 31 @classmethod 32 def set_log(cls, log_obj): 33 if log_obj is not None: 34 cls._log = log_obj 35 cls.info("INIT DeviceTest Log Successfully.") 36 37 @classmethod 38 def error(cls, content, error_no=None, is_traceback=False): 39 """ 40 Only the description information is printed in info mode, 41 and detailed error information is printed in debug mode 42 """ 43 error_no = error_no or '00000' 44 if is_traceback: 45 cls._log.debug(traceback.format_exc()) 46 cls._log.error(content, error_no=error_no) 47 48 @classmethod 49 def warn(cls, content): 50 cls._log.warning(content) 51 52 @classmethod 53 def warning(cls, content): 54 cls._log.warning(content) 55 56 @classmethod 57 def info(cls, content): 58 cls._log.info(content) 59 60 @classmethod 61 def debug(cls, content): 62 cls._log.debug(content) 63 64 @classmethod 65 def exception(cls, content, error_no=None): 66 error_no = error_no or '00000' 67 cls._log.exception(content, error_no=error_no) 68 69 70def create_dir(path): 71 """Creates a directory if it does not exist already. 72 Args: 73 path: The path of the directory to create. 74 """ 75 full_path = os.path.abspath(os.path.expanduser(path)) 76 if not os.path.exists(full_path): 77 os.makedirs(full_path, exist_ok=True) 78 79 80def _get_timestamp(time_format, delta=None): 81 now_time = get_cst_time() 82 if delta: 83 now_time = now_time + datetime.timedelta(seconds=delta) 84 return now_time.strftime(time_format)[:-3] 85 86 87def get_log_file_timestamp(delta=None): 88 """Returns a timestamp in the format used for log file names. 89 90 Default is current time. If a delta is set, the return value will be 91 the current time offset by delta seconds. 92 93 Params: 94 delta: Number of seconds to offset from current time; can be negative. 95 96 Returns: 97 A timestamp in log filen name format with an offset. 98 """ 99 return _get_timestamp("%m-%d-%Y_%H-%M-%S-%f", delta) 100 101 102def get_test_logger(log_path, tag, prefix=None, filename=None, 103 is_debug=False): 104 """Returns a logger object used for tests. 105 106 The logger object has a stream handler and a file handler. The stream 107 handler logs INFO level to the terminal, the file handler logs DEBUG 108 level to files. 109 110 Params: 111 log_path: Location of the log file. 112 TAG: Name of the logger's owner. 113 prefix: A prefix for each log line in terminal. 114 filename: Name of the log file. The default is the time the logger 115 is requested. 116 117 Returns: 118 A logger configured with one stream handler and one file handler 119 """ 120 log = logging.getLogger(tag) 121 if log.handlers: 122 return log 123 log.propagate = False 124 125 if not is_debug: 126 log.setLevel(logging.INFO) 127 else: 128 log.setLevel(logging.DEBUG) 129 # Log info to stream 130 log_line_format = "%(asctime)s.%(msecs).03d %(threadName)s-%(" \ 131 "thread)d %(levelname)s %(message)s" 132 log_line_time_format = "%Y-%m-%d %H:%M:%S" 133 terminal_format = log_line_format 134 if prefix: 135 terminal_format = "[{}] {}".format(prefix, log_line_format) 136 c_formatter = logging.Formatter(terminal_format, log_line_time_format) 137 ch_value = ConsoleHandler() 138 ch_value.setFormatter(c_formatter) 139 if not is_debug: 140 ch_value.setLevel(logging.INFO) 141 else: 142 ch_value.setLevel(logging.DEBUG) 143 144 # Log everything to file 145 f_formatter = logging.Formatter(log_line_format, log_line_time_format) 146 # All the logs of this test class go into one directory 147 if filename is None: 148 filename = get_log_file_timestamp() 149 if tag == "xDevice": 150 create_dir(log_path) 151 fh_vaule = logging.FileHandler( 152 os.path.join(log_path, filename), encoding="utf-8") 153 else: 154 log_path = os.path.join(log_path, filename) 155 create_dir(log_path) 156 fh_vaule = logging.FileHandler( 157 os.path.join(log_path, 'test_run_details.log'), encoding="utf-8") 158 fh_vaule.setFormatter(f_formatter) 159 if not is_debug: 160 fh_vaule.setLevel(logging.INFO) 161 else: 162 fh_vaule.setLevel(logging.DEBUG) 163 log.addHandler(fh_vaule) 164 log.addHandler(ch_value) 165 return log 166 167 168def kill_test_logger(_logger): 169 """Cleans up a test logger object created by get_test_logger. 170 171 Params: 172 logger: The logging object to clean up. 173 """ 174 for h_value in list(_logger.handlers): 175 _logger.removeHandler(h_value) 176 if isinstance(h_value, logging.FileHandler): 177 h_value.close() 178 179 180def create_latest_log_alias(actual_path): 181 """Creates a symlink to the latest test run logs. 182 183 Args: 184 actual_path: The source directory where the latest test run's logs are. 185 """ 186 link_path = os.path.join(os.path.dirname(actual_path), "latest") 187 if os.path.islink(link_path): 188 os.remove(link_path) 189 190 try: 191 os.symlink(actual_path, link_path) 192 except Exception as error: 193 print("xDeviceTest-31 create_latest_log_alias:" + str(error)) 194 195 196def logger(log_path, tag, prefix=None, filename=None, is_debug=False): 197 """Returns a logger and a reporter of the same name. 198 199 Params: 200 log_path: Location of the report file. 201 TAG: Name of the logger's owner. 202 prefix: A prefix for each log line in terminal. 203 filename: Name of the files. The default is the time the objects 204 are requested. 205 206 Returns: 207 A log object and a reporter object. 208 """ 209 if filename is None: 210 filename = get_log_file_timestamp() 211 create_dir(log_path) 212 create_latest_log_alias(log_path) 213 get_logger = get_test_logger(log_path, tag, prefix, filename, 214 is_debug=is_debug) 215 return get_logger, filename 216 217 218def get_log_line_timestamp(delta=None): 219 """ 220 Get a timestamp in the format used by log lines. 221 Default is current time. If a delta is set, the return value will be 222 the current time offset by delta seconds. 223 """ 224 return _get_timestamp("%Y-%m-%d %H:%M:%S.%f", delta) 225 226 227class ConsoleHandler(logging.Handler): 228 """ 229 A handler class which writes logging records, appropriately formatted, 230 to a stream. Note that this class does not close the stream, as 231 sys.stdout or sys.stderr may be used. 232 """ 233 234 terminator = '\n' 235 236 def __init__(self, stream=None): 237 """ 238 Initialize the handler. 239 """ 240 super().__init__() 241 self.stdout = sys.stdout 242 self.stderr = sys.stderr 243 self.stream = stream if stream is not None else sys.stderr 244 245 def flush(self): 246 247 self.acquire() 248 try: 249 if self.stream: 250 if hasattr(self.stream, "flush"): 251 self.stream.flush() 252 finally: 253 self.release() 254 255 def emit(self, emit_record): 256 257 try: 258 msg = self.format(emit_record) 259 if emit_record.levelno > logging.INFO: 260 self.stream = self.stderr 261 else: 262 self.stream = self.stdout 263 264 self.stream.write(msg) 265 self.stream.write(self.terminator) 266 self.flush() 267 except Exception: 268 self.handleError(emit_record) 269 270 271""" 272兼容release2脚本需要 273""" 274 275 276def print_info(msg): 277 DeviceTestLog._log.info(msg) 278 279 280def print_error(msg): 281 DeviceTestLog._log.error(msg) 282 283 284def print_debug(msg): 285 DeviceTestLog._log.debug(msg) 286 287 288def print_warn(msg): 289 DeviceTestLog._log.warning(msg) 290 291 292def print_trace(): 293 DeviceTestLog._log.error("".join( 294 traceback.format_exception(*sys.exc_info()))) 295