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

147 lines
6.8 KiB

#!/usr/bin/python
# program to execute automatically programs for a testsuite
# PARAM: --environment --application --tsdir [ testsuite, tstime ] --step
# main functions
# + execute_step() : testspec --> tdata.steps --> call( init_testsuite/testcase(),
# execute_testcase/testsuite(), collect_testcase/testsuite(), compare_testcase(), finish_testsuite() )
# ---------------------------------------------------
from datetime import datetime
2 years ago
import traceback
import basic.program
import basic.constants as B
# import tools.tdata_tool as tdata_tool
import tools.date_tool as date_tool
import tools.job_tool as job_tool
# import init_testcase
# import init_testsuite
# import execute_testcase
# import collect_testcase
# import compare_testcase
# import finish_testsuite
import model.factory
import model.catalog
import tools.job_tool
import tools.step_tool
PROGRAM_NAME = "test_executer"
myjob = None
testcases = []
testinstances = {}
def getTime():
time = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
return time
def startPyJob(job):
job.m.setMsg("# # # # # start executer # # # # # ")
verify = True
try:
steps = {}
if getattr(job.par, B.PAR_GRAN, "") == B.SUBJECT_TESTPLAN:
spec = model.factory.getTestplan(job, getattr(job.par, B.SUBJECT_PROJECT))
steps = getattr(spec, B.SUBJECT_STEPS, {})
elif getattr(job.par, B.PAR_GRAN, "") == B.SUBJECT_TESTSUITE:
spec = model.factory.getTestsuite(job, getattr(job.par, B.SUBJECT_PROJECT))
steps = getattr(spec, B.SUBJECT_STEPS, {})
elif getattr(job.par, B.PAR_GRAN, "") == B.SUBJECT_TESTCASE:
spec = model.factory.getTestcase(job, getattr(job.par, B.SUBJECT_PROJECT), name=getattr(job.par, B.SUBJECT_TESTCASE))
steps = getattr(spec, B.SUBJECT_STEPS, {})
else:
raise Exception("Object-collection steps is missing in {}".format(B.PAR_GRAN))
if not hasattr(job.par, B.PAR_STEP):
raise Exception("Parameter " + B.PAR_STEP + " is missing")
catalog = model.catalog.Catalog.getInstance()
programs = catalog.readDomain("programs", job)
stepnr = int(tools.step_tool.mapStepIDtoNr(job, getattr(job.par, B.PAR_STEP), getattr(job.par, B.PAR_GRAN)))
while stepnr < 3:
stepnr = int(tools.step_tool.mapStepIDtoNr(job, getattr(job.par, B.PAR_STEP), getattr(job.par, B.PAR_GRAN)))
executeStep(job, steps, programs, stepnr)
setattr(job.par, B.PAR_STEP, tools.step_tool.getNextStepID(job, stepnr))
if verify: print(" .. steps durchlaufen z58 .. " + getattr(job.par, B.PAR_STEP, "99"))
#setattr(job.par, "testcases", testcases)
if verify: print(" .. testcases z60 .. " + getattr(job.par, "testcases", "9-9"))
except Exception as e:
job.m.logDebug("+++++++++++++++++++++++++++++++++++++++++++++")
job.m.setFatal(str(e))
job.m.logDebug("+++++++++++++++++++++++++++++++++++++++++++++")
job.m.logDebug("execpt " + traceback.format_exc())
job.m.logDebug("+++++++++++++++++++++++++++++++++++++++++++++")
def executeStep(job, steps, programs, stepnr):
# TODO run over steps - definition ?!
"""
for arg in step["args"]:
if arg == "start":
print("start "+str(step["args"][arg]))
if "testsuite" in step["args"][arg]:
jobargs = {B.PAR_APP: job.par.application, B.PAR_ENV: job.par.environment,
B.PAR_TESTSUITE: job.par.usecase, B.PAR_TSTIME: date_tool.getActdate(date_tool.F_DIR)}
job_tool.start_child_process(job, jobargs)
elif B.PAR_TESTCASE in step["args"][arg]:
if step["comp"] in testcases:
jobargs = {B.PAR_APP: job.par.application, B.PAR_ENV: job.par.environment,
B.PAR_TCDIR: testcases[step["comp"]] }
else:
jobargs = {B.PAR_APP: job.par.application, B.PAR_ENV: job.par.environment,
B.PAR_TESTCASE: step["comp"], B.PAR_TCTIME: date_tool.getActdate(date_tool.F_DIR)}
job_tool.start_child_process(job, jobargs)
if arg == "report": # testsuite
#basic.program.Job.pushInstance(myjob)
pass
"""
verify = True
stepnr_par = int(stepnr)
for k in steps:
step = steps[k]
if verify: print("step " + k + " snr: " + step.stepnr + " =? " + getattr(job.par, B.PAR_STEP))
if int(step.stepnr) != stepnr_par:
# int(getattr(job.par, B.PAR_STEP)):
if verify: print("step " + k + " != snr ")
continue
if step.name in list(programs.keys()):
jobargs = {}
if job.par.gran == programs[step.name]["pardef"][B.PAR_GRAN]:
if verify: print("step " + k + " gleiche gran " + step.name)
# # testcase sollte direkt aufgerufen sein --> init_testcase()
# # aus steps nur dataref ermitteln!
if verify: print("\n======================== " + step.name + " ==========")
jobargs[B.PAR_PROGRAM] = step.name
for p in programs[step.name]["pardef"]:
jobargs[p] = getattr(job.par, p, "")
if verify: print("\n== " + str(jobargs))
job.m.logInfo("main program {} executed".format(step.name))
# tools.job_tool.start_child_process(job, jobargs)
elif job.par.gran == B.SUBJECT_TESTSUITE and B.SUBJECT_TESTCASE in step.name:
for tc in getattr(job.par, B.SUBJECT_TESTCASES):
print(tc)
# elif job.par.gran == testplan and testsuite == in step.name:
# for tc in getattr(job.par, testsuites):
continue
elif step.name in [getattr(job.par, B.PAR_VAR, "")]: # variant
if verify: print("step " + k + " gleiche variant " + step.name)
# jobargs[program] = execute_testcase
# select component from variant --> execute_testcase()
# for p in programs[step.name].pardef:
# jobargs[p] = getattr(job.par, p)
# tools.job_tool.start_child_process(job, jobargs)
job.m.logInfo("main program {} executed for variant {}.".format("exec_testcase", step.name))
continue
else:
if verify: print("step " + k + " unbekante Aktion " + step.name)
if verify: print(" .. steps durchlaufen .. ")
2 years ago
if __name__ == '__main__':
print(PROGRAM_NAME)
x = basic.program.Job(PROGRAM_NAME)
x.startJob()
2 years ago
try:
x.m.logDebug(str(vars(x.par)) + "\n" + str(vars(x.conf)))
if x.m.isRc("fatal"):
x.stopJob()
exit(x.m.rc * (-1) + 3)
startPy(x)
except:
x.m.setError(PROGRAM_NAME + " aborted")
finally:
x.stopJob()