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.

239 lines
8.3 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.db_abstract
import tools.git_tool
TABLE_NAME = "project"
""" system-name for this entity """
FIELD_ID = "prid"
FIELD_NAME = "name"
FIELD_DESCRIPTION = B.SUBJECT_DESCRIPTION
FIELD_REFERENCE = B.SUBJECT_REFERENCE
LIST_FIELDS = [FIELD_ID, FIELD_NAME, FIELD_DESCRIPTION, FIELD_REFERENCE]
""" list of object-attributes """
FILE_EXTENSION = D.DFILE_TYPE_YML
UNIQUE_FIELDS = [FIELD_NAME]
""" unique business field as human identifer """
IDENTIFYER_FIELDS = [FIELD_ID]
""" unique technical field as technical identifer """
class Project(model.entity.Entity):
prid = 0
name = ""
description = ""
reference = ""
def __int__(self, job, name=""):
self.job = job
if len(name) > 1:
self.getEntity(job, name)
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
"""
config = self.getConfig(job, B.SUBJECT_PROJECTS, "")
outList = list(config[B.SUBJECT_PROJECTS].keys())
return outList
def select_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
"""
outList = []
self.setDbAttributes(job, [TABLE_NAME])
dbi = basic.toolHandling.getDbTool(job, self, job.conf[B.TOPIC_NODE_DB]["type"])
data = dbi.selectRows(TABLE_NAME, job)
checkList = {}
for row in data[B.DATA_NODE_DATA]:
key = ""
for f in UNIQUE_FIELDS:
key += "_" + row[f]
if key in checkList:
continue
else:
checkList[key] = key
fields = []
for f in UNIQUE_FIELDS:
fields.append(row[f])
outList.append(fields)
return outList
def read_entity(self, job, name):
"""
reads the entity from the file-system
:param job:
:param name:
:return:
"""
print("name "+name)
config = self.getConfig(job, B.SUBJECT_PROJECTS, tools.config_tool.get_plain_filename(job, name))
for k in LIST_FIELDS:
if k not in config:
continue
setattr(self, k, config[k])
return self
def select_entity(self, job, name, row={}):
"""
reads the entity from the database
it should get the same result like read_entity
:param job:
:param name: unique field as string, unique fields as list
the unique-fields are defined in the class
:return: itself with filled object-attributes
"""
if row is None or len(row) == 0:
self.setDbAttributes(job, [TABLE_NAME])
dbi = basic.toolHandling.getDbTool(job, self, job.conf[B.TOPIC_NODE_DB]["type"])
if type(name) is list:
names = name
elif type(name) is str:
names = [name]
condition = "where "
for v in names:
condition += " and " + ""
data = dbi.selectRows(TABLE_NAME, job, "where username = \'" + names[0] + "\'")
if len(data[B.DATA_NODE_DATA]) > 1:
raise Exception("single selection with more than one result: "+names[0])
elif len(data[B.DATA_NODE_DATA]) == 1:
row = data[B.DATA_NODE_DATA][0]
else:
raise Exception("no result for: "+names[0])
for k in LIST_FIELDS:
if k not in row:
continue
setattr(self, k, row[k])
return self
def write_entity(self, job, name):
"""
writes the entity into the file-system
it similar to update_entity
:param job:
:param name:
:return:
"""
config = {}
config[model.user.TABLE_NAME] = {}
pathname = os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME], P.VAL_CONFIG,
P.VAL_USER, name + ".yml")
for k in LIST_FIELDS:
if getattr(self, k, "") == "" \
or k == FIELD_ID:
continue
config[model.user.TABLE_NAME][k] = getattr(self, k, "")
tools.file_tool.write_file_dict(job.m, job, pathname, config)
return self
def insert_entity(self, job, name, table="", rows={}):
"""
inserts the entity into the database
it similar to update_entity
:param job:
:param name:
:return:
"""
self.setDbAttributes(job, [TABLE_NAME])
dbi = basic.toolHandling.getDbTool(job, self, job.conf[B.TOPIC_NODE_DB]["type"])
condition = "where"
for f in UNIQUE_FIELDS:
# TODO other db-formats than string has to be implemented
condition += " and " + f + " = \'" + getattr(self, f, "") + "\'"
condition = condition.replace("where and", "where ")
data = dbi.selectRows(TABLE_NAME, job, condition)
if len(data[B.DATA_NODE_DATA]) > 0:
print("update statt insert")
return
if rows is None or len(rows) == 0:
insheader = dbi.getInsertFields(self.conf[B.DATA_NODE_DDL][table])
rows = []
row = {}
for f in insheader:
row[f] = getattr(self, f)
rows.append(row)
dbi.insertRows(job, table, rows)
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: single substring or list of name or dict of names with the keys as
:return:
"""
self.removeEntity(job, name, os.path.join(job.conf[B.TOPIC_PATH][P.ATTR_PATH_HOME], P.VAL_CONFIG, P.VAL_USER), "yml")
def delete_entity(self, job, name, table):
"""
deletes the entity into the database
it similar to update_entity
:param job:
:param name:
:return:
"""
self.setDbAttributes(job, [TABLE_NAME])
dbi = basic.toolHandling.getDbTool(job, self, job.conf[B.TOPIC_NODE_DB]["type"])
condition = "where"
for f in IDENTIFYER_FIELDS:
# TODO other db-formats than string has to be implemented
val = dbi.getDbValue(self.conf[B.DATA_NODE_DDL][table][f], getattr(self, f, ""))
condition += " and " + f + " = " + val + ""
condition = condition.replace("where and", "where ")
dbi.deleteRows(job, table, condition)
@staticmethod
def getConfig(job, subject, name):
"""
reads the entity from the database
it should get the same result like read_entity
:param job:
:param name:
:return:
"""
config = tools.config_tool.getConfig(job, P.KEY_BASIC, subject)
if config is not None:
if len(name) == 0:
return config
elif name in config[subject]:
return config[subject][name]
raise Exception("keine Config zu "+name)
@staticmethod
def getCurrentUser(job):
return os.environ.get("USERNAME")