# function for collecting and writing messages i.e. logging and calculate the return-code # ------------------------------------------------------------------------ """ there are two types of logging: * log everything relevant to document the test-run in relation of the test-system and the test-effort should be logged in order to archive these logfiles * debug everything to know how this automatism is running should log into a debug-file. This file can be removed after a few days the logging is parametrized by the log-level (fatal ... trace). Especially the debug- and trace-message can be switched on/of en detail in the functions and over special settings (parameter tool, component: these debugs should be switched on) there are three types of functions: * setFatal|Error|...() it logs, sets the top-message and return-code * logFatal|Error|...() it logs without any effect for the program-result * log()|debug(int) it logs depending on the special level-setting at the end of a program there are some results: * for each component there is a return-code, a top-message for documentation in the parameter-file, and a list of collected messages + for the main-program the result (return-code, top-message) will be summarized (the most relevant message will used) """ import basic.program import os import math from datetime import datetime import tools.path_tool #import tools.i18n_tool #import basic.text_const as T import basic.constants as B MTEXT_FATAL = "fatal" MTEXT_ERROR = "error" MTEXT_WARN = "warn" MTEXT_MSG = "msg" MTEXT_INFO = "info" MTEXT_DEBUG = "debug" MTEXT_TRACE = "trace" LIST_MTEXT = [MTEXT_FATAL, MTEXT_ERROR, MTEXT_WARN, MTEXT_MSG, MTEXT_INFO, MTEXT_DEBUG, MTEXT_TRACE, MTEXT_TRACE] LIMIT_FATAL = 2 LIMIT_ERROR = 4 LIMIT_WARN = 6 LIMIT_MSG = 8 LIMIT_INFO = 10 LIMIT_DEBUG = 12 LIMIT_TRACE = 14 LIMIT_XTRACE = 16 RC_FATAL = 3 RC_ERROR = 2 RC_WARN = 1 RC_MSG = 0 RC_INFO = 0 class TempMessage: """ simple implementation just to print first messages into temporary debugfile in order to get a message-object before the job information are set """ def __init__(self, job, logTime): # (self, componente, out, level): self.job = job self.level = LIMIT_DEBUG self.openDebug(job, logTime) def openDebug(self, job, logTime): path = os.path.join(B.HOME_PATH, "temp") if not os.path.exists(path): os.mkdir(path) self.debugpath = os.path.join(path, "debug_"+logTime+".txt") self.debugfile = open(self.debugpath, "w") def getLogLevel(self, tool="", comp=None): return 0 def logFatal(self, text): self.debug(LIMIT_FATAL, "FATAL: " + text) def logError(self, text): self.debug(LIMIT_ERROR, "ERROR: " + text) def setError(self, text): self.debug(LIMIT_ERROR, "ERROR: " + text) def logWarn(self, text): self.debug(LIMIT_WARN, "WARN: " + text) def logMsg(self, text): self.debug(LIMIT_MSG, "MSG: " + text) def logInfo(self, text): self.debug(LIMIT_INFO, text) def logDebug(self, prio, text=""): mprio = LIMIT_DEBUG mtext = str(prio) if len(text) > 1: mtext = text if isinstance(prio, int): mprio = int(prio) self.debug(mprio, mtext) def logTrace(self, prio, text): pass def logXTrace(self, prio, text): pass def debug(self, prio, text): """ eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt""" try: if prio <= self.level: self.debugfile.write(text + "\n") except: print("debug closed "+text) def closeMessage(self) : self.debug(LIMIT_INFO, "closeMessage ------------------------------------------- \n") self.debugfile.close() class Message: """ Ausgaben erfolgen prioritaeten-gesteuert anhand * Typ (fatal..trace) * Einstellung (a) ueber Parameter ODER (b) in Funktion Im Funktionskopf wird Einstellung gesetzt, z.B. verify=job.getDebugLevel (ggf keine Debug-Ausgabe) bzw. verify=job.getMessageLevel-1 (eingeschaltete Debug-Ausgabe) "fatal": "3", # Abbruchfehlker, wird immer in debug und log ausgegeben, setzt RC "error": "2", # Fehler, wird immer in debug und log ausgegeben, setzt RC "warn": "1", # Warnung, wird immer in debug und log ausgegeben, setzt RC "msg": "0", # Ergebnis, wird immer in debug und log ausgegeben, setzt RC "info": "0", # Info, wird ggf. in debug und log ausgegeben, setzt RC "debug": "0", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht "trace": "0", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht """ def __init__(self, job, level, logTime, compName=None): # (self, compName, out, level): self.job = job self.compName = compName # dezentrales Logsystem verify = LIMIT_DEBUG self.rc = RC_INFO if isinstance(level, str): i = 1 for l in LIST_MTEXT: if level.lower() == l: break i += 1 level = i * 2 if (level == 0): self.level = LIMIT_DEBUG else: self.level = level self.openDebug(job, logTime, compName) self.openLog(job, logTime, compName) self.topmessage = "" def openDebug(self, job, logTime, compName): path = job.conf[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG] if not os.path.exists(path): os.mkdir(path) logTime = logTime[0:11] + "0000" if compName is None: if hasattr(job.m, "debugpath"): self.debugpath = job.m.debugpath self.debugfile = job.m.debugfile return else: self.debugpath = os.path.join(path, "debug_"+logTime+".txt") else: self.debugpath = os.path.join(path, "debug_" + compName + "_" + logTime + ".txt") if os.path.exists(self.debugpath): self.debugfile = open(self.debugpath, "a") else: self.debugfile = open(self.debugpath, "w") def openLog(self, job, logTime, compName): if not hasattr(job, "par"): return pathPattern = job.programDef["logpath"] path = tools.path_tool.compose_path(job, pathPattern, None) parent = os.path.dirname(path) if not os.path.exists(parent): os.makedirs(parent) if compName is None: self.logpath = path self.logfile = open(path, "w") else: self.messages = [] def getErrortyp(self, prio): if prio <= LIMIT_FATAL: return "FATAL" elif prio <= LIMIT_ERROR: return "ERROR" elif prio <= LIMIT_WARN: return "WARN" elif prio <= LIMIT_MSG: return "MESSAGE" elif prio <= LIMIT_INFO: return "INFO" elif prio <= LIMIT_DEBUG: return "DEBUG" elif prio <= LIMIT_TRACE: return "TRACE" else: return "NDEF" def closeMessage(self) : self.debug(LIMIT_INFO, "closeMessage ------------------------------------------- \n") self.logfile.close() self.debugfile.close() def getLogLevel(self, tool="", comp=None): """ gets the increasing level depending on tool and component if these arguments matches with a job-parameter the level decreases :param tool: :param comp: :return: """ out = 0 job = self.job if not hasattr(job, "par"): return out if comp is not None and hasattr(job.par, "component") and comp.name in getattr(job.par, "component"): out += 2 if tool != "" and hasattr(job.par, "tool") and tool in getattr(job.par, "tool"): out += 2 return out def setRc(self, rc, text): if (int(rc) > self.rc) or (self.topmessage == ""): self.rc = rc i = rc * -1 + 3 if i < 0: return self.topmessage = LIST_MTEXT[i].upper() + ": " + text def isRc(self, returnCode): """ prueft, ob der gesetzte ReturnCode mit dem angefragten ReturnCode vereinbar ist Beispiel: gesetzt WARN, angefragt ERROR ist OK=True, anders herum: KO=False :param returnCode: :return: """ if isinstance(returnCode, int): rcId = returnCode else: rcId = ( LIST_MTEXT.index(returnCode.lower()) - 3 ) * (-1) if self.rc >= int(rcId) or rcId < 0: return True else: return False def getFinalReturncode(self): RETURN_TEXT = ["OK", "WARN", "ERROR", "FATAL"] return RETURN_TEXT[self.rc] def getFinalRc(self): RETURN_TEXT = ["OK", "WARN", "ERROR", "FATAL"] return int(self.rc) def setFatal(self, text): """ Routine zum Setzen des RC und gleichzeitigem Schreiben des Logs """ self.setRc(RC_FATAL, text) self.logFatal(text) def setError(self, text): """ Routine zum Setzen des RC und gleichzeitigem Schreiben des Logs """ self.setRc(RC_ERROR, text) self.logError(text) def setWarn(self, text): """ Routine zum Setzen des RC und gleichzeitigem Schreiben des Logs """ self.setRc(RC_WARN, text) self.logWarn(text) def setMsg(self, text): """ Routine zum Setzen des RC und gleichzeitigem Schreiben des Logs """ self.setRc(RC_MSG, text) self.logMsg(text) def getMessageText(self, job, text, args): return text def logFatal(self, prio, text=""): """ it logs a fatal error in logfile and debugfile - please use setFatal() in order to set the return-code FATAL means, the program can not be finished stable :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.log(LIMIT_FATAL, prio, "FATAL: " + text) self.debug(LIMIT_FATAL, prio, "FATAL: " + text) def logError(self, prio, text=""): """ it logs an error in logfile and debugfile - please use setError() in order to set the return-code ERROR means, the program can be finished incorrect but stable :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.log(LIMIT_ERROR, prio, "ERROR: " + text) self.debug(LIMIT_ERROR, prio, "ERROR: " + text) def logWarn(self, prio, text=""): """ it logs a warning in logfile and debugfile - please use setWarn() in order to set the return-code WARNING means, the program can be finished correct and stable but with points to check - especially founded business faults :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.log(LIMIT_WARN, prio, "WARN: " + text) self.debug(LIMIT_WARN, prio, "WARN: " + text) def logMsg(self, prio, text=""): """ it logs a message in logfile and debugfile - please use setMsg() in order to set the return-code MESSAGE means, the program can be finished without any points to check it manually afterwards in different to INFO it logs a working-result :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.log(LIMIT_MSG, prio, "MSG: " + text) self.debug(LIMIT_MSG, prio, "MSG: " + text) def logInfo(self, prio, text=""): """ it logs a message into logfile and debugfile without setting the return-code INFO means, the program can be finished without any points to check it manually afterwards in different to MESSAGE it logs just a working-step relating to the test-application :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.log(LIMIT_INFO, prio, text) self.debug(LIMIT_INFO, prio, text) def logDebug(self, prio, text=""): """ it logs a message into the debugfile without setting the return-code DEBUG means a working-step in different to INFO it logs a working-step without any relevance to the test-application :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.debug(LIMIT_DEBUG, prio, text) def logTrace(self, prio, text=""): """ it logs a message into the debugfile without setting the return-code TRACE means a working-step with some relevant data in different to DEBUG it logs a working-step with some relevant controlling-data :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.debug(LIMIT_TRACE, prio, text) def logXTrace(self, prio, text=""): """ it logs a message into the debugfile without setting the return-code XTRACE means a working-step with a lot of data in different to TRACE it logs not only the known relevant controlling-data :param prio: optional int [-2..+2] , it increases / decreases the necessary log-level in relation to the parametrized level :param text: must, if it is not set then get text from prio :return: """ self.debug(LIMIT_XTRACE, prio, text) def getLoggingArgs(self, mlevel, prio, text): verify = self.getLogLevel("msg_tool") if verify: self.debugfile.write(verify, "getLoggingArgs: " + str(mlevel)+ ", "+self.format2Str(prio)+", "+self.format2Str(text)) out = {} prefix = "" if isinstance(mlevel, int): out["mlevel"] = mlevel else: raise Exception("argument mlevel is not int "+str(mlevel)) if len(text) < 1: out["mprio"] = 0 if isinstance(prio, int): txt = str(prio) elif isinstance(prio, dict): txt = self.formatDict2Str(prio) elif isinstance(prio, str): txt = prio elif isinstance(prio, list): txt = self.formatList2Str(prio) else: txt = str(prio) else: if isinstance(prio, int): out["mprio"] = prio else: out["mprio"] = 0 prefix = text text = prio if isinstance(text, dict): txt = self.formatDict2Str(text) elif isinstance(text, str): txt = text elif isinstance(text, list): txt = self.formatList2Str(text) else: txt = str(text) if len(prefix) > 1: if ":" in prefix: out["mtext"] = prefix.strip() + " " + txt.strip() else: out["mtext"] = prefix.strip() + ": " + txt.strip() else: out["mtext"] = txt return out def format2Str(self, elem): if isinstance(elem, dict): return self.formatDict2Str(elem) elif isinstance(elem, str): return elem elif isinstance(elem, list): return self.formatList2Str(elem) else: return str(elem) def formatDict2Str(self, args): txt = "{" for k in args: if isinstance(args[k], dict): txt += k + ": " + self.formatDict2Str(args[k]) + ", " elif isinstance(args[k], str): txt += k + ": " + args[k] + ", " elif isinstance(args[k], list): txt += k + ": " + self.formatList2Str(args[k]) + ", " else: txt += k + ": " + str(args[k]) + ", " return txt[0:-2] + "}" def formatList2Str(self, args): txt = "[" for k in args: if isinstance(k, dict): txt += k + ": {" + self.formatDict2Str(k) + "}, " elif isinstance(k, str): txt += k + ", " elif isinstance(k, list): txt += self.formatList2Str(k) + ", " else: txt += str(k) + ", " return txt[0:-2] + "]" def log(self, mlevel, prio, text): args = self.getLoggingArgs(mlevel, prio, text) """ eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt""" if (args["mlevel"] + args["mprio"] > int(self.level)): return elif (self.compName is None): # and self.logfile.closed == False: try: self.logfile.write(args["mtext"] + "\n") except: pass else: self.messages.append(text) def debug(self, mlevel, prio, text=""): verify = self.getLogLevel("msg_tool") args = self.getLoggingArgs(mlevel, prio, text) if verify: self.debugfile.write("m.debug "+self.format2Str(args)+" >? "+str(self.level)+"\n") if (args["mlevel"] - args["mprio"] > int(self.level)): return if (args["mprio"] + 20) % 2 == 1: print(args["mtext"]) try: self.debugfile.write(args["mtext"] + "\n") except: raise Exception("debugfile closed: "+args["mtext"]) def resetLog(self): self.messages = [] def merge(self, submsg): self.setRc(submsg.getFinalRc(), submsg.topmessage) text = "\n".join(submsg.messages) self.logInfo("\n"+text+"\n")