Data-Test-Executer Framework speziell zum Test von Datenverarbeitungen mit Datengenerierung, Systemvorbereitungen, Einspielungen, ganzheitlicher diversifizierender Vergleich
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

492 lines
18 KiB

# 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.TOPIC_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("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")