Data-Test-Executer Framework speziell zum Test von Datenverarbeitungen mit Datengenerierung, Systemvorbereitungen, Einspielungen, ganzheitlicher diversifizierender Vergleich
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

293 lines
10 KiB

# GrundFunktionen zur Ablaufsteuerung
#
# --------------------------------------------------------
"""
1. Programm -- implementiert in Main-Klasse
2. Anwndung -- steuert zu pruefende System [ in basis_Config ]
3. application -- steuert zu pruefende Maschine [ in dir/applicationen ]
4. release -- steuert zu prufendes Release [ aus dir/release kann spez. release_Config geladen werden, dir/lauf/release ]
5. ~Verz -- Dokumentationsverzeichnis zu Testlauf/Testfall/Soll-Branch
6. zyklus -- optional unterscheidet echte und entwicklungsLaeufe
7. Programmspezifische Parameter
8. loglevel -- steuert Protokollierung; default debug (fatal/error/warn/msg/info/debug1/debug2/trace1/trace2)
10. Laufart -- steuert die Verarbeitung; default echt
- echt-auto Lauf aus Automatisierung (1-7)
- test Lauf ohne Ausfuehrungen am Testsystem, wohl aber in Testverzeichnissen
- echt-spez Wiederholung einer spezifischen Funktion (1-13)
- unit Ausfuehrung der Unittests
11. Modul -- schraenkt Verarbeitung auf parametriserte componenten ein
12. Funktion -- schraenkt Verarbeitung auf parametriserte Funktionen ein
13. Tool -- schraenkt Protokollierung/Verarbeitung auf parametriserte Tools ein
"""
import getpass
import os.path
import pprint
import sys
import json
import basic.program
import basic.constants as B
import tools.job_const as J
import check_environment
import check_specification
import check_configuration
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
import tools.path_tool
import tools.file_tool
import tools.date_tool as date_tool
import components.tools.job_tool
import model.application
import model.environment
import model.testplan
import model.testsuite
import model.testcase
import model.component
def hasModul(komp):
#job = Job.getInstance()
return False
def hasFunction(fct):
#job = Job.getInstance()
return False
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
def startJobProcesses(job):
""" function to open processes like db-connection """
components.tools.job_tool.startJobProcesses(job)
pass
def stopJobProcesses(job):
""" function to close processes like db-connection """
components.tools.job_tool.stopJobProcesses(job)
pass
def getUser(job=None):
return getpass.getuser()
def getUserProject(job=None):
# TODO move to objects.user
return "TESTPROJ"
def getActChildPath(job):
return os.path.join(job.conf[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG], job.username+".json")
def read_child_args(job):
path = getActChildPath(job)
out = None
if os.path.exists(path):
out = tools.file_tool.read_file_dict(job, path, job.m)
return out
def write_child_args(job, childArgs):
path = getActChildPath(job)
print("job "+str(job.__dict__))
print("path "+path)
print("args "+str(childArgs))
tools.file_tool.write_file_dict(job.m, job, path, childArgs)
def start_child_process(job, args):
"""
starts a new job in a child process
:param job:
:param args: job-args
:return:
"""
verifiy = job.getDebugLevel("job_tool")
job.m.logDebug(verifiy, "start_child_process")
job.m.logTrace(verifiy, args)
print("args "+str(args))
path = tools.path_tool.getActualJsonPath(job)
tools.file_tool.write_file_dict(job.m, job, path, args)
childArgs = {}
for p in args:
if p in B.LIST_MAIN_PAR:
childArgs[p] = args[p]
if p in [B.PAR_USER, B.PAR_STEP, B.PAR_PROJ, B.PAR_APP, B.PAR_COMP]:
childArgs[p] = args[p]
if p in [B.PAR_TESTCASE, B.PAR_TCTIME]:
childArgs[B.PAR_TCTIME] = date_tool.getActdate(date_tool.F_LOG)
childArgs[B.PAR_TESTCASE] = args[B.PAR_TESTCASE]
childArgs[B.PAR_GRAN] = B.PAR_TESTCASE
elif p in [B.PAR_TESTSUITE, B.PAR_TSTIME]:
childArgs[B.PAR_TSTIME] = date_tool.getActdate(date_tool.F_LOG)
childArgs[B.PAR_TESTSUITE] = args[B.PAR_TESTSUITE]
childArgs[B.PAR_GRAN] = B.PAR_TESTSUITE
elif p in [B.PAR_TESTPLAN]:
childArgs[B.PAR_TPTIME] = date_tool.getActdate(date_tool.F_LOG)
childArgs[B.PAR_TESTPLAN] = args[B.PAR_TESTPLAN]
childArgs[B.PAR_GRAN] = B.PAR_TESTPLAN
job.m.logInfo("process-programm " + args[B.PAR_PROGRAM] + " :: " + str(childArgs))
childJob = basic.program.Job(args[B.PAR_PROGRAM], "ulrichX", childArgs)
if hasattr(job.par, "tool"):
setattr(childJob.par, "tool", getattr(job.par, "tool"))
childJob.m.logDebug(verifiy, "start_child_process "+childJob.program)
childJob.m.logDebug(verifiy, args)
childJob.m.logDebug(verifiy, B.LIST_MAIN_PAR)
childJob.m.logDebug(verifiy, childArgs)
childJob.startJob()
if args[B.PAR_PROGRAM] == "init_testcase":
init_testcase.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "execute_testcase":
execute_testcase.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "collect_testcase":
collect_testcase.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "compare_testcase":
compare_testcase.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "init_testsuite":
init_testsuite.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "execute_testsuite":
print("execute_testsuite.startPyJob(childJob) not implemented")
elif args[B.PAR_PROGRAM] == "collect_testsuite":
print("collect_testsuite.startPyJob(childJob) not implemented")
elif args[B.PAR_PROGRAM] == "finish_testsuite":
finish_testsuite.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "test_executer":
test_executer.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "check_environment":
check_environment.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "check_configuration":
check_configuration.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "check_specification":
check_specification.startPyJob(childJob)
elif args[B.PAR_PROGRAM] == "clean_workspace":
clean_workspace.startPyJob(childJob)
childJob.stopJob(1)
return childJob
def select_gran(job, programDef):
"""
select one of testcase, testsuite or testplan
:return:
"""
return J.LIST_GRAN
def select_project(job, programDef):
"""
read projects and select one of them
:return:
"""
# the implemented projects which are configured in conf/basic.yml
return ["a", "b", "c"]
def select_application(job, programDef, project):
"""
read the application of the project and select one of them
:return:
"""
verify = job.getDebugLevel("job_tool")
job.debug(verify, "select_application "+project)
# the application are configured in comp/applications.yml which are optionally stored in the database
projList = [project]
apps = model.application.select_applications(job, projList)
return list(apps.keys())
def select_components(job, programDef, project, application):
"""
read the application of the project and select one of them
:return:
"""
verify = job.getDebugLevel("job_tool")
job.debug(verify, "select_components "+project)
# the application are configured in comp/applications.yml which are optionally stored in the database
comps = model.component.select_components(job, project, application)
return comps
def select_environment(job, programDef, project):
"""
read environments and select one of them
:return:
"""
projList = [project]
envs = model.environment.select_environments(job, projList)
return list(envs.keys())
def select_spec(job, programDef, gran, args):
"""
read the testspec of the granularity,application and select one of them with the variant
:return:
"""
print("select spec "+gran+" "+str(args))
if gran in [J.GRAN_TP.lower(), J.GRAN_TP]:
print("select tp spec " + gran)
return []
elif gran in [B.PAR_TESTSUITE, J.GRAN_TS]:
print("select ts spec " + gran)
out = model.testsuite.select_testsuites(job, [args[B.PAR_PROJ]], [args[B.PAR_APP]])
return list(out.keys())
elif gran in [B.PAR_TESTCASE, J.GRAN_TC]:
print("select tc spec " + gran)
out = model.testcase.select_testcases(job, [args[B.PAR_PROJ]], [args[B.PAR_APP]])
return list(out.keys())
print("select no spec "+gran)
def select_archiv(job, programDef, gran, project):
"""
read the testobjects of the granularity,project and select one of them with the time
:return:
"""
print("select archiv")
return ["a", "b", "c"]
def select_testtime(job, programDef, gran, args):
"""
read the testobjects of the granularity,project and select one of them with the time
:return:
"""
return ["a", "b", "c"]
def select_variant(job, programDef, gran, args):
"""
read the testobjects of the granularity,project and select one of them with the time
:return:
"""
print("select spec "+gran)
if gran == J.GRAN_TP.lower():
print("select tp spec " + gran)
return [""]
elif gran in [B.PAR_TESTSUITE, J.GRAN_TS]:
print("select ts spec " + gran)
out = model.testsuite.select_testsuite(job, args[B.PAR_PROJ], args[B.PAR_TESTSUITE])
if "variant" not in out:
return [""]
return list(out["variant"].keys())
elif gran in [B.PAR_TESTCASE, J.GRAN_TC]:
print("select tc spec " + gran + " " + str(args))
out = model.testcase.select_testcase(job, args[B.PAR_PROJ], args[B.PAR_TESTCASE])
if "variant" not in out:
return [""]
return list(out["variant"].keys())
else:
raise Exception("for gran "+gran+" variant can not be selected")
print("select no spec "+gran)
def select_testplan(job, programDef, args):
outList = []
return outList
def select_testsuite(job, programDef, args):
outList = []
return outList
def select_testcase(job, programDef, args):
outList = model.testcase.select_testcases(job, [args[B.PAR_PROJ]], [args[B.PAR_APP]])
return outList