diff --git a/basic/program.py b/basic/program.py index 5bc81f6..a1df6d0 100644 --- a/basic/program.py +++ b/basic/program.py @@ -262,7 +262,7 @@ class Job: if len(str(self.programDef[CTLG_PARSOURCE])) < 2: return None parpath = tools.path_tool.compose_path(self, self.programDef[CTLG_PARSOURCE], None) - if not os.path.join(parpath): + if parpath is None or not os.path.join(parpath): return None doc = tools.file_tool.read_file_dict(self, parpath, None) for k in doc.keys(): @@ -399,7 +399,7 @@ class Parameter: pardef = job.programDef[CTLG_PARDEF] for p in pardef: if len(p) > 1 and not hasattr(self, p): - job.m.setFatal("Parameter " + p + " is not set!") + raise Exception("Parameter {} is not set for {}!".format(p, self.program)) def setParameter(self, job): """ diff --git a/basic/step.py b/basic/step.py deleted file mode 100644 index 2c3546c..0000000 --- a/basic/step.py +++ /dev/null @@ -1,92 +0,0 @@ -#!/usr/bin/python -# -*- coding: utf-8 -*- -# --------------------------------------------------------------------------------------------------------- -# Author : Ulrich Carmesin -# Source : gitea.ucarmesin.de -# --------------------------------------------------------------------------------------------------------- -""" -this module implements the functionality of a test-step -which is defined in the test-specification -and is executed by any executer -there are 2 kinds of test-step -a) execute specific component in the job -b) execute specific test-entity in the test-suite-execution -""" -import basic.constants as B -import utils.data_const as D -import utils.i18n_tool - -LIST_ARGS = [ - "start", # for starting the specified main-program - "fct" # for calling the specified component-function -] - -class Step: - comp = "" - refLine = "" # in a: references the data-line(s) to be executed - execStep = "" # in a,b: executes only if the step is set in the job - args = {} - """ - the class contains each attribute of a test-step - * comp : the component which implements the step - * refLine : reference to data-lines which has to be executed - + execStep : - """ - def __init__(self): - self.comp = "" - self.refLine = "" - self.execStep = "" - self.variante = "" - self.args = {} - - def getStepText(self, job): - text = self.comp+D.CSV_DELIMITER+str(self.execStep)+D.CSV_DELIMITER+self.refLine - for k in self.args: - text += D.CSV_DELIMITER+k+":"+self.args[k] - return text+"\n" - - -def parseStep(job, fields): - step = Step() - step.comp = fields[D.STEP_COMP_I] - step.execStep = fields[D.STEP_EXECNR_I] - step.refLine = fields[D.STEP_REFNR_I] - step.variante = fields[D.STEP_VARIANT_I] - setattr(step, B.ATTR_DATA_REF, step.refLine) - i = 0 - if D.STEP_ARGS_I == D.STEP_LIST_I: - args = "" - for i in range(D.STEP_ARGS_I, len(fields)): - if len(fields[i]) < 1: - continue - if fields[i][0:1] == "#": - continue - args += "," + fields[i] - args = args[1:] - else: - i = D.STEP_ARGS_I - args = fields[D.STEP_ARGS_I] - - a = args.split(",") - for arg in a: - print("arg " + arg) - b = arg.split(":") - if len(b) < 2: - raise Exception(D.EXCP_MALFORMAT + " in arg["+str(i)+ "] " + str(fields)) - step.args[b[0]] = b[1] - if b[0] in LIST_ARGS: - setattr(step, b[0], b[1]) - i += 1 - # data[B.DATA_NODE_STEPS].append(step) - return step - -def getStepHeader(job): - text = "# " - text += utils.i18n_tool.I18n.getInstance(job).getText(f"{D.CSV_BLOCK_STEP=}", job) - text += ";"+utils.i18n_tool.I18n.getInstance(job).getText(f"{D.STEP_ATTR_COMP=}", job) - text += ";"+utils.i18n_tool.I18n.getInstance(job).getText(f"{D.STEP_ATTR_EXECNR=}", job) - text += ";"+utils.i18n_tool.I18n.getInstance(job).getText(f"{D.STEP_ATTR_REFNR=}", job) - text += ";"+utils.i18n_tool.I18n.getInstance(job).getText(f"{D.STEP_ATTR_ARGS=}", job) - return text + ";..;;;\n" - - diff --git a/execute_testcase.py b/execute_testcase.py index c02d452..0224943 100644 --- a/execute_testcase.py +++ b/execute_testcase.py @@ -5,16 +5,15 @@ # + craete_request() : testspec --> tdata.step --> archiv.request # + send_request() : archiv.request -- comp-config --> system.interface # --------------------------------------------------- -import sys -import os -import basic.step +import traceback + import basic.program as program import utils.tdata_tool import utils.report_tool import utils.path_tool import utils.file_tool import basic.componentHandling -import basic.message as message + # Press Umschalt+F10 to execute it or replace it with your code. # Press Double Shift to search everywhere for classes, files, tool windows, actions, and settings. @@ -22,22 +21,29 @@ import basic.message as message PROGRAM_NAME = "execute_testcase" def startPyJob(job): - cm = basic.componentHandling.ComponentManager.getInstance(job) - print("cm "+str(cm)) - cm.initComponents() - comps = cm.getComponents(PROGRAM_NAME) - job.m.setMsg("# Components initialized with these relevant components " + str(comps)) - tdata = utils.tdata_tool.getTestdata(job) - print(str(tdata)) - if not "_steps" in tdata: - raise Exception("no steps to execute in testdata") - for (step) in tdata["_steps"]: - if step.comp in comps: - comp = cm.getComponent(step.comp) - comp.execute_test(job, step, tdata) - job.m.merge(comp.m) - else: - job.m.setError(step.comp+" kann nicht aufgerufen werden!") + try: + cm = basic.componentHandling.ComponentManager.getInstance(job) + print("cm "+str(cm)) + cm.initComponents() + comps = cm.getComponents(PROGRAM_NAME) + job.m.setMsg("# Components initialized with these relevant components " + str(comps)) + tdata = utils.tdata_tool.getTestdata(job) + print(str(tdata)) + if not "_steps" in tdata: + raise Exception("no steps to execute in testdata") + for (step) in tdata["_steps"]: + if step.comp in comps: + comp = cm.getComponent(step.comp) + comp.execute_test(job, step, tdata) + job.m.merge(comp.m) + else: + job.m.setError(step.comp+" kann nicht aufgerufen werden!") + 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("+++++++++++++++++++++++++++++++++++++++++++++") # Press the green button in the gutter to run the script. diff --git a/init_testcase.py b/init_testcase.py index 2f2f381..9104cba 100644 --- a/init_testcase.py +++ b/init_testcase.py @@ -20,32 +20,39 @@ import basic.message as message PROGRAM_NAME = "init_testcase" def startPyJob(job): - cm = basic.componentHandling.ComponentManager.getInstance(job) - cm.initComponents() - comps = cm.getComponents(PROGRAM_NAME) - job.m.setMsg("# Components initialized with these relevant components " + str(comps)) - testdata = "" # tdata_tool.getTestdata(job) - print("------------------------------------------------------------") - for c in comps: - comp = cm.getComponent(c) - comp.m.logInfo("------- "+comp.name+" ----------------------------------------") - try: - if job.hasFunction("reset_TData"): - comp.reset_TData(job, B.PAR_TESTCASE) - if job.hasFunction("load_TData"): - comp.load_TData(job, B.PAR_TESTCASE, testdata) - if job.hasFunction("read_TData"): - comp.read_TData(job, path_tool.getKeyValue(job, P.KEY_PRECOND), B.PAR_TESTCASE) - except Exception as e: - txt = traceback.format_exc() - job.m.setFatal("Exception "+str(e)+"\n"+txt) - print("Exception " + str(e)+"\n"+txt) + try: + cm = basic.componentHandling.ComponentManager.getInstance(job) + cm.initComponents() + comps = cm.getComponents(PROGRAM_NAME) + job.m.setMsg("# Components initialized with these relevant components " + str(comps)) + testdata = "" # tdata_tool.getTestdata(job) + print("------------------------------------------------------------") + for c in comps: + comp = cm.getComponent(c) + comp.m.logInfo("------- "+comp.name+" ----------------------------------------") + try: + if job.hasFunction("reset_TData"): + comp.reset_TData(job, B.PAR_TESTCASE) + if job.hasFunction("load_TData"): + comp.load_TData(job, B.PAR_TESTCASE, testdata) + if job.hasFunction("read_TData"): + comp.read_TData(job, path_tool.getKeyValue(job, P.KEY_PRECOND), B.PAR_TESTCASE) + except Exception as e: + txt = traceback.format_exc() + job.m.setFatal("Exception "+str(e)+"\n"+txt) + print("Exception " + str(e)+"\n"+txt) - job.stopJob(0) - comp.m.logInfo("------- "+comp.name+" ----------------------------------------") - job.m.merge(comp.m) - print(str(comp)) - comp.conf["function"][PROGRAM_NAME] = comp.m.topmessage + job.stopJob(0) + comp.m.logInfo("------- "+comp.name+" ----------------------------------------") + job.m.merge(comp.m) + print(str(comp)) + comp.conf["function"][PROGRAM_NAME] = comp.m.topmessage + 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("+++++++++++++++++++++++++++++++++++++++++++++") # Press the green button in the gutter to run the script. if __name__ == '__main__': diff --git a/model/catalog.py b/model/catalog.py index 15860d3..1ced7e4 100644 --- a/model/catalog.py +++ b/model/catalog.py @@ -94,7 +94,7 @@ class Catalog: if not (isinstance(domain, str) or len(domain) < 1): raise Exception(EXP_KEY_MISSING, (domain)) if domain in self.catalog: - return + return self.catalog[domain] pathname = tools.config_tool.select_config_path(job, P.KEY_CATALOG, domain) if pathname is None: raise Exception(EXP_KEY_MISSING, (domain)) diff --git a/test_executer.py b/test_executer.py index 7d22ef4..3d4713b 100644 --- a/test_executer.py +++ b/test_executer.py @@ -18,6 +18,9 @@ import tools.job_tool as job_tool # import collect_testcase # import compare_testcase # import finish_testsuite +import model.factory +import model.catalog +import tools.job_tool PROGRAM_NAME = "test_executer" myjob = None @@ -30,14 +33,41 @@ def getTime(): 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) + while int(getattr(job.par, B.PAR_STEP, "99")) < 3: + executeStep(job, steps, programs) + snr = int(getattr(job.par, B.PAR_STEP, "99")) + setattr(job.par, B.PAR_STEP, str(snr+1)) + 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("+++++++++++++++++++++++++++++++++++++++++++++") - tdata = {} #tdata_tool.getTestdata(job) - if not hasattr(job.par, B.PAR_STEP): - raise Exception("Parameter " + B.PAR_STEP + " is missing") - testcases = getattr(job.par, B.PAR_TESTCASE) - for step in tdata[B.DATA_NODE_STEPS]: - if int(step.exexStep) != int(getattr(job.par, "step")): - continue +def executeStep(job, steps, programs): + # TODO run over steps - definition ?! + """ for arg in step["args"]: if arg == "start": print("start "+str(step["args"][arg])) @@ -56,42 +86,45 @@ def startPyJob(job): if arg == "report": # testsuite #basic.program.Job.pushInstance(myjob) pass - - #basic.program.Job.pushInstance(myjob) - setattr(job.par, "testcases", testcases) - # myjob.stopJob(1) - - -"""def startStepProgram(step, job, jobargs): - myjob = basic.program.Job("unit") # meaning temp - myjob.par.setParameterArgs(job, jobargs) - myjob.setProgram(step.start) - #myjob.pushInstance(myjob) - myjob.startJob() - try: - job.m.logInfo(step.start + " starting") - if step.start == "init_testcase": - init_testcase.startPyJob(myjob) - elif step.start == "execute_testcase": - execute_testcase.startPyJob(myjob) - elif step.start == "collect_testcase": - collect_testcase.startPyJob(myjob) - elif step.start == "compare_testcase": - compare_testcase.startPyJob(myjob) - job.m.logInfo(step.start + " finished") - except Exception as e: - txt1 = traceback.format_stack() - print("==================================================================0") - print(txt1) - print("==================================================================0") - txt2 = traceback.format_exc() - print(txt2) - print("==================================================================0") - job.m.setError(step.start + " aborted") - finally: - myjob.stopJob(1) - #myjob.popInstance(myjob) -""" + """ + verify = True + 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) != 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) diff --git a/tools/job_const.py b/tools/job_const.py index 46b2d48..d0556a8 100644 --- a/tools/job_const.py +++ b/tools/job_const.py @@ -54,3 +54,31 @@ CHOICE_TC = "Testfall auswaehlen" CHOICE_ARCHIV = "Testausfuehrung auswaehlen" CHOICE_TIME = "Testzeit auswaehlen" CHOICE_VARIANT = "Testvariante auswaehlen" + +# the sequence of a complete testexecution +STEP_INIT_TP = 10 +STEP_INIT_TS = 12 +STEP_INIT_TC = 14 +STEP_EXEC_TS1 = 22 +STEP_EXEC_TC1 = 24 +STEP_EXEC_TC2 = 26 +STEP_EXEC_TS2 = 28 +STEP_COLL_TC = 36 +STEP_COLL_TS = 38 +STEP_COMP_TC = 46 +STEP_COMP_TS = 48 +STEP_FIN_TC = 56 +STEP_FIN_TS = 58 +STEP_FIN_TP = 60 +STEP_BREAK = 99 +""" conditional break if basic configured as not full-automatized """ +STEPS_TESTPLAN = [ STEP_INIT_TP, STEP_INIT_TS, STEP_INIT_TC, + STEP_EXEC_TS1, STEP_EXEC_TC1, STEP_BREAK, STEP_EXEC_TC2, STEP_EXEC_TS2, + STEP_COLL_TC, STEP_COLL_TS, + STEP_COMP_TC, STEP_COMP_TS, STEP_FIN_TP ] +STEPS_TESTSUITE = [ STEP_INIT_TS, STEP_INIT_TC, + STEP_EXEC_TS1, STEP_EXEC_TC1, STEP_BREAK, STEP_EXEC_TC2, STEP_EXEC_TS2, + STEP_COLL_TC, STEP_COLL_TS, + STEP_COMP_TC, STEP_COMP_TS, STEP_FIN_TS ] +STEPS_TESTCASE = [ STEP_INIT_TC, STEP_EXEC_TC1, STEP_BREAK, STEP_EXEC_TC2, + STEP_COLL_TC, STEP_COMP_TC, STEP_FIN_TC ] diff --git a/tools/job_tool.py b/tools/job_tool.py index def9500..ae600dc 100644 --- a/tools/job_tool.py +++ b/tools/job_tool.py @@ -34,14 +34,14 @@ import clean_workspace import init_testsuite import test_executer -try: - import collect_testcase - import compare_testcase - import execute_testcase - import finish_testsuite - import init_testcase -except Exception as e: - pass +#try: +# import collect_testcase +# import compare_testcase +# import execute_testcase +# import finish_testsuite +# import init_testcase +#except Exception as e: +# pass import tools.path_tool import tools.file_tool import tools.date_tool as date_tool @@ -147,31 +147,46 @@ def start_child_process(job, args): childJob.m.logDebug(verifiy, childArgs) childJob.startJob() if args[B.PAR_PROGRAM] == "init_testcase": + import init_testcase init_testcase.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "execute_testcase": + import execute_testcase execute_testcase.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "collect_testcase": + import collect_testcase collect_testcase.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "compare_testcase": + import compare_testcase compare_testcase.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "init_testsuite": + import init_testsuite init_testsuite.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "execute_testsuite": + import execute_testcase print("execute_testsuite.startPyJob(childJob) not implemented") elif args[B.PAR_PROGRAM] == "collect_testsuite": + import collect_testcase print("collect_testsuite.startPyJob(childJob) not implemented") elif args[B.PAR_PROGRAM] == "finish_testsuite": + import finish_testsuite finish_testsuite.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "test_executer": + import test_executer test_executer.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "check_environment": + import check_environment check_environment.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "check_configuration": + import check_configuration check_configuration.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "check_specification": + import check_specification check_specification.startPyJob(childJob) elif args[B.PAR_PROGRAM] == "clean_workspace": + import clean_workspace clean_workspace.startPyJob(childJob) + else: + raise Exception("unkown program {}".format(args[B.PAR_PROGRAM])) childJob.stopJob(1) return childJob diff --git a/tools/step_tool.py b/tools/step_tool.py new file mode 100644 index 0000000..c5be125 --- /dev/null +++ b/tools/step_tool.py @@ -0,0 +1,22 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- +# --------------------------------------------------------------------------------------------------------- +# Author : Ulrich Carmesin +# Source : gitea.ucarmesin.de +# --------------------------------------------------------------------------------------------------------- +""" +this module implements the functionality of a test-step +which is defined in the test-specification +and is executed by any executer +there are 2 kinds of test-step +a) execute specific component in the job +b) execute specific test-entity in the test-suite-execution +""" +import basic.constants as B +import tools.data_const as D +# import utils.i18n_tool + +LIST_ARGS = [ + "start", # for starting the specified main-program + "fct" # for calling the specified component-function +] diff --git a/tools/value_tool.py b/tools/value_tool.py index a2db04a..db61e56 100644 --- a/tools/value_tool.py +++ b/tools/value_tool.py @@ -109,14 +109,18 @@ def get_key_value(job, key, comp=None): 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) + try: + 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) + except: + job.m.logError("Pattern konnte nicht aufgeloest werden " + pattern) + return None return None def compose_string(job, pattern, comp):