Browse Source

run jobs as unit-test

master
Ulrich Carmesin 3 years ago
parent
commit
803d3deef2
  1. 18
      basic/componentHandling.py
  2. 13
      basic/message.py
  3. 35
      basic/program.py
  4. 4
      components/component.py
  5. 4
      components/testa/Testa.py
  6. 3
      components/testa1/Testa1.py
  7. 3
      components/testa11/Testa11.py
  8. 3
      components/testa2/Testa2.py
  9. 3
      components/testa21/Testa21.py
  10. 3
      components/testb/Testb.py
  11. 3
      components/testb1/Testb1.py
  12. 21
      components/testexec.py
  13. 20
      test/test_job.py

18
basic/componentHandling.py

@ -70,7 +70,7 @@ class ComponentManager:
job.debug(verify, "getComponents " + compobjname)
return comps[compobjname]
def getComponents(mainfct):
def getComponents(self, mainfct):
job = basic.program.Job.getInstance()
verify = -2 + job.getDebugLevel("job_tool")
job.debug(verify, "getComponents " + mainfct)
@ -135,8 +135,9 @@ class ComponentManager:
print(vars(c.m))
if readedPar is not None:
print("createComponent 5 a " + name + " : " + str(readedPar))
for k in readedPar["components"][name].keys():
c.conf[k] = readedPar["components"][name][k]
if name in readedPar["comps"]:
for k in readedPar["comps"][name].keys():
c.conf[k] = readedPar["comps"][name][k]
comps[name] = c
self.createComponent(c, nr, suffix)
else:
@ -153,8 +154,11 @@ class ComponentManager:
print("createComponent 4 b " + componentName)
print(vars(c))
if readedPar is not None:
for k in readedPar["components"][name].keys():
c.conf[k] = readedPar["components"][name][k]
if name in readedPar["comps"]:
for k in readedPar["comps"][name].keys():
c.conf[k] = readedPar["comps"][name][k]
else:
print("comp fehlt "+name)
comps[name] = c
self.createComponent(c, i, suffix)
i = i + 1
@ -195,11 +199,11 @@ def getComponentDict():
for c in comps:
out[comps[c].name] = {}
print("getCompDict " + comps[c].name)
for k in comps[c].confs.keys():
for k in comps[c].conf.keys():
print("getCompDict " + k)
if isParameterSubnode(k): # "artifact" in k or "components" in k or "instance" in k:
print("getCompDict -b- " + k)
out[comps[c].name][k] = copy.deepcopy(comps[c].confs[k])
out[comps[c].name][k] = copy.deepcopy(comps[c].conf[k])
return out

13
basic/message.py

@ -84,6 +84,14 @@ class Message:
elif job.program == "unit":
self.logfile = self.debugfile
else:
self.setLogdir(logTime)
self.topmessage = ""
self.messages = []
print("message initialisiert mit level " + str(self.level))
def setLogdir(self, logTime):
job = basic.program.Job.getInstance()
basedir = job.par.basedir
basedir = basedir.replace("base", "log")
# basedir = utils.path_tool.composePath(basedir, None)
@ -92,10 +100,6 @@ class Message:
logpath = basedir + "/protokoll_" + logTime + ".txt"
self.logDebug("logfile " + logpath)
self.logfile = open(logpath, "w")
self.topmessage = ""
self.messages = []
print("message initialisiert mit level " + str(self.level))
def initErrorTyp(self):
self.CONST_ERRTYP = {
@ -225,7 +229,6 @@ class Message:
def debug(self, prio, text):
""" eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt"""
if (int(prio) < int(self.level)+1):
print ("debugfile " + str(self.debugfile))
self.debugfile.write(text + "\n")
def merge(self, submsg):

35
basic/program.py

@ -20,42 +20,49 @@ jobdef = {
"pfilesource" : "",
"pfiletarget" : "envparfile",
"basedir": "envbase",
"dirname": "envdir",
"logdir": "{job.par.envdir}/{log}/log_{time}.txt" },
"check_environment": {
"pardef": "",
"pfilesource" : "",
"pfiletarget" : "envparfile",
"basedir": "envbase",
"dirname": "envdir",
"logdir": "{job.par.envdir}/{log}/log_{time}.txt" },
"init_testset": {
"pardef": "tsdir,tdtyp,tdsrc,tdname",
"pfilesource" : "envparfile",
"pfiletarget" : "tsparfile",
"basedir": "tsbase",
"dirname": "tsdir",
"logdir": "{job.par.tsdir}/{log}/log_{tstime}.txt" },
"init_testcase": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource" : "tsparfile",
"pfilesource" : "envparfile",
"pfiletarget" : "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt" },
"test_system": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt"},
"finish_testcase": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tcdir",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt"},
"finish_testset": {
"pardef": "tcdir,tdtyp,tdsrc,tdname",
"pfilesource": "tsparfile",
"pfiletarget": "tcparfile",
"basedir": "tcbase",
"dirname": "tsdir",
"logdir": "{job.par.tcdir}/{log}/log_{tctime}.txt"}
}
@ -96,6 +103,12 @@ class Job:
Job(program)
return Job.__instance
def setProgram(self, program):
self.program = program
self.par.program = program
self.par.setBasedir(program)
self.m.setLogdir("logTime")
def getInstance():
if (Job.__instance is not None):
return Job.__instance
@ -196,21 +209,18 @@ class Parameter:
def __init__ (self, program):
print ("# init Parameter for " + program)
self.program = program
if program == "unit":
self.basedir = "debugs"
return
elif jobdef[program]:
self.setBasedir(program)
print (f"# Parameter initialisiert {self.program} mit basedir {self.basedir}")
if (program != "unit"): self.setParameter()
def setBasedir(self, program):
if jobdef[program]:
self.basedir = jobdef[program]["basedir"]
elif "estcase" in program:
self.basedir = "tcbase"
elif "estset" in program:
self.basedir = "tsbase"
elif "nviron" in program:
self.basedir = "envbase"
if hasattr(self, jobdef[program]["dirname"]):
utils.path_tool.extractPath(self.basedir, getattr(self, jobdef[program]["dirname"]))
else:
self.basedir = "debugs"
print (f"# Parameter initialisiert {self.program} mit basedir {self.basedir}")
self.setParameter()
def checkParameter(self):
job = Job.getInstance()
@ -220,7 +230,6 @@ class Parameter:
print(p)
if len(p) > 1 and not hasattr(self, p):
job.m.setFatal("Parameter " + p + " is not set!")
print (f"job initialisiert {self.program}")
def setParameter(self):
"""

4
components/component.py

@ -95,6 +95,10 @@ class Component():
(5) the summary result is returned
:return:
"""
job = basic.program.Job.getInstance()
verify = job.getDebugLevel(self.name)
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() " + str(self.name))
pass
def getHitscore(self, typ, rs, tg):

4
components/testa/Testa.py

@ -1,7 +1,7 @@
import components.component
import components.testexec
class Testa(components.component.Component):
class Testa(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testa')

3
components/testa1/Testa1.py

@ -1,6 +1,7 @@
import components.component
import components.testexec
class Testa1(components.component.Component):
class Testa1(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testa1')

3
components/testa11/Testa11.py

@ -1,6 +1,7 @@
import components.component
import components.testexec
class Testa11(components.component.Component):
class Testa11(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testa1')

3
components/testa2/Testa2.py

@ -1,6 +1,7 @@
import components.component
import components.testexec
class Testa2(components.component.Component):
class Testa2(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testa1')

3
components/testa21/Testa21.py

@ -1,5 +1,6 @@
import components.component
import components.testexec
class Testa21(components.component.Component):
class Testa21(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testa1')

3
components/testb/Testb.py

@ -1,5 +1,6 @@
import components.component
import components.testexec
class Testb(components.component.Component):
class Testb(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testb')

3
components/testb1/Testb1.py

@ -1,5 +1,6 @@
import components.component
import components.testexec
class Testb1(components.component.Component):
class Testb1(components.component.Component, components.testexec.Testexecuter):
def __init__(self):
print('init Testa1')

21
components/testexec.py

@ -46,8 +46,8 @@ class Testexecuter():
job = basic.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logInfo("--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.reset_TData(granularity)
self.m.setMsg("prepare_system for " + self.name +" "+ granularity + " is OK")
self.m.logInfo("--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
pass
@ -60,7 +60,8 @@ class Testexecuter():
"""
job = basic.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.debug(verify, "--- "+str(inspect.currentframe().f_code.co_name)+"() started at "
+ datetime.now().strftime("%Y%m%d_%H%M%S")+" for " + str(self.name).upper())
if "log" in self.conf["artifact"]:
self.m.logInfo("log rotate in "+ self.name)
if "flaskdb" in self.conf["artifact"]:
@ -69,8 +70,8 @@ class Testexecuter():
self.m.logInfo("lob is deleted with flaskdb "+ self.name)
if "file" in self.conf["artifact"]:
self.m.logInfo("rm files in "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.setMsg("resetInstance for " + self.name + " is OK")
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
def load_TData(self, granularity, testdata):
"""
@ -82,12 +83,12 @@ class Testexecuter():
"""
job = basic.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
if "testdata" in self.conf:
if self.conf["testdata"] == "flaskdb":
self.m.logInfo("insert flaskdb-content " + self.name)
self.m.setMsg("data loaded for " + self.name + " is OK")
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
def read_TData(self, granularity):
"""
@ -97,13 +98,13 @@ class Testexecuter():
"""
job = basic.program.Job.getInstance()
verify = -1+job.getDebugLevel(self.name)
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() started at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
if "flaskdb" in self.conf["artifact"]:
self.m.logInfo("select flaskdb-content "+ self.name)
if "lob" in self.conf["artifact"]:
self.m.logInfo("check lob if is deleted with flaskdb "+ self.name)
self.m.setMsg("checkInstance for " + self.name + " is OK")
self.m.logDebug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
self.m.setMsg("readInstance for " + self.name + " is OK")
self.m.debug(verify, "--- " + str(inspect.currentframe().f_code.co_name) + "() finished at " + datetime.now().strftime("%Y%m%d_%H%M%S") + " for " + str(self.name).upper())
def test_System(self, granularity):
"""

20
test/test_job.py

@ -2,9 +2,12 @@ import unittest
import os
from basic.program import Job
from basic.componentHandling import ComponentManager
import init_testcase
HOME_PATH = "/home/ulrich/6_Projekte/Programme/holtz"
PYTHON_CMD = "python"
class MyTestCase(unittest.TestCase):
def test_parameter(self):
def xtest_parameter(self):
job = Job("unit")
args = { "application" : "TEST" , "environment" : "ENV01", "modus" : "unit", "loglevel" : "debug", "tool" : "job_tool"}
job.par.setParameterArgs(args)
@ -14,9 +17,9 @@ class MyTestCase(unittest.TestCase):
self.assertEqual(job.getDebugLevel("file_tool"), 23)
self.assertEqual(job.getDebugLevel("job_tool"), 23)
args = { "application" : "TEST" , "environment" : "ENV01", "modus" : "unit", "loglevel" : "debug",
"tool" : "job_tool", "tsdir": "/home/ulrich/6_Projekte/Programme/holtz/test/lauf/V0.1/startjob/2021-08-21_18-ß2-01"}
"tool" : "job_tool", "tsdir": os.path.join(HOME_PATH, "test", "lauf", "V0.1", "startjob", "2021-08-21_18-ß2-01")}
job.par.setParameterArgs(args)
def test_components(self):
def xtest_components(self):
print("# # # # tetsComponents # # # # #")
job = Job.resetInstance("unit")
args = { "application" : "TEST" , "environment" : "ENV01", "modus" : "unit", "loglevel" : "debug", "tool" : "job_tool"}
@ -26,10 +29,19 @@ class MyTestCase(unittest.TestCase):
cm.createComponents("testa", 1, "")
def test_run(self):
os.system("python /home/ulrich/6_Projekte/Programme/holtz/check_environment.py -a TEST -e ENV01")
# os.system("python /home/ulrich/6_Projekte/Programme/holtz/check_environment.py -a TEST -e ENV01")
# os.system("python /home/basic/6_Projekte/Programme/pythonProject/init_testset.py -a TEST -e ENV01 "
# "-ts /home/basic/6_Projekte/Programme/holtz/test/lauf/V0.1/implement_2021-08-28_23-50-51 -dt csv -ds implement -dn firstunit")
#os.system(PYTHON_CMD+" "+os.path.join(HOME_PATH,"init_testcase.py")+" -a TEST -e ENV01 "
# "-tc "+os.path.join(HOME_PATH,"test","lauf","V0.1","TC0001","2021-08-28_23-50-51")+" -dt csv -ds implement -dn TC0001")
job = Job("unit")
args = { "application" : "TEST" , "environment" : "ENV01", "modus" : "unit", "loglevel" : "debug",
"tool" : "job_tool", "tcdir": os.path.join(HOME_PATH,"test","lauf","V0.1","TC0001","2021-08-28_23-50-51")}
job.par.setParameterArgs(args)
job.setProgram("init_testcase")
# os.system("python /home/basic/6_Projekte/Programme/pythonProject/init_testset.py")
init_testcase.start(job)
job.stopJob(1)
if __name__ == '__main__':
unittest.main()

Loading…
Cancel
Save