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.

331 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
2 years ago
import tools.job_const as J
import check_environment
import check_specification
import check_configuration
2 years ago
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
1 year ago
#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)
2 years ago
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)
2 years ago
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
print("process-programm " + args[B.PAR_PROGRAM] + " :: " + str(childArgs))
job.m.logInfo("process-programm " + args[B.PAR_PROGRAM] + " :: " + str(childArgs))
childJob = basic.program.Job(args[B.PAR_PROGRAM], "ulrich", 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)
2 years ago
elif args[B.PAR_PROGRAM] == "clean_workspace":
import clean_workspace
2 years ago
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]
1 year ago
# 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]
1 year ago
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)
1 year ago
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)
testsuite = model.factory.getTestsuite(job, args[B.PAR_PROJ], name=args[B.PAR_TESTSUITE])
if hasattr(testsuite, B.SUBJECT_VARIANTS):
outVariants = list(getattr(testsuite, B.SUBJECT_VARIANTS).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())
elif hasattr(testcase, B.SUBJECT_STEPS):
outVariants = list(getattr(testcase, B.SUBJECT_STEPS).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
2 years ago
def select_testplan(job, programDef, args):
outList = []
return outList
def select_testsuite(job, programDef, args):
outList = []
outList = model.factory.getTestsuite(job, args[B.PAR_PROJ]).get_unique_names(job, project=args[B.PAR_PROJ], application=args[B.PAR_APP])
2 years ago
return outList
def select_testcase(job, programDef, args):
1 year ago
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])
#