Browse Source

start_dialog

refactor
Ulrich 2 years ago
parent
commit
0bae1ed3e7
  1. 5
      basic/message.py
  2. 14
      basic/program.py
  3. 1
      execute_testcase.py
  4. 20
      init_testcase.py
  5. 318
      start_dialog.py
  6. 26
      test/test_71job.py
  7. 12
      utils/date_tool.py
  8. 1
      utils/dbshive_tool.py
  9. 12
      utils/job_tool.py
  10. 7
      utils/path_tool.py

5
basic/message.py

@ -250,7 +250,10 @@ class Message:
def debug(self, prio, text):
""" eigentliche Schreibroutine: hierin wird debug-Level beruecksichtgigt"""
if (int(prio) < int(self.level)+1) : #and self.debugfile.closed == False:
self.debugfile.write(text + "\n")
try:
self.debugfile.write(text + "\n")
except:
print("debug closed "+text)
def resetLog(self):
self.messages = []

14
basic/program.py

@ -51,11 +51,11 @@ jobdef = {
"unit": {
"pardef": "",
"pfilesource": "",
"pfiletarget": "envparfile",
"basedir": "envbase",
"dirname": "envdir",
"pfiletarget": "",
"basedir": "workbase",
"dirname": "workdir",
"loglevel": basic.message.LIMIT_DEBUG,
"logpath": "{job.par.envdir}/{log}/log_{job.start:H}.txt" },
"logpath": "{job.conf.data}/workspace/unittest_{job.start:H}.txt" },
"check_environment": {
"pardef": "",
"pfilesource": "",
@ -321,9 +321,11 @@ class Job:
self.dumpParameter()
print("stopJob " + str(self.m.messages) + ", " + str(self.m.debugfile))
self.m.logInfo("# # " + self.m.topmessage + " # # # ")
self.m.logInfo("# # # Stop Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # " + self.ende.strftime("%d.%m.%Y %H:%M:%S") + " # # # ")
footer = "# # # Stop Job " + utils.date_tool.formatParsedDate(str(self.start), utils.date_tool.F_LOG)
footer += " # " + utils.date_tool.formatParsedDate(str(self.ende), utils.date_tool.F_LOG) + " # # # "
self.m.logInfo(footer)
self.m.debug(basic.message.LIMIT_INFO, "# # " + self.m.topmessage + " # # # ")
self.m.debug(basic.message.LIMIT_INFO, "# # # Stop Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # " + self.ende.strftime("%d.%m.%Y %H:%M:%S") + " # # # RC: " + str(self.m.getFinalRc()))
self.m.debug(basic.message.LIMIT_INFO, footer + " # # # RC: " + str(self.m.getFinalRc()))
self.m.closeMessage()
rc = self.m.getFinalRc()
print ("rc " + str(rc))

1
execute_testcase.py

@ -28,6 +28,7 @@ def startPyJob(job):
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"]:

20
init_testcase.py

@ -6,6 +6,7 @@
# + load_testcase() : testspec --> tdata --> system.data
# + select_testcase() : system.data --> data --> archiv.result
# ---------------------------------------------------import os
import traceback
import basic.program as program
import utils.tdata_tool
import basic.componentHandling
@ -28,12 +29,19 @@ def startPyJob(job):
for c in comps:
comp = cm.getComponent(c)
comp.m.logInfo("------- "+comp.name+" ----------------------------------------")
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, utils.path_tool.getKeyValue(job, P.KEY_PRECOND), B.PAR_TESTCASE)
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, utils.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))

318
start_dialog.py

@ -11,60 +11,150 @@
"""
import os.path
import json
import re
import basic.program
import basic.constants as B
import collect_testcase
import compare_testcase
import copy_appdummy
import execute_testcase
import finish_testsuite
import init_testcase
import init_testsuite
import test_executer
import utils.file_tool
import utils.data_const as D
import utils.date_tool
import utils.path_tool
import utils.path_const as P
tempJob = {}
PROGRAM_NAME = "unit"
PROGRAM_NAME = "service"
DLG_TESTCASE = "Testfall"
DLG_TESTSUITE = "Testsuite"
DLG_COMPLETE = "Komplettausfuehrung"
LIST_DLG_GRAN = [DLG_TESTCASE, DLG_TESTSUITE]
DLG_START_QUESTION = "was soll getestet werden"
# DLG_TESTPLAN = "Testplan"
DLG_ENVIRONMENT = "Umgebung"
DLG_APPLICATION = "Anwendung"
DLG_REDO = "wiederholen"
DLG_CONTINUE = "fortsetzen"
DLG_DUMMY_STEP = "Dummy-Schritt"
DLG_NEWJOB = "neuer Job"
JSON_FILE = "actualJob.json"
JOB_NR = {
"init_testsuite": {
"jobnr": "0" },
"execute_testsuite": {
"jobnr": "1"},
"collect_testsuite": {
"jobnr": "2"},
"compare_testsuite": {
"jobnr": "3"},
"finish_testsuite": {
"jobnr": "4"},
"init_testcase": {
"jobnr": "5" },
"execute_testcase": {
"jobnr": "6" },
"collect_testcase": {
"jobnr": "7" },
"compare_testcase": {
"jobnr": "8" },
DLG_TESTSUITE : {
"start": "init_testsuite",
"init_testsuite": {
"jobnr": "0" },
"execute_testsuite": {
"jobnr": "1"},
"collect_testsuite": {
"jobnr": "2"},
"compare_testsuite": {
"jobnr": "3"},
"finish_testsuite": {
"jobnr": "4"}
},
DLG_TESTCASE: {
"start": "init_testcase",
"init_testcase": {
"jobnr": "5" },
"execute_testcase": {
"jobnr": "6" },
"collect_testcase": {
"jobnr": "7" },
"copy_appdummy": {
"jobnr": "8" },
"compare_testcase": {
"jobnr": "9" },
},
"check_environment": {
"jobnr": "9" },
"jobnr": "10" },
"test_executer": {
"jobnr": "10"},
"jobnr": "11"},
}
JOB_LIST = [
"init_testsuite",
"execute_testsuite",
"collect_testsuite",
"compare_testsuite",
"finish_testsuite",
"init_testcase",
"execute_testcase",
"collect_testcase",
"compare_testcase",
"check_environment",
"test_executer"
"init_testsuite", # 0
"execute_testsuite", # 1
"collect_testsuite", # 2
"compare_testsuite", # 3
"finish_testsuite", # 4
"init_testcase", # 5
"execute_testcase", # 6
"collect_testcase", # 7
"copy_appdummy", # 8
"compare_testcase", # 9
"check_environment", # 10
"test_executer" # 11
]
appList = []
envList = ["ENV01"]
testList = {"TEST": [
"TC0001", "TST001"
]}
envList = []
entities = {}
entities[DLG_TESTCASE] = {}
entities[DLG_TESTSUITE] = {}
def readContext(job):
for k in job.conf.confs["applications"]:
for k in job.conf.confs[B.SUBJECT_APPS]:
appList.append(k)
path = job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ENV]
if os.path.exists(path):
for d in os.listdir(path):
print ("-- "+d)
if not os.path.isdir(os.path.join(path, d)):
continue
if d[0:1] == "_":
continue
envList.append(d)
path = job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_TDATA]
if os.path.exists(path):
for d in os.listdir(path):
print("tdata path "+d)
if not os.path.isdir(os.path.join(path, d)):
print("continue a")
continue
if d[0:1] == "_":
print("continue b")
continue
specpath = os.path.join(path, d, D.DFILE_TESTCASE_NAME + ".csv")
readSpec(job, d, DLG_TESTCASE, specpath)
specpath = os.path.join(path, d, D.DFILE_TESTSUITE_NAME + ".csv")
readSpec(job, d, DLG_TESTSUITE, specpath)
def readSpec(job, testentity, testgran, specpath):
print("spec "+specpath)
if not os.path.isfile(specpath):
print("continue c")
return
text = utils.file_tool.readFileText(job, specpath, job.m)
print("-----------\n"+text+"\n------------------")
if re.match(r".*?depricated;[jJyY]", text):
return
if re.match(r".*\nhead:application;", text):
print("## app gematcht")
res = re.search(r".*head:application;(.+)\n", text)
apps = res.group(1).replace(";", ",").split(",")
print("# "+str(apps))
for a in apps:
if len(a) < 1:
break
if a not in entities[testgran]:
entities[testgran][a] = []
print(a+" in "+testentity+" "+testgran+" -- "+str(entities))
entities[testgran][a].append(testentity)
def printProc(job, process):
print("--------------------------------------------------")
for k in process:
print("| {0:15s} : {1}".format(k, process[k]))
print("--------------------------------------------------")
def restartActualProcess(job):
@ -74,61 +164,105 @@ def restartActualProcess(job):
"""
path = os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG], JSON_FILE)
if os.path.exists(path):
actProc = utils.file_tool.readFileDict(job, path)
jobNr = int(JOB_NR[actProc["job"]]["jobnr"])
question = "Prozess "
choiceList = ""
if jobNr < 8 and jobNr != 4:
question += "F_ortsetzen | "
choiceList += "f | "
question += "W_iederholen | N_euen Prozess | X_exit"
choiceList = "w | n | x ?"
print("+-----------------------------------------------")
for k in actProc:
print('| {:6s} : {:60s}'.format(k, actProc[k]))
print("+-----------------------------------------------")
print(question)
choice = input("Auswahl "+choiceList)
choiceList.replace(" | ","")
if choice.lower() not in choiceList[:-1]:
print("FEHLER : falsche Auswahl")
elif choice.lower() == "x":
job.stopJob(0)
elif choice.lower() == "w":
actProc = utils.file_tool.readFileDict(job, path, job.m)
print("restartActJob "+str(actProc))
printProc(job, actProc)
step = int(actProc["step"])
if actProc["program"] == "test_executer":
if step > 5:
dialogProcess(job)
else:
actProc["step"] = str(step+1)
startProcess(job, actProc)
selection = [DLG_NEWJOB, DLG_REDO]
nr = int(JOB_NR[actProc["gran"]][actProc["program"]]["jobnr"])
if (actProc["gran"] == DLG_TESTSUITE and nr < 4) or (actProc["gran"] == DLG_TESTCASE and nr < 9):
selection.append(DLG_CONTINUE)
if nr == 7:
selection.append(DLG_DUMMY_STEP)
choice = getChoice(job, selection, DLG_ENVIRONMENT)
print(choice)
if choice == DLG_REDO:
startProcess(job, actProc)
elif choice == DLG_DUMMY_STEP:
actProc["program"] = JOB_LIST[nr+1]
startProcess(job, actProc)
elif choice.lower() == "f":
actProc["job"] = JOB_NR[int(jobNr)+1]
elif choice == DLG_CONTINUE:
if nr == 7:
nr = 9
else:
nr += 1
print (" act nr "+str(nr))
actProc["step"] = str(step + 1)
actProc["program"] = JOB_LIST[nr]
startProcess(job, actProc)
elif choice.lower() == "n":
createProcess(job)
else:
print("unbekannte Situation")
elif choice == DLG_NEWJOB:
dialogProcess(job)
else:
createProcess(job)
dialogProcess(job)
def createProcess(job):
def dialogProcess(job):
"""
dialog for selection and starting a process
:param job:
:return:
"""
process = {}
index = 0
print("create new process")
process["app"] = getChoice(job, appList, "Anwendung")
selection = []
selection = LIST_DLG_GRAN
if DLG_TESTCASE + " - " + DLG_COMPLETE not in selection:
selection.append(DLG_TESTCASE + " - " + DLG_COMPLETE)
if DLG_TESTSUITE + " - " + DLG_COMPLETE not in selection:
selection.append(DLG_TESTSUITE + " - " + DLG_COMPLETE)
choice = getChoice(job, LIST_DLG_GRAN, DLG_START_QUESTION)
if DLG_COMPLETE in choice:
process["gran"] = choice[0:-3-len(DLG_COMPLETE)]
process["program"] = "test_executer"
process["step"] = 1
else:
process["gran"] = choice
process["program"] = JOB_NR[process["gran"]]["start"]
process["step"] = 1
if len(appList) == 1:
process["app"] = appList[0]
else:
process["app"] = getChoice(job, appList, DLG_ENVIRONMENT)
#
if len(envList) == 1:
process["env"] = envList[0]
else:
process["env"] = getChoice(job, envList, "Umgebung")
process["env"] = getChoice(job, envList, DLG_ENVIRONMENT)
#
process["tc"] = getChoice(job, testList[process["app"]], "Testfall")
if len(entities[process["gran"]][process["app"]]) == 1:
process["entity"] = entities[process["gran"]][process["app"]][0]
else:
process["entity"] = getChoice(job, entities[process["gran"]][process["app"]], process["gran"])
print(str(process))
setattr(job.par, B.PAR_ENV, process["env"])
setattr(job.par, B.PAR_APP, process["app"])
if process["gran"] == DLG_TESTCASE:
setattr(job.par, B.PAR_TESTCASE, process["entity"])
setattr(job.par, B.PAR_TCTIME, utils.date_tool.getActdate(utils.date_tool.F_DIR))
path = utils.path_tool.composePattern(job, "{"+P.P_TCBASE+"}", None)
process[B.PAR_TCDIR] = path
elif process["gran"] == DLG_TESTSUITE:
setattr(job.par, B.PAR_TESTSUITE, process["entity"])
setattr(job.par, B.PAR_TSTIME, utils.date_tool.getActdate(utils.date_tool.F_DIR))
path = utils.path_tool.composePattern(job, "{"+P.P_TSBASE+"}", None)
process[B.PAR_TSDIR] = path
startProcess(job, process)
def getChoice(job, choiselist, description):
index = 0
print("+------------- "+description+" ----------")
print('| {:2d} : {:60s}'.format(0, "exit"))
print('| | {:2d} : {:60s}'.format(0, "exit"))
for k in choiselist:
index += 1
print('| {:2d} : {:60s}'.format(index, k))
print('| | {:2d} : {:60s}'.format(index, k))
print("+-----------------------------------------------")
choice = input("Auswahl 1-" + str(index) + ": ")
if not choice.isnumeric():
@ -145,9 +279,45 @@ def getChoice(job, choiselist, description):
def startProcess(job, process):
print(str(process))
path = os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG], JSON_FILE)
utils.file_tool.writeFileDict(job.m, job, path, process)
jobargs = {}
jobargs[B.PAR_APP] = process["app"]
jobargs[B.PAR_ENV] = process["env"]
if B.PAR_STEP in process:
jobargs[B.PAR_STEP] = process[B.PAR_STEP]
if B.PAR_TCDIR in process:
jobargs[B.PAR_TCDIR] = process[B.PAR_TCDIR]
jobargs[B.PAR_TESTCASE] = process["entity"]
elif B.PAR_TSDIR in process:
jobargs[B.PAR_TSDIR] = process[B.PAR_TSDIR]
jobargs[B.PAR_TESTSUITE] = process["entity"]
print("process-programm "+process["program"])
myjob = basic.program.Job(process["program"], jobargs)
myjob.startJob()
if process["program"] == "init_testcase":
init_testcase.startPyJob(myjob)
elif process["program"] == "execute_testcase":
execute_testcase.startPyJob(myjob)
elif process["program"] == "collect_testcase":
collect_testcase.startPyJob(myjob)
elif process["program"] == "compare_testcase":
compare_testcase.startPyJob(myjob)
elif process["program"] == "copy_appdummy":
copy_appdummy.startPyJob(myjob)
elif process["program"] == "init_testsuite":
init_testsuite.startPyJob(myjob)
elif process["program"] == "execute_testsuite":
print("execute_testsuite.startPyJob(myjob) not implemented")
elif process["program"] == "collect_testsuite":
print("collect_testsuite.startPyJob(myjob) not implemented")
elif process["program"] == "finish_testsuite":
finish_testsuite.startPyJob(myjob)
elif process["program"] == "test_executer":
test_executer.startPyJob(myjob)
restartActualProcess(job)
if __name__ == '__main__':
job = basic.program.Job(PROGRAM_NAME)
job = basic.program.Job(PROGRAM_NAME, {})
readContext(job)
restartActualProcess(job)

26
test/test_71job.py

@ -2,6 +2,8 @@ import unittest
import os
import inspect
import shutil
import execute_testcase
import utils.path_tool
from basic.program import Job
from basic.componentHandling import ComponentManager
@ -18,7 +20,7 @@ HOME_PATH = test.constants.HOME_PATH
PYTHON_CMD = "python"
TEST_FUNCTIONS = ["test_tdata", "test_getCsvSpec_data", "test_getCsvSpec_tree", "test_getCsvSpec_key",
"test_getCsvSpec_conf", "test_extractPattern", "test_parseCsv", "test_smokeTestcase"]
TEST_FUNCTIONS = ["test_00create"]
TEST_FUNCTIONS = ["test_smokeTestcase"]
class MyTestCase(unittest.TestCase):
@ -169,16 +171,15 @@ class MyTestCase(unittest.TestCase):
"collect_testcase", "compare_testcase"]
programs = ["execute_testcase"]
testcase = "TC0001"
timexec = "2022-08-18_14-23-34"
timexec = "2022-09-17_08-23-34"
fct = "read_TData"
if "init_testcase" in programs:
program = "init_testcase"
job = Job("unit")
args = {B.PAR_APP: "TESTAPP", B.PAR_ENV: "ENV01", "modus": "unit",
B.PAR_TCDIR: os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], testcase, timexec),
ujob = Job("unit", {})
args = {B.PAR_APP: "TESTAPP", B.PAR_ENV: "ENV01",
B.PAR_TCDIR: os.path.join(ujob.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], testcase, timexec),
"step": 1}
job.par.setParameterArgs(job, args)
job.setProgram(program)
job = Job("init_testcase", args)
if "copy_precondition" in programs:
copyAppData(job, program, "tdpreexec")
pass
@ -189,11 +190,14 @@ class MyTestCase(unittest.TestCase):
program = "execute_testcase"
step = basic.step.Step()
step.start = "execute_testcase"
job = Job("unit")
args = {B.PAR_APP: "TESTAPP", B.PAR_ENV: "ENV01", "modus": "unit",
B.PAR_TCDIR: os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], testcase, timexec),
ujob = Job("unit", {})
args = {B.PAR_APP: "TESTAPP", B.PAR_ENV: "ENV01",
B.PAR_TCDIR: os.path.join(ujob.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], testcase, timexec),
"step": 2}
test_executer.startStepProgram(step, job, args)
job = Job("execute_testcase", args)
job.startJob()
execute_testcase.startPyJob(job)
job.stopJob(1)
print("fertig")
def copyAppData(job, program, tdsource):

12
utils/date_tool.py

@ -32,6 +32,10 @@ def getFormatDatetupel(dtupel, format):
return getFormatdate(datetime.datetime(dtupel[0], dtupel[1], dtupel[2],
dtupel[3], dtupel[4], dtupel[5]) ,format)
def formatParsedDate(instring, format):
dtupel = parseDate(instring)
print ("---------------"+str(dtupel))
return getFormatDatetupel(dtupel, format)
def parseFormula(instring):
"""
@ -116,6 +120,14 @@ def parseDate(instring):
print(instring)
if instring[0:2] == "{(" and instring[-2:] == ")}":
return parseFormula(instring)
if re.match(r"\d{8}_\d{6}", instring):
year = int(instring[0:4])
mon = int(instring[4:6])
day = int(instring[6:8])
hour = int(instring[9:11])
min = int(instring[11:13])
sec = int(instring[13:])
return (year, mon, day, hour, min, sec)
if len(instring) > 8:
for d in ["_", " "]:
if d in instring and instring.find(d) > 8:

1
utils/dbshive_tool.py

@ -32,6 +32,7 @@ class DbFcts(utils.db_abstract.DbFcts):
tdata = {}
dry = 0
verify = -1+job.getDebugLevel("db_tool")
attr = self.getDbAttributes(B.SVAL_NULL)
cmd = "SELECT "+",".join(self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER])
cmd += " FROM "+table
sqls = self.comp.composeSqlClauses(job, cmd)

12
utils/job_tool.py

@ -19,7 +19,9 @@
12. Funktion -- schraenkt Verarbeitung auf parametriserte Funktionen ein
13. Tool -- schraenkt Protokollierung/Verarbeitung auf parametriserte Tools ein
"""
from basic.program import Job
import basic.program
def hasModul(komp):
#job = Job.getInstance()
return False
@ -29,3 +31,11 @@ def hasFunction(fct):
def hasTool(tool):
#job = Job.getInstance()
return False
def createJob(parentJob, jobargs):
job = basic.program.Job("temp") # meaning temp
job.par.setParameterArgs(job, jobargs)
job.startJob()
return

7
utils/path_tool.py

@ -12,7 +12,7 @@ import utils.config_tool
import re
import basic.constants as B
import utils.path_const as P
import utils.date_tool
TOOL_NAME = "path_tool"
def getHome():
@ -70,6 +70,10 @@ def getKeyValue(job, key, comp=None):
a = key[4:].split(":")
val = getattr(job, a[0])
# only date with hours
if a[0] in ["start"]:
print("++++++++++++++"+str(val))
val = utils.date_tool.formatParsedDate(str(val), utils.date_tool.F_LOG)
print("++++++++++++++"+val)
if len(a) > 1 and a[1] == "H":
val = val[0:-4]+"00"
if verify: job.debug(verify, val)
@ -137,6 +141,7 @@ def composePattern(job, pattern, comp):
for pat in l:
if verbose: print(str(max) + ": " + pattern + ": " + pat)
pit = getKeyValue(job, pat[1:-1], comp)
if verbose: print(str(pit) + ": " + pattern + ": " + pat)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)
pattern = pattern.replace(pat, pit)
#job.debug(verify, str(max) + ": " + pattern + ": " + pat + ": " + pit)

Loading…
Cancel
Save