Compare commits

...

3 Commits

  1. 24
      basic/compexec.py
  2. 28
      basic/program.py
  3. 91
      basic/step.py
  4. 9
      execute_testcase.py
  5. 3
      init_testcase.py
  6. 0
      start_dialog.py
  7. 3
      test/test_job.py
  8. 28
      test/test_tdata.py
  9. 52
      test_executer.py
  10. 12
      utils/config/path.yml
  11. 0
      utils/filexml_tool.py
  12. 5
      utils/tdata_tool.py

24
basic/testexec.py → basic/compexec.py

@ -229,6 +229,30 @@ class Testexecuter():
def get_Response(self, granularity): def get_Response(self, granularity):
pass pass
def execute_test(self, job, step, tdata):
"""
the function executes a teststep. The exact step with partial steps are defined in the component-configuration
under teststeps and the step-attribute.
:param job:
:param step: the step object
:param tdata:
:return:
"""
if step.fct in self.conf["teststeps"]:
for s in self.conf["teststeps"][step.fct]:
stepconf = self.conf["teststeps"][step.fct][s]
if stepconf["tool"] == "file":
print("file-tool")
elif stepconf["tool"] == "api":
print("api-tool")
elif stepconf["tool"] == "cli":
print("cli-tool")
elif stepconf["tool"] == "db":
print("db-tool")
else:
print("nichts da")
def finish_Test(self, granularity): def finish_Test(self, granularity):
""" """
initialization-routine for finish-step initialization-routine for finish-step

28
basic/program.py

@ -278,17 +278,41 @@ class Job:
def hasElement(self, parameter, elem): 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): if hasattr(self.par, parameter):
print (parameter + " in Parameter") print (parameter + " in Parameter")
if getattr(self.par, parameter).find(elem) >= 0: if getattr(self.par, parameter).find(elem) >= 0:
return True return True
return False return False
return True return True
def hascomponente(self, komp): def hascomponente(self, comp):
return self.hasElement("componente", komp) """
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): 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) return self.hasElement("function", fct)
def hasTool(self, tool): 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) return self.hasElement("tool", tool)
def getMessageLevel(self, errtyp, elem): def getMessageLevel(self, errtyp, elem):
if (not hasattr(self, "m")) or (self.m is None): if (not hasattr(self, "m")) or (self.m is None):

91
basic/step.py

@ -0,0 +1,91 @@
#!/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
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
"""
def __init__(self):
self.comp = ""
self.refLine = ""
self.execStep = ""
self.args = {}
def parseOldStep(job, fields):
step = {}
step[B.DATA_NODE_COMP] = fields[D.STEP_COMP_I]
step[B.ATTR_EXEC_REF] = fields[D.STEP_EXECNR_I]
step[B.ATTR_DATA_REF] = fields[D.STEP_REFNR_I]
step[B.ATTR_STEP_ARGS] = {}
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:
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 + "" + l)
step[B.ATTR_STEP_ARGS][b[0]] = b[1]
# data[B.DATA_NODE_STEPS].append(step)
return step
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]
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:
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 + "" + str(fields))
step.args[b[0]] = b[1]
if b[0] in LIST_ARGS:
setattr(step, b[0], b[1])
# data[B.DATA_NODE_STEPS].append(step)
return step

9
execute_testcase.py

@ -1,6 +1,7 @@
# This is a sample Python script. # This is a sample Python script.
import sys# import sys#
import os import os
import basic.step
import basic.program as program import basic.program as program
import utils.tdata_tool import utils.tdata_tool
import utils.report_tool import utils.report_tool
@ -24,12 +25,12 @@ def startPyJob(job):
if not "_steps" in tdata: if not "_steps" in tdata:
raise Exception("no steps to execute in testdata") raise Exception("no steps to execute in testdata")
for (step) in tdata["_steps"]: for (step) in tdata["_steps"]:
if step["comp"] in comps: if step.comp in comps:
comp = cm.getComponent(step["comp"]) comp = cm.getComponent(step.comp)
comp.execute_testcase(step, tdata) comp.execute_test(job, step, tdata)
job.m.merge(comp.m) job.m.merge(comp.m)
else: else:
job.m.setError(step["comp"]+" kann nicht aufgerufen werden!") job.m.setError(step.comp+" kann nicht aufgerufen werden!")
# Press the green button in the gutter to run the script. # Press the green button in the gutter to run the script.

3
init_testcase.py

@ -23,8 +23,11 @@ def startPyJob(job):
for c in comps: for c in comps:
comp = cm.getComponent(c) comp = cm.getComponent(c)
comp.m.logInfo("------- "+comp.name+" ----------------------------------------") comp.m.logInfo("------- "+comp.name+" ----------------------------------------")
if job.hasFunction("reset_TData"):
comp.reset_TData(B.PAR_TESTCASE) comp.reset_TData(B.PAR_TESTCASE)
if job.hasFunction("load_TData"):
comp.load_TData(B.PAR_TESTCASE, testdata) comp.load_TData(B.PAR_TESTCASE, testdata)
if job.hasFunction("read_TData"):
comp.read_TData("vorher", B.PAR_TESTCASE) comp.read_TData("vorher", B.PAR_TESTCASE)
comp.m.logInfo("------- "+comp.name+" ----------------------------------------") comp.m.logInfo("------- "+comp.name+" ----------------------------------------")
job.m.merge(comp.m) job.m.merge(comp.m)

0
start_dialog.py

3
test/test_job.py

@ -52,6 +52,7 @@ class MyTestCase(unittest.TestCase):
programs = ["init_testcase"] programs = ["init_testcase"]
testcase = "TC0001" testcase = "TC0001"
timexec = "2022-06-28_21-23-34" timexec = "2022-06-28_21-23-34"
fct = "read_TData"
# os.system("python "+os.path.join(HOME_PATH, "check_environment.py")+" -a TEST -e ENV01") # os.system("python "+os.path.join(HOME_PATH, "check_environment.py")+" -a TEST -e ENV01")
# os.system("python "+os.path.join(HOME_PATH, "init_testsuite.py")+" -a TEST -e ENV01 " # os.system("python "+os.path.join(HOME_PATH, "init_testsuite.py")+" -a TEST -e ENV01 "
# "-ts "+os.path.join(HOME_PATH, "test","lauf","V0.1","implement_2021-08-28_23-50-51")+" -dt csv -ds implement -dn firstunit") # "-ts "+os.path.join(HOME_PATH, "test","lauf","V0.1","implement_2021-08-28_23-50-51")+" -dt csv -ds implement -dn firstunit")
@ -64,7 +65,7 @@ class MyTestCase(unittest.TestCase):
if "init_testcase" in programs: if "init_testcase" in programs:
program = "init_testcase" program = "init_testcase"
job = Job("unit") job = Job("unit")
args = { B.PAR_APP: "TESTAPP", B.PAR_ENV: "ENV01", "modus": "unit", args = { B.PAR_APP: "TESTAPP", B.PAR_ENV: "ENV01", "modus": "unit", B.PAR_FCT: fct,
B.PAR_TCDIR: os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], testcase, timexec), B.PAR_TCDIR: os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], testcase, timexec),
"step": 1 } "step": 1 }
# "usecase": "TST001", "tstime": "2022-03-17_17-28"} # "usecase": "TST001", "tstime": "2022-03-17_17-28"}

28
test/test_tdata.py

@ -19,7 +19,7 @@ OS_SYSTEM = test.constants.OS_SYSTEM
# here you can select single testfunction for developping the tests # here you can select single testfunction for developping the tests
TEST_FUNCTIONS = ["test_tdata", "test_getCsvSpec_data", "test_getCsvSpec_tree", "test_getCsvSpec_key", TEST_FUNCTIONS = ["test_tdata", "test_getCsvSpec_data", "test_getCsvSpec_tree", "test_getCsvSpec_key",
"test_getCsvSpec_conf", "test_extractPattern", "test_parseCsv"] "test_getCsvSpec_conf", "test_extractPattern", "test_parseCsv"]
TEST_FUNCTIONS = ["test_getCsvSpec_data"] # TEST_FUNCTIONS = ["test_getCsvSpec_data"]
class MyTestCase(unittest.TestCase): class MyTestCase(unittest.TestCase):
mymsg = "--------------------------------------------------------------" mymsg = "--------------------------------------------------------------"
@ -119,9 +119,21 @@ class MyTestCase(unittest.TestCase):
cnttest += 3 cnttest += 3
for step in tdata[B.DATA_NODE_STEPS]: for step in tdata[B.DATA_NODE_STEPS]:
print(step) print(step)
self.assertIn(B.DATA_NODE_COMP, step) self.assertEqual(hasattr(step, B.DATA_NODE_COMP), True)
self.assertIn(B.ATTR_DATA_REF, step) # self.assertEqual(hasattr(step, B.ATTR_DATA_REF), True)
self.assertIn(B.ATTR_STEP_ARGS, step) self.assertEqual(hasattr(step, B.ATTR_STEP_ARGS), True)
cnttest += 3
specLines = [
"step:1;testa;1;1;table:_lofts;action:export;;;;;",
"#;;;;;;"
]
tdata = {}
tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_DATA)
print(tdata)
self.assertEqual(1, len(tdata))
self.assertIn(B.DATA_NODE_STEPS, tdata)
self.assertIsInstance(tdata[B.DATA_NODE_STEPS], list)
self.assertEqual(2, len(tdata[B.DATA_NODE_STEPS][0].args))
cnttest += 3 cnttest += 3
if B.DATA_NODE_TABLES in tests: if B.DATA_NODE_TABLES in tests:
specLines = [ specLines = [
@ -216,7 +228,7 @@ class MyTestCase(unittest.TestCase):
print(str(tdata)) print(str(tdata))
self.assertEqual(1, len(tdata)) self.assertEqual(1, len(tdata))
self.assertEqual(1, len(tdata["_tables"])) self.assertEqual(1, len(tdata["_tables"]))
self.assertEqual(2, len(tdata["_tables"]["capital"])) self.assertEqual(3, len(tdata["_tables"]["capital"]))
self.assertEqual(3, len(tdata["_tables"]["capital"]["_keys"])) self.assertEqual(3, len(tdata["_tables"]["capital"]["_keys"]))
cnttest += 4 cnttest += 4
specLines = [ specLines = [
@ -231,9 +243,9 @@ class MyTestCase(unittest.TestCase):
#tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_TREE) #tdata = t.parseCsvSpec(job.m, specLines, D.CSV_SPECTYPE_TREE)
print(str(tdata)) print(str(tdata))
self.assertEqual(1, len(tdata)) self.assertEqual(1, len(tdata))
self.assertNotIn("capital", tdata["_tables"]) self.assertIn("capital", tdata["_tables"])
self.assertEqual(1, len(tdata["_tables"])) self.assertEqual(2, len(tdata["_tables"]))
self.assertEqual(2, len(tdata["_tables"]["country"])) self.assertEqual(3, len(tdata["_tables"]["country"]))
self.assertEqual(2, len(tdata["_tables"]["country"]["_keys"])) self.assertEqual(2, len(tdata["_tables"]["country"]["_keys"]))
cnttest += 4 cnttest += 4
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)

52
test_executer.py

@ -1,10 +1,13 @@
from datetime import datetime from datetime import datetime
import traceback
import basic.program import basic.program
import basic.constants as B import basic.constants as B
import utils.tdata_tool import utils.tdata_tool
import init_testcase import init_testcase
import init_testsuite import init_testsuite
import execute_testcase import execute_testcase
import collect_testcase
import compare_testcase
import finish_testsuite import finish_testsuite
PROGRAM_NAME = "test_executer" PROGRAM_NAME = "test_executer"
@ -16,16 +19,18 @@ def getTime():
time = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") time = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
return time return time
def start(myjob): def startPy(pjob):
myjob = pjob
myjob.m.setMsg("# # # # # start executer # # # # # ") myjob.m.setMsg("# # # # # start executer # # # # # ")
tdata = utils.tdata_tool.getTestdata() tdata = utils.tdata_tool.getTestdata()
job = basic.program.Job("unit") job = basic.program.Job("unit")
if not hasattr(myjob.par, B.PAR_STEP):
raise Exception("Parameter " + B.PAR_STEP + " is missing")
testcases = getattr(myjob.par, B.PAR_TESTCASE) testcases = getattr(myjob.par, B.PAR_TESTCASE)
if not hasattr(myjob.par, "step"):
raise Exception("Parameter step is missing")
for step in tdata[B.DATA_NODE_STEPS]: for step in tdata[B.DATA_NODE_STEPS]:
if int(step["_nr"]) != int(getattr(myjob.par, "step")): if int(step.exexStep) != int(getattr(myjob.par, "step")):
continue continue
for arg in step["args"]: for arg in step["args"]:
if arg == "start": if arg == "start":
if "testsuite" in step["args"][arg]: if "testsuite" in step["args"][arg]:
@ -72,15 +77,48 @@ def start(myjob):
# myjob.stopJob(1) # myjob.stopJob(1)
def startStepProgram(step, job, jobargs):
myjob = basic.program.Job("unit") # meaning temp
myjob.par.setParameterArgs(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)
if __name__ == '__main__': if __name__ == '__main__':
print(PROGRAM_NAME) print(PROGRAM_NAME)
x = basic.program.Job(PROGRAM_NAME) x = basic.program.Job(PROGRAM_NAME)
#m = message.Message(3)
#m.closeMessage()
x.startJob() x.startJob()
try:
x.m.logDebug(str(vars(x.par)) + "\n" + str(vars(x.conf))) x.m.logDebug(str(vars(x.par)) + "\n" + str(vars(x.conf)))
if x.m.isRc("fatal"): if x.m.isRc("fatal"):
x.stopJob() x.stopJob()
exit(x.m.rc * (-1) + 3) exit(x.m.rc * (-1) + 3)
start(x) startPy(x)
except:
x.m.setError(PROGRAM_NAME + " aborted")
finally:
x.stopJob() x.stopJob()

12
utils/config/path.yml

@ -14,15 +14,19 @@ pattern:
sumfile: xxx sumfile: xxx
backup: backup backup: backup
reffile: Herkunft.txt reffile: Herkunft.txt
appdir: APP
tc: testfall tc: testfall
ts: testlauf ts: testlauf
debugname: debug debugname: debug
logname: log logname: log
preexec: env-pre-exec # only for dry unit-test
postexec: env-post-exec # only for dry unit-test
debugs: "{job.conf.home}/test/log" debugs: "{job.conf.home}/test/log"
# environment # environment
envbase: "{job.conf.environment}/{job.par.environment}" envbase: "{job.conf.environment}/{job.par.environment}"
envlog: "{envbase}/{log}" envlog: "{envbase}/{log}"
envparfile: "{envbase}/{parfile}" envparfile: "{envbase}/{parfile}"
envappdir: "{envbase}/{appdir}/{comp.name}"
# testcase # testcase
tcbase: "{job.conf.archiv}/{job.par.testcase}/{job.par.tctime}" tcbase: "{job.conf.archiv}/{job.par.testcase}/{job.par.tctime}"
tclog: "{tcbase}/{log}" tclog: "{tcbase}/{log}"
@ -33,6 +37,14 @@ pattern:
tcrundiff: "{tcresult}/{rundiff}" tcrundiff: "{tcresult}/{rundiff}"
tcprecond: "{tcresult}/{precond}" tcprecond: "{tcresult}/{precond}"
tcpostcond: "{tcresult}/{postcond}" tcpostcond: "{tcresult}/{postcond}"
# testdata
tdbase: "{job.conf.testdata}/{job.par.testcase}"
tdresult: "{tdbase}/{result}"
tdparfile: "{tdbase}/{parfile}"
tdprecond: "{tdresult}/{precond}"
tdpostcond: "{tdresult}/{postcond}"
tdpreexec: "{tdbase}/{preexec}/{comp.name}" # only for dry unit-test
tdpostexec: "{tdbase}/{postexec}/{comp.name}" # only for dry unit-test
# testset # testset
tsbase: "{job.conf.archiv}/{ts}/{job.par.usecase}_{job.par.tstime}" tsbase: "{job.conf.archiv}/{ts}/{job.par.usecase}_{job.par.tstime}"
tslog: "{tsbase}/{log}" tslog: "{tsbase}/{log}"

0
utils/filexml_tool.py

5
utils/tdata_tool.py

@ -27,6 +27,7 @@ import utils.file_tool
import basic.constants as B import basic.constants as B
import utils.data_const as D import utils.data_const as D
import utils.date_tool import utils.date_tool
import basic.step
TOOL_NAME = "tdata_tool" TOOL_NAME = "tdata_tool"
""" name of the tool in order to switch debug-info on """ """ name of the tool in order to switch debug-info on """
@ -118,6 +119,7 @@ def getCsvSpec(msg, filename, type):
def parseCsvSpec(msg, lines, type): def parseCsvSpec(msg, lines, type):
job = basic.program.Job.getInstance()
data = {} data = {}
header = [] header = []
h = [] # from a[] h = [] # from a[]
@ -140,6 +142,8 @@ def parseCsvSpec(msg, lines, type):
if (a[0].lower() == D.CSV_BLOCK_STEP): if (a[0].lower() == D.CSV_BLOCK_STEP):
if (not B.DATA_NODE_STEPS in data): if (not B.DATA_NODE_STEPS in data):
data[B.DATA_NODE_STEPS] = [] data[B.DATA_NODE_STEPS] = []
step = basic.step.parseStep(job, fields)
"""
step = {} step = {}
step[B.DATA_NODE_COMP] = fields[D.STEP_COMP_I] step[B.DATA_NODE_COMP] = fields[D.STEP_COMP_I]
step[B.ATTR_EXEC_REF] = fields[D.STEP_EXECNR_I] step[B.ATTR_EXEC_REF] = fields[D.STEP_EXECNR_I]
@ -163,6 +167,7 @@ def parseCsvSpec(msg, lines, type):
if len(b) < 2: if len(b) < 2:
raise Exception(D.EXCP_MALFORMAT + "" + l) raise Exception(D.EXCP_MALFORMAT + "" + l)
step[B.ATTR_STEP_ARGS][b[0]] = b[1] step[B.ATTR_STEP_ARGS][b[0]] = b[1]
"""
data[B.DATA_NODE_STEPS].append(step) data[B.DATA_NODE_STEPS].append(step)
continue continue
elif (a[0].lower() == D.CSV_BLOCK_OPTION): elif (a[0].lower() == D.CSV_BLOCK_OPTION):

Loading…
Cancel
Save