# 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 sys import basic.program import basic.constants as B try: import collect_testcase import compare_testcase import execute_testcase import finish_testsuite import init_testcase import init_testsuite import test_executer except Exception as e: pass import tools.path_tool import tools.file_tool import components.tools.job_tool PROG_TC_INIT = "init_testcase" PROG_TC_EXECUTE = "execute_testcase" PROG_TC_COLLECT = "collect_testcase" PROG_TC_COMPARE = "compare_testcase" PROG_TS_INIT = "init_testsuite" PROG_TS_EXECUTE = "execute_testsuite" PROG_TS_COLLECT = "collect_testsuite" PROG_TS_FINISH = "finish_testsuite" PROG_TEST_EXECUTER = "test_executer" LIST_TC_PROGS = [PROG_TC_INIT, PROG_TC_EXECUTE, PROG_TC_COLLECT, PROG_TC_COMPARE] LIST_TS_PROGS = [PROG_TS_INIT, PROG_TS_EXECUTE, PROG_TS_COLLECT, PROG_TS_FINISH] PROG_CHECK_ENV = "check_environment" PROG_CHECK_CONF = "check_configuration" PROG_CHECK_SPEC = "check_specification" PROG_UNZIP_RESULT = "unzip_result" PROG_DECLARE_RESULT = "declare_result" PROG_CLEAN_WORKSPACE = "clean_workspace" PROG_MAKE_WORKSPACE = "make_workspace" PROG_UNIT_TESTER = "unit_tester" LIST_SERVICE_PROG = [PROG_CHECK_ENV, PROG_CHECK_CONF, PROG_CHECK_SPEC, PROG_UNZIP_RESULT, PROG_DECLARE_RESULT, PROG_CLEAN_WORKSPACE, PROG_MAKE_WORKSPACE, PROG_UNIT_TESTER] PROC_TP_EXECUTION = "Testplan ausfuehren" PROC_TS_EXECUTION = "Testsuite ausfuehren" PROC_TS_STEPWISE = "Testsuite Job fuer Job ausfuehren" PROC_TC_EXECUTION = "Testfall ausfuehren" PROC_TC_STEPWISE = "Testfall Job fuer Job ausfuehren" PROC_REDO_EXECUTION = "Job der Testausfuehrung wiederholen" PROC_SINGLE_JOB = "Einzelnes Programm ausfuehren" LIST_PROC = [PROC_TP_EXECUTION, PROC_TS_EXECUTION, PROC_TS_STEPWISE, PROC_TC_EXECUTION, PROC_TC_STEPWISE, PROC_REDO_EXECUTION, PROC_SINGLE_JOB] GRAN_TP = "Testplan" GRAN_TS = "Testsuite" GRAN_TC = "Testfall" LIST_GRAN = [GRAN_TP, GRAN_TS, GRAN_TC] 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 "xxx" 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) if os.path.exists(path): sys.path.remove() tools.file_tool.writeFileDict(job, path, childArgs) def start_child_process(job, args): """ starts a new job in a child process :param job: :param args: job-args :return: """ print(str(args)) path = tools.path_tool.getActualJsonPath(job) print("------- "+path) tools.file_tool.writeFileDict(job.m, job, path, args) childArgs = {} childArgs[B.PAR_APP] = args[B.PAR_APP] childArgs[B.PAR_ENV] = args[B.PAR_ENV] if B.PAR_STEP in args: childArgs[B.PAR_STEP] = args[B.PAR_STEP] if B.PAR_TCDIR in args: childArgs[B.PAR_TCDIR] = args[B.PAR_TCDIR] childArgs[B.PAR_TESTCASE] = args["entity"] elif B.PAR_TSDIR in args: childArgs[B.PAR_TSDIR] = args[B.PAR_TSDIR] childArgs[B.PAR_TESTSUITE] = args["entity"] print("process-programm " + args[B.PAR_PROGRAM]) childJob = basic.program.Job(args[B.PAR_PROGRAM], 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) return childJob def select_gran(job, programDef): """ select one of testcase, testsuite or testplan :return: """ return LIST_GRAN def select_environment(job, programDef): """ read environments and select one of them :return: """ return "" def select_project(job, programDef): """ read projects and select one of them :return: """ # the implemented projects which are configured in conf/basic.yml return [""] def select_application(job, programDef, project): """ read the application of the project and select one of them :return: """ # the application are configured in comp/applications.yml which are optionally stored in the database return "" def select_spec(job, programDef, gran, application): """ read the testspec of the granularity,application and select one of them with the variant :return: """ return "" def select_archiv(job, programDef, gran, project): """ read the testobjects of the granularity,project and select one of them with the time :return: """ return ""