#!/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 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 .. ") if __name__ == '__main__': print(PROGRAM_NAME) x = basic.program.Job(PROGRAM_NAME) x.startJob() 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()