Browse Source

start_dialog first changes

refactor
Ulrich 2 years ago
parent
commit
82c4448838
  1. 2
      basic/constants.py
  2. 16
      basic/program.py
  3. 12
      catalog/programs.csv
  4. 2
      finish_testsuite.py
  5. 0
      model/__init__.py
  6. 0
      model/application.py
  7. 2
      model/catalog.py
  8. 345
      start_dialog.py
  9. 68
      test/test_03path.py
  10. 8
      test/test_07catalog.py
  11. 2
      tools/config_tool.py
  12. 4
      tools/file_abstract.py
  13. 10
      tools/file_tool.py
  14. 2
      tools/filecsv_fcts.py
  15. 186
      tools/job_tool.py
  16. 4
      tools/value_tool.py
  17. 6
      unit_tester.py

2
basic/constants.py

@ -31,6 +31,7 @@ SVAL_NULL = "null"
# -------------------------------------------------------------
# parameter with arguments
PAR_PROGRAM = 'program'
PAR_USER = 'user'
PAR_PROJ = 'project'
""" definition of the project which will be tested """
PAR_APP = 'application'
@ -63,6 +64,7 @@ PAR_STEP = 'step'
PAR_DESCRIPT = 'desription'
PAR_STRING = 'parstring'
""" """
PAR_TESTPLAN = "testplan"
PAR_TESTCASE = "testcase"
PAR_TESTCASES = "testcases"
PAR_TESTSUITE = "usecase"

16
basic/program.py

@ -17,7 +17,7 @@ from datetime import datetime
import basic.constants as B
import basic.message
import basic.componentHandling
import objects.catalog
import model.catalog
import tools.value_tool
import tools.date_tool
import tools.path_tool
@ -54,7 +54,7 @@ class SimpleJob:
path = tools.path_tool.getBasisConfigPath()
self.conf = getConfiguration(self, path)
self.jobid = str(100000)
catalog = objects.catalog.Catalog.getInstance()
catalog = model.catalog.Catalog.getInstance()
self.programDef = catalog.getValue(self, CTLG_NAME, program, "")
if args is not None:
@ -99,7 +99,7 @@ class Job:
__instances = []
__jobid = 100000
__logtime = "20200101_000000"
#catalog = objects.catalog.Catalog.getInstance()
#catalog = model.catalog.Catalog.getInstance()
def __init__ (self, program, username="", args=None):
"""
@ -139,7 +139,7 @@ class Job:
self.username = os.getlogin()
path = tools.path_tool.getBasisConfigPath()
self.conf = getConfiguration(self, path)
catalog = objects.catalog.Catalog.getInstance()
catalog = model.catalog.Catalog.getInstance()
self.programDef = catalog.getValue(self, CTLG_NAME, program, "")
try:
path = tools.config_tool.getConfigPath(self, P.KEY_BASIC, B.BASIS_FILE)
@ -175,7 +175,7 @@ class Job:
if hasattr(self, "conf"):
conf = self.conf
conf["configpath"].append(path)
doc = tools.file_tool.readFileDict(self, path, None)
doc = tools.file_tool.read_file_dict(self, path, None)
if "basic" in doc:
for k, v in doc["basic"].items():
if k not in conf:
@ -268,7 +268,7 @@ class Job:
parpath = tools.path_tool.compose_path(self, self.programDef[CTLG_PARSOURCE], None)
if not os.path.join(parpath):
return None
doc = tools.file_tool.readFileDict(self, parpath, None)
doc = tools.file_tool.read_file_dict(self, parpath, None)
for k in doc.keys():
output[k] = copy.deepcopy(doc[k])
return output
@ -356,7 +356,7 @@ def getConfiguration(job, path):
if hasattr(job, "conf"):
conf = job.conf
conf["configpath"].append(path)
doc = tools.file_tool.readFileDict(job, path, None)
doc = tools.file_tool.read_file_dict(job, path, None)
if "basic" in doc:
for k, v in doc["basic"].items():
if k not in conf:
@ -401,7 +401,7 @@ class Parameter:
def checkParameter(self, job):
print (f"Parameter initialisiert {self.program}")
pardef = job.programDef[CTLG_PARDEF]
for p in pardef["par"]:
for p in pardef:
if len(p) > 1 and not hasattr(self, p):
job.m.setFatal("Parameter " + p + " is not set!")

12
catalog/programs.csv

@ -11,11 +11,11 @@ table:programs;name;objtype;objname;time;env;app;variant;pardef;pfilesource;pfil
;collect_testcase;tc;m;m;m;m;o;;tcparfile;tcparfile;tcdir;tcbase;info;{job.par.tcdir}/{log}/{job.program}_{job.start}.txt
;compare_testcase;tc;m;m;m;m;o;;tcparfile;tcparfile;tcdir;tcbase;info;{job.par.tcdir}/{log}/{job.program}_{job.start}.txt
;check_environment;env;;;m;;o;;;envparfile;envdir;envbase;info;{job.par.envdir}/{log}/log_{job.start}.txt
;check_specification;tp,ts,tc;o;;;;;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;check_configuration;ws;o;;;o;;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;make_workspace;ws;;;_;;;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;unit_tester;ws;o;;d;;;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;check_specification;tp,ts,tc;o;;;;n;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;check_configuration;ws;o;;;o;n;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;make_workspace;ws;;;_;;n;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;unit_tester;ws;o;;d;;n;"{""par"":{""application"":""service"",""environment"":""Testserver"",""variant"":""""}}";;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;declare_expection;ts,tc;m;m;m;m;o;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;clean_workspace;ws;o ;;_;;;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;clean_workspace;ws;o ;;_;;n;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;start_dialog;ws;input;;input;input;input;;;;wsdir;wsbase;info;{job.par.wsdir}/{log}/log_{job.start}.txt
;web_start;ws;;;;;;;;;wsdir;wsbase;warn;{job.par.wsdir}/{log}/log_{job.start}.txt
;web_start;ws;;;;;n;;;;wsdir;wsbase;warn;{job.par.wsdir}/{log}/log_{job.start}.txt

1 _key name
11 collect_testcase tc m m m m o tcparfile tcparfile tcdir tcbase info {job.par.tcdir}/{log}/{job.program}_{job.start}.txt
12 compare_testcase tc m m m m o tcparfile tcparfile tcdir tcbase info {job.par.tcdir}/{log}/{job.program}_{job.start}.txt
13 check_environment env m o envparfile envdir envbase info {job.par.envdir}/{log}/log_{job.start}.txt
14 check_specification tp,ts,tc o n wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
15 check_configuration ws o o n wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
16 make_workspace ws _ n wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
17 unit_tester ws o d n {"par":{"application":"service","environment":"Testserver","variant":""}} wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
18 declare_expection ts,tc m m m m o wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
19 clean_workspace ws o _ n wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
20 start_dialog ws input input input input wsdir wsbase info {job.par.wsdir}/{log}/log_{job.start}.txt
21 web_start ws n wsdir wsbase warn {job.par.wsdir}/{log}/log_{job.start}.txt

2
finish_testsuite.py

@ -27,7 +27,7 @@ def startPyJob(job):
text = report.reportTestsuite()
path = os.path.join(job.par.tsdir, "Result.html")
utils.file_tool.writeFileText(job.m, job, path, text)
archivFolder = job.conf.confs["paths"]["archiv"]
archivFolder = job.conf["paths"]["archiv"]
tsFolder = os.path.join(archivFolder, "Testlauf")
tarfile = utils.zip_tool.openNewTarFile(job, tsFolder, job.par.usecase+"_"+job.par.tstime+".tar.gz")
utils.zip_tool.appendFolderIntoTarFile(job, tsFolder, job.par.usecase+"_"+job.par.tstime, tarfile)

0
objects/__init__.py → model/__init__.py

0
basic/application.py → model/application.py

2
objects/catalog.py → model/catalog.py

@ -102,7 +102,7 @@ class Catalog:
msg = job.m
else:
msg = None
data = tools.file_tool.readFileDict(job, pathname, msg)
data = tools.file_tool.read_file_dict(job, pathname, msg)
if hasattr(job, "m"):
job.m.debug(12, "domain " + domain + " readed from " + pathname)
self.catalog[domain] = data[B.DATA_NODE_TABLES][domain][B.DATA_NODE_KEYS]

345
start_dialog.py

@ -1,258 +1,52 @@
#!/usr/bin/python
# program to execute programs for a testcases or for a testsuite
# PARAM from INPUT: --granularity --application --environment --testcase/testsuite
# main functions
# + input_param() : cache-actjob --> user-input --> local-param
# + start_job() : local-param --> cache-actjob --> start-param
# ---------------------------------------------------
#
#
# ----------------------------------------------------------
"""
This program is created for the business-test on the level of unit testing.
On this level there is an intensive test-set up to a carthesian product possible.
The test-specification is written in a yaml-config in the data-directory.
workflow:
1. generate test-cases with all inputs into a csv-file of the spec-yaml
2. run the csv-file and collect all results into a new result-csv-file with all inputs and outputs
3. check the result-csv-file and declare it as a target-results
OR instead 2 and 3 on test-first-strategy:
3. create manually the target-result with your inputs and the target system-outputs
4. run and compare the actual-result with the target-result
5. at the end you get an xls-sheet like your result-csv-file
but additionally with comparsion-result as detailed result and the only counts of passed and failed tests
as minimum-result which can be inserted into management-tools
"""
import os.path
import json
import re
# Press the green button in the gutter to run the script.
import importlib
import os, glob
import io
import sys
import basic.program
import basic.constants as B
import utils.job_tool
import utils.file_tool
import utils.data_const as D
import utils.date_tool
import utils.path_tool
import utils.path_const as P
tempJob = {}
PROGRAM_NAME = "service"
import tools.job_tool
import model.catalog
DLG_TESTCASE = "Testfall"
DLG_TESTSUITE = "Testsuite"
DLG_COMPLETE = "Komplettausfuehrung"
LIST_DLG_GRAN = [DLG_TESTCASE, DLG_TESTSUITE]
DLG_START_QUESTION = "was soll getestet werden"
# DLG_TESTPLAN = "Testplan"
DLG_ENVIRONMENT = "Umgebung"
DLG_APPLICATION = "Anwendung"
DLG_REDO = "wiederholen"
DLG_CONTINUE = "fortsetzen"
DLG_DUMMY_STEP = "Dummy-Schritt"
DLG_NEWJOB = "neuer Job"
PROGRAM_NAME = "start_dialog"
JOB_NR = {
DLG_TESTSUITE : {
"start": "init_testsuite",
"init_testsuite": {
"jobnr": "0" },
"execute_testsuite": {
"jobnr": "1"},
"collect_testsuite": {
"jobnr": "2"},
"compare_testsuite": {
"jobnr": "3"},
"finish_testsuite": {
"jobnr": "4"}
},
DLG_TESTCASE: {
"start": "init_testcase",
"init_testcase": {
"jobnr": "5" },
"execute_testcase": {
"jobnr": "6" },
"collect_testcase": {
"jobnr": "7" },
"copy_appdummy": {
"jobnr": "8" },
"compare_testcase": {
"jobnr": "9" },
},
"check_environment": {
"jobnr": "10" },
"test_executer": {
"jobnr": "11"},
}
VERIFY = False
JOB_LIST = [
"init_testsuite", # 0
"execute_testsuite", # 1
"collect_testsuite", # 2
"compare_testsuite", # 3
"finish_testsuite", # 4
"init_testcase", # 5
"execute_testcase", # 6
"collect_testcase", # 7
"copy_appdummy", # 8
"compare_testcase", # 9
"check_environment", # 10
"test_executer" # 11
]
appList = []
envList = []
entities = {}
entities[DLG_TESTCASE] = {}
entities[DLG_TESTSUITE] = {}
def readContext(job):
for k in job.conf.confs[B.SUBJECT_APPS]:
appList.append(k)
path = job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_ENV]
if os.path.exists(path):
for d in os.listdir(path):
print ("-- "+d)
if not os.path.isdir(os.path.join(path, d)):
continue
if d[0:1] == "_":
continue
envList.append(d)
path = job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_TDATA]
if os.path.exists(path):
for d in os.listdir(path):
print("tdata path "+d)
if not os.path.isdir(os.path.join(path, d)):
print("continue a")
continue
if d[0:1] == "_":
print("continue b")
continue
specpath = os.path.join(path, d, D.DFILE_TESTCASE_NAME + ".csv")
readSpec(job, d, DLG_TESTCASE, specpath)
specpath = os.path.join(path, d, D.DFILE_TESTSUITE_NAME + ".csv")
readSpec(job, d, DLG_TESTSUITE, specpath)
def startPyJob(job):
# check if an actual job is stored
childArgs = tools.job_tool.read_child_args(job)
if childArgs is None:
childArgs = initDialog(job)
pass
def readSpec(job, testentity, testgran, specpath):
print("spec "+specpath)
if not os.path.isfile(specpath):
print("continue c")
return
text = utils.file_tool.readFileText(job, specpath, job.m)
print("-----------\n"+text+"\n------------------")
if re.match(r".*?depricated;[jJyY]", text):
return
if re.match(r".*\nhead:application;", text):
print("## app gematcht")
res = re.search(r".*head:application;(.+)\n", text)
apps = res.group(1).replace(";", ",").split(",")
print("# "+str(apps))
for a in apps:
if len(a) < 1:
break
if a not in entities[testgran]:
entities[testgran][a] = []
print(a+" in "+testentity+" "+testgran+" -- "+str(entities))
entities[testgran][a].append(testentity)
def printProc(job, process):
print("--------------------------------------------------")
for k in process:
print("| {0:15s} : {1}".format(k, process[k]))
print("--------------------------------------------------")
def restartActualProcess(job):
"""
check if an actual process is open
:return:
"""
path = utils.path_tool.getActualJsonPath(job)
if os.path.exists(path):
actProc = utils.file_tool.readFileDict(job, path, job.m)
print("restartActJob "+str(actProc))
printProc(job, actProc)
step = int(actProc["step"])
if actProc["program"] == "test_executer":
if step > 5:
dialogProcess(job)
else:
actProc["step"] = str(step+1)
utils.job_tool.startProcess(job, actProc)
restartActualProcess(job)
selection = [DLG_NEWJOB, DLG_REDO]
nr = int(JOB_NR[actProc["gran"]][actProc["program"]]["jobnr"])
if (actProc["gran"] == DLG_TESTSUITE and nr < 4) or (actProc["gran"] == DLG_TESTCASE and nr < 9):
selection.append(DLG_CONTINUE)
if nr == 7:
selection.append(DLG_DUMMY_STEP)
choice = getChoice(job, selection, DLG_ENVIRONMENT)
print(choice)
if choice == DLG_REDO:
utils.job_tool.startProcess(job, actProc)
restartActualProcess(job)
elif choice == DLG_DUMMY_STEP:
actProc["program"] = JOB_LIST[nr+1]
utils.job_tool.startProcess(job, actProc)
restartActualProcess(job)
elif choice == DLG_CONTINUE:
if nr == 7:
nr = 9
else:
nr += 1
print (" act nr "+str(nr))
actProc["step"] = str(step + 1)
actProc["program"] = JOB_LIST[nr]
utils.job_tool.startProcess(job, actProc)
elif choice == DLG_NEWJOB:
dialogProcess(job)
else:
dialogProcess(job)
def dialogProcess(job):
def getChoice(job, choiselist, description):
"""
dialog for selection and starting a process
to choise one of the list or exit the program
:param job:
:param choiselist:
:param description:
:return:
"""
process = {}
index = 0
print("create new process")
selection = []
selection = LIST_DLG_GRAN
if DLG_TESTCASE + " - " + DLG_COMPLETE not in selection:
selection.append(DLG_TESTCASE + " - " + DLG_COMPLETE)
if DLG_TESTSUITE + " - " + DLG_COMPLETE not in selection:
selection.append(DLG_TESTSUITE + " - " + DLG_COMPLETE)
choice = getChoice(job, LIST_DLG_GRAN, DLG_START_QUESTION)
if DLG_COMPLETE in choice:
process["gran"] = choice[0:-3-len(DLG_COMPLETE)]
process["program"] = "test_executer"
process["step"] = 1
else:
process["gran"] = choice
process["program"] = JOB_NR[process["gran"]]["start"]
process["step"] = 1
if len(appList) == 1:
process["app"] = appList[0]
else:
process["app"] = getChoice(job, appList, DLG_ENVIRONMENT)
#
if len(envList) == 1:
process["env"] = envList[0]
else:
process["env"] = getChoice(job, envList, DLG_ENVIRONMENT)
#
if len(entities[process["gran"]][process["app"]]) == 1:
process["entity"] = entities[process["gran"]][process["app"]][0]
else:
process["entity"] = getChoice(job, entities[process["gran"]][process["app"]], process["gran"])
print(str(process))
setattr(job.par, B.PAR_ENV, process["env"])
setattr(job.par, B.PAR_APP, process["app"])
if process["gran"] == DLG_TESTCASE:
setattr(job.par, B.PAR_TESTCASE, process["entity"])
setattr(job.par, B.PAR_TCTIME, utils.date_tool.getActdate(utils.date_tool.F_DIR))
path = utils.path_tool.composePattern(job, "{"+P.P_TCBASE+"}", None)
process[B.PAR_TCDIR] = path
elif process["gran"] == DLG_TESTSUITE:
setattr(job.par, B.PAR_TESTSUITE, process["entity"])
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
utils.job_tool.startProcess(job, process)
restartActualProcess(job)
def getChoice(job, choiselist, description):
index = 0
print("+------------- "+description+" ----------")
print('| | {:2d} : {:60s}'.format(0, "exit"))
@ -265,6 +59,7 @@ def getChoice(job, choiselist, description):
print("FEHLER Fehleingabe "+choice)
getChoice(job, choiselist, description)
elif int(choice) < 1:
job.stopJob(0)
exit(0)
elif int(choice) > index:
print("FEHLER Fehleingabe "+choice)
@ -273,8 +68,70 @@ def getChoice(job, choiselist, description):
return choiselist[int(choice) - 1]
def initDialog(job, args={}):
"""
dialog to initialize a child-process
:param job:
:return:
"""
# which process
if "proc" not in args:
args["proc"] = getChoice(job, tools.job_tool.LIST_PROC, "Welchen Prozess starten")
args["gran"] = ""
args[B.PAR_USER] = tools.job_tool.getUser()
args[B.PAR_PROJ] = tools.job_tool.getUserProject()
args[B.PAR_STEP] = ""
if args["proc"] == tools.job_tool.PROC_TP_EXECUTION:
args["gran"] = B.PAR_TESTPLAN
args[B.PAR_STEP] = "1"
args[B.PAR_PROGRAM] = tools.job_tool.PROG_TEST_EXECUTER
elif args["proc"] in [tools.job_tool.PROC_TS_STEPWISE, tools.job_tool.PROC_TS_EXECUTION]:
args["gran"] = B.PAR_TESTSUITE
args[B.PAR_STEP] = "1"
if args["proc"] == tools.job_tool.PROC_TS_EXECUTION:
args[B.PAR_PROGRAM] = tools.job_tool.PROG_TEST_EXECUTER
else:
args[B.PAR_PROGRAM] = tools.job_tool.PROG_TS_INIT
elif args["proc"] in [tools.job_tool.PROC_TC_STEPWISE, tools.job_tool.PROC_TC_EXECUTION]:
args["gran"] = B.PAR_TESTCASE
args[B.PAR_STEP] = "1"
if args["proc"] == tools.job_tool.PROC_TC_EXECUTION:
args[B.PAR_PROGRAM] = tools.job_tool.PROG_TEST_EXECUTER
else:
args[B.PAR_PROGRAM] = tools.job_tool.PROG_TC_INIT
elif args["proc"] == tools.job_tool.PROC_REDO_EXECUTION:
args[B.PAR_PROGRAM] = getChoice(job, tools.job_tool.LIST_TS_PROGS + tools.job_tool.LIST_TC_PROGS,
"Welches Programm starten")
elif args["proc"] == tools.job_tool.PROC_SINGLE_JOB:
args[B.PAR_PROGRAM] = getChoice(job, tools.job_tool.LIST_SERVICE_PROG, "Welches Programm starten")
else:
args["gran"] = ""
catalog = model.catalog.Catalog.getInstance()
programDef = catalog.getValue(job, basic.program.CTLG_NAME, args[B.PAR_PROGRAM], "")
for p in programDef[basic.program.CTLG_PARDEF]:
if p in args:
print("p "+p+" ist "+args[p])
else:
args[p] = programDef[basic.program.CTLG_PARDEF][p]
print("p "+p+" jetzt " + args[p])
print(str(args))
def childDialog(job, args):
"""
dialog to manage the child-process
:param job:
:return:
"""
pass
if __name__ == '__main__':
job = basic.program.Job(PROGRAM_NAME, {})
readContext(job)
restartActualProcess(job)
job = basic.program.Job(PROGRAM_NAME)
# TODO to move to job_tool
args = {}
if isinstance(job.programDef[basic.program.CTLG_PARDEF], dict):
for p in job.programDef[basic.program.CTLG_PARDEF]:
args[p] = job.programDef[basic.program.CTLG_PARDEF][p]
job.setParameter(args)
job.startJob()
startPyJob(job)
job.stopJob(0)

68
test/test_03path.py

@ -1,21 +1,20 @@
import unittest
import os
import inspect
import utils.path_tool
import tools.path_tool
import basic.program
import basic.constants as B
import test.constants
import test.testtools
import utils.path_const as P
import tools.path_const as P
HOME_PATH = test.constants.HOME_PATH
DATA_PATH = test.constants.DATA_PATH
OS_SYSTEM = test.constants.OS_SYSTEM
# here you can select single testfunction for developping the tests
TEST_FUNCTIONS = ["test_key", "test_rejoinPath", "test_rejoinPath", "test_composePath", "test_composePattern",
"test_extractPath", "test_extractPattern"]
#TEST_FUNCTIONS = [ "test_extractPath"]
TEST_FUNCTIONS = ["test_rejoinPath", "test_rejoinPath", "test_extractPattern"]
#TEST_FUNCTIONS = ["test_key", "test_extractPath", "test_composePath"]
verbose = True
class MyTestCase(unittest.TestCase):
@ -34,25 +33,6 @@ class MyTestCase(unittest.TestCase):
paths = pt.getInstance()
paths.pattern[P.P_ENVBASE] = "{job.conf.environment}/{job.par.environment}"
def test_key(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
res = utils.path_tool.getKeyValue(job, "job.par."+B.PAR_ENV, None)
self.assertEqual(res, test.testtools.DEFAULT_ENV)
cnttest += 1
for par in [B.ATTR_PATH_ARCHIV, B.ATTR_PATH_ENV]:
res = utils.path_tool.getKeyValue(job, "job.conf."+par, None)
if verbose: print("assertIn "+par+": "+res+" -- "+DATA_PATH)
self.assertIn(DATA_PATH, res)
cnttest += 1
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
def test_rejoinPath(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
@ -60,13 +40,13 @@ class MyTestCase(unittest.TestCase):
if actfunction not in TEST_FUNCTIONS:
return
if OS_SYSTEM == "linux":
res = utils.path_tool.rejoinPath("home", "ulrich")
res = tools.path_tool.rejoinPath("home", "ulrich")
self.assertEqual(res, "/home/ulrich")
cnttest += 1
res = utils.path_tool.rejoinPath("opt")
res = tools.path_tool.rejoinPath("opt")
self.assertEqual(res, "/opt")
cnttest += 1
res = utils.path_tool.rejoinPath("http://domain.com", "application", "function")
res = tools.path_tool.rejoinPath("http://domain.com", "application", "function")
self.assertEqual(res, "http://domain.com/application/function")
cnttest += 1
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
@ -81,41 +61,21 @@ class MyTestCase(unittest.TestCase):
# set the relevant datastrutures
job = test.testtools.getJob()
comp = test.testtools.getComp(job, "testrest")
pt = utils.path_tool.PathConf(job)
pt = tools.path_tool.PathConf(job)
self.setTestPaths(job)
self.setPathConfig(pt)
# tests
path = utils.path_tool.composePath(job, P.P_ENVBASE, None)
path = tools.path_tool.compose_path(job, P.P_ENVBASE, None)
self.assertIn(job.conf[B.SUBJECT_PATH][B.ATTR_PATH_ENV], path)
self.assertIn(getattr(job.par, B.PAR_ENV), path)
cnttest += 2
path = utils.path_tool.composePath(job, P.P_TCLOG, None)
path = tools.path_tool.compose_path(job, P.P_TCLOG, None)
self.assertIn(job.conf[B.SUBJECT_PATH][B.ATTR_PATH_ARCHIV], path)
self.assertIn(getattr(job.par, B.PAR_TESTCASE), path)
cnttest += 2
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
def test_composePattern(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
# set the relevant datastrutures
job = test.testtools.getJob()
comp = test.testtools.getComp(job)
pt = utils.path_tool.PathConf(job)
self.setTestPaths(job)
self.setPathConfig(pt)
# tests
path = utils.path_tool.composePattern(job, "{"+P.P_ENVBASE+"}", None)
self.assertIn(job.conf[B.SUBJECT_PATH][B.ATTR_PATH_ENV], path)
self.assertIn(getattr(job.par, B.PAR_ENV), path)
cnttest += 2
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
def test_extractPattern(self):
global mymsg
actfunction = str(inspect.currentframe().f_code.co_name)
@ -123,9 +83,9 @@ class MyTestCase(unittest.TestCase):
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
pt = utils.path_tool.PathConf(job)
pt = tools.path_tool.PathConf(job)
job = test.testtools.getJob("ts", test.testtools.DEFAULT_APP, test.testtools.DEFAULT_ENV, "2021-08-21_18-ß2-01")
r = utils.path_tool.extractPattern(job, "tsbase" )
r = tools.path_tool.extractPattern(job, "tsbase" )
self.assertEqual(r[0][1], "job.conf.archiv")
self.assertEqual(r[1][1], P.KEY_TESTSUITE)
self.assertEqual(r[1][2], "testlauf")
@ -140,8 +100,8 @@ class MyTestCase(unittest.TestCase):
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
pt = utils.path_tool.PathConf(job)
r = utils.path_tool.extractPath(job, "tsbase" , os.path.join(DATA_PATH, "lauf", "testlauf", "startjob_2021-08-21_10-02-01"))
pt = tools.path_tool.PathConf(job)
r = tools.path_tool.extractPath(job, "tsbase" , os.path.join(DATA_PATH, "lauf", "testlauf", "startjob_2021-08-21_10-02-01"))
if verbose: print("r " + str(r))
self.assertEqual(job.par.usecase, "startjob")
self.assertEqual(job.par.tstime, "2021-08-21_10-02-01")

8
test/test_07catalog.py

@ -10,7 +10,7 @@ import basic.message
import basic.program
import test.constants
import test.testtools
import objects.catalog
import model.catalog
HOME_PATH = test.constants.HOME_PATH
OS_SYSTEM = test.constants.OS_SYSTEM
@ -30,7 +30,7 @@ class MyTestCase(unittest.TestCase):
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
catalog = objects.catalog.Catalog.getInstance()
catalog = model.catalog.Catalog.getInstance()
self.assertIsNotNone(catalog)
cnttest += 1
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
@ -43,7 +43,7 @@ class MyTestCase(unittest.TestCase):
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
catalog = objects.catalog.Catalog.getInstance()
catalog = model.catalog.Catalog.getInstance()
self.assertRaises(Exception, catalog.readDomain, ("xxx", job))
cnttest += 1
try:
@ -67,7 +67,7 @@ class MyTestCase(unittest.TestCase):
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
catalog = objects.catalog.Catalog.getInstance()
catalog = model.catalog.Catalog.getInstance()
res = catalog.getValue(job, "countries", "key", "")
self.assertEqual(res, "")
self.assertEqual(job.m.rc, basic.message.RC_ERROR)

2
tools/config_tool.py

@ -233,7 +233,7 @@ def getConfig(job, modul, name, subname=""):
job.debug(verify, "getConfig " + pathname)
if len(pathname) < 1:
return confs
doc = tools.file_tool.readFileDict(job, pathname, msg)
doc = tools.file_tool.read_file_dict(job, pathname, msg)
if modul == D.DDL_FILENAME:
# in csv the root is the subname
# from the Dict-structure of DDL_FILENAME pick the substructure of the subname

4
tools/file_abstract.py

@ -78,7 +78,7 @@ class FileFcts():
schema = o["schema"]
archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/request", self.comp), filename) # ergebnisse/comp/request )
#txt = self.createDict()
tools.file_tool.writeFileText(self.comp.m, job, archivpath, txt)
tools.file_tool.write_tile_text(self.comp.m, job, archivpath, txt)
def send_request(self, job, step):
archivpath = ""
@ -94,7 +94,7 @@ class FileFcts():
fct = basic.toolHandling.getCliTool(job, self.comp)
fct.copy(self.job, archivpath, envpath)
elif technique == "api":
txt = tools.file_tool.readFileText(job, archivpath, self.comp.m)
txt = tools.file_tool.read_file_text(job, archivpath, self.comp.m)
fct = basic.toolHandling.getApiTool(job, self.comp)
response = fct.send(self.job, self.comp, txt)
archivpath = os.path.join(tools.path_tool.compose_path(job, "{tcresult}/response", self.comp), filename)

10
tools/file_tool.py

@ -181,14 +181,14 @@ def detectFileEncode(job, path, msg): # return ""
return 'utf-8'
def readFileLines(job, path, msg):
lines = readFileText(job, path, msg)
def read_file_lines(job, path, msg):
lines = read_file_text(job, path, msg)
if isinstance(lines, (str)):
return lines.splitlines()
return []
def readFileText(job, path, msg):
def read_file_text(job, path, msg):
# job = basic.program.Job.getInstance()
verify = int(job.getDebugLevel("file_tool"))
if not os.path.exists(path):
@ -207,7 +207,7 @@ def getModTime(job, filepath):
return out
def readFileDict(job, path, msg):
def read_file_dict(job, path, msg):
"""
reads and gets general a dict from any kind of filetyp
:param path: with extension of filetype
@ -258,7 +258,7 @@ def castOrderedDict(res, job=None, key=""):
return doc
def writeFileText(msg, job, path, text, enc="utf-8"):
def write_tile_text(msg, job, path, text, enc="utf-8"):
# job = basic.program.Job.getInstance()
verify = int(job.getDebugLevel("file_tool"))
mkPaths(job, path, msg)

2
tools/filecsv_fcts.py

@ -24,7 +24,7 @@ class FileFcts(tools.file_abstract.FileFcts):
:param text:
:return:
"""
lines = tools.file_tool.readFileLines(self.job, path, self.getMsg())
lines = tools.file_tool.read_file_lines(self.job, path, self.getMsg())
return self.parseCsv(self.getMsg(), self.job, lines)
def parseCsv(self, msg, job, lines, ttype=""):

186
tools/job_tool.py

@ -19,6 +19,10 @@
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:
@ -35,6 +39,44 @@ 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()
@ -62,41 +104,115 @@ def stopJobProcesses(job):
components.tools.job_tool.stopJobProcesses(job)
pass
def startProcess(job, process):
print(str(process))
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, 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)
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 ""

4
tools/value_tool.py

@ -10,7 +10,7 @@ import re
import basic.constants as B
import tools.path_const as P
import tools.date_tool
import objects.catalog
import model.catalog
TOOL_NAME = "value_tool"
@ -86,7 +86,7 @@ def get_key_value(job, key, comp=None):
pass
elif DOM_CTLG in key:
a = key.split(".")
catalog = objects.catalog.Catalog.getInstance()
catalog = model.catalog.Catalog.getInstance()
if len(a) == 3:
return catalog.getValue(job, a[1], a[2])
if len(a) == 4:

6
unit_tester.py

@ -65,9 +65,9 @@ if __name__ == '__main__':
job = basic.program.Job(PROGRAM_NAME)
# TODO to move to job_tool
args = {}
if isinstance(job.programDef[basic.program.CTLG_PARDEF], dict) and "par" in job.programDef[basic.program.CTLG_PARDEF]:
for p in job.programDef[basic.program.CTLG_PARDEF]["par"]:
args[p] = job.programDef[basic.program.CTLG_PARDEF]["par"][p]
if isinstance(job.programDef[basic.program.CTLG_PARDEF], dict):
for p in job.programDef[basic.program.CTLG_PARDEF]:
args[p] = job.programDef[basic.program.CTLG_PARDEF][p]
job.setParameter(args)
job.startJob()
startPyJob(job)

Loading…
Cancel
Save