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.
 
 
 

257 lines
8.2 KiB

# ---------------------------------------------------------------------------------------------------------
# Author : Ulrich Carmesin
# Source : gitea.ucarmesin.de
# ---------------------------------------------------------------------------------------------------------
import os
import basic.toolHandling
import basic.constants as B
import model.entity
import tools.path_const as P
import tools.data_const as D
import tools.config_tool
import tools.file_tool
import tools.git_tool
TABLE_NAME = "component"
""" system-name for this entity """
FIELD_ID = "tbid"
FIELD_NAME = D.FIELD_NAME
FIELD_DESCRIPTION = B.SUBJECT_DESCRIPTION
FIELD_REFERENCE = B.SUBJECT_REFERENCE
FIELD_ATTRIBUTES = B.NODE_ATTRIBUTES
FIELD_PROJECT = B.SUBJECT_PROJECT
FIELD_APPLICATION = B.SUBJECT_APP
LIST_FIELDS = [FIELD_ID, FIELD_NAME, FIELD_DESCRIPTION, FIELD_REFERENCE]
""" list of object-attributes """
LIST_NODES = [B.DATA_NODE_HEADER, B.DATA_NODE_DATA, B.DATA_NODE_FIELDS]
LIST_SUBTABLES = {}
DEFAULT_FIELD = ""
DEFAULT_TYPE = "string"
DEFAULT_FORMAT = "vchar(256)"
DEFAULT_INDEX = "N"
DEFAULT_CONSTRAINT = "nullable"
DEFAULT_AGGREGAT = ""
DEFAULT_GENERIC = ""
DEFAULT_KEY = ""
DEFAULT_ACCEPTANCE = ""
DEFAULT_ALIAS = ""
DEFAULT_DESCRIPTION = ""
DEFAULTS = {
D.DDL_FIELD : DEFAULT_FIELD,
D.DDL_TYPE : DEFAULT_TYPE,
D.DDL_FORMAT : DEFAULT_FORMAT,
D.DDL_INDEX : DEFAULT_INDEX,
D.DDL_CONSTRAINT : DEFAULT_CONSTRAINT,
D.DDL_AGGREGAT : DEFAULT_AGGREGAT,
D.DDL_GENERIC : DEFAULT_GENERIC,
D.DDL_KEY : DEFAULT_KEY,
D.DDL_ACCEPTANCE : DEFAULT_ACCEPTANCE,
D.DDL_ALIAS : DEFAULT_ALIAS,
D.DDL_DESCRIPTION : DEFAULT_DESCRIPTION
}
def select_tables(job, project="", application="", component=""):
outList = []
appl = tools.config_tool.getConfig(job, P.KEY_BASIC, B.SUBJECT_APPS)
path = os.path.join(job.conf[B.TOPIC_PATH][B.ATTR_PATH_COMPS], "catalog", "tables")
for p in os.listdir(path):
if p[-4:] not in [".csv", ".yml", ".xml", "json"]:
continue
table = p[:-4]
outList.append(table)
return outList
class Table(model.entity.Entity):
tbid = 0
name = ""
project = ""
application = ""
component = ""
fieldnames = []
fielddef = {}
def __init__(self, job, project="", application="", component="", name=""):
"""
to be initialized by readSpec
:param job:
"""
self.job = job
if len(project) > 1:
self.project = project
if len(application) > 1:
self.application = application
if len(component) > 1:
self.component = component
if len(name) > 1:
self.read_ddl(job, name)
def get_schema(self, tableName="", tableObject=None):
"""
gets schema/ddl-informations in order to create the database
"""
sql = ""
sqlTable = ""
sqlSub = ""
dbi = basic.toolHandling.getDbTool(self.job, None, self.job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE])
sqlTable += dbi.getCreateTable(tableName)
tableId = ""
for f in self.fieldnames:
if f[0:1] == "_":
continue
fo = self.fielddef[f]
if D.DDL_INDEX in fo and len(fo[D.DDL_INDEX]) > 0:
a = fo[D.DDL_INDEX].split(":")
if a[0] == "I":
sqlSub += dbi.getSchemaIndex(tableName, fo[D.DDL_FIELD]) + "\n"
elif a[0] == "S":
attrList = []
attr = {"attr":fo[D.DDL_FIELD], "atype": fo[D.DDL_TYPE]}
attrList.append(attr)
for i in range(2, len(a)):
if i % 2 == 1:
continue
if a[i] == "attr":
attr = {"attr":B.NODE_ATTRIBUTES, "atype": D.TYPE_TEXT}
elif i+1 < len(a):
attr = {"attr": a[i], "atype": a[i+1]}
attrList.append(attr)
sqlSub += dbi.getSchemaSubtable(a[1], attrList) + "\n"
sqlSub += dbi.getSchemaIndex(dbi.getSubtableName(a[1], fo[D.DDL_FIELD]), tableId) + "\n"
continue
sqlTable += dbi.getSchemaAttribut(fo[D.DDL_FIELD], fo[D.DDL_TYPE]) + ","
if fo[D.DDL_TYPE] == D.TYPE_PK:
tableId = fo[D.DDL_FIELD]
sql = sqlTable[0:-1]+");\n"+sqlSub
"""
# print(sql)
"""
return sql
def read_unique_names(self, job, project, application, gran, args):
"""
reads the entity-names from file-storage
:param job:
:param opt. project: select-criteria if used and defined
:param opt. application: select-criteria if used and defined
:param opt. gran: granularity values testcase / testsuite / testplan
:param opt. args additional args
:return: list of entity-names
"""
path = os.path.join(job.conf[B.TOPIC_PATH][B.ATTR_PATH_COMPS], P.KEY_CATALOG, P.VAL_TABLES)
outList = self.getDirlist(job, path, "csv")
return outList
def read_entity(self, job, name):
config = self.getConfig(job, P.KEY_CATALOG, name, tools.config_tool.get_plain_filename(job, name))
return self.setAttributes(job, config, name, LIST_FIELDS, LIST_NODES, LIST_SUBTABLES)
def getFieldList(self):
"""
returns a list of scalar attributes
:return: LIST_FIELDS
"""
return LIST_FIELDS
def getNodeList(self):
"""
returns a list of sub-nodes - which can be persisted in a clob-field
:return: LIST_NODES
"""
return LIST_NODES
def getSubtableList(self):
"""
returns a list of sub-tables
:return: LIST_SUBTABLES
"""
return LIST_SUBTABLES
def getName(self):
"""
returns the name - maybe build from other attributes
:return:
"""
return self.name
def getIDName(self):
"""
it returns the name as unique-id - maybe build from few attributes
:return:
"""
return self.name
def read_ddl(self, job, name):
ddl = tools.config_tool.getConfig(job, D.DDL_FILENAME, self.component, name)
self.fieldnames = []
fielddef = {}
if B.DATA_NODE_TABLES in ddl and name in ddl[B.DATA_NODE_TABLES]:
ddl = ddl[B.DATA_NODE_TABLES][name]
for k in ddl[B.DATA_NODE_DATA]:
if not isinstance(k, dict) \
or D.DDL_FIELD not in k:
continue
ddl[k[D.DDL_FIELD]] = k
for f in ddl:
if f in [B.DATA_NODE_HEADER, B.DATA_NODE_FIELDS, B.DATA_NODE_DATA]:
continue
field = {}
self.fieldnames.append(f)
for k in D.LIST_DDL_ATTR:
if k in ddl[f]:
field[k] = ddl[f][k]
else:
field[k] = DEFAULTS[k]
fielddef[f] = field
self.fielddef = fielddef
def select_entity(self, job, name):
"""
reads the entity from the database
it should get the same result like read_entity
:param job:
:param name:
:return:
"""
self.read_entity(job, name)
# raise Exception(B.EXCEPT_NOT_IMPLEMENT)
def write_entity(self, job, name):
"""
writes the entity into the database
it similar to update_entity
:param job:
:param name:
:return:
"""
raise Exception(B.EXCEPT_NOT_IMPLEMENT)
def update_entity(self, job, name):
"""
writes the entity into the database
it similar to update_entity
:param job:
:param name:
:return:
"""
raise Exception(B.EXCEPT_NOT_IMPLEMENT)
def remove_entity(self, job, name):
"""
removes the entity from the file-system
it similar to delete_entity
:param job:
:param name:
:return:
"""
raise Exception(B.EXCEPT_NOT_IMPLEMENT)
def delete_entity(self, job, name):
"""
deletes the entity into the database
it similar to update_entity
:param job:
:param name:
:return:
"""
raise Exception(B.EXCEPT_NOT_IMPLEMENT)