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.
349 lines
14 KiB
349 lines
14 KiB
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
# ---------------------------------------------------------------------------------------------------------
|
|
# Author : Ulrich Carmesin
|
|
# Source : gitea.ucarmesin.de
|
|
# https://ucarmesin.de/index.php/it/testautomatisierung-fuer-daten-test/225-konfiguration-der-testanwendung
|
|
# ---------------------------------------------------------------------------------------------------------
|
|
import sys
|
|
import basic.constants as B
|
|
|
|
try:
|
|
import basic.program
|
|
except ImportError:
|
|
print("ImportError: " + str(ImportError.with_traceback()))
|
|
pass
|
|
import basic.componentHandling
|
|
import tools.path_tool
|
|
import tools.file_tool
|
|
import os.path
|
|
import basic.constants as B
|
|
import tools.data_const as D
|
|
import tools.path_const as P
|
|
|
|
COMP_FILES = [D.DDL_FILENAME]
|
|
CONFIG_FORMAT = [D.DFILE_TYPE_YML, D.DFILE_TYPE_JSON, D.DFILE_TYPE_CSV]
|
|
|
|
def getExistingPath(job, pathnames):
|
|
if isinstance(pathnames, str):
|
|
pathnames = [pathnames]
|
|
for p in pathnames:
|
|
if p[-1:] == ".":
|
|
p = p[0:-1]
|
|
for format in CONFIG_FORMAT:
|
|
pathname = p+"."+format
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
return None
|
|
|
|
def getConfigPath(job, 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
|
|
"""
|
|
if job is None:
|
|
verify = False # job = basic.program.Job.getInstance()
|
|
else:
|
|
verify = job.getDebugLevel("config_tool")-4
|
|
if verify: job.debug(verify, "getConfig " + modul + ", " + name)
|
|
if modul == P.KEY_TOOL:
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
|
|
P.VAL_CONFIG, P.KEY_TOOL+"_"+name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, P.KEY_TOOL+"_"+name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM],
|
|
P.VAL_UTIL, P.VAL_CONFIG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_ENV],
|
|
job.par.environment, P.VAL_CONFIG, P.KEY_TOOL+"_"+ name))
|
|
if path is not None:
|
|
return path
|
|
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
|
|
elif modul == P.KEY_COMP:
|
|
for format in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, P.KEY_COMP+"_" + name + "."+format)
|
|
if verify: job.debug(verify, "4 " + pathname)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
for format in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(name), "CONFIG." + format)
|
|
if verify: job.debug(verify, "5 " + pathname)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
if verify: job.debug(verify, "6 " + pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, modul+", "+name)
|
|
elif modul in COMP_FILES:
|
|
# for example DATASTRUCURE or the table
|
|
pathnames = []
|
|
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(name), modul))
|
|
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(subname), modul))
|
|
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, modul))
|
|
pathnames.append(os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, subname))
|
|
configpath = getExistingPath(job, pathnames)
|
|
if configpath is not None:
|
|
return configpath
|
|
for format in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf["paths"][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["paths"][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["paths"][P.ATTR_PATH_COMPONENTS],
|
|
P.VAL_CONFIG , name + "."+format)
|
|
if verify: job.debug(verify, "4 " + pathname)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
elif modul == P.KEY_TESTCASE:
|
|
for format in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_TDATA],
|
|
name, D.DFILE_TESTCASE_NAME + "."+format)
|
|
if verify: job.debug(verify, "4 " + pathname)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
elif modul == P.KEY_TESTSUITE:
|
|
for format in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf["paths"][P.ATTR_PATH_TDATA],
|
|
name, D.DFILE_TESTSUITE_NAME + "." + format)
|
|
if verify: job.debug(verify, "4 " + pathname)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
elif modul == P.KEY_CATALOG:
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_TDATA],
|
|
P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_COMPONENTS],
|
|
P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf["paths"][P.ATTR_PATH_PROGRAM],
|
|
P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
else:
|
|
pathname = tools.path_tool.composePath(job, P.P_TCPARFILE)
|
|
if verify: job.debug(verify, "7 " + pathname)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
pathname = tools.path_tool.composePath(job, P.P_TSPARFILE)
|
|
if verify: 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["paths"][P.ATTR_PATH_RELEASE],
|
|
P.VAL_CONFIG, "basis."+format)
|
|
if verify: 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["paths"][P.ATTR_PATH_ENV],
|
|
P.VAL_CONFIG, "basis."+format)
|
|
if verify: 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["paths"][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, "basis."+format)
|
|
if verify: 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):
|
|
if (isinstance(o, dict)):
|
|
if (name in o.keys()):
|
|
return o[name]
|
|
elif (isinstance(o, list)):
|
|
pass
|
|
elif hasattr(o, name):
|
|
return getattr(o, name)
|
|
return False
|
|
|
|
|
|
def hasAttr(o, name):
|
|
if (isinstance(o, dict)):
|
|
if (name in o.keys()):
|
|
return True
|
|
elif (isinstance(o, list)):
|
|
pass
|
|
|
|
|
|
elif hasattr(o, name):
|
|
return True
|
|
return False
|
|
|
|
|
|
def getConfig(job, modul, name, subname=""):
|
|
if job is None:
|
|
verify = 24
|
|
else:
|
|
verify = job.getDebugLevel("config_tool")-4
|
|
msg = None
|
|
if hasattr(job, "m"): msg = job.m
|
|
pathname = getConfigPath(job, modul, name, subname)
|
|
confs = {}
|
|
job.debug(verify, "getConfig " + pathname)
|
|
if len(pathname) < 1:
|
|
return confs
|
|
doc = tools.file_tool.readFileDict(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
|
|
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 getAttribute(comp, path, attr, job):
|
|
attrList = getAttributeList(comp, path, job)
|
|
if attr in attrList:
|
|
return attrList[attr]
|
|
else:
|
|
return ""
|
|
|
|
|
|
def getAttributeList(comp, path, job):
|
|
"""
|
|
gets a concrete attribute-list for an arteifact-element from the config-attributes from the connection-attributes
|
|
https://ucarmesin.de/index.php/it/testautomatisierung-fuer-daten-test/225-konfiguration-der-testanwendung#konfigurationshierarchie
|
|
:param comp:
|
|
:param path: artifact-type.artifact-name for example: DB.person
|
|
:return: list of all attributes for the artifact-element
|
|
"""
|
|
attrList = {}
|
|
a = path.split(".")
|
|
artType = a[0]
|
|
artName = a[1]
|
|
if B.SUBJECT_CONN not in comp.conf:
|
|
raise Exception ("Environment is not configured")
|
|
if artType in comp.conf[B.SUBJECT_CONN]:
|
|
if artName in comp.conf[B.SUBJECT_CONN][artType]:
|
|
for attr, val in comp.conf[B.SUBJECT_CONN][artType][artName].items():
|
|
if attr not in B.LIST_ATTR[artType]:
|
|
continue
|
|
attrList[attr] = val
|
|
for attr, val in comp.conf[B.SUBJECT_CONN][artType].items():
|
|
if attr not in B.LIST_ATTR[artType]:
|
|
continue
|
|
if attr in attrList:
|
|
continue
|
|
attrList[attr] = val
|
|
if artType in comp.conf[B.SUBJECT_ARTS]:
|
|
if artName in comp.conf[B.SUBJECT_ARTS][artType]:
|
|
for attr, val in comp.conf[B.SUBJECT_ARTS][artType][artName].items():
|
|
if attr not in B.LIST_ATTR[artType]:
|
|
continue
|
|
if attr in attrList:
|
|
continue
|
|
attrList[attr] = val
|
|
for attr, val in comp.conf[B.SUBJECT_ARTS][artType].items():
|
|
if attr not in B.LIST_ATTR[artType]:
|
|
continue
|
|
if attr in attrList:
|
|
continue
|
|
attrList[attr] = val
|
|
return attrList
|
|
|
|
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
|
|
https://ucarmesin.de/index.php/it/testautomatisierung-fuer-daten-test/225-konfiguration-der-testanwendung#konfigurationshierarchie
|
|
: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
|