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.

298 lines
10 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 = "user"
""" system-name for this entity """
FIELD_ID = "id"
FIELD_USERNAME = "username"
FILE_EXTENSION = D.DFILE_TYPE_YML
UNIQUE_FIELDS = [FIELD_USERNAME]
""" unique business field as human identifer """
IDENTIFYER_FIELDS = [FIELD_ID]
""" unique technical field as technical identifer """
class User(model.entity.Entity):
FIELD_ID = "id"
FIELD_USERNAME = "username"
FIELD_NAME = "name"
FIELD_FAMNAME = "famname"
FIELD_EMAIL = "email"
FIELD_PASSWORD = "password"
FIELD_PROJECT = B.SUBJECT_PROJECT
FIELD_ROLE = "role"
LIST_FIELDS = [FIELD_ID, FIELD_ROLE, FIELD_PROJECT, FIELD_PASSWORD, FIELD_EMAIL, FIELD_FAMNAME, FIELD_NAME,
FIELD_USERNAME]
""" list of object-attributes """
LIST_NODES = [B.NODE_ATTRIBUTES]
LIST_SUBTABLES = {}
UNIQUE_FIELDS = [FIELD_USERNAME]
id = 0
username = ""
name = ""
famname = ""
email = ""
password = ""
project = ""
role = ""
attributes = ""
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
"""
outList = []
path = os.path.join(job.conf[B.TOPIC_PATH][B.ATTR_PATH_HOME], P.VAL_CONFIG, "user")
for k in os.listdir(path):
filename = tools.config_tool.get_plain_filename(job, k)
if "default" == filename:
continue
outList.append(filename)
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 = model.user.User.getUserConfig(job, tools.config_tool.get_plain_filename(job, name))
for k in self.LIST_FIELDS:
if k not in config:
continue
setattr(self, k, config[k])
return self
@staticmethod
def rebuild_data(job, data: dict) -> dict:
"""
gets the subtable-tag from filecsv and sets the subtables in order to workable entity-elements
:param job:
:param data:
:return:
"""
data = tools.file_type.popSubjectsNode(job, data)
data = tools.file_type.popNameNode(job, data)
outdata = {}
for k in data:
if k == "sysuser":
sysuser = User.getCurrentUser(job)
outdata[sysuser] = data[k]
outdata[sysuser][FIELD_USERNAME] = sysuser
else:
outdata[k] = data[k]
return outdata
def check_data(self, job, data: dict) -> dict:
"""
it checks the data for the specific form
:param job:
:param tdata:
:param ttype:
:return:
"""
import tools.file_type
checkNodes = {}
checkNodes[tools.file_type.MUST_NODES] = []
checkNodes[tools.file_type.MUSTNT_NODES] = [B.DATA_NODE_DATA, B.DATA_NODE_HEADER, B.DATA_NODE_FIELDS, B.DATA_NODE_KEYS] + B.LIST_SUBJECTS
checkNodes[tools.file_type.OPT_NODES] = []
return tools.file_type.check_nodes(job, data, checkNodes)
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 self.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 self.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"
if B.DATA_NODE_DDLKEYS in self.conf[B.DATA_NODE_DDL][table]:
keys = self.conf[B.DATA_NODE_DDL][table][B.DATA_NODE_DDLKEYS]
else:
keys = self.conf[B.DATA_NODE_DDL][table]
for f in IDENTIFYER_FIELDS:
# TODO other db-formats than string has to be implemented
val = dbi.getDbValue(keys[f], getattr(self, f, ""))
condition += " and " + f + " = " + val + ""
condition = condition.replace("where and", "where ")
dbi.deleteRows(job, table, condition)
@staticmethod
def getUserConfig(job, 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_USER, name, ttype=B.SUBJECT_USER)
if config is not None:
return config
if name == model.user.User.getCurrentUser(job):
config = tools.config_tool.getConfig(job, P.KEY_USER, "default", ttype=B.SUBJECT_USER)
if "user" in config:
config = config["user"]
if config is not None:
config["username"] = name
return config
raise Exception("keine Config zu "+name)
@staticmethod
def getCurrentUser(job):
return os.environ.get("USERNAME")