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.
329 lines
12 KiB
329 lines
12 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
|
|
|
|
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
|
|
import model.factory
|
|
|
|
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.TOPIC_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":
|
|
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
|
|
|
|
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)
|
|
apps = model.factory.getApplication(job).read_unique_names(job, project, "", "", {})
|
|
return apps # 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.factory.getEnvironment().read_unique_names(job, project, "", "", "")
|
|
# model.environment.select_environments(job, projList)
|
|
return envs
|
|
|
|
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.factory.getTestcase(job, args[B.PAR_PROJ], args[B.PAR_APP]).read_unique_names(
|
|
job, args[B.PAR_PROJ], args[B.PAR_APP], "", {})
|
|
return out
|
|
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:
|
|
"""
|
|
outVariants = []
|
|
print("select spec "+gran)
|
|
if gran == J.GRAN_TP.lower():
|
|
setParameterArgs(job, args, [B.PAR_PROJ])
|
|
print("select tp spec " + gran)
|
|
return [""]
|
|
elif gran in [B.PAR_TESTSUITE, J.GRAN_TS]:
|
|
setParameterArgs(job, args, [B.PAR_PROJ])
|
|
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]:
|
|
setParameterArgs(job, args, [B.PAR_PROJ])
|
|
print("select tc spec " + gran + " " + str(args))
|
|
testcase = model.factory.getTestcase(job, args[B.PAR_PROJ], name=args[B.PAR_TESTCASE])
|
|
print(str(testcase))
|
|
#if "variant" not in testcase:
|
|
# return [""]
|
|
if hasattr(testcase, B.SUBJECT_VARIANTS):
|
|
outVariants = list(getattr(testcase, B.SUBJECT_VARIANTS).keys())
|
|
#return list(testcase[B.SUBJECT_VARIANTS].keys())
|
|
else:
|
|
raise Exception("for gran "+gran+" variant can not be selected")
|
|
if len(outVariants) == 0:
|
|
outVariants.append("defaultVariant")
|
|
print("select no spec "+gran)
|
|
return outVariants
|
|
|
|
def select_testplan(job, programDef, args):
|
|
outList = []
|
|
return outList
|
|
|
|
def select_testsuite(job, programDef, args):
|
|
outList = []
|
|
return outList
|
|
|
|
def select_testcase(job, programDef, args):
|
|
import model.testcase
|
|
print("testcase args "+str(args))
|
|
# outList = model.testcase.select_testcases(job, [args[B.PAR_PROJ]], [args[B.PAR_APP]])
|
|
outList = model.factory.getTestcase(job, args[B.PAR_PROJ]).get_unique_names(job, project=args[B.PAR_PROJ], application=args[B.PAR_APP])
|
|
return outList
|
|
|
|
def setParameterArgs(job, args, parameter):
|
|
if not hasattr(job, "par"):
|
|
job.setParameter({})
|
|
# setattr(job, "par", object)
|
|
for p in parameter:
|
|
if hasattr(job.par, p):
|
|
continue
|
|
if p in args:
|
|
setattr(job.par, p, args[p])
|
|
|
|
#
|