Testskripter
/
datestex
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

refactor job and message

refactor
Ulrich 2 years ago
parent
a66f1b068a
commit
4a3ae36410
11 changed files with 780 additions and 552 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 391
      basic/message.py
  2. 408
      basic/program.py
  3. 2
      objects/catalog.py
  4. 21
      test/constants.py
  5. 3
      test/test_10job.py
  6. 165
      test/test_11message.py
  7. 62
      tools/config_tool.py
  8. 10
      tools/file_abstract.py
  9. 19
      tools/path_const.py
  10. 130
      tools/path_tool.py
  11. 109
      tools/value_tool.py

391
basic/message.py
View File

@ -25,22 +25,93 @@ import tools.path_tool
#import basic.text_const as T
import basic.constants as B
LIMIT_FATAL = 0
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
LIMIT_ERROR = 4
LIMIT_WARN = 8
LIMIT_MSG = 12
LIMIT_INFO = 16
LIMIT_DEBUG = 20
LIMIT_TRACE = 24
RC_OFFSET = 4
RC_FATAL = 8
RC_ERROR = 6
RC_WARN = 5
RC_MSG = 4
RC_INFO = 3
RC_DEBUG = 2
RC_TRACE = 1
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()
class Message:
"""
@ -48,92 +119,70 @@ class Message:
* 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": "4", # Abbruchfehlker, wird immer in debug und log ausgegeben, setzt RC
"error": "8", # Fehler, wird immer in debug und log ausgegeben, setzt RC
"warn": "12", # Warnung, wird immer in debug und log ausgegeben, setzt RC
"msg": "16", # Ergebnis, wird immer in debug und log ausgegeben, setzt RC
"info": "20", # Info, wird ggf. in debug und log ausgegeben, setzt RC
"debug": "24", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht
"trace": "28", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht
"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, componente):
# (self, componente, out, level):
self.job = job
self.componente = componente # dezantrales Logsystem
verify = LIMIT_DEBUG
self.initErrorTyp()
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
# init debugfile - except for each component
if componente is not None: # use general debugfile
self.openDebug(job, logTime, componente)
self.openLog(job, logTime, componente)
self.topmessage = ""
print("message initialisiert mit level " + str(self.level))
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
self.debug(verify, "> > > debugfile uebernommen zu " + str(componente))
return
else:
debugpath = job.conf["paths"]["debugs"] + "/debug_" + logTime[0:-4] + "00.txt"
print ("debugpathx "+debugpath)
if os.path.exists(debugpath):
self.debugfile = open(debugpath, "a")
self.debugpath = os.path.join(path, "debug_"+logTime+".txt")
else:
self.debugfile = open(debugpath, "w")
self.debug(verify, "> > > debugfile geoeffnet zu " + job.program + " mit " + debugpath)
# init logfile - except for components or unittest
#self.logDebug("logfile " + str(componente) + ", " + str(job.par.basedir))
if componente is not None: #
self.logfile = self.debugfile
elif job.program == "unit":
self.logfile = self.debugfile
elif hasattr(job.par, B.PAR_MODUS) and getattr(job.par, B.PAR_MODUS) == "unit":
self.logfile = self.debugfile
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.setLogdir(job, logTime)
self.topmessage = ""
self.messages = []
print("message initialisiert mit level " + str(self.level))
self.debugfile = open(self.debugpath, "w")
def setLogdir(self, job, logTime):
basedir = job.par.basedir
basedir = basedir.replace("base", "log")
os.makedirs(basedir, exist_ok=True)
# basedir = tools.path_tool.composePath(basedir, None)
# basedir = tools.path_tool.composePath(job, basedir, None)
logpath = job.getLogpath()
basedir = os.path.dirname(logpath)
os.makedirs(basedir, exist_ok=True)
# logpath = os.path.join(basedir , job.program + "_" + logTime + ".txt")
self.logDebug("logfile " + logpath)
if os.path.exists(logpath):
self.logfile = open(logpath, "a")
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.logfile = open(logpath, "w")
def initErrorTyp(self):
self.CONST_ERRTYP = {
"fatal": "4", # wird immer in debug und log ausgegeben, setzt RC
"error": "8", # wird immer in debug und log ausgegeben, setzt RC
"warn": "12", # wird immer in debug und log ausgegeben, setzt RC
"msg": "16", # wird immer in debug und log ausgegeben, setzt RC
"info": "20", # wird immer in debug und log ausgegeben, setzt RC
"debug": "24", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht
"trace": "28", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht
"rc1": "fatal",
"rc2": "error",
"rc3": "warn",
"rc4": "info",
"rc5": "debug",
"rc6": "trace",
"rc8": "fatal",
"1": "fatal",
"2": "error",
"3": "warn",
"4": "info",
"5": "debug",
"6": "trace"
}
self.messages = []
def getErrortyp(self, prio):
if prio <= LIMIT_FATAL:
@ -160,16 +209,15 @@ class Message:
def setRc(self, rc, text):
job = self.job #basic.program.Job.getInstance()
verify = -0+LIMIT_DEBUG
self.debug(verify, "setRc " + str(rc) + " " + str(self.rc)+ "\n")
if (int(rc) > self.rc):
self.rc = rc
self.topmessage = self.CONST_ERRTYP["rc"+str(rc)].upper() + ": " + text
# TODO quickfix
self.topmessage = ": " + text
elif (int(rc) == self.rc):
self.topmessage = self.CONST_ERRTYP["rc"+str(rc)].upper() + ": " + text
self.topmessage = ": " + text
def isRc(self, rc):
rcId = int(int(self.CONST_ERRTYP[rc]) / 4 - RC_OFFSET)
rcId = 0 # int(int(self.CONST_ERRTYP[rc]) / 4 - RC_OFFSET)
print("< < < isRc " + str(self.rc) + " <=? " + str(rcId))
if self.rc <= int(rcId):
print("return True")
@ -179,10 +227,7 @@ class Message:
return False
def getFinalRc(self):
if (self.rc <= RC_OFFSET):
return 0
else:
return int(int(self.rc) - RC_OFFSET)
return int(int(self.rc))
def setFatal(self, text):
""" Routine zum Setzen des RC und gleichzeitigem Schreiben des Logs """
@ -202,54 +247,156 @@ class Message:
def setMsg(self, text):
""" Routine zum Setzen des RC und gleichzeitigem Schreiben des Logs """
self.setRc(RC_MSG, text)
self.logInfo(text)
self.logMsg(text)
def getMessageText(self, job, text, args):
return text
def logFatal(self, text):
self.log(LIMIT_FATAL, "FATAL: " + text)
self.debug(LIMIT_FATAL, "FATAL: " + 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, text):
self.log(LIMIT_ERROR, "ERROR: " + text)
self.debug(LIMIT_ERROR, "ERROR: " + 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, text):
self.log(LIMIT_WARN, "WARN: " + 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, text):
self.log(LIMIT_MSG, text)
self.log(LIMIT_MSG, "MSG: " + 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, text):
self.log(LIMIT_INFO, 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, text):
self.debug(LIMIT_DEBUG, 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, text):
self.debug(LIMIT_TRACE, 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 log(self, 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
def log(self, mlevel, prio, text):
args = self.getLoggingArgs(mlevel, prio, text)
""" eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt"""
if (int(prio) <= int(self.level)) and (self.componente is None): # and self.logfile.closed == False:
if (args["mlevel"] + args["mprio"] > int(self.level)):
return
elif (self.componente is None): # and self.logfile.closed == False:
try:
self.logfile.write(text + "\n")
self.logfile.write(args["mtext"] + "\n")
except:
pass
elif (int(prio) <= int(self.level)):
self.messages.append(text)
else:
self.debug(prio, self.getErrortyp(prio) + ": " + text)
self.messages.append(text)
def debug(self, prio, text):
""" eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt"""
if (int(prio) < int(self.level)+1) : #and self.debugfile.closed == False:
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(text + "\n")
self.debugfile.write(args["mtext"] + "\n")
except:
print("debug closed "+text)
raise Exception("debugfile closed: "+args["mtext"])
def resetLog(self):
self.messages = []

408
basic/program.py
View File

@ -9,12 +9,16 @@
#import sys, getopt
import argparse
import copy
import time
import yaml
import os
from datetime import datetime
import basic.constants as B
import basic.message
import basic.componentHandling
import objects.catalog
import tools.value_tool
import tools.date_tool
import tools.path_tool
import tools.file_tool
@ -26,128 +30,13 @@ import tools.job_tool
LIMIT_INFO = 16 #basic.message.LIMIT_INFO
LIMIT_DEBUG = 12 #basic.message.LIMIT_DEBUG
jobdef = {
"webflask": {
"pardef": "",
"pfilesource": "",
"pfiletarget": "",
"basedir": "workbase",
"dirname": "workdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.conf.data}/workspace/webflask_{job.start:H}.txt" },
"declare_result": {
"pardef": "",
"pfilesource": "",
"pfiletarget": "",
"basedir": "workbase",
"dirname": "workdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.conf.data}/workspace/reorg_{job.start:H}.txt" },
"service": {
"pardef": "",
"pfilesource": "",
"pfiletarget": "",
"basedir": "workbase",
"dirname": "workdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.conf.data}/workspace/service_{job.start:H}.txt" },
"unit": {
"pardef": "",
"pfilesource": "",
"pfiletarget": "",
"basedir": "workbase",
"dirname": "workdir",
"loglevel": LIMIT_DEBUG,
"logpath": "{job.conf.data}/workspace/unittest_{job.start:H}.txt" },
"check_environment": {
"pardef": "",
"pfilesource": "",
"pfiletarget": "envparfile",
"basedir": "envbase",
"dirname": "envdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.envdir}/{log}/log_{job.start}.txt" },
"test_executer": {
"pardef": "",
"pfilesource": "tsparfile",
"pfiletarget": "tsparfile",
"basedir": "tsbase",
"dirname": "tsdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tsdir}/{log}/{job.program}_{job.start}.txt" },
"init_testsuite": {
"pardef": "tsdir", # ,tdtyp,tdsrc,tdname",
"pfilesource": "envparfile",
"pfiletarget": "tsparfile",
"basedir": "tsbase",
"dirname": "tsdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tsdir}/{log}/{job.program}_{job.start}.txt" },
"init_testcase": {
"pardef": "tcdir", # ",tdtyp,tdsrc,tdname",
"pfilesource": "envparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt" },
"execute_testcase": {
"pardef": "tcdir", # ",tdtyp,tdsrc,tdname",
"pfilesource": "tcparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt" },
"collect_testcase": {
"pardef": "tcdir", # ",tdtyp,tdsrc,tdname",
"pfilesource": "tcparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt" },
"copy_appdummy": {
"pardef": "tcdir", # ",tdtyp,tdsrc,tdname",
"pfilesource": "tcparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt"},
"compare_testcase": {
"pardef": "tcdir", # ",tdtyp,tdsrc,tdname",
"pfilesource": "tcparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt" },
"test_system": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tsparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt"},
"finish_testcase": {
"pardef": "tcdir",
"pfilesource": "tcparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt"},
"finish_testsuite": {
"pardef": "tsdir",
"pfilesource": "tsparfile",
"pfiletarget": "tssarfile",
"basedir": "tsbase",
"dirname": "tsdir",
"loglevel": LIMIT_INFO,
"logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt"}
}
CTLG_NAME = "programs"
CTLG_PARDEF = "pardef"
CTLG_PARSOURCE = "pfilesource"
CTLG_PARTARGET = "pfiletarget"
CTLG_BASEDIR = "basedir"
CTLG_LOGPATH = "logpath"
CTLG_LOGLEVEL = "loglevel"
def setGlobal():
@ -168,7 +57,7 @@ DEFAULT_TIME = "2022-08-29_17-29-59"
def createJob(pprg="", pgran="", papp="", penv="", ptstamp="", pmode=""):
"""
this creates a Job-Object with the main arguments.
this creates a Job-Object with the main arguments ## same as in testtools?
:param pprg: program-name
:param pgran: tc|ts|
:param papp: application
@ -213,7 +102,7 @@ def createJob(pprg="", pgran="", papp="", penv="", ptstamp="", pmode=""):
path = T.DATA_PATH + "/workspace/"
job = basic.program.Job(prgname)
# job.conf.confs[B.SUBJECT_PATH]["components"] = T.COMP_PATH
# job.conf[B.SUBJECT_PATH]["components"] = T.COMP_PATH
args = {"application": app, "environment": env, "modus": mode, gran + "time": tstamp,
gran + "dir": path,
"step": 1}
@ -222,13 +111,65 @@ def createJob(pprg="", pgran="", papp="", penv="", ptstamp="", pmode=""):
job.par.setParameterArgs(job, args)
return job
class SimpleJob:
"""
the simple job is just used for test issues
"""
__logtime = "20200101_000000"
def __init__(self, program, username="", args=None):
self.program = program
self.username = username
path = tools.path_tool.getBasisConfigPath()
self.conf = getConfiguration(self, path)
self.jobid = str(100000)
catalog = objects.catalog.Catalog.getInstance()
self.programDef = catalog.getValue(self, CTLG_NAME, program, "")
if args is not None:
if "par" in args:
self.par = Parameter(self, args["par"])
for k in args:
if k == "par":
continue
setattr(self, k, args[k])
else:
self.par == Parameter(self, None)
if not hasattr(self, "start"):
logTime = tools.date_tool.getActdate(tools.date_tool.F_LOG)
while logTime <= SimpleJob.__logtime:
time.sleep(1)
logTime = tools.date_tool.getActdate(tools.date_tool.F_LOG)
self.start = logTime
def getDebugLevel(self, arg):
return 12
def debug(self, prio, text):
pass
def getParameter(self, parameter):
if hasattr(self.par, parameter) and getattr(self.par, parameter) is not None:
return getattr(self.par, parameter)
else:
val = tools.value_tool.compose_pattern(self, parameter, None)
if val is None:
self.m.logError("Parameter "+parameter+" nicht in job.par ")
return
setattr(self.par, parameter, val)
return val
class SimpleParameter():
def __init__(self, job, args=None):
self.program = job.program
class Job:
__instance = None
__instances = []
__jobid = 100000
__logtime = "20200101_000000"
#catalog = objects.catalog.Catalog.getInstance()
def __init__ (self, program, args=None):
def __init__ (self, program, username="", args=None):
"""
initializing the job-object as a global structure for the actual doing
type of job:
@ -250,39 +191,72 @@ class Job:
# - job.par with parameter-args or cli-args
# - job.msg
print("################# init Job ## " + program + " #################")
# logtime muss unique sein denn logDateien und Verzeichnisse muessen eindeutig sein
logTime = tools.date_tool.getActdate(tools.date_tool.F_LOG)
while logTime <= Job.__logtime:
time.sleep(1)
logTime = tools.date_tool.getActdate(tools.date_tool.F_LOG)
self.start = logTime
self.m = basic.message.TempMessage(self, logTime)
Job.__jobid += 1
self.jobid = str(Job.__jobid)
if len(program) < 3:
print("FATAL: programname is missing")
exit(3)
self.program = program
conf = Configuration(self, program)
self.conf = conf.confs
if len(username) < 2:
self.username = os.getlogin()
path = tools.path_tool.getBasisConfigPath()
self.conf = getConfiguration(self, path)
catalog = objects.catalog.Catalog.getInstance()
self.programDef = catalog.getValue(self, CTLG_NAME, program, "")
try:
path = tools.config_tool.getConfigPath(self, P.KEY_BASIC, B.BASIS_FILE)
print("comps.basispath "+path)
self.conf.setConfiguration(self, path)
# conf = self.readConfiguration(path)
self.conf = self.getConfiguration(path)
except:
pass # the special path is not necessary
logTime = tools.date_tool.getActdate(tools.date_tool.F_LOG)
self.start = logTime
self.m = basic.message.Message(self, jobdef[program]["loglevel"], logTime, None)
print("FATAL: config-file could not be loaded")
exit(3)
if args is not None:
self.setParameter(args)
self.m = basic.message.Message(self, self.programDef[CTLG_LOGLEVEL], logTime, None)
def altinit(self, program, args):
appl = tools.config_tool.getConfig(self, P.KEY_BASIC, B.SUBJECT_APPS)
if appl is not None:
self.conf.confs[B.SUBJECT_APPS] = appl[B.SUBJECT_APPS]
if B.SUBJECT_PROJECTS in self.conf.confs:
for k in self.conf.confs[B.SUBJECT_PROJECTS]:
self.conf[B.SUBJECT_APPS] = appl[B.SUBJECT_APPS]
if B.SUBJECT_PROJECTS in self.conf:
for k in self.conf[B.SUBJECT_PROJECTS]:
if k not in appl[B.SUBJECT_PROJECTS]:
raise Exception("Workspace has project "+k+" which is not configured")
self.conf.confs[B.SUBJECT_PROJECTS][k] = appl[B.SUBJECT_PROJECTS][k]
self.conf[B.SUBJECT_PROJECTS][k] = appl[B.SUBJECT_PROJECTS][k]
else:
self.conf.confs[B.SUBJECT_PROJECTS] = appl[B.SUBJECT_PROJECTS]
self.conf[B.SUBJECT_PROJECTS] = appl[B.SUBJECT_PROJECTS]
par = Parameter(self, program, args)
self.par = par
def getLogpath(self):
path = tools.path_tool.composePattern(self, jobdef[self.program]["logpath"], None)
path = tools.path_tool.compose_path(self, self.programDef[CTLG_LOGPATH], None)
return path
def getConfiguration(self, path):
conf = {}
conf["configpath"] = []
if hasattr(self, "conf"):
conf = self.conf
conf["configpath"].append(path)
doc = tools.file_tool.readFileDict(self, path, None)
if "basic" in doc:
for k, v in doc["basic"].items():
if k not in conf:
conf[k] = v
else:
for k, v in doc.items():
if k not in conf:
conf[k] = v
return conf
def murks(self, par, program):
self.par = par
dirpath = self.par.getDirParameter()
@ -298,18 +272,18 @@ class Job:
def setProgram(self, program):
self.program = program
basedir = jobdef[program]["basedir"]
basedir = self.programDef[CTLG_BASEDIR]
self.basedir = basedir
if (self.par is not None) and self.par in B.LIST_MAIN_PAR:
setattr(self.par, "program", program)
setattr(self.par, "basedir", basedir)
parstring = getattr(self.par, "parstring")
setattr(self.par, B.PAR_PROGRAM, program)
setattr(self.par, B.PAR_BASEDIR, basedir)
parstring = getattr(self.par, B.PAR_STRING)
parstring = parstring[parstring.find("--"):]
parstring = "python "+program+" "+parstring
setattr(self.par, "parstring", parstring)
if not hasattr(self.par, jobdef[program]["dirname"]):
setattr(self.par, jobdef[program]["dirname"],
tools.path_tool.composePattern(self, "{"+basedir+"}", None))
if not hasattr(self.par, self.programDef[B.PAR_DIRNAME]):
setattr(self.par, self.programDef[B.PAR_DIRNAME],
tools.path_tool.compose_path(self, "{"+basedir+"}", None))
print(parstring)
self.par.setParameterLoaded(self)
@ -349,9 +323,9 @@ class Job:
def dumpParameter(self):
if len(jobdef[self.program]["pfiletarget"]) < 2:
if len(self.programDef[CTLG_PARTARGET]) < 2:
return
parpath = tools.path_tool.composePath(self, jobdef[self.program]["pfiletarget"], None)
parpath = tools.path_tool.compose_path(self, self.programDef[CTLG_PARTARGET], None)
output = {}
cconf = basic.componentHandling.getComponentDict()
output["par"] = self.par.__dict__
@ -372,9 +346,9 @@ class Job:
def loadParameter(self):
output = {}
if len(str(jobdef[self.program]["pfilesource"])) < 2:
if len(str(self.programDef[CTLG_PARSOURCE])) < 2:
return None
parpath = tools.path_tool.composePath(self, jobdef[self.program]["pfilesource"], None)
parpath = tools.path_tool.compose_path(self, self.programDef[CTLG_PARSOURCE], None)
print("parpath "+parpath)
if not os.path.join(parpath):
return None
@ -385,17 +359,23 @@ class Job:
return output
def setParameter(self, args):
self.par = Parameter(self, args)
def getParameter(self, parameter):
if hasattr(self.par, parameter):
if hasattr(self.par, parameter) and getattr(self.par, parameter) is not None:
return getattr(self.par, parameter)
elif "xxxtime" in parameter:
neu = tools.date_tool.getActdate(tools.date_tool.F_DIR)
# setattr(self.par, parameter, neu)
return neu
else:
print("Parameter "+parameter+" nicht in job.par ")
for attr, value in self.par.__dict__.items():
print(attr, value)
val = tools.value_tool.compose_pattern(self, parameter, None)
if val is None:
self.m.logError("Parameter "+parameter+" nicht in job.par ")
return
setattr(self.par, parameter, val)
return val
def hasElement(self, parameter, elem):
@ -440,11 +420,12 @@ class Job:
return basic.message.LIMIT_DEBUG
elif elem.find("tool") > 1 and hasattr(self, "par"):
if not hasattr(self.par, "tool") or getattr(self.par, "tool").find(elem) <= 0:
return int(self.m.CONST_ERRTYP[errtyp]) -1
return 4
else:
return int(self.m.CONST_ERRTYP[errtyp])
return 4
else:
return int(self.m.CONST_ERRTYP[errtyp])
# TODO quickfix
return 4
def getInfoLevel(self, elem):
return self.getMessageLevel("info", elem)
def getDebugLevel(self, elem):
@ -458,6 +439,24 @@ class Job:
else:
print(text)
def getConfiguration(job, path):
conf = {}
conf["configpath"] = []
if hasattr(job, "conf"):
conf = job.conf
conf["configpath"].append(path)
doc = tools.file_tool.readFileDict(job, path, None)
if "basic" in doc:
for k, v in doc["basic"].items():
if k not in conf:
conf[k] = v
else:
for k, v in doc.items():
if k not in conf:
conf[k] = v
return conf
# ------------------------------------------------------------------------------------------------------------------
class Parameter:
"""
@ -468,45 +467,32 @@ class Parameter:
comparison with a parameter-file from the work-folder
"""
print ("class Parameter")
def __init__ (self, job, program, args=None):
print ("# init Parameter for " + program)
self.program = program
def __init__ (self, job, args=None):
print ("# init Parameter for " + job.program)
self.program = job.program
if args is not None and isinstance(args, dict):
self.setParameterArgs(job, args)
else:
self.setParameter(job)
self.basedir = jobdef[program]["basedir"]
job.basedir = jobdef[program]["basedir"]
self.basedir = job.programDef[B.PAR_BASEDIR]
job.basedir = job.programDef[B.PAR_BASEDIR]
print (f"# Parameter initialisiert {self.program} mit basedir {self.basedir}")
job.par = self
if not hasattr(self, jobdef[program]["dirname"]):
setattr(self, jobdef[program]["dirname"],
tools.path_tool.composePattern(job, "{"+self.basedir+"}", None))
# Parameter aus Verzeichnis extraieren bzw. Verzeichnis aus Parameter erstellen
if not hasattr(self, job.programDef[B.PAR_DIRNAME]):
setattr(self, job.programDef[B.PAR_DIRNAME],
tools.value_tool.compose_pattern(job, self.basedir, None))
else:
tools.path_tool.extractPath(job, jobdef[program]["basedir"], getattr(self, jobdef[program]["dirname"]))
tools.path_tool.extractPath(job, job.programDef[B.PAR_BASEDIR],
getattr(self, job.programDef[B.PAR_DIRNAME]))
# Abgleich mit zuvor gespeicherten Parametern
if len(job.programDef["pfilesource"]) > 2:
self.setParameterLoaded(job)
def setBasedir(self, job, program):
if jobdef[program]:
self.basedir = jobdef[program]["basedir"]
if hasattr(self, jobdef[program]["dirname"]):
tools.path_tool.extractPath(self.basedir, getattr(self, jobdef[program]["dirname"]))
elif self.basedir == "workbase":
home = tools.path_tool.getHome()
dirpath = os.path.join(home, "data", "workspace")
setattr(self, jobdef[program]["dirname"], dirpath)
elif program != "unit":
# compose after setargs
dirpath = tools.path_tool.composePattern(job, "{"+jobdef[program]["basedir"]+"}", None)
setattr(self, jobdef[program]["dirname"], dirpath)
else:
self.basedir = "debugs"
def checkParameter(self, job):
print (f"Parameter initialisiert {self.program}")
pardef = jobdef[job.program]["pardef"]
pardef = self.programDef[CTLG_PARDEF]
for p in pardef.split(","):
print(p)
if len(p) > 1 and not hasattr(self, p):
@ -578,9 +564,10 @@ class Parameter:
dirpath = self.getDirParameter()
#if dirpath is not None:
# tools.path_tool.extractPath(job, dirpath[0], dirpath[1])
if hasattr(self, "application") and self.application in job.conf.confs[B.SUBJECT_APPS]:
if B.ATTR_APPS_PROJECT in job.conf.confs[B.SUBJECT_APPS][self.application]:
setattr(self, B.ATTR_APPS_PROJECT, job.conf.confs[B.SUBJECT_APPS][self.application][B.ATTR_APPS_PROJECT])
if hasattr(self, "application") and B.SUBJECT_APPS in job.conf \
and getattr(self, "application") in job.conf[B.SUBJECT_APPS]:
if B.ATTR_APPS_PROJECT in job.conf[B.SUBJECT_APPS][self.application]:
setattr(self, B.ATTR_APPS_PROJECT, job.conf[B.SUBJECT_APPS][self.application][B.ATTR_APPS_PROJECT])
proj = getattr(self, B.ATTR_APPS_PROJECT)
@ -605,52 +592,3 @@ class Parameter:
if len(a) == 2:
return self[a[0]][a[1]]
return
# ------------------------------------------------------------------------------------------------------------------
class Configuration:
def __init__ (self, job, program):
self.program = program
path = tools.path_tool.getBasisConfigPath()
print ("conf initialisieren "+self.program+" > "+path)
self.setConfiguration(job, path)
return
def debug(self, verify, text):
if hasattr(self, "m"):
self.m.debug(verify, text)
def setConfiguration(self, job, path):
if not hasattr(self, "confs"):
self.confs = {}
self.confs["configpath"] = []
doc = tools.file_tool.readFileDict(job, path, None)
self.confs["configpath"].append(path)
if "basic" in doc:
for k, v in doc["basic"].items():
if k not in self.confs:
self.confs[k] = v
else:
for k, v in doc.items():
if k not in self.confs:
self.confs[k] = v
def setConfig(self, path, val):
a = path.split(".")
if len(a) == 1:
self.confs[a[0]] = val
elif len(a) == 2:
self.confs[a[0]][a[1]] = val
elif len(a) == 3:
self.confs[a[0]][a[1]][a[2]] = val
def getPath(self, key):
return self.confs.get(B.SUBJECT_PATH).get(key)
def getJobConf(self, key):
a = key.split(":")
if len(a) == 1:
return self.confs[a[0]]
if len(a) == 2:
return self.confs[a[0]][a[1]]

2
objects/catalog.py
View File

@ -103,6 +103,8 @@ class Catalog:
else:
msg = None
data = tools.file_tool.readFileDict(job, pathname, msg)
if hasattr(job, "m"):
job.m.debug(12, "domain " + domain + " readed from " + pathname)
self.catalog[domain] = data[B.DATA_NODE_TABLES][domain][B.DATA_NODE_KEYS]
return data

21
test/constants.py
View File

@ -7,21 +7,10 @@
"""
constants
"""
import basic.constants as B
import os
home = os.getcwd()
prgdir = ""
if home[-5:] == "xtests" and home[-9:] != "program":
home = home[0:-5]
if home[-10:] == "components":
home = home[0:-11]
if home[-9:] == "program":
prgdir = home[-6:]
home = home[0:-7]
elif home[-7:] == "program":
prgdir = home[-7:]
home = home[0:-8]
HOME_PATH = home
DATA_PATH = os.path.join(home, "data")
PROG_PATH = os.path.join(home, prgdir)
COMP_PATH = os.path.join(home, prgdir, "components")
HOME_PATH = B.HOME_PATH
DATA_PATH = os.path.join(HOME_PATH, "data")
PROG_PATH = os.path.join(HOME_PATH, "program")
COMP_PATH = os.path.join(HOME_PATH, "program", "components")
OS_SYSTEM = "linux"

3
test/test_10job.py
View File

@ -32,6 +32,9 @@ class MyTestCase(unittest.TestCase):
job = basic.program.Job(PROGRAM_NAME)
print(str(job.__dict__))
self.checkSimpleJob(job)
sjob = basic.program.SimpleJob(PROGRAM_NAME)
print(str(sjob.__dict__))
self.checkSimpleJob(sjob)
def checkSimpleJob(self, job):
self.assertIn("conf", job.__dict__)

165
test/test_11message.py
View File

@ -0,0 +1,165 @@
import unittest
import os
import inspect
import shutil
import tools.path_tool
import basic.program
from basic.componentHandling import ComponentManager
import test.constants
import basic.constants as B
import test.constants as T
import basic.componentHandling
import tools.file_tool
import basic.message
HOME_PATH = test.constants.HOME_PATH
PYTHON_CMD = "python"
TEST_FUNCTIONS = ["test_00init", "test_05loginigArgs", "test_10set", "test_11log", "test_20close"]
PROGRAM_NAME = "clean_workspace"
NOT_TO_LOG = ["xx1xx", "xx4xx"]
TO_LOG = ["yy1yy", "yy2yy", "yy3yy", "yy4yy", "yy5yy", "yy6yy", "xx0xx", "xx8xx"]
NOT_TO_DEBUG = ["xx4xx", "xx1xx", "xx1xx", "xx1xx", "xx1xx", "xx1xx", "xx1xx"]
TO_DEBUG = ["xx9xx", "xx22xx", "xx35xx"]
class MyTestCase(unittest.TestCase):
mymsg = "--------------------------------------------------------------"
def test_00init(self):
global mymsg
global msgObject
global jobObject
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
# simple job instantiate - without parameter and only simple messaging
args = {}
args["par"] = {}
args["par"]["mode"] = "unit"
args["par"]["wsdir"] = os.path.join(B.HOME_PATH, "workspace")
job = basic.program.SimpleJob(PROGRAM_NAME, "unittest", args)
jobObject = job
print(str(job.__dict__))
tlogTime = "20220101_123456"
# temporary Message - a simple implementation at initialization of the job
tmsg = basic.message.TempMessage(job, tlogTime)
print(str(tmsg.__dict__))
self.checkSimpleMessage(tmsg, "debug", tlogTime)
self.assertEqual(basic.message.LIMIT_DEBUG, getattr(tmsg, "level"))
setattr(job, "m", tmsg)
msg = basic.message.Message(job, "trace", tlogTime, None)
setattr(job, "m", msg)
print(str(msg.__dict__))
self.checkSimpleMessage(msg, "debug", tlogTime)
self.checkSimpleMessage(msg, "log", tlogTime)
self.assertEqual(basic.message.LIMIT_TRACE, getattr(msg, "level"))
msgObject = msg #
def test_05loginigArgs(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
res = msgObject.getLoggingArgs(12, 1, "text")
self.assertEqual(res["mlevel"], 12)
self.assertEqual(res["mprio"], 1)
self.assertEqual(res["mtext"], "text")
res = msgObject.getLoggingArgs(12, "text", "")
self.assertEqual(res["mlevel"], 12)
self.assertEqual(res["mprio"], 0)
self.assertEqual(res["mtext"], "text")
res = msgObject.getLoggingArgs(12, "text", "ERROR: ")
self.assertEqual(res["mlevel"], 12)
self.assertEqual(res["mprio"], 0)
self.assertEqual(res["mtext"], "ERROR: text")
def test_10set(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
print("test_set "+str(msgObject.__dict__))
msgObject.setMsg("yy1yy result-msg")
self.assertEqual(basic.message.RC_MSG, msgObject.getFinalRc())
msgObject.setWarn("yy2yy warn-msg")
self.assertEqual(basic.message.RC_WARN, msgObject.getFinalRc())
msgObject.setError("yy3yy error-msg")
self.assertEqual(basic.message.RC_ERROR, msgObject.getFinalRc())
msgObject.setWarn("yy4yy warn-msg")
self.assertEqual(basic.message.RC_ERROR, msgObject.getFinalRc())
msgObject.setError("yy5yy error-msg")
self.assertEqual(basic.message.RC_ERROR, msgObject.getFinalRc())
msgObject.setFatal("yy6yy fatal-msg")
self.assertEqual(basic.message.RC_FATAL, msgObject.getFinalRc())
def test_11log(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
print("test_log "+str(msgObject.__dict__))
i = 0
# auf verschiedenen Ebenen loggen ohne und mit Weitergabe von Prio
for level in [basic.message.LIMIT_INFO, basic.message.LIMIT_DEBUG, basic.message.LIMIT_TRACE, basic.message.LIMIT_XTRACE]:
setattr(msgObject, "level", level)
msgObject.logInfo("xx"+str(i)+"xx info at level "+str(msgObject.level))
i += 1
msgObject.logDebug("xx" + str(i) + "xx debug at level " + str(msgObject.level))
i += 1
msgObject.logTrace("xx" + str(i) + "xx trace at level " + str(msgObject.level))
i += 1
msgObject.logXTrace("xx" + str(i) + "xx xtrace at level " + str(msgObject.level))
i += 1
msgObject.logInfo(2, "xx"+str(i)+"xx info++ at level "+str(msgObject.level))
i += 1
msgObject.logDebug(2, "xx" + str(i) + "xx debug++ at level " + str(msgObject.level))
i += 1
msgObject.logTrace(2, "xx" + str(i) + "xx trace++ at level " + str(msgObject.level))
i += 1
msgObject.logXTrace(2, "xx" + str(i) + "xx xtrace++ at level " + str(msgObject.level))
i += 1
msgObject.logInfo(-2, "xx"+str(i)+"xx info-- at level "+str(msgObject.level))
i += 1
msgObject.logDebug(-2, "xx" + str(i) + "xx debug-- at level " + str(msgObject.level))
i += 1
msgObject.logTrace(-2, "xx" + str(i) + "xx trace-- at level " + str(msgObject.level))
i += 1
msgObject.logXTrace(-2, "xx" + str(i) + "xx xtrace-- at level " + str(msgObject.level))
i += 1
def test_20close(self):
"""
:return:
"""
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
print("test_close "+str(msgObject.__dict__))
self.assertEqual(False, msgObject.logfile.closed)
self.assertEqual(False, msgObject.debugfile.closed)
msgObject.closeMessage()
self.assertEqual(True, msgObject.logfile.closed)
self.assertEqual(True, msgObject.debugfile.closed)
# now check what is logged
logtext = tools.file_tool.readFileText(jobObject, msgObject.logpath, None)
debugtext = tools.file_tool.readFileText(jobObject, msgObject.debugpath, None)
for x in TO_LOG:
regex = r".*" + x + ""
self.assertIn(x, logtext)
self.assertIn(x, debugtext)
def checkSimpleMessage(self, msg, prefix, logTime):
for x in [prefix+"path", prefix+"file", "level"]:
print(x)
self.assertIn(x, msg.__dict__)
self.assertIn(logTime, getattr(msg, "debugpath"))
if __name__ == '__main__':
unittest.main()

62
tools/config_tool.py
View File

@ -58,32 +58,32 @@ def getConfigPath(job, modul, name, subname=""):
verify = job.getDebugLevel("config_tool")-4
if verify: job.debug(verify, "getConfig " + modul + ", " + name)
if modul == P.KEY_TOOL:
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
P.VAL_CONFIG, P.KEY_TOOL+"_"+name))
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
P.VAL_CONFIG, name))
if path is not None:
return path
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_HOME],
P.VAL_CONFIG, P.KEY_TOOL+"_"+name))
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_HOME],
P.VAL_CONFIG, name))
if path is not None:
return path
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM],
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_PROGRAM],
P.VAL_UTIL, P.VAL_CONFIG, name))
if path is not None:
return path
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_ENV],
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_ENV],
job.par.environment, P.VAL_CONFIG, P.KEY_TOOL+"_"+ name))
if path is not None:
return path
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
elif modul == P.KEY_COMP:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_HOME],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_HOME],
P.VAL_CONFIG, P.KEY_COMP+"_" + name + "."+format)
if verify: job.debug(verify, "4 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
basic.componentHandling.getComponentFolder(name), "CONFIG." + format)
if verify: job.debug(verify, "5 " + pathname)
if os.path.exists(pathname):
@ -93,88 +93,90 @@ def getConfigPath(job, modul, name, subname=""):
elif modul in COMP_FILES:
# for example DATASTRUCURE or the table
pathnames = []
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
pathnames.append(os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
basic.componentHandling.getComponentFolder(name), modul))
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
pathnames.append(os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
basic.componentHandling.getComponentFolder(subname), modul))
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, modul))
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, subname))
pathnames.append(os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, modul))
pathnames.append(os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, subname))
configpath = getExistingPath(job, pathnames)
if configpath is not None:
return configpath
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
basic.componentHandling.getComponentFolder(name), modul+"."+format)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
basic.componentHandling.getComponentFolder(name), subname+"."+format)
if os.path.exists(pathname):
return pathname
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
elif modul == P.KEY_BASIC:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
P.VAL_CONFIG , name + "."+format)
if verify: job.debug(verify, "4 " + pathname)
if os.path.exists(pathname):
return pathname
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
P.VAL_CONFIG, name))
if path is not None:
return path
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_HOME],
P.VAL_CONFIG, name))
if path is not None:
return path
elif modul == P.KEY_TESTCASE:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_TDATA],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_TDATA],
name, D.DFILE_TESTCASE_NAME + "."+format)
if verify: job.debug(verify, "4 " + pathname)
if os.path.exists(pathname):
return pathname
elif modul == P.KEY_TESTSUITE:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_TDATA],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_TDATA],
name, D.DFILE_TESTSUITE_NAME + "." + format)
if verify: job.debug(verify, "4 " + pathname)
if os.path.exists(pathname):
return pathname
elif modul == P.KEY_CATALOG:
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_TDATA],
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_TDATA],
P.KEY_CATALOG, name))
if path is not None:
return path
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
P.KEY_CATALOG, name))
if path is not None:
return path
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM],
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_PROGRAM],
P.KEY_CATALOG, name))
if path is not None:
return path
raise Exception(P.EXP_CONFIG_MISSING, name)
else:
pathname = tools.path_tool.composePath(job, P.P_TCPARFILE)
pathname = tools.path_tool.compose_path(job, P.P_TCPARFILE)
if verify: job.debug(verify, "7 " + pathname)
if os.path.exists(pathname):
return pathname
pathname = tools.path_tool.composePath(job, P.P_TSPARFILE)
pathname = tools.path_tool.compose_path(job, P.P_TSPARFILE)
if verify: job.debug(verify, "8 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_RELEASE],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_RELEASE],
P.VAL_CONFIG, "basis."+format)
if verify: job.debug(verify, "9 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_ENV],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_ENV],
P.VAL_CONFIG, "basis."+format)
if verify: job.debug(verify, "9 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_HOME],
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_HOME],
P.VAL_CONFIG, "basis."+format)
if verify: job.debug(verify, "9 " + pathname)
if os.path.exists(pathname):

10
tools/file_abstract.py
View File

@ -27,7 +27,7 @@ class FileFcts():
def getMsg(self):
if self.comp is not None:
return self.comp.m
if self.job is not None:
if self.job is not None and hasattr(self.job, "m"):
return self.job.m
return None
@ -76,7 +76,7 @@ class FileFcts():
continue
mapping = o["mapping"]
schema = o["schema"]
archivpath = os.path.join(tools.path_tool.composePattern(job, "{tcresult}/request", self.comp), filename) # ergebnisse/comp/request )
archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/request", self.comp), filename) # ergebnisse/comp/request )
#txt = self.createDict()
tools.file_tool.writeFileText(self.comp.m, job, archivpath, txt)
@ -84,20 +84,20 @@ class FileFcts():
archivpath = ""
filename = step.args["filename"]
technique = step.args["technique"]
archivpath = os.path.join(tools.path_tool.composePattern(job, "{tcresult}/request", self.comp), filename)
archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/request", self.comp), filename)
if technique == "cli":
for o in self.comp.conf[B.SUBJECT_ARTS][B.TOPIC_NODE_FILE]:
if o["name"] != filename:
continue
envpath = o["envpath"]
envpath = tools.path_tool.composePattern(job, envpath, self.comp)
envpath = tools.path_tool.compose_path(job, envpath, self.comp)
fct = basic.toolHandling.getCliTool(job, self.comp)
fct.copy(self.job, archivpath, envpath)
elif technique == "api":
txt = tools.file_tool.readFileText(job, archivpath, self.comp.m)
fct = basic.toolHandling.getApiTool(job, self.comp)
response = fct.send(self.job, self.comp, txt)
archivpath = os.path.join(tools.path_tool.composePattern(job, "{tcresult}/response", self.comp), filename)
archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/response", self.comp), filename)
"""
get_response:

19
tools/path_const.py
View File

@ -1,4 +1,6 @@
import basic.constants as B
import tools.config_tool
# -------------------------------------------------------------
# values and keywords
KEY_PRECOND = "precond"
@ -113,3 +115,20 @@ EXP_COMP_MISSING = "Component is missing for {}"
""" excetion for the case that a specific component doesnt exist, 1 parameter (context) """
EXP_CONFIG_MISSING = "Configuration is missing for {}"
""" excetion for the case that a specific configuration is missing, 1 parameter (context) """
class PathConf:
"""
this class contains the structure-informations of the testrelevant directories
"""
__instance = None
def __init__(self, job=None):
confs = tools.config_tool.getConfig(job, "tool", "path")
self.pattern = confs["pattern"]
PathConf.__instance = self
@staticmethod
def getInstance(job = None):
if (PathConf.__instance is None):
PathConf(job)
return PathConf.__instance

130
tools/path_tool.py
View File

@ -6,14 +6,16 @@
# ---------------------------------------------------------------------------------------------------------
""" In diesem Modul werden alle Funktionen zusammengefasst zur Generierung und Ermittlung von pathsn """
import os.path
import sys
import basic.program
import tools.config_tool
import re
import basic.constants as B
import tools.path_const as P
import tools.date_tool
import tools.value_tool
import getpass
from tools.path_const import PathConf
TOOL_NAME = "path_tool"
def getHome():
@ -54,116 +56,24 @@ def getActualJsonPath(job):
# return path
return path
def getKeyValue(job, key, comp=None):
"""
this function gets the value for the key which relates to an attribute in the job or in the component
:param key:
:param comp:
:return:
"""
#job = basic.program.Job.getInstance()
try:
verify = job.getDebugLevel(TOOL_NAME)-4
except:
verify = False
pt = PathConf.getInstance(job)
if verify: job.debug(verify, "getKeyValue " + key)
if 'job.par' in key:
val = job.getParameter(key[8:])
return val
elif 'job.conf' in key:
val = job.conf.confs[B.SUBJECT_PATH][key[9:]]
if verify: job.debug(verify, val)
return val
elif 'job.' in key:
a = key[4:].split(":")
val = getattr(job, a[0])
# only date with hours
if a[0] in ["start"]:
print("++++++++++++++"+str(val))
val = tools.date_tool.formatParsedDate(str(val), tools.date_tool.F_LOG)
print("++++++++++++++"+val)
if len(a) > 1 and a[1] == "H":
val = val[0:-4]+"00"
if verify: job.debug(verify, val)
return val
# return job.conf.paths[key[9:]]
elif 'comp.' in key:
if comp is None:
raise Exception(P.EXP_COMP_MISSING.format(key))
if tools.config_tool.hasAttr(comp.conf, key[5:]):
return tools.config_tool.getAttr(comp.conf, key[5:])
if tools.config_tool.hasAttr(comp, key[5:]):
return tools.config_tool.getAttr(comp, key[5:])
return ""
elif 'env.' in key:
if key[4:] in comp.conf["conn"]:
return comp.conf["conn"][key[4:]]
pass
elif key in pt.pattern:
return pt.pattern[key]
elif "time" in key and hasattr(job, "start"):
return getattr(job, "start")
else:
return "xx-"+key+"-xx"
def composePath(job, pathname, comp):
def compose_path(job, pathname, comp):
"""
this function composes a concrete path by the structured pathname
- the key of pathname is declared in path_const and the structure is configurated in config/path.yml.
- the key of pathname is declared in path_const and the structure is configurated in config/value.yml.
:param pathname - plain keyword
:param comp:
:return:
"""
#job = basic.program.Job.getInstance()
verify = job.getDebugLevel(TOOL_NAME)
pt = PathConf.getInstance(job)
job.debug(verify, "composePath " + pathname + " zu " + str(pt) + "mit ")
job.debug(verify, str(pt.pattern))
if pt.pattern[pathname]:
return composePattern(job, pt.pattern[pathname], comp)
job.debug(verify, "composePath " + pathname)
if "{" in pathname:
return tools.value_tool.compose_pattern(job, pathname, comp)
else:
job.debug(verify, "in Pattern nicht vorhanden: " + pathname)
def composePattern(job, pattern, comp):
"""
the function composes the pattern to the standardarized path with the attributes
which are stored in the job and the component
- the key of pathname is declared in path_const and the structure is configurated in config/path.yml.
:param pattern: - keyword surroundet with {}
:param comp:
:return: path
"""
#job = basic.program.Job.getInstance()
try:
verify = job.getDebugLevel(TOOL_NAME)
except:
verify = False
verbose = not False
#job.debug(verify, "composePattern " + pattern)
max=5
l = re.findall('\{.*?\}', pattern)
#job.debug(verify, l)
print(l)
for pat in l:
if verbose: print(str(max) + ": " + pattern + ": " + pat)
pit = getKeyValue(job, pat[1:-1], comp)
if verbose: print(str(pit) + ": " + pattern + ": " + pat)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
pattern = pattern.replace(pat, pit)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
while ("{" in pattern):
max = max-1
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
pattern = composePattern(job, pattern, comp)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
if (max < 3) :
break
return pattern
def rejoinPath(a, b="", c="", d="", e="", f=""):
"""
this function concatenates the arguments to a path in the correct format for the operating-system
@ -229,7 +139,7 @@ def extractPattern(job, pathtyp, comp=None):
pre = work[0:i]
pat = work[i+1:j]
job.debug(verify, work + " von " + str(i) + "-" + str(j) + " pre " + pre + "pat " + pat)
pit = getKeyValue(job, pat, comp)
pit = tools.value_tool.get_key_value(job, pat, comp)
tup = (pre, pat, pit)
out.append(tup)
work = work[j+1:]
@ -298,23 +208,3 @@ def extractPath(job, pathtyp, path):
i = i +1
class PathConf:
"""
this class contains the structure-informations of the testrelevant directories
"""
__instance = None
def __init__(self, job=None):
#print('init pathConf')
confs = tools.config_tool.getConfig(job, "tool", "path")
self.pattern = confs["pattern"]
#print(self.pattern)
PathConf.__instance = self
@staticmethod
def getInstance(job = None):
#print("PathConf getInstance " + str(PathConf.__instance))
if (PathConf.__instance is None):
PathConf(job)
#print("PathConf getInstance " + str(PathConf.__instance))
return PathConf.__instance

109
tools/value_tool.py
View File

@ -13,6 +13,7 @@ import re
import basic.constants as B
import tools.path_const as P
import tools.date_tool
import objects.catalog
import getpass
TOOL_NAME = "value_tool"
@ -21,8 +22,29 @@ DOM_PAR = "par"
DOM_COMP = "comp"
DOM_CONF = "conf"
DOM_ENV = "env"
DOM_CTLG = "ctlg"
def getKeyValue(job, key, comp=None):
class ValueConf:
"""
this class contains the structure-informations of the testrelevant directories
"""
__instance = None
def __init__(self, job=None):
#print('init pathConf')
confs = tools.config_tool.getConfig(job, "tool", "value")
self.pattern = confs["pattern"]
#print(self.pattern)
ValueConf.__instance = self
@staticmethod
def getInstance(job = None):
#print("PathConf getInstance " + str(PathConf.__instance))
if (ValueConf.__instance is None):
ValueConf(job)
#print("PathConf getInstance " + str(PathConf.__instance))
return ValueConf.__instance
def get_key_value(job, key, comp=None):
"""
this function gets the value for the key which relates to an attribute in the job or in the component
:param key:
@ -37,26 +59,19 @@ def getKeyValue(job, key, comp=None):
#pt = PathConf.getInstance(job)
if verify: job.debug(verify, "getKeyValue " + key)
if DOM_JOB == key[0:3]:
if DOM_PAR in key[4:7]:
val = job.getParameter(key[8:])
a = key.split(".")
if DOM_PAR == a[1]:
val = job.getParameter(a[2])
return val
elif DOM_CONF in key[4:8]:
val = job.conf.confs[B.SUBJECT_PATH][key[9:]]
elif DOM_CONF == a[1]:
val = job.conf[B.SUBJECT_PATH][a[2]]
if verify: job.debug(verify, val)
return val
elif 'job.' in key:
a = key[4:].split(":")
val = getattr(job, a[0])
# only date with hours
if a[0] in ["start"]:
print("++++++++++++++"+str(val))
val = tools.date_tool.formatParsedDate(str(val), tools.date_tool.F_LOG)
print("++++++++++++++"+val)
if len(a) > 1 and a[1] == "H":
val = val[0:-4]+"00"
if verify: job.debug(verify, val)
return val
# return job.conf.paths[key[9:]]
elif len(a) == 2:
if not hasattr(job, a[1]):
job.m.setError("key, job has not attribute "+a[1])
return ""
return getattr(job, a[1])
elif DOM_COMP in key:
if comp is None:
raise Exception(P.EXP_COMP_MISSING.format(key))
@ -69,8 +84,66 @@ def getKeyValue(job, key, comp=None):
if key[4:] in comp.conf["conn"]:
return comp.conf["conn"][key[4:]]
pass
elif DOM_CTLG in key:
a = key.split(".")
catalog = objects.catalog.Catalog.getInstance()
if len(a) == 3:
return catalog.getValue(job, a[1], a[2])
if len(a) == 4:
return catalog.getValue(job, a[1], a[2], a[3])
pass
elif "time" in key and hasattr(job, "start"):
return getattr(job, "start")
else:
vc = ValueConf.getInstance(job)
if key in vc.pattern:
return vc.pattern[key]
pc = P.PathConf.getInstance(job)
if key in pc.pattern:
return pc.pattern[key]
job.m.setError("key is not defined " + key)
return "xx-"+key+"-xx"
def compose_pattern(job, pattern, comp):
if "{" in pattern and "}" in pattern:
return compose_string(job, pattern,comp)
vc = ValueConf.getInstance(job)
if pattern in vc.pattern:
return compose_string(job, "{" + pattern + "}", comp)
pc = P.PathConf.getInstance(job)
if pattern in pc.pattern:
return compose_string(job, "{" + pattern + "}", comp)
return None
def compose_string(job, pattern, comp):
"""
the function composes the pattern to the standardarized path with the attributes
which are stored in the job and the component
- the key of pathname is declared in path_const and the structure is configurated in config/value.yml.
:param pattern: - keyword surroundet with {}
:param comp:
:return: path
"""
try:
verify = job.getDebugLevel(TOOL_NAME)
except:
verify = False
verbose = not False
max=5
l = re.findall('\{.*?\}', pattern)
#job.debug(verify, l)
if verbose: print("l: " + str(l))
for pat in l:
if verbose: print(str(max) + ": " + pattern + ": " + pat)
pit = get_key_value(job, pat[1:-1], comp)
if verbose: print(str(pit) + ": " + pattern + ": " + pat)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
pattern = pattern.replace(pat, pit)
while ("{" in pattern):
max = max-1
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
pattern = compose_string(job, pattern, comp)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
if (max < 3) :
break
return pattern

Write Preview
Loading…
Cancel
Save