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.

293 lines
9.7 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"
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 = {}
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):
id = 0
username = ""
name = ""
famname = ""
email = ""
password = ""
project = ""
role = ""
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
"""
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 LIST_FIELDS:
if k not in config:
continue
setattr(self, k, config[k])
return self
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 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 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)
if config is not None:
return config
if name == model.user.User.getCurrentUser(job):
config = tools.config_tool.getConfig(job, P.KEY_USER, "default")
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")