datestex/utils/tdata_tool.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------------------------------------------
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
"""
the issue of this tool is to transform extern data to the internal structure and the internal structure into extern data - i.e. mostly test-results.
* * * * * * * *
the testdata have several elements
* parameter (-td --tdata) : to identify which testdata should be loaded
* source (flaskdb: dbname / dir: filename) : always structured in a table (easy to specify) with columns
  * node : where the rows are
  * action : what should be done - default insert
  + fields : dates in relation of a reference<day or a formula
* interface : configured in components and used in comparison with attributes to each field:
  * ignored - if it should be ignored on differences, it is necessary on technical ID-fields
  * id-field - necessary
* * * * * * * *
the testdata itself which are written in different artifacts of modern applications are mostly stored as tree
- so as xml, json, always with plain data in the leaf. So the intern structure should be also a tree - in python: dictionary.
"""
import os.path

import basic.program
import utils.file_tool
import basic.constants as B

DATA_SOURCE_DIR = "dir"
DATA_SOURCE_CSV = "csv"

CSV_HEADER_START = ["node", "table", "tabelle"]
CSV_DELIMITER = ";"

CSV_SPECTYPE_DATA = "data"
CSV_SPECTYPE_TREE = "tree"

def getSourcefile(filename):
    pass

def getTestdata():
    """
    get the testdata from one of the possible soources
    * dir: each file in the specific testarchiv
    * csv: specific file
    * db:  specific db with a testcase-catalogue
    :return:
    """
    job = basic.program.Job.getInstance()
    reftyp = getattr(job.par, "tdtyp")
    source = getattr(job.par, "tdsrc")
    criteria = getattr(job.par, "tdname")
    tdata = {"reftyp": reftyp, "source": source, "criteria": criteria}
    if reftyp == "flaskdb":
        # read data-structure with sourcename
        # connect to source
        # select with all data with datastructure
        job.m.setInfo("Test-Data readed from " + reftyp + " for " + criteria)
    elif reftyp == "csv":
        # read file in testdata
        job.m.logInfo("Test-Data readed from " + reftyp + " for " + criteria)
    elif reftyp == "dir":
        filename = os.path.join(job.conf.getJobConf("paths:testdata"), source, "testspec.csv")
        data = getCsvSpec(job.m, filename, CSV_SPECTYPE_DATA)
        for k in data:
            tdata[k] = data[k]
    else:
        job.m.setFatal("test-Data: reftyp " + reftyp + " is not implemented")
    return tdata

def getCsvSpec(msg, filename, type):
    """
    get data from a csv-file
    a) data : like a table with data-array of key-value-pairs
    b) tree : as a tree - the rows must be unique identified by the first column
    """
    data = {}
    header = []
    lines = utils.file_tool.readFileLines(msg, filename)
    status = "start"
    for l in lines:
        fields = l.split(CSV_DELIMITER)
        # check empty line, comment
        if (len(l.strip().replace(CSV_DELIMITER,"")) < 1):
            status = "start"
            continue
        if (fields[0][0:1] == "#"):
            continue
        a = fields[0].split(":")
        # keywords option, step, table
        if (a[0] not in data):
            data[a[0]] = {}
        if (a[0].lower() == "step"):
            if (not "steps" in data):
                data["steps"] = []
            step = {}
            step["comp"] = fields[1]
            step["todo"] = fields[2]
            step["args"] = fields[3]
            data["steps"].append(step)
            continue
        elif (a[0].lower() == "option"):
            data[a[0]][a[1]] = fields[1]
            continue
        elif (a[0].lower() in CSV_HEADER_START):
            i = 0
            for f in fields:
                i += 1
                if i == 1: continue
                header.append(f)
            status = CSV_SPECTYPE_DATA
            continue
        elif (status == CSV_SPECTYPE_DATA):
            # check A-col for substructure
            if (a[0] not in data):
                data[a[0]] = {}
                if len(a) == 1 and type == CSV_SPECTYPE_DATA:
                    data[a[0]][B.DATA_NODE_DATA] = []
                # its a component
            if len(a) > 1 and a[1] not in data[a[0]]:
                data[a[0]][a[1]] = {}
                if len(a) == 2 and type == CSV_SPECTYPE_DATA:
                    data[a[0]][a[1]]["_data"] = []
            if len(a) > 2 and a[2] not in data[a[0]][a[1]]:
                data[a[0]][a[1]][a[2]] = {}
                if len(a) == 3 and type == CSV_SPECTYPE_DATA:
                    data[a[0]][a[1]][a[1]][B.DATA_NODE_DATA] = []
            # fill data
            row = {}
            i = 1
            for f in header:
                row[f] = fields[i]
                i += 1
            if len(a) == 1 and type == CSV_SPECTYPE_DATA:
                data[a[0]][B.DATA_NODE_DATA].append(row)
            elif len(a) == 1 and type == CSV_SPECTYPE_DATA:
                data[a[0]] = {f: row}
            elif len(a) == 2 and type == CSV_SPECTYPE_DATA:
                data[a[0]][a[1]][B.DATA_NODE_DATA].append(row)
            elif len(a) == 1 and type == CSV_SPECTYPE_DATA:
                data[a[0]][a[1]][B.DATA_NODE_DATA] = {f: row}
            elif len(a) == 3 and type == CSV_SPECTYPE_DATA:
                data[a[0]][a[1]][a[2]] = row
            elif len(a) == 1 and type == CSV_SPECTYPE_DATA:
                data[a[0]][a[1]][a[2]] = {f: row}
    return data

def readCsv(msg, filename, comp):
    job = basic.program.Job.getInstance()
    verify = -1+job.getDebugLevel("tdata_tool")
    job.debug(verify, "readCsv " + filename)
    fields = []
    nodes = []
    columns = []
    output = {}
    state = 0
    data = []
    cnt = 0
    lines = utils.file_tool.readFileLines(msg, filename)
    startCols = 1
    for line in lines:
        fields = line.split(';')
        testline = line.replace(";", "")
        job.debug(verify, str(state) + " line " + line + " :" + str(len(fields)) + ": " + str(fields))
        if len(testline) < 2 and state < 1:
            state = 0
        elif fields[0].lower() in CSV_HEADER_START:
            state = 2
            columns = []
            cnt = len(fields)
            job.debug(verify, str(state) + " cnt " + str(cnt))
            j = 0
            for i in range(1, cnt-1):
                if fields[0][0:1] == "_":
                    startCols += 1
                    continue
                job.debug(verify, str(i) + " cnt " + str(fields[i]))
                if len(fields[i]) > 0:
                    columns.append(fields[i])
                    j = j + 1
                cnt = j
            job.debug(verify, str(state) + " " + str(cnt) + " cols " + str(columns))
        elif state >= 2 and len(testline) > 2:
            if state == 2:
                nodes = fields[0].split(":")
                job.debug(verify, str(state) + " nodes " + str(nodes))
            state = 3
            row = {}
            for i in range(startCols, cnt-1):
                row[columns[i-2]] = fields[i]
            job.debug(verify, str(state) + " row " + str(row))
            data.append(row)
        elif state == 3:
            job.debug(verify, "structure " + str(state) + ": " + str(nodes))
            output = setSubnode(0, nodes, data, output)
            data = []
            state = 0
    output = setSubnode(0, nodes, data, output)
    return output

def setSubnode(i, nodes, data, tree):
    print("setSubnode " + str(i) + ": " + ": " + str(tree))
    if i >= len(nodes):
        print("setSubnode a " + str(i))
        tree["data"] = data
    elif tree is not None and nodes[i] in tree.keys():
        print("setSubnode b " + str(i))
        tree[nodes[i]] = setSubnode(i+1, nodes, data, tree[nodes[i]])
    else:
        print("setSubnode c " + str(i))
        tree[nodes[i]] = setSubnode((i + 1), nodes, data, {})
    return tree

def getDataStructure(comp):
    # gets data-structure from the vml in the component-folder
    job = basic.program.Job.getInstance()
    verify = -1+job.getDebugLevel("tdata_tool")
    job.debug(verify, "getDataStructure " + comp)

def normalizeDataRow(dstruct, xpathtupel, row, referencedate):
    # normalize data of the row if necessary
    # raw-value is saved as new field with _raw as suffix
    job = basic.program.Job.getInstance()
    verify = -1+job.getDebugLevel("tdata_tool")
    job.debug(verify, "calcDataRow " + row)

def writeDataTable(filename, tdata, comp):
    """
    writes the testdata into a csv-file for documentation of the test-run
    :param teststatus:
    :param tdata:
    :param comp: if specific else None
    :return:
    """
    job = basic.program.Job.getInstance()
    verify = -1+job.getDebugLevel("tdata_tool")
    job.debug(verify, "writeDataTable " + str(comp))
    text = "table"
    for f in tdata[B.DATA_NODE_HEADER]:
        text += ";"+f
    for r in tdata[B.DATA_NODE_DATA]:
        text += "\n"
        for f in tdata[B.DATA_NODE_HEADER]:
            if f in r:
                text += ";"+r[f]
            else:
                text += ";"
    text += "\n"
    utils.file_tool.writeFileText(comp.m, filename, text)