Compare commits
5 Commits
6cc11e2bc5
...
2443134bfc
| Author | SHA1 | Date |
|---|---|---|
 Ulrich Carmesin
|
2443134bfc | 2 years ago |
|
Ulrich Carmesin
|
830589fead | 2 years ago |
|
Ulrich Carmesin
|
3307c789db | 2 years ago |
|
Ulrich Carmesin
|
cae2beaa60 | 2 years ago |
|
Ulrich Carmesin
|
a86a816e98 | 2 years ago |
35 changed files with 1001 additions and 188 deletions
@ -0,0 +1,146 @@ |
|||||
|
|
||||
|
""" |
||||
|
|
||||
|
""" |
||||
|
import os.path |
||||
|
import json |
||||
|
import basic.program |
||||
|
import basic.constants as B |
||||
|
import utils.file_tool |
||||
|
tempJob = {} |
||||
|
|
||||
|
PROGRAM_NAME = "unit" |
||||
|
JSON_FILE = "actualJob.json" |
||||
|
JOB_NR = { |
||||
|
"init_testsuite": { |
||||
|
"jobnr": "0" }, |
||||
|
"execute_testsuite": { |
||||
|
"jobnr": "1"}, |
||||
|
"collect_testsuite": { |
||||
|
"jobnr": "2"}, |
||||
|
"compare_testsuite": { |
||||
|
"jobnr": "3"}, |
||||
|
"finish_testsuite": { |
||||
|
"jobnr": "4"}, |
||||
|
"init_testcase": { |
||||
|
"jobnr": "5" }, |
||||
|
"execute_testcase": { |
||||
|
"jobnr": "6" }, |
||||
|
"collect_testcase": { |
||||
|
"jobnr": "7" }, |
||||
|
"compare_testcase": { |
||||
|
"jobnr": "8" }, |
||||
|
"check_environment": { |
||||
|
"jobnr": "9" }, |
||||
|
"test_executer": { |
||||
|
"jobnr": "10"}, |
||||
|
} |
||||
|
JOB_LIST = [ |
||||
|
"init_testsuite", |
||||
|
"execute_testsuite", |
||||
|
"collect_testsuite", |
||||
|
"compare_testsuite", |
||||
|
"finish_testsuite", |
||||
|
"init_testcase", |
||||
|
"execute_testcase", |
||||
|
"collect_testcase", |
||||
|
"compare_testcase", |
||||
|
"check_environment", |
||||
|
"test_executer" |
||||
|
] |
||||
|
|
||||
|
appList = [] |
||||
|
envList = ["ENV01"] |
||||
|
testList = {"TEST": [ |
||||
|
"TC0001", "TST001" |
||||
|
]} |
||||
|
|
||||
|
def readContext(job): |
||||
|
for k in job.conf.confs["applications"]: |
||||
|
appList.append(k) |
||||
|
|
||||
|
|
||||
|
def restartActualProcess(job): |
||||
|
""" |
||||
|
check if an actual process is open |
||||
|
:return: |
||||
|
""" |
||||
|
path = os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_DEBUG], JSON_FILE) |
||||
|
if os.path.exists(path): |
||||
|
actProc = utils.file_tool.readFileDict(path) |
||||
|
jobNr = int(JOB_NR[actProc["job"]]["jobnr"]) |
||||
|
question = "Prozess " |
||||
|
choiceList = "" |
||||
|
if jobNr < 8 and jobNr != 4: |
||||
|
question += "F_ortsetzen | " |
||||
|
choiceList += "f | " |
||||
|
question += "W_iederholen | N_euen Prozess | X_exit" |
||||
|
choiceList = "w | n | x ?" |
||||
|
print("+-----------------------------------------------") |
||||
|
for k in actProc: |
||||
|
print('| {:6s} : {:60s}'.format(k, actProc[k])) |
||||
|
print("+-----------------------------------------------") |
||||
|
print(question) |
||||
|
choice = input("Auswahl "+choiceList) |
||||
|
choiceList.replace(" | ","") |
||||
|
if choice.lower() not in choiceList[:-1]: |
||||
|
print("FEHLER : falsche Auswahl") |
||||
|
elif choice.lower() == "x": |
||||
|
job.stopJob(0) |
||||
|
elif choice.lower() == "w": |
||||
|
startProcess(job, actProc) |
||||
|
elif choice.lower() == "f": |
||||
|
actProc["job"] = JOB_NR[int(jobNr)+1] |
||||
|
startProcess(job, actProc) |
||||
|
elif choice.lower() == "n": |
||||
|
createProcess(job) |
||||
|
else: |
||||
|
print("unbekannte Situation") |
||||
|
else: |
||||
|
createProcess(job) |
||||
|
|
||||
|
|
||||
|
def createProcess(job): |
||||
|
process = {} |
||||
|
index = 0 |
||||
|
print("create new process") |
||||
|
process["app"] = getChoice(job, appList, "Anwendung") |
||||
|
# |
||||
|
if len(envList) == 1: |
||||
|
process["env"] = envList[0] |
||||
|
else: |
||||
|
process["env"] = getChoice(job, envList, "Umgebung") |
||||
|
# |
||||
|
process["tc"] = getChoice(job, testList[process["app"]], "Testfall") |
||||
|
print(str(process)) |
||||
|
|
||||
|
|
||||
|
def getChoice(job, choiselist, description): |
||||
|
index = 0 |
||||
|
print("+------------- "+description+" ----------") |
||||
|
print('| {:2d} : {:60s}'.format(0, "exit")) |
||||
|
for k in choiselist: |
||||
|
index += 1 |
||||
|
print('| {:2d} : {:60s}'.format(index, k)) |
||||
|
print("+-----------------------------------------------") |
||||
|
choice = input("Auswahl 1-" + str(index) + ": ") |
||||
|
if not choice.isnumeric(): |
||||
|
print("FEHLER Fehleingabe "+choice) |
||||
|
getChoice(job, choiselist, description) |
||||
|
elif int(choice) < 1: |
||||
|
exit(0) |
||||
|
elif int(choice) > index: |
||||
|
print("FEHLER Fehleingabe "+choice) |
||||
|
getChoice(job, choiselist, description) |
||||
|
else: |
||||
|
return choiselist[int(choice) - 1] |
||||
|
|
||||
|
|
||||
|
def startProcess(job, process): |
||||
|
print(str(process)) |
||||
|
|
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
job = basic.program.Job(PROGRAM_NAME) |
||||
|
readContext(job) |
||||
|
restartActualProcess(job) |
||||
@ -1,4 +1,27 @@ |
|||||
HOME_PATH = "/home/ulrich/6_Projekte/Programme/datest" |
#!/usr/bin/python |
||||
DATA_PATH = "/home/ulrich/6_Projekte/Programme/data" |
# -*- coding: utf-8 -*- |
||||
COMP_PATH = "/home/ulrich/6_Projekte/Programme/data/components" |
# --------------------------------------------------------------------------------------------------------- |
||||
|
# Author : Ulrich Carmesin |
||||
|
# Source : gitea.ucarmesin.de |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
""" |
||||
|
constants |
||||
|
""" |
||||
|
import os |
||||
|
home = os.getcwd() |
||||
|
prgdir = "" |
||||
|
if home[-4:] == "test" and home[-6:] != "datest": |
||||
|
home = home[0:-5] |
||||
|
if home[-10:] == "components": |
||||
|
home = home[0:-11] |
||||
|
if home[-6:] == "datest": |
||||
|
prgdir = home[-6:] |
||||
|
home = home[0:-7] |
||||
|
elif home[-7:] == "program": |
||||
|
prgdir = home[-7:] |
||||
|
home = home[0:-8] |
||||
|
HOME_PATH = home |
||||
|
DATA_PATH = os.path.join(home, "data") |
||||
|
PROG_PATH = os.path.join(home, prgdir) |
||||
|
COMP_PATH = os.path.join(home, prgdir, "components") |
||||
OS_SYSTEM = "linux" |
OS_SYSTEM = "linux" |
||||
@ -0,0 +1,27 @@ |
|||||
|
import unittest |
||||
|
import basic.program |
||||
|
import utils.conn_tool |
||||
|
|
||||
|
TEST_FUNCTIONS = ["test_01conn"] |
||||
|
#TEST_FUNCTIONS = ["test_01conn"] |
||||
|
verbose = True |
||||
|
|
||||
|
class MyTestCase(unittest.TestCase): |
||||
|
def runTest(self): |
||||
|
self.test_conn() |
||||
|
|
||||
|
def test_01conn(self): |
||||
|
if verbose: print("# # # # testComponents # # # # #") |
||||
|
job = basic.program.Job("unit") |
||||
|
args = { "application" : "TEST" , "environment" : "ENV01", "modus" : "unit", "loglevel" : "debug", "tool" : "job_tool"} |
||||
|
job.par.setParameterArgs(args) |
||||
|
self.assertEqual(True, True) |
||||
|
conn = utils.conn_tool.getConnection("testcm", 1) |
||||
|
if verbose: print("test-conn " + str(conn)) |
||||
|
conns = utils.conn_tool.getConnections("testcrmdb") |
||||
|
if verbose: print("test-conn " + str(conns)) |
||||
|
conns = utils.conn_tool.getConnections("testprd") |
||||
|
if verbose: print("test-conn " + str(conns)) |
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
unittest.main() |
||||
@ -0,0 +1,96 @@ |
|||||
|
#!/usr/bin/python |
||||
|
# -*- coding: utf-8 -*- |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
# Author : Ulrich Carmesin |
||||
|
# Source : gitea.ucarmesin.de |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
import unittest |
||||
|
import os |
||||
|
import inspect |
||||
|
import utils.path_tool |
||||
|
import basic.program |
||||
|
import basic.constants as B |
||||
|
import test.constants |
||||
|
import test.testtools |
||||
|
import utils.path_const as P |
||||
|
import utils.i18n_tool |
||||
|
import basic.constants as B |
||||
|
|
||||
|
HOME_PATH = test.constants.HOME_PATH |
||||
|
OS_SYSTEM = test.constants.OS_SYSTEM |
||||
|
|
||||
|
# here you can select single testfunction for developping the tests |
||||
|
TEST_FUNCTIONS = ["test_01class", "test_02text", "test_03aliasList"] |
||||
|
#TEST_FUNCTIONS = [ "test_03aliasList"] |
||||
|
verbose = False |
||||
|
|
||||
|
EXP_KEY_MISSING = "key is missing {}" |
||||
|
EXP_KEY_DOESNT_EXIST = "key doesnt exist in domain {}" |
||||
|
|
||||
|
class MyTestCase(unittest.TestCase): |
||||
|
mymsg = "--------------------------------------------------------------" |
||||
|
|
||||
|
def test_01class(self): |
||||
|
global mymsg |
||||
|
actfunction = str(inspect.currentframe().f_code.co_name) |
||||
|
cnttest = 0 |
||||
|
if actfunction not in TEST_FUNCTIONS: |
||||
|
return |
||||
|
job = test.testtools.getJob() |
||||
|
i18n = utils.i18n_tool.I18n.getInstance() |
||||
|
self.assertIsNotNone(i18n) |
||||
|
cnttest += 1 |
||||
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
||||
|
|
||||
|
|
||||
|
def test_02text(self): |
||||
|
global mymsg |
||||
|
actfunction = str(inspect.currentframe().f_code.co_name) |
||||
|
cnttest = 0 |
||||
|
if actfunction not in TEST_FUNCTIONS: |
||||
|
return |
||||
|
job = test.testtools.getJob() |
||||
|
job.conf.confs["language"] = "de" |
||||
|
i18n = utils.i18n_tool.I18n.getInstance() |
||||
|
res = i18n.getText(f"{EXP_KEY_MISSING=}", job) |
||||
|
if verbose: print("RESULT "+res) |
||||
|
self.assertEqual(res, "Schluesselwort fehlt {}") |
||||
|
cnttest += 1 |
||||
|
res = i18n.getText(f"{EXP_KEY_DOESNT_EXIST=}", job) |
||||
|
if verbose: print("RESULT "+res) |
||||
|
self.assertEqual(res, "key doesnt exist in domain {}") |
||||
|
cnttest += 1 |
||||
|
res = i18n.getText(f"{B.ATTR_APPS_PROJECT=}", job) |
||||
|
if verbose: print("RESULT "+res) |
||||
|
self.assertEqual(res, "project") |
||||
|
cnttest += 1 |
||||
|
job.conf.confs["language"] = "fr" |
||||
|
i18n = utils.i18n_tool.I18n.getInstance() |
||||
|
self.assertRaises(Exception, i18n.getText, (f"{EXP_KEY_MISSING=}", job)) |
||||
|
cnttest += 1 |
||||
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
||||
|
|
||||
|
|
||||
|
def test_03aliasList(self): |
||||
|
global mymsg |
||||
|
actfunction = str(inspect.currentframe().f_code.co_name) |
||||
|
cnttest = 0 |
||||
|
if actfunction not in TEST_FUNCTIONS: |
||||
|
return |
||||
|
job = test.testtools.getJob() |
||||
|
job.conf.confs["language"] = "de" |
||||
|
i18n = utils.i18n_tool.I18n.getInstance() |
||||
|
# i18n.getText("EXP_KEY_MISSING", EXP_KEY_MISSING, job) |
||||
|
res = i18n.getAliasList(f"{EXP_KEY_MISSING=}", job) |
||||
|
if verbose: print("RESULT "+str(res)) |
||||
|
self.assertEqual(res, ['key is missing {}', 'Schluesselwort fehlt {}']) |
||||
|
cnttest += 1 |
||||
|
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
||||
|
|
||||
|
def test_zzz(self): |
||||
|
if verbose: print(MyTestCase.mymsg) |
||||
|
|
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
verbose = True |
||||
|
unittest.main() |
||||
@ -1,23 +0,0 @@ |
|||||
import unittest |
|
||||
import basic.program |
|
||||
import utils.conn_tool |
|
||||
|
|
||||
class MyTestCase(unittest.TestCase): |
|
||||
def runTest(self): |
|
||||
self.test_conn() |
|
||||
|
|
||||
def test_conn(self): |
|
||||
print("# # # # tetsComponents # # # # #") |
|
||||
job = basic.program.Job("unit") |
|
||||
args = { "application" : "TEST" , "environment" : "ENV01", "modus" : "unit", "loglevel" : "debug", "tool" : "job_tool"} |
|
||||
job.par.setParameterArgs(args) |
|
||||
self.assertEqual(True, True) |
|
||||
conn = utils.conn_tool.getConnection("testb", 1) |
|
||||
print("test-conn " + str(conn)) |
|
||||
conns = utils.conn_tool.getConnections("testb") |
|
||||
print("test-conn " + str(conns)) |
|
||||
conns = utils.conn_tool.getConnections("testa") |
|
||||
print("test-conn " + str(conns)) |
|
||||
|
|
||||
if __name__ == '__main__': |
|
||||
unittest.main() |
|
||||
@ -0,0 +1,102 @@ |
|||||
|
#!/usr/bin/python |
||||
|
# -*- coding: utf-8 -*- |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
# Author : Ulrich Carmesin |
||||
|
# Source : gitea.ucarmesin.de |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
import basic.program |
||||
|
import utils.config_tool |
||||
|
import utils.db_abstract |
||||
|
import mysql.connector |
||||
|
import basic.constants as B |
||||
|
|
||||
|
class DbFcts(utils.db_abstract.DbFcts): |
||||
|
""" |
||||
|
This interface defines each necessary connection to any kind of database. |
||||
|
The specific technique how to connect to the concrete DBMS has to be implemented in the specific tool. |
||||
|
""" |
||||
|
|
||||
|
def __init__(self): |
||||
|
pass |
||||
|
|
||||
|
def selectRows(self, table): |
||||
|
""" method to select rows from a database |
||||
|
statement written in sql """ |
||||
|
tdata = {} |
||||
|
job = basic.program.Job.getInstance() |
||||
|
verify = -1+job.getDebugLevel("db_tool") |
||||
|
cmd = "SELECT * FROM "+table+";" |
||||
|
#mycursor = self.getConnector() |
||||
|
#mycursor.execute(cmd) |
||||
|
#myresult = mycursor.fetchall() |
||||
|
tdata[B.DATA_NODE_HEADER] = [] |
||||
|
for f in self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER]: |
||||
|
tdata[B.DATA_NODE_HEADER].append(f) |
||||
|
myresult = [] |
||||
|
for x in myresult: |
||||
|
print(x) |
||||
|
self.comp.m.logInfo(cmd) |
||||
|
return tdata |
||||
|
|
||||
|
def deleteRows(self, table): |
||||
|
""" method to delete rows from a database |
||||
|
statement written in sql """ |
||||
|
job = basic.program.Job.getInstance() |
||||
|
verify = -1+job.getDebugLevel("db_tool") |
||||
|
cmd = "DELETE FROM "+table+";" |
||||
|
self.comp.m.logInfo(cmd) |
||||
|
|
||||
|
def updateRows(self, statement): |
||||
|
""" method to delete rows from a database |
||||
|
statement written in sql """ |
||||
|
raise Exception(B.EXCEPT_NOT_IMPLEMENT) |
||||
|
|
||||
|
def insertRows(self, table, rows): |
||||
|
""" method to insert rows into a database |
||||
|
the rows will be interpreted by the ddl of the component |
||||
|
""" |
||||
|
job = basic.program.Job.getInstance() |
||||
|
verify = -1+job.getDebugLevel("db_tool") |
||||
|
cmd = "INSERT INTO "+table+";" |
||||
|
header = "" |
||||
|
for h in self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER]: |
||||
|
print(h) |
||||
|
header += ", "+h |
||||
|
cmd += " (" + header[1:]+" ) " |
||||
|
rowvalues = "" |
||||
|
for r in rows: |
||||
|
print("r-----------------") |
||||
|
print(r) |
||||
|
rowvalues = "" |
||||
|
cmd += "\n ( " |
||||
|
for h in self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_HEADER]: |
||||
|
print("h "+h) |
||||
|
if (h in r): |
||||
|
rowvalues += ", "+self.getDbValue(self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_DATA][h], r[h]) |
||||
|
else: |
||||
|
rowvalues += ", "+self.getDbValue(self.comp.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_DATA][h], "") |
||||
|
print("rv " + rowvalues) |
||||
|
cmd += rowvalues[1:]+" )," |
||||
|
cmd = cmd[0:-1]+";" |
||||
|
self.comp.m.logInfo(cmd) |
||||
|
|
||||
|
def getConnector(self): |
||||
|
""" add-on-method to get the connector |
||||
|
this method should only called by the class itself """ |
||||
|
job = basic.program.Job.getInstance() |
||||
|
mydb = mysql.connector.connect( |
||||
|
host = "localhost", |
||||
|
user = "datest", |
||||
|
password = "Advent!2021", |
||||
|
database = "datest" |
||||
|
) |
||||
|
return mysql |
||||
|
|
||||
|
@staticmethod |
||||
|
def execStatement(self, comp, conn, statement): |
||||
|
""" add-on-method to execute the statement |
||||
|
this method should only called by the class itself """ |
||||
|
raise Exception(B.EXCEPT_NOT_IMPLEMENT) |
||||
|
|
||||
|
|
||||
|
|
||||
@ -0,0 +1,95 @@ |
|||||
|
#!/usr/bin/python |
||||
|
# -*- coding: utf-8 -*- |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
# Author : Ulrich Carmesin |
||||
|
# Source : gitea.ucarmesin.de |
||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
import utils.config_tool |
||||
|
import utils.path_const as P |
||||
|
import basic.program |
||||
|
|
||||
|
DEFAULT_LANGUAGE = "en" |
||||
|
EXP_KEY_MISSING = "key is missing {}" |
||||
|
EXP_KEY_DOESNT_EXIST = "key doesnt exist in domain {}" |
||||
|
|
||||
|
class I18n: |
||||
|
__instance = None |
||||
|
cache = {} |
||||
|
""" |
||||
|
in this class there should be managed each defined key-value-pairs |
||||
|
the pairs ara loaded from the path testdata/catalog: |
||||
|
* initially the csv-file catalog.csv |
||||
|
* on demand other csv-files in the path |
||||
|
""" |
||||
|
|
||||
|
def __init__(self): |
||||
|
self.cache = {} |
||||
|
self.cache = utils.config_tool.getConfig(P.KEY_TOOL, "i18n") |
||||
|
I18n.__instance = self |
||||
|
pass |
||||
|
|
||||
|
|
||||
|
@staticmethod |
||||
|
def getInstance(): |
||||
|
if I18n.__instance == None: |
||||
|
return I18n() |
||||
|
return I18n.__instance |
||||
|
|
||||
|
|
||||
|
def getText(self, key, job=None): |
||||
|
""" |
||||
|
this function gets the text depending on language which is set in job.conf |
||||
|
:param domain: |
||||
|
:param key: |
||||
|
:return: |
||||
|
if not (isinstance(langue, str) and len(langue)): |
||||
|
raise Exception(EXP_KEY_MISSING, (domain, key)) |
||||
|
if not (isinstance(key, str) and len(key)): |
||||
|
raise Exception(EXP_KEY_MISSING, (langue, key)) |
||||
|
|
||||
|
return self.cache[langue][key] |
||||
|
""" |
||||
|
if job is None: |
||||
|
jon = basic.program.Job.getInstance() |
||||
|
language = job.conf.confs["language"] |
||||
|
if language not in self.cache: |
||||
|
raise Exception(EXP_KEY_MISSING, (key)) |
||||
|
print(key) |
||||
|
out = self.extractText(key) |
||||
|
key = self.extractKey(key) |
||||
|
if key in self.cache[language]: |
||||
|
out = self.cache[language][key] |
||||
|
elif key in self.cache[DEFAULT_LANGUAGE]: |
||||
|
out = self.cache[DEFAULT_LANGUAGE][key] |
||||
|
return out |
||||
|
|
||||
|
def getAliasList(self, key, job=None): |
||||
|
if job is None: |
||||
|
jon = basic.program.Job.getInstance() |
||||
|
out = [] |
||||
|
key = self.extractKey(key) |
||||
|
for language in self.cache: |
||||
|
out.append(self.cache[language][key]) |
||||
|
return out |
||||
|
|
||||
|
def extractKey(self, key): |
||||
|
if "=" in key: |
||||
|
i = key.find("=") |
||||
|
x = key[0:i] |
||||
|
if "." in x: |
||||
|
i = x.find(".") |
||||
|
y = x[i + 1:] |
||||
|
else: |
||||
|
y = x |
||||
|
key = y |
||||
|
print("= in key " + y) |
||||
|
else: |
||||
|
print("!= in key") |
||||
|
y = key |
||||
|
return key |
||||
|
|
||||
|
def extractText(self, key): |
||||
|
if "=" in key: |
||||
|
i = key.find("=") |
||||
|
return key[i + 2:-1] |
||||
|
return "" |
||||
Loading…
Reference in new issue