Testskripter
/
datestex
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

job tools

refactor
Ulrich 2 years ago
parent
0bae1ed3e7
commit
a78543222b
7 changed files with 97 additions and 88 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      basic/constants.py
  2. 7
      basic/program.py
  3. 31
      basic/step.py
  4. 65
      start_dialog.py
  5. 5
      utils/data_const.py
  6. 64
      utils/job_tool.py
  7. 9
      utils/path_tool.py

4
basic/constants.py
View File

@ -34,6 +34,8 @@ PAR_APP = 'application'
""" definition of the application which will be tested """
PAR_ENV = 'environment'
""" definition of the environment where the application runs which will be tested """
PAR_VAR = 'variant'
""" definition of a variant dataflow especially the channel """
PAR_REL = 'release'
""" definition of the release of the application which will be tested """
PAR_TSDIR = 'tsdir'
@ -67,7 +69,7 @@ PAR_DB_WHERE = "dbwhere"
""" optional parameter with a where-clause """
PAR_DB_PARTITION = "dbparts"
""" optional parameter for partitions of a partitioned tables """
LIST_MAIN_PAR = [PAR_APP, PAR_ENV, PAR_VAR, PAR_REL, PAR_TSDIR, PAR_TSDIR]
# -------------------------------------------------------------
# attributes

7
basic/program.py
View File

@ -22,6 +22,7 @@ import utils.file_tool
import utils.config_tool
import test.constants as T
import utils.path_const as P
import utils.job_tool
jobdef = {
"webflask": {
@ -289,7 +290,7 @@ class Job:
self.program = program
basedir = jobdef[program]["basedir"]
self.basedir = basedir
if (self.par is not None):
if (self.par is not None) and self.par in B.LIST_MAIN_PAR:
setattr(self.par, "program", program)
setattr(self.par, "basedir", basedir)
parstring = getattr(self.par, "parstring")
@ -309,6 +310,7 @@ class Job:
logTime = self.start.strftime("%Y%m%d_%H%M%S")
self.m = basic.message.Message(self, basic.message.LIMIT_DEBUG, logTime, None)
print("prog-68 " + str(self.m.rc))
utils.job_tool.startJobProcesses(self)
self.par.setParameterLoaded(self)
self.m.logInfo("# # # Start Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # # # ")
self.m.debug(basic.message.LIMIT_INFO, "# # # Start Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # # # ")
@ -317,6 +319,7 @@ class Job:
def stopJob(self, reboot=0):
utils.job_tool.stopJobProcesses(self)
self.ende = datetime.now()
self.dumpParameter()
print("stopJob " + str(self.m.messages) + ", " + str(self.m.debugfile))
@ -329,6 +332,7 @@ class Job:
self.m.closeMessage()
rc = self.m.getFinalRc()
print ("rc " + str(rc))
print(footer)
if reboot == 0:
exit(rc)
@ -519,6 +523,7 @@ class Parameter:
parser = argparse.ArgumentParser()
parser.add_argument('-a', '--'+B.SUBJECT_APPS, required=True, action='store')
parser.add_argument('-e', '--'+B.PAR_ENV, required=True, action='store')
parser.add_argument('-v', '--'+B.PAR_VAR, required=False, action='store')
parser.add_argument('-r', '--release', action='store')
parser.add_argument('-ts', '--'+B.PAR_TSDIR, action='store')
parser.add_argument('-tc', '--'+B.PAR_TCDIR, action='store')

31
basic/step.py
View File

@ -28,11 +28,15 @@ class Step:
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):
@ -41,38 +45,13 @@ class Step:
text += D.CSV_DELIMITER+k+":"+self.args[k]
return text+"\n"
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]
step.variante = fields[D.STEP_VARIANT_I]
setattr(step, B.ATTR_DATA_REF, step.refLine)
if D.STEP_ARGS_I == D.STEP_LIST_I:
args = ""

65
start_dialog.py
View File

@ -15,14 +15,7 @@ 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.job_tool
import utils.file_tool
import utils.data_const as D
import utils.date_tool
@ -46,7 +39,6 @@ DLG_CONTINUE = "fortsetzen"
DLG_DUMMY_STEP = "Dummy-Schritt"
DLG_NEWJOB = "neuer Job"
JSON_FILE = "actualJob.json"
JOB_NR = {
DLG_TESTSUITE : {
"start": "init_testsuite",
@ -162,7 +154,7 @@ def restartActualProcess(job):
check if an actual process is open
:return:
"""
path = os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG], JSON_FILE)
path = utils.path_tool.getActualJsonPath(job)
if os.path.exists(path):
actProc = utils.file_tool.readFileDict(job, path, job.m)
print("restartActJob "+str(actProc))
@ -173,7 +165,8 @@ def restartActualProcess(job):
dialogProcess(job)
else:
actProc["step"] = str(step+1)
startProcess(job, actProc)
utils.job_tool.startProcess(job, actProc)
restartActualProcess(job)
selection = [DLG_NEWJOB, DLG_REDO]
nr = int(JOB_NR[actProc["gran"]][actProc["program"]]["jobnr"])
@ -184,10 +177,12 @@ def restartActualProcess(job):
choice = getChoice(job, selection, DLG_ENVIRONMENT)
print(choice)
if choice == DLG_REDO:
startProcess(job, actProc)
utils.job_tool.startProcess(job, actProc)
restartActualProcess(job)
elif choice == DLG_DUMMY_STEP:
actProc["program"] = JOB_LIST[nr+1]
startProcess(job, actProc)
utils.job_tool.startProcess(job, actProc)
restartActualProcess(job)
elif choice == DLG_CONTINUE:
if nr == 7:
nr = 9
@ -196,7 +191,7 @@ def restartActualProcess(job):
print (" act nr "+str(nr))
actProc["step"] = str(step + 1)
actProc["program"] = JOB_LIST[nr]
startProcess(job, actProc)
utils.job_tool.startProcess(job, actProc)
elif choice == DLG_NEWJOB:
dialogProcess(job)
else:
@ -254,7 +249,8 @@ def dialogProcess(job):
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)
utils.job_tool.startProcess(job, process)
restartActualProcess(job)
def getChoice(job, choiselist, description):
index = 0
@ -277,45 +273,6 @@ def getChoice(job, choiselist, description):
return choiselist[int(choice) - 1]
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, {})

5
utils/data_const.py
View File

@ -89,8 +89,9 @@ LIST_BLOCK_CONST = ["CSV_BLOCK_HEAD", "CSV_BLOCK_OPTION", "CSV_BLOCK_STEP", "CSV
STEP_COMP_I = 1
STEP_EXECNR_I = 2
STEP_REFNR_I = 3
STEP_ARGS_I = 4
STEP_LIST_I = 4
STEP_VARIANT_I = 4
STEP_ARGS_I = 5
STEP_LIST_I = 5
STEP_ATTR_COMP = "component"
STEP_ATTR_EXECNR = "exec-step"
STEP_ATTR_REFNR = "reference-nr"

64
utils/job_tool.py
View File

@ -20,8 +20,17 @@
13. Tool -- schraenkt Protokollierung/Verarbeitung auf parametriserte Tools ein
"""
import basic.program
import basic.constants as B
import collect_testcase
import compare_testcase
import execute_testcase
import finish_testsuite
import init_testcase
import init_testsuite
import test_executer
import utils.path_tool
import utils.file_tool
import components.utils.job_tool
def hasModul(komp):
#job = Job.getInstance()
return False
@ -33,9 +42,56 @@ def hasTool(tool):
return False
def createJob(parentJob, jobargs):
job = basic.program.Job("temp") # meaning temp
job.par.setParameterArgs(job, jobargs)
job.startJob()
return
def startJobProcesses(job):
""" function to open processes like db-connection """
components.utils.job_tool.startJobProcesses(job)
pass
def stopJobProcesses(job):
""" function to close processes like db-connection """
components.utils.job_tool.stopJobProcesses(job)
pass
def startProcess(job, process):
print(str(process))
path = utils.path_tool.getActualJsonPath(job)
print("------- "+path)
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"] == "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)

9
utils/path_tool.py
View File

@ -13,6 +13,7 @@ import re
import basic.constants as B
import utils.path_const as P
import utils.date_tool
import getpass
TOOL_NAME = "path_tool"
def getHome():
@ -45,6 +46,14 @@ def getBasisConfigPath():
raise Exception("no basis-configuration found")
def getActualJsonPath(job):
username = getpass.getuser()
path = os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG], username+"Job.json")
print("------ path "+path)
#if os.path.exists(path):
# return path
return path
def getKeyValue(job, key, comp=None):
"""
this function gets the value for the key which relates to an attribute in the job or in the component

Write Preview
Loading…
Cancel
Save