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.
 
 
 

333 lines
11 KiB

import os
import basic.constants as B
import tools.data_const as D
import model.factory
MUST_NODES = "must"
MUSTNT_NODES = "mustnt"
OPT_NODES = "optional"
def rebuild_tdata(job, tdata: dict, tableAttr: dict, ttype:str) -> dict:
"""
it rebuilds the data into the specific form
:param job:
:param tdata:
:param ttype:
:return:
"""
if ttype == D.CSV_SPECTYPE_DDL:
return DatatypeDDL.rebuild_data(job, tdata, tableAttr)
elif ttype == D.CSV_SPECTYPE_DATA:
return rebuildSpec(job, tdata)
elif ttype == D.CSV_SPECTYPE_CONF:
return rebuildConfig(job, tdata)
elif ttype == D.CSV_SPECTYPE_CTLG:
return DatatypeCatalog.rebuild_data(job, tdata, tableAttr)
# return rebuildCatalog(job, tdata)
elif ttype == D.CSV_SPECTYPE_TREE:
return rebuildCatalog(job, tdata)
elif ttype + "s" in B.LIST_SUBJECTS or ttype == B.SUBJECT_USER:
enty = model.factory.get_entity_object(job, ttype, {})
return enty.rebuild_data(job, tdata)
elif ttype in ["basic", "tool"]:
return tdata
else:
raise Exception("ttype is not defined " + ttype)
def insert_fields(job, tdata, fields: list, ttype: str) -> dict:
if ttype == D.CSV_SPECTYPE_DDL:
return tdata
elif ttype == D.CSV_SPECTYPE_DATA:
return tdata
elif ttype == D.CSV_SPECTYPE_CONF:
return tdata
elif ttype == D.CSV_SPECTYPE_CTLG:
return tdata
elif ttype == D.CSV_SPECTYPE_TREE:
return tdata
elif ttype in B.LIST_SUBJECTS:
return tdata
else:
job.m.logError("ttype is not defined " + ttype)
return tdata
def buildRow(job, tdata, fields: list) -> dict:
"""
build simple rows from fields
:param job:
:param tdata:
:param fields:
:return:
"""
row = {}
# TODO ? welcher Fall ?
#if ttype == D.CSV_SPECTYPE_DATA and ":" not in fields[0] and D.DATA_ATTR_ALIAS in tableDict:
# fields = [tableDict[D.DATA_ATTR_ALIAS]] + fields
# simple fields of a table:
# table:name;col1;col2;...
# ;val1;val2;...
i = 1
for f in tdata[B.DATA_NODE_HEADER]:
# --> still not used
# TODO arguments - as json or simplified "arg1:val1,arg2:val2,.."
if f in [B.DATA_NODE_ARGS, "args"]:
arguments = {}
row[B.DATA_NODE_ARGS] = arguments
if B.DATA_NODE_ARGS in row:
a = fields[i].split(":")
row[B.DATA_NODE_ARGS][a[0]] = a[1]
# <-- still not used
else:
row[f] = fields[i]
i += 1
for arg in fields[len(tdata[B.DATA_NODE_HEADER])+1:]:
# args := arg1:val1
if len(arg) == 0 or arg.strip()[0:1] == "#":
continue
print("arg "+arg)
a = arg.split(":")
row[B.DATA_NODE_ARGS][a[0]] = a[1]
return row
def check_tdata(job, tdata: dict, ttype:str) -> dict:
"""
it checks the data for the specific form
:param job:
:param tdata:
:param ttype:
:return:
"""
if ttype == D.CSV_SPECTYPE_DDL:
return DatatypeDDL.check_data(job, tdata)
elif ttype == D.CSV_SPECTYPE_DATA:
return checkSpec(job, tdata)
elif ttype == D.CSV_SPECTYPE_CONF:
return checkConfig(job, tdata)
elif ttype == D.CSV_SPECTYPE_COMP:
return checkComp(job, tdata)
elif ttype == D.CSV_SPECTYPE_CTLG:
return DatatypeCatalog.check_data(job, tdata)
elif ttype == D.CSV_SPECTYPE_TREE:
return checkCatalog(job, tdata)
elif ttype + "s" in B.LIST_SUBJECTS or ttype == B.SUBJECT_USER:
enty = model.factory.get_entity_object(job, ttype, {})
return enty.check_data(job, tdata)
elif ttype in ["basic"]:
return tdata
else:
job.m.logError("ttype is not defined " + ttype)
return tdata
def rebuildConfig(job, tdata: dict) -> dict:
return tdata
def insertConfig(job, tdata, fields: list) -> dict:
# TODO test it - configuration exists as a tree (yml, json, ...)
return tdata
def checkConfig(job, tdata: dict) -> dict:
checkNodes = {}
checkNodes[MUST_NODES] = [B.DATA_NODE_HEADER, B.DATA_NODE_DATA]
checkNodes[MUSTNT_NODES] = [B.DATA_NODE_FIELDS, B.DATA_NODE_KEYS]
checkNodes[OPT_NODES] = []
return check_nodes(job, tdata, checkNodes)
def rebuildComp(job, tdata: dict) -> dict:
return tdata
def insertComp(job, tdata, fields: list) -> dict:
# TODO test it - comp-configuration exists as a tree (yml, json, ...)
row = buildRow(job, tdata, fields)
tdata[fields[1]] = row
return tdata
def checkComp(job, tdata: dict) -> dict:
checkNodes = {}
checkNodes[MUST_NODES] = [B.SUBJECT_ARTIFACTS, B.SUBJECT_STEPS, "functions", B.SUBJECT_DATATABLES]
checkNodes[MUSTNT_NODES] = [B.DATA_NODE_DATA]
checkNodes[OPT_NODES] = []
return check_nodes(job, tdata, checkNodes)
class DatatypeCatalog():
"""
structure:
* B.DATA_NODE_HEADER : list of ddl-attributes
* B.DATA_NODE_FIELDS : list of field-names
* B.DATA_NODE_KEYS : fields with attributes (header X fields)
"""
@staticmethod
def rebuild_data(job, data: dict, tableAttr: dict) -> dict:
data = popTablesNode(job, data)
data = popNameNode(job, data)
data = addTableAttr(job, data, tableAttr)
return data
@staticmethod
def check_data(job, data: dict) -> dict:
checkNodes = {}
checkNodes[MUST_NODES] = [B.DATA_NODE_HEADER, B.DATA_NODE_KEYS]
checkNodes[MUSTNT_NODES] = [B.DATA_NODE_DATA]
checkNodes[OPT_NODES] = [B.DATA_NODE_FIELDS]
check_nodes(job, data, checkNodes)
return data
def popSubjectsNode(job, data: dict) -> dict:
if len(data) == 1:
key = list(data.keys())[0]
if key in B.LIST_SUBJECTS:
data = data[key]
return data
def popTablesNode(job, data: dict) -> dict:
if B.DATA_NODE_TABLES not in data:
return data
outdata = {}
# if double-DATA_NODE_TABLES
if B.DATA_NODE_TABLES in data and B.DATA_NODE_TABLES in data[B.DATA_NODE_TABLES]:
for k in data[B.DATA_NODE_TABLES][B.DATA_NODE_TABLES]:
if k in data[B.DATA_NODE_TABLES]:
print("Error")
else:
data[B.DATA_NODE_TABLES][k] = data[B.DATA_NODE_TABLES][B.DATA_NODE_TABLES][k]
data[B.DATA_NODE_TABLES].pop(B.DATA_NODE_TABLES)
if len(data[B.DATA_NODE_TABLES]) > 1:
job.m.setError("Mehr als eine Tabelle")
return data
elif len(data[B.DATA_NODE_TABLES]) == 0:
job.m.setError("Keine Tabelle")
return outdata
else:
for k in data[B.DATA_NODE_TABLES]:
outdata[k] = data[B.DATA_NODE_TABLES][k]
return outdata
def popNameNode(job, data: dict) -> dict:
outdata = {}
for k in data:
if "_"+D.FIELD_NAME in data[k] and k == data[k]["_"+D.FIELD_NAME]:
for l in data[k]:
outdata[l] = data[k][l]
else:
outdata[k] = data[k]
return outdata
def addTableAttr(job, data: dict, tableAttr: dict) -> dict:
for k in tableAttr:
if k == "_hit":
continue
data[k] = tableAttr[k]
return data
def rebuildCatalog(job, tdata: dict) -> dict:
return tdata
def insertCatalog(job, tdata, fields: list) -> dict:
row = buildRow(job, tdata, fields)
tdata[D.CSV_NODETYPE_KEYS][fields[tdata[D.DATA_ATTR_KEY]].strip()] = row
return tdata
def checkCatalog(job, tdata: dict) -> dict:
checkNodes = {}
checkNodes[MUST_NODES] = [B.DATA_NODE_HEADER, B.DATA_NODE_FIELDS, B.DATA_NODE_KEYS]
checkNodes[MUSTNT_NODES] = [B.DATA_NODE_DATA]
checkNodes[OPT_NODES] = []
return check_nodes(job, tdata, checkNodes)
class DatatypeDDL():
"""
structure:
* B.DATA_NODE_HEADER : list of field-names = column-names for table content
* B.DATA_NODE_DATA : list of rows = tabel-content
* B.DATA_NODE_DDLFIELDS : list of ddl-attributes - optional because it is B.DATA_NODE_KEYS.keys()
* B.DATA_NODE_DDLKEYS : fields with attributes (header X fields)
"""
@staticmethod
def rebuild_data(job, data: dict, tableAttr: dict) -> dict:
data = popTablesNode(job, data)
data = popNameNode(job, data)
data = buildKeys(job, data)
data["fieldnames"] = data[B.DATA_NODE_FIELDS]
data["fielddef"] = data[B.DATA_NODE_KEYS]
data[B.DATA_NODE_DDLFIELDS] = data[B.DATA_NODE_HEADER]
data[B.DATA_NODE_DDLKEYS] = data[B.DATA_NODE_KEYS]
data[B.DATA_NODE_HEADER] = data[B.DATA_NODE_FIELDS]
data.pop(B.DATA_NODE_KEYS)
data.pop(B.DATA_NODE_FIELDS)
data = addTableAttr(job, data, tableAttr)
return data
@staticmethod
def check_data(job, data: dict) -> dict:
checkNodes = {}
checkNodes[MUST_NODES] = [B.DATA_NODE_HEADER, B.DATA_NODE_DDLFIELDS, B.DATA_NODE_DDLKEYS]
checkNodes[MUSTNT_NODES] = []
checkNodes[OPT_NODES] = [B.DATA_NODE_DATA]
return check_nodes(job, data, checkNodes)
def buildKeys(job, data: dict) -> dict:
fields = []
keys = {}
if B.DATA_NODE_DATA in data and len(data[B.DATA_NODE_DATA]) > 1:
pass
else:
data.pop(B.DATA_NODE_DATA)
for k in data:
if k[:1] == "_":
continue
fields.append(k)
keys[k] = data[k]
for k in fields:
data.pop(k)
data[B.DATA_NODE_FIELDS] = fields
data[B.DATA_NODE_KEYS] = keys
return data
def insertDDL(job, tdata, fields: list) -> dict:
return tdata
def rebuildSpec(job, tdata: dict) -> dict:
return tdata
def insertSpec(job, tdata, fields: list) -> dict:
tdata[B.DATA_NODE_DATA].append(fields)
return tdata
def checkSpec(job, tdata: dict) -> dict:
checkNodes = {}
checkNodes[MUST_NODES] = [B.DATA_NODE_HEADER, B.DATA_NODE_DATA]
checkNodes[MUSTNT_NODES] = [B.DATA_NODE_FIELDS, B.DATA_NODE_KEYS]
checkNodes[OPT_NODES] = []
return check_nodes(job, tdata, checkNodes)
def check_nodes(job, config: dict, checkNodes: dict):
mustNodes = checkNodes[MUST_NODES]
mustntNodes = checkNodes[MUSTNT_NODES]
optionalNodes = checkNodes[OPT_NODES]
a = str(config["_path"]).split(os.path.sep)
b = a[-1].split(".")
path = config["_path"]
"""
if b[0] in config:
config = config[b[0]]
config["_path"] = path
if len(config) == 2:
for x in B.LIST_SUBJECTS:
if x[:-1] in config:
config = config[x[:-1]]
break
"""
for n in mustNodes:
if n not in config:
raise Exception("must-node doesnt exist "+n+" "+path)
for n in mustntNodes:
if n not in config:
continue
if len(config[n]) == 0:
job.m.logWarn("empty mustnt-node "+n+" "+path)
else:
raise Exception("must-node doesnt exist "+n+" "+path)
return config