Browse Source

status 20220814

master
Ulrich Carmesin 2 years ago
parent
commit
a86a816e98
  1. 4
      basic/component.py
  2. 3
      basic/componentHandling.py
  3. 17
      basic/message.py
  4. 4
      basic/program.py
  5. 17
      basic/toolHandling.py
  6. 146
      start_dialog.py
  7. 5
      test/test_toolhandling.py
  8. 87
      utils/file_abstract.py
  9. 26
      utils/file_tool.py

4
basic/component.py

@ -8,7 +8,7 @@
# ---------------------------------------------------------------------
from datetime import datetime
import basic.testexec
import basic.compexec
import basic.message
import basic.program
import inspect
@ -24,7 +24,7 @@ class CompData:
#class Component(components.sysmonitor.SystemMonitor, components.testexec.Testexecuter, components.report.Report,
# components.maintain.Maintainer, components.catalog, threading.Thread):
class Component(basic.testexec.Testexecuter):
class Component(basic.compexec.Testexecuter):
"""
A component represents an application of the system-under-test or a data-artifact which is created from the system-under-test.
As the representation it has to knowlegde of the url, which other components depends on this component.

3
basic/componentHandling.py

@ -179,6 +179,7 @@ class ComponentManager:
:param nr: number if component is multiple
:return: instance of the component with all necessary attributes
"""
job = basic.program.Job.getInstance()
cmodul = importlib.import_module(getComponentPath(compName))
class_ = getattr(cmodul, getComponentClass(compName))
c = class_()
@ -189,7 +190,7 @@ class ComponentManager:
name = compName
i = 0
c.name = name
c.m = basic.message.Message(basic.message.LIMIT_DEBUG, "logTime", name)
c.m = basic.message.Message(job, basic.message.LIMIT_DEBUG, "logTime", name)
c.conf = utils.config_tool.mergeConn(c.m, confs["conf"], conns[i])
c.conf[B.SUBJECT_CONN] = conns[i]
c.init()

17
basic/message.py

@ -53,14 +53,9 @@ class Message:
"debug": "24", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht
"trace": "28", # wird nur in debug ausgegeben, wenn log-level hoechstens auf eingestelltem job-level steht
"""
def __init__(self, level, logTime, componente):
def __init__(self, job, level, logTime, componente):
# (self, componente, out, level):
self.componente = componente # dezantrales Logsystem
#self.level = level # Vorgabe zu level zum Filtern, ob auszugebe
job = basic.program.Job.getInstance()
# print(vars(job))
# print(globals())
# exit(5)
verify = LIMIT_DEBUG
self.initErrorTyp()
self.rc = RC_INFO
@ -87,23 +82,23 @@ class Message:
elif job.program == "unit":
self.logfile = self.debugfile
else:
self.setLogdir(logTime)
self.setLogdir(job, logTime)
self.topmessage = ""
self.messages = []
print("message initialisiert mit level " + str(self.level))
def setLogdir(self, logTime):
job = basic.program.Job.getInstance()
def setLogdir(self, job, logTime):
basedir = job.par.basedir
basedir = basedir.replace("base", "log")
# basedir = utils.path_tool.composePath(basedir, None)
basedir = utils.path_tool.composePath(basedir, None)
os.makedirs(basedir, exist_ok=True)
logpath = basedir + "/protokoll_" + logTime + ".txt"
logpath = os.path.join(basedir , job.program + "_" + logTime + ".txt")
self.logDebug("logfile " + logpath)
self.logfile = open(logpath, "w")
def initErrorTyp(self):
self.CONST_ERRTYP = {
"fatal": "4", # wird immer in debug und log ausgegeben, setzt RC

4
basic/program.py

@ -129,7 +129,7 @@ class Job:
if program == "unit": # no job will be started
self.start = datetime.now()
logTime = self.start.strftime("%Y%m%d_%H%M%S")
self.m = basic.message.Message(basic.message.LIMIT_DEBUG, logTime, None)
self.m = basic.message.Message(self, basic.message.LIMIT_DEBUG, logTime, None)
print("prog-50 " + str(self.par.basedir))
@staticmethod
@ -214,7 +214,7 @@ class Job:
self.start = datetime.now()
print("prog-68 " + str(self.par.basedir))
logTime = self.start.strftime("%Y%m%d_%H%M%S")
self.m = basic.message.Message(basic.message.LIMIT_DEBUG, logTime, None)
self.m = basic.message.Message(self, basic.message.LIMIT_DEBUG, logTime, None)
print("prog-68 " + str(self.m.rc))
self.par.setParameterLoaded()
self.m.logInfo("# # # Start Job " + self.start.strftime("%d.%m.%Y %H:%M:%S") + " # # # ")

17
basic/toolHandling.py

@ -4,7 +4,8 @@
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
import importlib, os
import importlib
import os
import basic.program
import basic.constants as B
# -------------------------------------------------
@ -91,3 +92,17 @@ def getApiTool(comp):
c = class_()
c.setComp(comp)
return c
def getFileTool(job, comp=None):
verify = int(job.getDebugLevel("db_tool"))
filetype = getCompAttr(comp, B.TOPIC_NODE_FILE, B.ATTR_TYPE, "")
toolname = "file"+filetype+"_tool"
filepath = os.path.join(job.conf.confs[B.SUBJECT_PATH][B.ATTR_PATH_PROGRAM], "utils", toolname+".py")
#comp.m.debug(verify, "toolname "+filepath)
if not os.path.exists(filepath):
raise FileNotFoundError("file for tool "+toolname+" does not exist "+filepath)
cmodul = importlib.import_module("utils."+toolname)
class_ = getattr(cmodul, "FileFcts")
c = class_()
c.setComp(job, comp)
return c

146
start_dialog.py

@ -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)

5
test/test_toolhandling.py

@ -40,11 +40,16 @@ class MyTestCase(unittest.TestCase):
self.assertRegex(str(type(tool)), 'dbmysql_tool.DbFcts')
tool = basic.toolHandling.getCliTool(comp)
self.assertRegex(str(type(tool)), 'clissh_tool.CliFcts')
comp.conf[B.SUBJECT_CONN][B.TOPIC_NODE_FILE] = {}
comp.conf[B.SUBJECT_CONN][B.TOPIC_NODE_FILE][B.ATTR_TYPE] = "xml"
tool = basic.toolHandling.getFileTool(job, comp)
self.assertRegex(str(type(tool)), 'filexml_tool.FileFcts')
comp.conf[B.SUBJECT_CONN][B.TOPIC_NODE_DB][B.ATTR_TYPE] = "dxx"
comp.conf[B.SUBJECT_CONN][B.TOPIC_NODE_CLI][B.ATTR_TYPE] = "sxx"
self.assertRaises(FileNotFoundError, basic.toolHandling.getDbTool, comp)
self.assertRaises(FileNotFoundError, basic.toolHandling.getCliTool, comp)
if __name__ == '__main__':
unittest.main()

87
utils/file_abstract.py

@ -4,12 +4,26 @@
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
import os
import basic.program
import basic.catalog
import utils.config_tool
import basic.constants as B
import basic.toolHandling
import utils.data_const as D
import utils.date_tool
import utils.file_tool
import utils.path_tool
"""
# TODO Beschreibung file-tools
* fileXml_tool
* fileJson_tool
* fileYaml_tool
* fileCsv_tool
* fileHtml_tool
...
"""
class FileFcts():
"""
@ -18,6 +32,9 @@ class FileFcts():
def __init__(self):
pass
def setComp(self, job, comp=None):
self.job = job
self.comp = comp
def parseText(self, text):
"""
@ -28,4 +45,70 @@ class FileFcts():
def file2dict(self):
pass
pass
# Funktionen
#=== init_testcase ===
def removeFiles(self):
utils.file_tool.removeFiles(self.comp.m, "envpath", "pattern", self.comp.conf["conn"])
def copyFiles(self):
fileList = []
srcpath = ""
envpath = ""
pattern = ""
utils.file_tool.copyFiles(self.job, fileList, srcpath, envpath, pattern)
def readEnvFiles(self):
envpath = ""
pattern = ""
fileList = utils.file_tool.getFiles(self.comp.m, envpath, pattern, self.comp.conf["conn"])
# === execute_testcase ===
def create_request(self, tdata, step):
mapping = ""
schema = ""
archivpath = ""
filename = step.args["filename"]
txt = ""
for o in self.comp.conf[B.SUBJECT_ARTS][B.TOPIC_NODE_FILE]:
if o["name"] != filename:
continue
mapping = o["mapping"]
schema = o["schema"]
archivpath = os.path.join(utils.path_tool.composePattern("{tcresult}/request", self.comp), filename) # ergebnisse/comp/request )
#txt = self.createDict()
utils.file_tool.writeFileText(self.comp.m, archivpath, txt)
def send_request(self, step):
archivpath = ""
filename = step.args["filename"]
technique = step.args["technique"]
archivpath = os.path.join(utils.path_tool.composePattern("{tcresult}/request", self.comp), filename)
if technique == "cli":
for o in self.comp.conf[B.SUBJECT_ARTS][B.TOPIC_NODE_FILE]:
if o["name"] != filename:
continue
envpath = o["envpath"]
envpath = utils.path_tool.composePattern(envpath, self.comp)
fct = basic.toolHandling.getCliTool(self.comp)
fct.copy(self.job, archivpath, envpath)
elif technique == "api":
txt = utils.file_tool.readFileText(archivpath, self.comp.m)
fct = basic.toolHandling.getApiTool(self.comp)
response = fct.send(self.job, self.comp, txt)
archivpath = os.path.join(utils.path_tool.composePattern("{tcresult}/response", self.comp), filename)
"""
get_response:
- channel (sync/async)
... implement
- archivpath ( ergebnisse/comp/response )
- envpath ( ./log) / envconn ( = in Request empfangen )
=== collect_testcase ===
- envpath
- pattern
> readfiles
"""

26
utils/file_tool.py

@ -50,6 +50,32 @@ def getFiles(msg, path, pattern, conn):
return out
def removeFiles(msg, path, pattern, conn):
"""
search filenames in the directory and removes it
- if conn is set search remote
:param msg: -- msg-Objekt
:param path: -- path - String
:param pattern: -- filename as Pattern
:param conn:
:return: Array filenames
"""
pass
def copyFiles(job, fileList, source, target, comp):
"""
copies files from source to target
:param job:
:param fileList:
:param source:
:param target:
:param comp:
:return:
"""
pass
def getRemoteFiles(msg, path, pattern, conn):
"""
search filenames in the directory - if conn is set search remote

Loading…
Cancel
Save