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.

407 lines
15 KiB

3 years ago
# 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)
"""
3 years ago
import basic.program
3 years ago
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
3 years ago
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]
LIMIT_FATAL = 2
3 years ago
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 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()
3 years ago
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.getDebugLevel-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
3 years ago
"""
2 years ago
def __init__(self, job, level, logTime, componente):
3 years ago
# (self, componente, out, level):
self.job = job
3 years ago
self.componente = componente # dezantrales 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
3 years ago
if (level == 0):
self.level = LIMIT_DEBUG
else:
self.level = level
self.openDebug(job, logTime, componente)
self.openLog(job, logTime, componente)
3 years ago
self.topmessage = ""
print("message initialisiert mit level " + str(self.level))
2 years ago
def openDebug(self, job, logTime, componente):
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 componente 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_" + componente.name + "_" + 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, componente):
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 componente is None:
self.logpath = path
self.logfile = open(path, "w")
else:
self.messages = []
3 years ago
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 setRc(self, rc, text):
job = self.job #basic.program.Job.getInstance()
3 years ago
if (int(rc) > self.rc):
self.rc = rc
# TODO quickfix
self.topmessage = ": " + text
3 years ago
elif (int(rc) == self.rc):
self.topmessage = ": " + text
3 years ago
def isRc(self, rc):
rcId = 0 # int(int(self.CONST_ERRTYP[rc]) / 4 - RC_OFFSET)
3 years ago
print("< < < isRc " + str(self.rc) + " <=? " + str(rcId))
if self.rc <= int(rcId):
print("return True")
return True
else:
print("return False")
return False
def getFinalRc(self):
return int(int(self.rc))
3 years ago
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)
3 years ago
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)
3 years ago
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)
3 years ago
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)
3 years ago
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)
3 years ago
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)
3 years ago
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)
3 years ago
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):
out = {}
if isinstance(mlevel, int):
out["mlevel"] = mlevel
else:
raise Exception("argument mlevel is not int "+str(mlevel))
a = text.split(":")
cat = ""
txt = ""
if len(a[0]) > 1 and a[0].lower() in LIST_MTEXT:
cat = a[0]
if len(a) > 1 and len(a[1]) > 1:
txt = a[1]
elif len(a[0]) > 1 and a[0].lower() not in LIST_MTEXT:
txt = a[0]
if isinstance(prio, int) and len(txt) > 1:
out["mprio"] = prio
elif len(txt) < 1 and isinstance(prio, str):
txt = prio
out["mprio"] = 0
if len(txt) < 1:
raise Exception("argument text is not fount " + str(mlevel) + ", " + str(prio) + ", " + str(text))
if len(cat) > 1:
out["mtext"] = cat + ": " + txt
else:
out["mtext"] = txt
return out
3 years ago
def log(self, mlevel, prio, text):
args = self.getLoggingArgs(mlevel, prio, text)
3 years ago
""" eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt"""
if (args["mlevel"] + args["mprio"] > int(self.level)):
return
elif (self.componente is None): # and self.logfile.closed == False:
try:
self.logfile.write(args["mtext"] + "\n")
except:
pass
3 years ago
else:
self.messages.append(text)
3 years ago
def debug(self, mlevel, prio, text=""):
args = self.getLoggingArgs(mlevel, prio, text)
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"])
3 years ago
def resetLog(self):
self.messages = []
3 years ago
def merge(self, submsg):
self.setRc(submsg.getFinalRc(), submsg.topmessage)
text = "\n".join(submsg.messages)
self.logInfo("\n"+text+"\n")