datestex/tools/config_tool.py

#!/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))
    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 select_config_path(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")
    if verify: job.debug(verify, "getConfig " + modul + ", " + name)
    if modul == P.KEY_TOOL:
        return getToolPath(job, name, subname)
    elif modul == P.KEY_COMP:
        for format in CONFIG_FORMAT:
            pathname = os.path.join(job.conf[B.TOPIC_PATH][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[B.TOPIC_PATH][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:
        return getCompPath(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 verify: job.debug(verify, "7 " + pathname)
        if os.path.exists(pathname):
            return pathname
        pathname = tools.path_tool.compose_path(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[B.TOPIC_PATH][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[B.TOPIC_PATH][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[B.TOPIC_PATH][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 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 getCompPath(job, name, subname, filename):
    # for example DATASTRUCURE or the table
    pathnames = []
    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))
    configpath = getExistingPath(job, pathnames)
    if configpath is not None:
        return configpath
    if name == B.ATTR_INST_TESTSERVER:
        print(name)
        for format in CONFIG_FORMAT:
            pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_PROGRAM],
                                    P.ATTR_PATH_MODEL, subname + "." + format)
            print(pathname)
            if os.path.exists(pathname):
                return pathname

    for format in CONFIG_FORMAT:
        pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_COMPONENTS],
                                basic.componentHandling.getComponentFolder(name), filename + "." + format)
        print(pathname)
        if os.path.exists(pathname):
            return pathname
    for format 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 + "." + format)
        if os.path.exists(pathname):
            return pathname
    raise Exception(P.EXP_CONFIG_MISSING, filename + ", " + name)

def getBasicPath(job, name):
    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

def getEnvironmentPath(job, name):
    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

def getTestPath(job, name, subdir, filename):
    for format in CONFIG_FORMAT:
        pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_TDATA], job.par.project,
                                subdir, name, filename + "." + format)
        if os.path.exists(pathname):
            return pathname

def getCatalogPath(job, name):
    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, name):
    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 pathname
    return None

def getModelPath(job, name):
    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 pathname
    return None


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, comp):
    if attribute == B.ATTR_CONN_DBTYPE:
        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, name, subname=""):
    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 None
    job.debug(verify, "getConfig " + pathname)
    if len(pathname) < 1:
        return confs
    if modul == D.DDL_FILENAME:
        doc = tools.file_tool.read_file_dict(job, pathname, msg, D.CSV_SPECTYPE_DATA)
        # 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
    else:
        doc = tools.file_tool.read_file_dict(job, pathname, msg)
    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.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_ARTIFACT]:
        if artName in comp.conf[B.SUBJECT_ARTIFACT][artType]:
            for attr, val in comp.conf[B.SUBJECT_ARTIFACT][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_ARTIFACT][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_ARTIFACT]:
            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
        for a in conf[B.SUBJECT_ARTIFACT][topic]:
            if topic not in conn:
                continue
            if a in list:
                if a in conn[topic]:
                    conf[B.SUBJECT_ARTIFACT][topic][a] = conn[topic][a]
            else:
                for b in conf[B.SUBJECT_ARTIFACT][topic][a]:
                    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_ARTIFACT][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_ARTIFACT][topic][a] = conn[topic][a]
    return conf