How to use get_log_name method in autotest

Best Python code snippet using autotest_python

logger_server.py

Source:logger_server.py

1#!/usr/bin/env python2# -*-coding:utf-8-*-3# @Time : 2017/11/1 ~ 2019/9/14# @Author : Allen Woo5import json6import os7import socket8import time9import logging10from logging.handlers import TimedRotatingFileHandler11from apps.configs.sys_config import LOG_PATH, SOCKET_PORT12from apps.utils.osr_async.osr_async import async_process13class LoggerClientUDP:14    def __init__(self,15                 set_level=logging.INFO,16                 logfile="{}/{}.log".format(LOG_PATH, time.time()),17                 get_log_name='logger',18                 formatter='%(asctime)s %(levelname)s %(message)s',19                 **kwargs20                 ):21        self.sk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)22        self.set = {23            "kwargs": {24                "set_level": set_level,25                "logfile": logfile,26                "get_log_name": get_log_name,27                "formatter": formatter28            }29        }30        self.sp = "__+++++udposrlog+++++__"31    def debug(self, log, **kwargs):32        self.set["log_level"] = "debug"33        self.send_log(log=log)34    def info(self, log, **kwargs):35        self.set["log_level"] = "info"36        self.send_log(log=log)37    def warning(self, log, **kwargs):38        self.set["log_level"] = "warning"39        self.send_log(log=log)40    def error(self, log, **kwargs):41        self.set["log_level"] = "error"42        self.send_log(log=log)43    def critical(self, log, **kwargs):44        self.set["log_level"] = "critical"45        self.send_log(log=log)46    def exception(self, log, **kwargs):47        self.set["log_level"] = "error"48        self.send_log(log=log)49    def send_log(self, log):50        set = json.dumps(self.set)51        log = "{}{}{}".format(set, self.sp, log)52        self.sk.sendto(log.encode(), ("127.0.0.1", SOCKET_PORT))53    def __del__(self):54        self.sk.close()55class LogServerUDP:56    def init_app(self):57        self.sk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)58        try:59            self.sk.bind(("127.0.0.1", SOCKET_PORT))60            self.failed = False61            print(" * [pid:{}] Logger Server...".format(os.getpid()))62        except:63            self.failed = True64            return False65        self.loggers = {}66        self.self_logfile = "{}/logger_server.log".format(LOG_PATH)67        self.get_log_name = "logger_server"68        self.kwargs = {69            "set_level": logging.DEBUG,70            "logfile": self.self_logfile,71            "get_log_name": self.get_log_name,72            "formatter": '%(asctime)s %(levelname)s %(message)s'73        }74        self.set_logger(self.kwargs)75        self.sp =  "__+++++udposrlog+++++__"76        return True77    def set_logger(self, kwargs):78        set_level = kwargs.get("set_level")79        logfile = kwargs.get("logfile")80        get_log_name = kwargs.get("get_log_name")81        formatter = kwargs.get("formatter")82        if not os.path.exists(os.path.split(logfile)[0]):83            os.makedirs(os.path.split(logfile)[0])84        # According to the time85        file_handler = TimedRotatingFileHandler(logfile, "midnight", 1, 7)86        file_handler.suffix = "%Y-%m-%d"87        # According to the size88        # file_handler = RotatingFileHandler(filename, maxBytes=10*1024*1024, backupCount=3)89        file_handler.setLevel(set_level)90        _formatter = logging.Formatter(formatter)91        file_handler.setFormatter(_formatter)92        logging.getLogger(get_log_name).addHandler(file_handler)93        logging.getLogger(get_log_name).setLevel(logging.INFO)94        logg = logging.getLogger(get_log_name)95        self.loggers[get_log_name] = {96            "critical": logg.critical,97            "error": logg.error,98            "warning": logg.warning,99            "info": logg.info,100            "debug": logg.debug,101        }102        return self.loggers[get_log_name]103    @async_process()104    def log_server(self):105        if self.failed:106            return107        while True:108            data = self.sk.recv(10240)109            data = data.decode()110            data = data.split(self.sp)111            sets = data[0]112            log = data[1]113            sets = json.loads(sets)114            kwargs = sets["kwargs"]115            logg = self.loggers.get(kwargs["get_log_name"])116            if not logg:117                logg = self.set_logger(kwargs)118            logg[sets["log_level"]](log)119    def __del__(self):...

logger.py

Source:logger.py

...57    return "       "58class Logger(object):59    def set_log_name(self, name):60        self.__logname = name61    def get_log_name(self):62        if hasattr(self,"__logname") and self.__logname :63            return self.__logname64        else:65            return "%s.%s"%(self.__module__,self.__class__.__name__)66    def log_debug(self, msg, *args, **kwargs): 67        # msg = "%s %s"%(objaddr(self),msg)68        mylogger = logging.getLogger(self.get_log_name())69        mylogger.debug(msg, *args, **kwargs)70    def log_info(self, msg, *args, **kwargs): 71        # msg = "%s  %s"%(objaddr(self),msg)72        mylogger = logging.getLogger(self.get_log_name())73        mylogger.info(msg, *args, **kwargs)74    def log_warn(self, msg, *args, **kwargs): 75        # msg = "%s  %s"%(objaddr(self),msg)76        mylogger = logging.getLogger(self.get_log_name())77        mylogger.warn(msg, *args, **kwargs)78    def log_error(self, msg, *args, **kwargs): 79        # msg = "%s  %s"%(objaddr(self),msg)80        mylogger = logging.getLogger(self.get_log_name())81        mylogger.error(msg, *args, **kwargs)82    def log_critical(self, msg, *args, **kwargs): 83        # msg = "%s  %s"%(objaddr(self),msg)84        mylogger = logging.getLogger(self.get_log_name())85        mylogger.critical(msg, *args, **kwargs)86    def log_exception(self, msg, *args, **kwargs):87        # msg = "%s  %s"%(objaddr(self),msg)88        mylogger = logging.getLogger(self.get_log_name())89        mylogger.exception(msg, *args, **kwargs)90    def log_deprecated(self, msg, *args, **kwargs):91        # msg = "%s  %s"%(objaddr(self),msg)92        mylogger = logging.getLogger(self.get_log_name())93        mylogger.log(100,msg, *args, **kwargs)94def new_logger(name):95    l = Logger()96    l.set_log_name(name)97    ...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.