#!/usr/bin/python # -*- coding: utf-8 -*- # --------------------------------------------------------------------------------------------------------- # Author : Ulrich Carmesin # Source : gitea.ucarmesin.de # --------------------------------------------------------------------------------------------------------- # Template Batchrahmen # #import sys, getopt import argparse import copy import yaml import os from datetime import datetime import basic.constants as B import basic.message import basic.message import basic.componentHandling import utils.date_tool import utils.path_tool import utils.file_tool import utils.config_tool import test.constants as T import utils.path_const as P jobdef = { "webflask": { "pardef": "", "pfilesource": "", "pfiletarget": "", "basedir": "workbase", "dirname": "workdir", "loglevel": basic.message.LIMIT_INFO, "logpath": "{job.conf.data}/workspace/webflask_{job.start:H}.txt" }, "declare_result": { "pardef": "", "pfilesource": "", "pfiletarget": "", "basedir": "workbase", "dirname": "workdir", "loglevel": basic.message.LIMIT_INFO, "logpath": "{job.conf.data}/workspace/reorg_{job.start:H}.txt" }, "service": { "pardef": "", "pfilesource": "", "pfiletarget": "", "basedir": "workbase", "dirname": "workdir", "loglevel": basic.message.LIMIT_INFO, "logpath": "{job.conf.data}/workspace/service_{job.start:H}.txt" }, "unit": { "pardef": "", "pfilesource": "", "pfiletarget": "envparfile", "basedir": "envbase", "dirname": "envdir", "loglevel": basic.message.LIMIT_DEBUG, "logpath": "{job.par.envdir}/{log}/log_{job.start:H}.txt" }, "check_environment": { "pardef": "", "pfilesource": "", "pfiletarget": "envparfile", "basedir": "envbase", "dirname": "envdir", "loglevel": basic.message.LIMIT_INFO, "logpath": "{job.par.envdir}/{log}/log_{job.start}.txt" }, "test_executer": { "pardef": "", "pfilesource": "tsparfile", "pfiletarget": "tsparfile", "basedir": "tsbase", "dirname": "tsdir", "loglevel": basic.message.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": basic.message.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": basic.message.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": basic.message.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": basic.message.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": basic.message.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": basic.message.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": basic.message.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": basic.message.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": basic.message.LIMIT_INFO, "logpath": "{job.par.tcdir}/{log}/{job.program}_{job.start}.txt"} } def setGlobal(): pass EXCP_CANT_POP = "cant pop this job from the instances" DEFAULT_ARCHIV_DIR = T.DATA_PATH + "/lauf" DEFAULT_GRAN = "ws" DEFAULT_PRG = "webflask" DEFAULT_GRAN = "ws" DEFAULT_APP = "WEBFLASK" DEFAULT_ENV = "Workspace" DEFAULT_MODE = "test" 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. :param pprg: program-name :param pgran: tc|ts| :param papp: application :param penv: environment :param ptstamp: timestamp - part of specific testfolder :param pmode: if it is a productive or development execution :return: """ if len(pprg) < 1: prgname = DEFAULT_PRG else: prgname = pprg if len(pgran) < 1: gran = DEFAULT_GRAN else: gran = pgran if len(pgran) < 1: gran = DEFAULT_GRAN else: gran = pgran if len(papp) < 1: app = DEFAULT_APP else: app = papp if len(penv) < 1: env = DEFAULT_ENV else: env = penv if len(ptstamp) < 1: tstamp = DEFAULT_TIME else: tstamp = ptstamp if len(pmode) < 1: mode = DEFAULT_MODE else: mode = pmode if gran == "tc": path = DEFAULT_ARCHIV_DIR + "/TC0001/" + tstamp elif gran == "ts": path = DEFAULT_ARCHIV_DIR + "/testlauf/TST001_" + tstamp else: path = T.DATA_PATH + "/workspace/" job = basic.program.Job(prgname) # job.conf.confs[B.SUBJECT_PATH]["components"] = T.COMP_PATH args = {"application": app, "environment": env, "modus": mode, gran + "time": tstamp, gran + "dir": path, "step": 1} print(str(args)) # "usecase": "TST001", "tstime": "2022-03-17_17-28"} job.par.setParameterArgs(job, args) return job class Job: __instance = None __instances = [] __jobid = 100000 def __init__ (self, program, args=None): """ initializing the job-object as a global structure for the actual doing type of job: a) (single) job for testing work [--args] work-folder is the testcase-folder, logging into testcase-log b) (single) job for reorganization of the testing area [--args] working-folder is the workspace or the environment-folder, logging is there c) webflask-job for the web-server --args work-folder is the web-workspace, if real jobs created these jobs runs as testing resp. reorganization-job d) unit-testing job --args! work-folder is the debug-area if real jobs for unit-testing created these jobs runs with unit-mode like a unit-test :param program: :param args: optional with arguments otherwise the cli-parameter """ # postcondition # - job.conf with basis config # - job.par with parameter-args or cli-args # - job.msg print ("################# init Job ## " + program + " #################") Job.__jobid += 1 self.jobid = str(Job.__jobid) self.program = program conf = Configuration(self, program) self.conf = conf try: path = utils.config_tool.getConfigPath(self, P.KEY_BASIC, B.BASIS_FILE) print("comps.basispath "+path) self.conf.setConfiguration(self, path) except: pass # the special path is not necessary appl = utils.config_tool.getConfig(self, P.KEY_BASIC, B.SUBJECT_APPS) print(appl) if appl is not None: self.conf.confs[B.SUBJECT_APPS] = appl[B.SUBJECT_APPS] par = Parameter(self, program, args) self.par = par logTime = utils.date_tool.getActdate(utils.date_tool.F_LOG) self.start = logTime self.m = basic.message.Message(self, jobdef[program]["loglevel"], logTime, None) def getLogpath(self): path = utils.path_tool.composePattern(self, jobdef[self.program]["logpath"], None) return path def murks(self, par, program): self.par = par dirpath = self.par.getDirParameter() setGlobal() if dirpath is not None: utils.path_tool.extractPath(dirpath[0], dirpath[1]) if program == "unit": # no job will be started self.start = datetime.now() logTime = self.start.strftime("%Y%m%d_%H%M%S") self.m = basic.message.Message(self, basic.message.LIMIT_DEBUG, logTime, None) print("prog-50 " + str(self.par.basedir)) def setProgram(self, program): self.program = program basedir = jobdef[program]["basedir"] self.basedir = basedir if (self.par is not None): setattr(self.par, "program", program) setattr(self.par, "basedir", basedir) parstring = getattr(self.par, "parstring") 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"], utils.path_tool.composePattern(self, "{"+basedir+"}", None)) print(parstring) self.par.setParameterLoaded(self) def startJob(self): self.start = datetime.now() print("prog-68 " + str(self.par.basedir)) logTime = self.start.strftime("%Y%m%d_%H%M%S") self.m = basic.message.Message(self, basic.message.LIMIT_DEBUG, logTime, None) print("prog-68 " + str(self.m.rc)) self.par.setParameterLoaded(self) self.m.logInfo("# # # Start Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # # # ") self.m.debug(basic.message.LIMIT_INFO, "# # # Start Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # # # ") self.par.checkParameter(self) self.m.logInfo(self.par.parstring) def stopJob(self, reboot=0): self.ende = datetime.now() self.dumpParameter() print("stopJob " + str(self.m.messages) + ", " + str(self.m.debugfile)) self.m.logInfo("# # " + self.m.topmessage + " # # # ") self.m.logInfo("# # # Stop Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # " + self.ende.strftime("%d.%m.%Y %H:%M:%S") + " # # # ") self.m.debug(basic.message.LIMIT_INFO, "# # " + self.m.topmessage + " # # # ") self.m.debug(basic.message.LIMIT_INFO, "# # # Stop Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # " + self.ende.strftime("%d.%m.%Y %H:%M:%S") + " # # # RC: " + str(self.m.getFinalRc())) self.m.closeMessage() rc = self.m.getFinalRc() print ("rc " + str(rc)) if reboot == 0: exit(rc) def dumpParameter(self): parpath = utils.path_tool.composePath(self, jobdef[self.program]["pfiletarget"], None) output = {} cconf = basic.componentHandling.getComponentDict() output["par"] = self.par.__dict__ if len(cconf) < 1: utils.file_tool.writeFileDict(self.m, self, parpath, output) return output[B.SUBJECT_COMPS] = {} for c in cconf: output[B.SUBJECT_COMPS][c] = {} for x in ["function", "conn"]: if x not in cconf[c]: continue output[B.SUBJECT_COMPS][c][x] = cconf[c][x] if x == B.SUBJECT_CONN and "passwd" in cconf[c][x]: cconf[B.SUBJECT_COMPS][c][x]["passwd"] = "xxxxx" utils.file_tool.writeFileDict(self.m, self, parpath, output) def loadParameter(self): output = {} if len(str(jobdef[self.program]["pfilesource"])) < 2: return None parpath = utils.path_tool.composePath(self, jobdef[self.program]["pfilesource"], None) print("parpath "+parpath) if not os.path.join(parpath): return None print("parpath "+parpath) doc = utils.file_tool.readFileDict(self, parpath, None) for k in doc.keys(): output[k] = copy.deepcopy(doc[k]) return output def getParameter(self, parameter): if hasattr(self.par, parameter): return getattr(self.par, parameter) elif "xxxtime" in parameter: neu = utils.date_tool.getActdate(utils.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) def hasElement(self, parameter, elem): """ the function searches in an optional job.parameter (a) true, if the parameter does not exist (DEFAULT) (b) true, if the element is member of the parameter (c) false, if parameter exists and elem is not member of the parameter :param parameter: :param elem: :return: """ if hasattr(self.par, parameter): print (parameter + " in Parameter") if getattr(self.par, parameter).find(elem) >= 0: return True return False return True def hascomponente(self, comp): """ it searches if comp is member of the optional job.parameter (default = each is member) :param comp: :return: """ return self.hasElement("componente", comp) def hasFunction(self, fct): """ it searches if fct is member of the optional job.parameter (default = each is member) :param fct: :return: """ return self.hasElement("function", fct) def hasTool(self, tool): """ it searches if tool is member of the optional job.parameter (default = each is member) :param tool: :return: """ return self.hasElement("tool", tool) def getMessageLevel(self, errtyp, elem): if (not hasattr(self, "m")) or (self.m is None): return basic.message.LIMIT_DEBUG elif elem.find("tool") > 1: if not hasattr(self.par, "tool") or getattr(self.par, "tool").find(elem) <= 0: return int(self.m.CONST_ERRTYP[errtyp]) -1 else: return int(self.m.CONST_ERRTYP[errtyp]) else: return int(self.m.CONST_ERRTYP[errtyp]) def getInfoLevel(self, elem): return self.getMessageLevel("info", elem) def getDebugLevel(self, elem): return self.getMessageLevel("debug", elem) def getTraceLevel(self, elem): return self.getMessageLevel("trace", elem) def debug(self, prio, text): #print("job.debug "+str(prio)+" "+text) if hasattr(self, "m"): self.m.debug(prio, text) else: print(text) # ------------------------------------------------------------------------------------------------------------------ class Parameter: """ parameter with source either the list args or the args of the main-program content of parameter: + single arguments + dirname of work-folder either from dir-parameter or composed from single 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 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"] 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"], utils.path_tool.composePattern(job, "{"+self.basedir+"}", None)) else: utils.path_tool.extractPath(job, jobdef[program]["basedir"], getattr(self, jobdef[program]["dirname"])) self.setParameterLoaded(job) def setBasedir(self, job, program): if jobdef[program]: self.basedir = jobdef[program]["basedir"] if hasattr(self, jobdef[program]["dirname"]): utils.path_tool.extractPath(self.basedir, getattr(self, jobdef[program]["dirname"])) elif self.basedir == "workbase": home = utils.path_tool.getHome() dirpath = os.path.join(home, "data", "workspace") setattr(self, jobdef[program]["dirname"], dirpath) elif program != "unit": # compose after setargs dirpath = utils.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"] for p in pardef.split(","): print(p) if len(p) > 1 and not hasattr(self, p): job.m.setFatal("Parameter " + p + " is not set!") def setParameter(self, job): """ 1. Programm -- implementiert in Main-Klasse 2. anwndung -- steuert zu pruefende System [ in basis_Config ] 3. amgebung -- steuert zu pruefende Maschine [ in dir/applicationen ] 4. release -- steuert zu prufendes Release [ aus dir/release kann spez. release_Config geladen werden, dir/lauf/release ] 5. ~Verz -- Dokumentationsverzeichnis zu Testlauf/Testfall/Soll-Branch 6. zyklus -- optional unterscheidet echte und entwicklungsLaeufe 7. Programmspezifische Parameter 8. loglevel -- steuert Protokollierung; default debug (fatal/error/warn/msg/info/debug1/debug2/trace1/trace2) 10. modus -- steuert die Verarbeitung; default echt - echt-auto Lauf aus Automatisierung (1-7) - test Lauf ohne Ausfuehrungen am Testsystem, wohl aber in Testverzeichnissen - echt-spez Wiederholung einer spezifischen Funktion (1-13) - unit Ausfuehrung der Unittests 11. componente -- schraenkt Verarbeitung auf parametriserte componenten ein 12. funktion -- schraenkt Verarbeitung auf parametriserte Funktionen ein 13. tool -- schraenkt Protokollierung/Verarbeitung auf parametriserte Tools ein """ # args = str(sys.argv) # print ("Job-Programm %s : " % args) parser = argparse.ArgumentParser() parser.add_argument('-a', '--'+B.SUBJECT_APPS, required=True, action='store') parser.add_argument('-e', '--'+B.PAR_ENV, required=True, action='store') parser.add_argument('-r', '--release', action='store') parser.add_argument('-ts', '--'+B.PAR_TSDIR, action='store') parser.add_argument('-tc', '--'+B.PAR_TCDIR, action='store') parser.add_argument('-ws', '--'+B.PAR_WSDIR, action='store') parser.add_argument('-rs', '--rsdir', action='store') parser.add_argument('-dt', '--tdtyp', action='store') # PAR_TDTYP parser.add_argument('-ds', '--tdsrc', action='store') # PAR_TDSRC parser.add_argument('-dn', '--tdname', action='store') # PAR_TDNAME parser.add_argument('-l', '--loglevel', action='store') # PAR_LOG parser.add_argument('-m', '--modus', action='store') # PAR_MODUS parser.add_argument('-c', '--component', action='store') # PAR_COMP parser.add_argument('-f', '--function', action='store') # PAR_FCT parser.add_argument('-t', '--tool', action='store') # PAR_TOOL parser.add_argument('-s', '--'+B.PAR_STEP, action='store') # PAR_STEP # parser.add_argument('-t', '--typ', default='einzel', action='store') # parser.add_argument('-d', '--dir', action='store') args = parser.parse_args() self.setParameterArgs(job, args) def getDirParameter(self): if hasattr(self, "tcdir"): return ("tcbase", str(self.tcdir)) if hasattr(self, "tsdir"): return ("tsbase", str(self.tsdir)) return None def setParameterArgs(self, job, args): print("setParArgs " + str(type(args))) self.parstring = "python " + self.program if "dict" in str(type(args)): for k in args: self.setJobAttr(k, args[k]) else: for k in vars(args): if getattr(args, k) is not None: self.setJobAttr(k , getattr(args, k)) dirpath = self.getDirParameter() #if dirpath is not None: # utils.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]) proj = getattr(self, B.ATTR_APPS_PROJECT) def setParameterLoaded(self, job): #job = Job.getInstance() print("setParLoaded "+job.program) readedPar = job.loadParameter() if readedPar is not None: print(str(readedPar)) for k in readedPar["par"].keys(): if not hasattr(self, k): self.setJobAttr(k, readedPar["par"][k]) def setJobAttr(self, key, val): setattr(self, key, val) self.parstring = self.parstring + " --" + key + " " + str(val) def getJobPar(self, key): a = key.split(":") if len(a) == 1: return self[a[0]] if len(a) == 2: return self[a[0]][a[1]] return # ------------------------------------------------------------------------------------------------------------------ class Configuration: def __init__ (self, job, program): self.program = program path = utils.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 = utils.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]]