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.
 
 
 

133 lines
4.8 KiB

import os.path
import re
import basic.component
import basic.constants as B
import basic.toolHandling
import tools.config_tool
import tools.data_const as D
import tools.file_tool
import tools.filecsv_fcts
import model.table
import model.factory
import tools.value_tool
import tools.data_tool
COMP_NAME = B.ATTR_INST_TESTSERVER
# class Testserver(basic.component.Component):
class Testserver:
"""
the Testserver represents the workspace with all resources for the automation
"""
tables = {}
__instance = None
__writeDB = True
def __init__(self, job):
"""
collect all resources into this object
:param job:
"""
print('init '+COMP_NAME)
self.m = job.m
self.conf = {}
if B.TOPIC_NODE_DB in job.conf:
self.conf[B.TOPIC_CONN] = {}
self.conf[B.TOPIC_CONN][B.TOPIC_NODE_DB] = {}
for attr in B.LIST_DB_ATTR:
if attr in job.conf[B.TOPIC_NODE_DB]:
self.conf[B.TOPIC_CONN][B.TOPIC_NODE_DB][attr] = job.conf[B.TOPIC_NODE_DB][attr]
# TODO was muss auf dem Testserver initial geladen werden?
self.model = {}
Testserver.__instance = self
for s in B.LIST_SUBJECTS:
self.model[tools.data_tool.getSingularKeyword(s)] = model.factory.get_entity_object(job, s, {})
pass
@staticmethod
def getInstance(job):
if Testserver.__instance == None:
return Testserver(job)
def createAdminDBTables(self, job):
"""
creates the complete data-model in the database. it contains:
* the model for administration
* the model of each project:
* * root-tables - defined in testcases TODO wie allgemein deklariert, special components/config
* * comp-artifacts - it could contain build-rules for building from testcase-spec
:param job:
:return:
"""
tables = {}
if B.TOPIC_NODE_DB in job.conf:
self.dbi = basic.toolHandling.getDbTool(job, self, job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE])
else:
return "No DB in job-config"
# the model for administration
for m in self.model.keys():
print("\n==== model " + m)
self.createDBTable(job, B.ATTR_INST_TESTSERVER, B.ATTR_INST_TESTSERVER, m)
enty = self.model[m]
for t in enty.getSubtableNames():
print("subtable "+t)
self.createDBTable(job, B.ATTR_INST_TESTSERVER, B.ATTR_INST_TESTSERVER, t)
def createProjectDBTables(self, job):
"""
creates the complete data-model in the database. it contains:
* the model for administration
* the model of each project:
* * root-tables - defined in testcases TODO wie allgemein deklariert, special components/config
* * comp-artifacts - it could contain build-rules for building from testcase-spec
:param job:
:return:
"""
tables = {}
path = os.path.join(job.conf[B.TOPIC_PATH][B.ATTR_PATH_PROGRAM], "model")
fct = basic.toolHandling.getFileTool(job, None, "csv")
for m in sorted(os.listdir(path)):
if not re.match(r".*?\.csv", m):
print("sonstig "+m)
continue
print("model "+m)
modelPath = os.path.join(path, m)
modelDoc = fct.load_file(modelPath, D.CSV_SPECTYPE_DATA)
table = model.table.Table(job, project="", application="", component=COMP_NAME, name=m[:-4])
sql = table.get_schema(tableName=m[:-4], tableObject=table) # [B.DATA_NODE_TABLES][m[:-4]]
job.m.logInfo(sql)
tables[m[:-4]] = modelDoc
for s in sql.split(";\n"):
if len(s) < 3:
continue
try:
self.dbi.execStatement(s+";", job.conf[B.TOPIC_NODE_DB])
print("SQL executed: "+s)
except Exception as e:
raise Exception("Fehler bei createSchema "+s)
def createDBTable(self, job, project, context, tablename):
"""
creates a table in the database
:param job:
:return:
"""
args = {}
args["context"] = context
table = model.table.Table(job)
table = table.read_entity(job, tablename, args=args)
sql = table.get_schema(tablename, model.table.TYPE_ADMIN) # [B.DATA_NODE_TABLES][m[:-4]]
job.m.logInfo(sql)
for s in sql.split(";\n"):
if len(s) < 3:
continue
try:
if self.__writeDB:
self.dbi.execStatement(s + ";", job.conf[B.TOPIC_NODE_DB])
print("SQL executed: " + s)
except Exception as e:
raise Exception("Fehler bei createSchema " + s)