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.

488 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