Data-Test-Executer Framework speziell zum Test von Datenverarbeitungen mit Datengenerierung, Systemvorbereitungen, Einspielungen, ganzheitlicher diversifizierender Vergleich
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

257 lines
9.8 KiB

#!/usr/bin/python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------------------------------------------
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
import sys
import basic.constants as B
try:
import basic.program
except ImportError:
print("ImportError: " + str(ImportError.with_traceback()))
pass
import basic.componentHandling
import utils.path_tool
import utils.file_tool
import os.path
import basic.constants as B
import utils.data_const as D
import utils.path_const as P
COMP_FILES = [D.DDL_FILENAME]
CONFIG_FORMAT = [D.DFILE_TYPE_YML, D.DFILE_TYPE_JSON, D.DFILE_TYPE_CSV]
def getConfigPath(modul, name, subname=""):
"""
gets the most specified configuration of different sources
Parameter:
* typ -- (basic, comp, tool)
* name -- the specific class
sources:
* programm <<
* install <<
* environ << basis-conf
* release << basis-conf
* testset << parameter/environ
* testcase << parameter
the parameter-files could be one of these file-types:
* yaml, json, csv
"""
job = basic.program.Job.getInstance()
verify = job.getDebugLevel("config_tool")-4
job.debug(verify, "getConfig " + modul + ", " + name)
#TODO path rejoin, config as const
if modul == P.KEY_TOOL:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_HOME),
P.VAL_CONFIG, P.KEY_TOOL+"_"+name+"."+format)
job.debug(verify, "1 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_PROGRAM),
P.VAL_UTIL, P.VAL_CONFIG, name+"."+format)
job.debug(verify, "2 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_ENV),
job.par.environment, P.VAL_CONFIG, P.KEY_TOOL+"_"+ name+"."+format)
job.debug(verify, "3 " + pathname)
if os.path.exists(pathname):
return pathname
job.debug(verify, "3x " + pathname)
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
elif modul == P.KEY_COMP:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_HOME),
P.VAL_CONFIG, P.KEY_COMP+"_" + name + "."+format)
job.debug(verify, "4 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_COMPONENTS),
basic.componentHandling.getComponentFolder(name), "CONFIG." + format)
job.debug(verify, "5 " + pathname)
if os.path.exists(pathname):
return pathname
job.debug(verify, "6 " + pathname)
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
elif modul in COMP_FILES:
# for example DATASTRUCURE or the table
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_COMPONENTS),
basic.componentHandling.getComponentFolder(name), modul+"."+format)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_COMPONENTS),
basic.componentHandling.getComponentFolder(name), subname+"."+format)
if os.path.exists(pathname):
return pathname
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
elif modul == P.KEY_BASIC:
for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_COMPONENTS),
name + "."+format)
job.debug(verify, "4 " + pathname)
if os.path.exists(pathname):
return pathname
else:
pathname = utils.path_tool.composePath(P.P_TCPARFILE)
job.debug(verify, "7 " + pathname)
if os.path.exists(pathname):
return pathname
pathname = utils.path_tool.composePath(P.P_TSPARFILE)
job.debug(verify, "8 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_RELEASE),
P.VAL_CONFIG, "basis."+format)
job.debug(verify, "9 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_ENV),
P.VAL_CONFIG, "basis."+format)
job.debug(verify, "9 " + pathname)
if os.path.exists(pathname):
return pathname
for format in CONFIG_FORMAT:
if len(subname) > 1:
pathname = os.path.join(job.conf.getPath(P.ATTR_PATH_HOME),
P.VAL_CONFIG, "basis."+format)
job.debug(verify, "9 " + pathname)
if os.path.exists(pathname):
return pathname
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
def getConfValue(attribute, comp):
if attribute == B.ATTR_CONN_DBTYPE:
if not hasAttr(comp.conf[B.SUBJECT_CONN], "dbtype"):
if hasAttr(comp.conf[B.SUBJECT_CONN], "types") and hasAttr(comp.conf[B.SUBJECT_CONN]["types"], "dbtype"):
dbtype = comp.conf[B.SUBJECT_CONN]["types"]["dbtype"]
else:
raise LookupError("dbtype is not set in comp " + comp.name)
else:
dbtype = comp.conf["conn"]["dbtype"]
return ""
return ""
def getAttr(o, name):
#print("hasAttr " + str(type(o))+" "+name)
if (isinstance(o, dict)):
if (name in o.keys()):
#print("hasAttr dict ok " + str(type(o)))
return o[name]
#print("hasAttr dict "+str(type(o)))
elif (isinstance(o, list)):
pass
#print("hasAttr list "+str(type(o)))
elif hasattr(o, name):
#print("hasAttr class ok "+str(type(o)))
return getattr(o, name)
return False
def hasAttr(o, name):
#print("hasAttr " + str(type(o))+" "+name)
if (isinstance(o, dict)):
if (name in o.keys()):
#print("hasAttr dict ok " + str(type(o)))
return True
#print("hasAttr dict "+str(type(o)))
elif (isinstance(o, list)):
pass
#print("hasAttr list "+str(type(o)))
elif hasattr(o, name):
#print("hasAttr class ok "+str(type(o)))
return True
return False
def getConfig(modul, name, subname=""):
job = basic.program.Job.getInstance()
verify = job.getDebugLevel("config_tool")-4
msg = None
if hasattr(job, "m"): msg = job.m
pathname = getConfigPath(modul, name, subname)
confs = {}
job.debug(verify, "getConfig " + pathname)
if len(pathname) < 1:
return confs
doc = utils.file_tool.readFileDict(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
keys = list(doc.keys())
if subname not in keys and len(keys) == 1:
doc0 = doc[keys[0]]
doc = doc0
keys = list(doc.keys())
if subname in keys:
doc0 = doc[subname]
doc = doc0
for i, v in doc.items():
confs[i] = v
return confs
def mergeConn(msg, conf, conn):
"""
merges the config-attributes from the connection-attributes
because the connection-attributes has to overwrite the config-attributes if the subject is configured
:param conf:
:param conn:
:return:
"""
if B.SUBJECT_INST not in conf:
conf[B.SUBJECT_INST] = {}
for a in conn[B.SUBJECT_INST]:
conf[B.SUBJECT_INST][a] = conn[B.SUBJECT_INST][a]
for topic in [B.TOPIC_NODE_DB, B.TOPIC_NODE_CLI, B.TOPIC_NODE_API, B.TOPIC_NODE_FILE]:
if topic not in conf[B.SUBJECT_ARTS]:
continue
if topic == B.TOPIC_NODE_DB:
list = B.LIST_DB_ATTR
if topic == B.TOPIC_NODE_CLI:
list = B.LIST_CLI_ATTR
if topic == B.TOPIC_NODE_API:
list = B.LIST_API_ATTR
if topic == B.TOPIC_NODE_FILE:
list = B.LIST_FILE_ATTR
print(" --- merge-conn " + topic + " " + str(list))
for a in conf[B.SUBJECT_ARTS][topic]:
if topic not in conn:
continue
if a in list:
if a in conn[topic]:
conf[B.SUBJECT_ARTS][topic][a] = conn[topic][a]
else:
for b in conf[B.SUBJECT_ARTS][topic][a]:
print(" --- merge-conn b " + topic + " " + a+" "+b)
if b not in list:
msg.logError("not-topic-attribute in topic-connection: "+topic+", "+b)
continue
if a not in conn[topic]:
continue
if b in conn[topic][a]:
conf[B.SUBJECT_ARTS][topic][a][b] = conn[topic][a][b]
for a in list:
if topic not in conn:
break
if topic not in conn:
continue
if a in conn[topic]:
conf[B.SUBJECT_ARTS][topic][a] = conn[topic][a]
return conf