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.
487 lines
18 KiB
487 lines
18 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
|
|
import tools.value_tool
|
|
|
|
try:
|
|
import basic.program
|
|
except ImportError:
|
|
print("ImportError: " + str(ImportError))
|
|
pass
|
|
import basic.componentHandling
|
|
import tools.path_tool
|
|
import tools.file_tool
|
|
import tools.data_tool
|
|
import os.path
|
|
import basic.constants as B
|
|
import tools.data_const as D
|
|
import tools.path_const as P
|
|
|
|
TABLE_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 f in CONFIG_FORMAT:
|
|
pathname = p + "." + f
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
return None
|
|
|
|
|
|
def select_config_path(job: any, modul: str, name: str, subname="") -> str:
|
|
"""
|
|
gets the most specified configuration of different sources
|
|
:param job:
|
|
:param modul:
|
|
:param name:
|
|
:param subname:
|
|
:return:
|
|
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")
|
|
if modul == P.KEY_TOOL:
|
|
return getToolPath(job, name, subname)
|
|
elif modul == P.KEY_COMP:
|
|
for f in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, P.KEY_COMP + "_" + name + "." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
for f in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(name), "CONFIG." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, modul + ", " + name)
|
|
elif modul in TABLE_FILES:
|
|
return getTablePath(job, name, subname, modul)
|
|
elif modul == P.KEY_BASIC:
|
|
return getBasicPath(job, name)
|
|
elif modul == P.KEY_TESTCASE:
|
|
return getTestPath(job, name, B.SUBJECT_TESTCASES, D.DFILE_TESTCASE_NAME)
|
|
elif modul == P.KEY_TESTSUITE:
|
|
return getTestPath(job, name, B.SUBJECT_TESTSUITES, D.DFILE_TESTSUITE_NAME)
|
|
elif modul == P.KEY_CATALOG:
|
|
return getCatalogPath(job, name)
|
|
elif modul == P.KEY_ENV:
|
|
return getEnvironmentPath(job, name)
|
|
elif modul == P.KEY_USER:
|
|
return getUserPath(job, name)
|
|
elif modul == P.KEY_MODEL:
|
|
return getModelPath(job, name)
|
|
else:
|
|
pathname = tools.path_tool.compose_path(job, P.P_TCPARFILE)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
pathname = tools.path_tool.compose_path(job, P.P_TSPARFILE)
|
|
if os.path.exists(pathname):
|
|
return pathname
|
|
for f in CONFIG_FORMAT:
|
|
if len(subname) > 1:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_RELEASE],
|
|
P.VAL_CONFIG, "basis." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
for f in CONFIG_FORMAT:
|
|
if len(subname) > 1:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_ENV],
|
|
P.VAL_CONFIG, "basis." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
for f in CONFIG_FORMAT:
|
|
if len(subname) > 1:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, "basis." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, modul + ", " + name)
|
|
|
|
|
|
def getToolPath(job, name, subname):
|
|
if subname != "":
|
|
envdir = subname
|
|
elif hasattr(job.par, "environment"):
|
|
envdir = job.par.environment
|
|
else:
|
|
envdir = "xxxxx"
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_ENV],
|
|
envdir, P.VAL_CONFIG, P.KEY_TOOL + "_" + name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
P.VAL_CONFIG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM],
|
|
P.VAL_TOOLS, P.VAL_CONFIG, name))
|
|
if path is not None:
|
|
return path
|
|
raise Exception(P.EXP_CONFIG_MISSING, envdir + ", " + name)
|
|
|
|
|
|
def getTablePath(job: any, name: str, subname: str, filename: str) -> str:
|
|
"""
|
|
reads the ddl of the table depending on context (=name)
|
|
a) component: the ddl is read from specific or general component-folder
|
|
b) testcase: the ddl is read from general component-folder
|
|
c) testserver: the ddl is read from model-folder
|
|
:param job:
|
|
:param name: name of the context testserver/testdata or name of the component
|
|
:param subname: name of the table
|
|
:param filename: dont use
|
|
:return:
|
|
"""
|
|
pathnames = []
|
|
if name == B.ATTR_INST_TESTSERVER:
|
|
path = tools.value_tool.compose_string(job, job.conf[B.TOPIC_PATH][P.ATTR_PATH_MODEL], None)
|
|
pathnames.append(os.path.join(path, subname))
|
|
else:
|
|
path = tools.value_tool.compose_string(job, job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS], None)
|
|
pathnames.append(os.path.join(path, P.KEY_CATALOG, P.VAL_TABLES, subname))
|
|
pass
|
|
"""
|
|
pathnames.append(os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(name), filename))
|
|
pathnames.append(os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(subname), filename))
|
|
pathnames.append(os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, filename))
|
|
pathnames.append(os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM], P.VAL_BASIC, subname))
|
|
"""
|
|
print("pathnames " + str(pathnames))
|
|
configpath = getExistingPath(job, pathnames)
|
|
print("configpath " + str(configpath))
|
|
if configpath is not None:
|
|
return configpath
|
|
if name == B.ATTR_INST_TESTSERVER:
|
|
# print(name)
|
|
for f in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM],
|
|
P.ATTR_PATH_MODEL, subname + "." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
for f in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(name), filename + "." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
for f in CONFIG_FORMAT:
|
|
if len(subname) > 1:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
basic.componentHandling.getComponentFolder(name), subname + "." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, filename + ", " + name)
|
|
|
|
|
|
def getBasicPath(job: any, name: str) -> str:
|
|
"""
|
|
the function searchs the config-file in the basic-context.
|
|
:param job:
|
|
:param name:
|
|
:return:
|
|
"""
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME],
|
|
P.VAL_CONFIG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
P.VAL_CONFIG, name))
|
|
if path is not None:
|
|
return path
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
|
|
|
|
def getEnvironmentPath(job: any, name: str) -> str:
|
|
"""
|
|
the function searchs the config-file in the environment-context.
|
|
:param job:
|
|
:param name:
|
|
:return:
|
|
"""
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_ENV], name,
|
|
P.VAL_CONFIG, B.SUBJECT_ENVIRONMENT))
|
|
if path is not None:
|
|
return path
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
|
|
|
|
def getTestPath(job: any, name: str, subdir: str, filename: str) -> str:
|
|
for f in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_TDATA], job.par.project,
|
|
subdir, name, filename + "." + f)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
|
|
|
|
def getCatalogPath(job: any, name: str) -> str:
|
|
"""
|
|
the function searchs the config-file in the catalog-context.
|
|
:param job:
|
|
:param name:
|
|
:return:
|
|
"""
|
|
if hasattr(job, "par") and hasattr(job.par, B.SUBJECT_PROJECT):
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_TDATA],
|
|
getattr(job.par, B.SUBJECT_PROJECT), P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_TDATA],
|
|
P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
|
|
P.KEY_CATALOG, P.VAL_TABLES, name))
|
|
if path is not None:
|
|
return path
|
|
path = getExistingPath(job, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM],
|
|
P.KEY_CATALOG, name))
|
|
if path is not None:
|
|
return path
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
|
|
|
|
def getUserPath(job: any, name: str) -> str:
|
|
"""
|
|
the function searchs the config-file in the user-context.
|
|
:param job:
|
|
:param name:
|
|
:return:
|
|
"""
|
|
for ext in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME], P.VAL_CONFIG,
|
|
P.VAL_USER, name + "." + ext)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
|
|
|
|
def getModelPath(job: any, name: str) -> str:
|
|
"""
|
|
the function searchs the config-file in the model-context.
|
|
:param job:
|
|
:param name:
|
|
:return:
|
|
"""
|
|
for ext in CONFIG_FORMAT:
|
|
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM], P.ATTR_PATH_MODEL, name + "." + ext)
|
|
if os.path.exists(pathname):
|
|
return str(pathname)
|
|
raise Exception(P.EXP_CONFIG_MISSING, name)
|
|
|
|
|
|
def get_plain_filename(job, path):
|
|
"""
|
|
the function extracts the plain filename without directory-name and extension
|
|
:param job:
|
|
:param path:
|
|
:return:
|
|
"""
|
|
names = list(os.path.split(path))
|
|
filename = names.pop()
|
|
for ext in CONFIG_FORMAT:
|
|
if filename[-len(ext):] == ext:
|
|
filename = filename[:-len(ext) - 1]
|
|
return filename
|
|
return filename
|
|
|
|
|
|
def getConfValue(attribute: str, comp: any) -> str:
|
|
if attribute == B.ATTR_TYPE:
|
|
if not hasAttr(comp.conf[B.TOPIC_CONN], "dbtype"):
|
|
if hasAttr(comp.conf[B.TOPIC_CONN], "types") and hasAttr(comp.conf[B.TOPIC_CONN]["types"], "dbtype"):
|
|
dbtype = comp.conf[B.TOPIC_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: str, name: str, subname: str = "", ttype: str = D.CSV_SPECTYPE_DATA) -> dict:
|
|
"""
|
|
searches for configration and read it
|
|
:param job:
|
|
:param modul:
|
|
:param name:
|
|
:param subname:
|
|
:param ttype: type of config / csv-type
|
|
:return:
|
|
"""
|
|
if job is None:
|
|
verify = 0
|
|
else:
|
|
verify = job.getDebugLevel("config_tool")
|
|
msg = None
|
|
if hasattr(job, "m"): msg = job.m
|
|
pathname = select_config_path(job, modul, name, subname)
|
|
confs = {}
|
|
if pathname is None:
|
|
return {}
|
|
if len(pathname) < 1:
|
|
return confs
|
|
if ttype == "" and modul in ["tool", "comp"]:
|
|
ttype = modul
|
|
doc = tools.file_tool.read_file_dict(job, pathname, msg, ttype)
|
|
# TODO ?! unnoetiges verschiebe
|
|
for i, v in doc.items():
|
|
confs[i] = v
|
|
return confs
|
|
|
|
|
|
def getAttribute(comp, path, attr, job) -> str:
|
|
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.TOPIC_CONN not in comp.conf:
|
|
raise Exception("Environment is not configured")
|
|
if artType in comp.conf[B.TOPIC_CONN]:
|
|
if artName in comp.conf[B.TOPIC_CONN][artType]:
|
|
for attr, val in comp.conf[B.TOPIC_CONN][artType][artName].items():
|
|
if attr not in B.LIST_ATTR[artType]:
|
|
continue
|
|
attrList[attr] = val
|
|
for attr, val in comp.conf[B.TOPIC_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_ARTIFACTS]:
|
|
if artName in comp.conf[B.SUBJECT_ARTIFACTS][artType]:
|
|
for attr, val in comp.conf[B.SUBJECT_ARTIFACTS][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_ARTIFACTS][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.TOPIC_INST not in conf:
|
|
conf[B.TOPIC_INST] = {}
|
|
for a in conn[B.TOPIC_INST]:
|
|
conf[B.TOPIC_INST][a] = conn[B.TOPIC_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_ARTIFACTS]:
|
|
continue
|
|
if topic == B.TOPIC_NODE_DB:
|
|
attrlist = B.LIST_DB_ATTR
|
|
elif topic == B.TOPIC_NODE_CLI:
|
|
attrlist = B.LIST_CLI_ATTR
|
|
elif topic == B.TOPIC_NODE_API:
|
|
attrlist = B.LIST_API_ATTR
|
|
elif topic == B.TOPIC_NODE_FILE:
|
|
attrlist = B.LIST_FILE_ATTR
|
|
else:
|
|
attrlist = []
|
|
for a in conf[B.SUBJECT_ARTIFACTS][topic]:
|
|
if topic not in conn:
|
|
continue
|
|
if a in attrlist:
|
|
if a in conn[topic]:
|
|
conf[B.SUBJECT_ARTIFACTS][topic][a] = conn[topic][a]
|
|
else:
|
|
for b in conf[B.SUBJECT_ARTIFACTS][topic][a]:
|
|
if b not in attrlist:
|
|
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_ARTIFACTS][topic][a][b] = conn[topic][a][b]
|
|
for a in attrlist:
|
|
if topic not in conn:
|
|
break
|
|
if topic not in conn:
|
|
continue
|
|
if a in conn[topic]:
|
|
conf[B.SUBJECT_ARTIFACTS][topic][a] = conn[topic][a]
|
|
return conf
|
|
|