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" |
|||
DATA_PATH = "/home/ulrich/6_Projekte/Programme/data" |
|||
COMP_PATH = "/home/ulrich/6_Projekte/Programme/data/components" |
|||
#!/usr/bin/python |
|||
# -*- coding: utf-8 -*- |
|||
# --------------------------------------------------------------------------------------------------------- |
|||
# 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" |
|||
@ -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