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.

462 lines
18 KiB

import dpath.util
import json
import yaml
import re
import datetime
from tools.reader import fcts as reader
"""
Register-Meldung erstellen
"""
class register():
def __init__(self):
self.absender = Absender()
self.vrfelder = Vrfelder()
pass
def schreibeRegister(self, content, regAdr):
ausgabe = {}
content["vrfelder"] = self.vrfelder
content["absender"] = self.absender.keys[regAdr]
#
ausgabe = self.schreibeProtokoll(ausgabe, content)
print( "ausgabe vorher: " + str( ausgabe))
ausgabe = self.schreibeAbsender(ausgabe, content, regAdr)
print("ausgabe vorher: "+str(ausgabe))
ausgabe = self.schreibeEmpfaenger(ausgabe, content)
print("ausgabe vorher: "+str(ausgabe))
ausgabe = self.schreibeMetadaten(ausgabe, content)
print("ausgabe vorher: "+str(ausgabe))
ausgabe = self.schreibeSegmentkopf(ausgabe, content)
print("ausgabe vorher: "+str(ausgabe))
ausgabe["nachricht"]["datensegment"] = []
for row in content["tabelle"]["person"]["_data"]:
print ("foreach row")
print (row)
satz = {}
satz = self.schreibeSegmentsatz(satz, content, row["_lfdNR"])
# print("ausgabe vorher: "+str(ausgabe))
ausgabe["nachricht"]["datensegment"].append(satz["datensegment"])
return ausgabe
""" Fuellt und ordnet Protokollknoten ein"""
def schreibeProtokoll(self, ausgabe, content):
ausgabe["protokoll"] = self.setProtokoll(content)["protokoll"]
return ausgabe
def setProtokoll(self, content):
ausgabe = {}
d = datetime.datetime.now()
ausgabe = fcts.setMerkmal(ausgabe, '/protokoll/dokumentinstanz/datum', d.strftime("%Y%m%d"))
ausgabe = fcts.setMerkmal(ausgabe, '/protokoll/dokumentinstanz/uhrzeit', d.strftime("%H%M%S"))
ausgabe = fcts.setMerkmal(ausgabe, '/protokoll/dokumentinstanz/anwendung/anwendungsname', "Anwendungsname")
ausgabe = fcts.setMerkmal(ausgabe, '/protokoll/dokumentinstanz/anwendung/version', "Version")
ausgabe = fcts.setMerkmal(ausgabe, '/protokoll/dokumentinstanz/anwendung/hersteller', "StBA")
ausgabe = fcts.setMerkmal(ausgabe, '/protokoll/dokumentinstanz/ressourceID[@klasse="SDFMETA"]', "--RessourceID")
return ausgabe
""" Fuellt und ordnet Absenderknoten ein"""
def schreibeAbsender(self, ausgabe, content, regAdr):
ausgabe["absender"] = self.setAbsender(content, regAdr)
return ausgabe
def setAbsender(self, content, regAdr):
ausgabe = {}
for feld in Absender.FELDER:
ausgabe = fcts.setMerkmalContent(ausgabe, content, 0, feld)
return ausgabe
""" Fuellt und ordnet Empfaengerknoten ein"""
def schreibeEmpfaenger(self, ausgabe, content):
ausgabe["empfaenger"] = self.setEmpfaenger(content)["empfaenger"]
return ausgabe
def setEmpfaenger(self, content):
ausgabe = {}
ausgabe = fcts.setMerkmal(ausgabe, '/empfaenger/kennung[@klasse="STAID"]', '99')
return ausgabe
""" Fuellt und ordnet Metadatenknoten ein"""
def schreibeMetadaten(self, ausgabe, content):
nachricht = self.setMetadaten(content)
for knoten in nachricht:
if knoten != "datensegment":
ausgabe[knoten] = nachricht[knoten]
return ausgabe
def setMetadaten(self, content):
ausgabe = {}
d = datetime.datetime.now()
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/erhebung/kennung[@klasse="ERHID"]', 'ERHEBUNG_KENNUNG')
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/erhebung/ressource[@klasse="SDFMETA"]', 'ERHEBUNG_RESSOURCE_1')
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/erhebung/ressource[@klasse="SDF-ERZEUGUNG"]', 'ERHEBUNG_RESSOURCE_2')
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/berichtszeitraum/jahr', d.strftime("%Y"))
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/berichtsempfaenger/kennung[@klasse ="STAID"]', 'BERICHTSEMPFÄNGER_KENNUNG')
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/datenattribute/attribute::dezimalzeichen', 'DEZIMAL_ZEICHEN')
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/datenattribute/attribute::tausender-trennzeichen', 'TAUSENDER_TRENNZEICHEN')
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/hmm[@name="BerichtseinheitID"]/wert', 'BERICHTSEINHEIT_ID')
return ausgabe
def schreibeSegmentkopf(self, ausgabe, content, ):
ausgabe = fcts.setMerkmal(ausgabe, '/nachricht/datensegment/meldungsID', 'MELDUNGS_ID')
return ausgabe
def schreibeSegmentsatz(self, ausgabe, content, index):
for feld in Person.FELDER:
ausgabe = fcts.setMerkmalContent(ausgabe, content, index, feld)
return ausgabe
class Absender():
FELDER = ("ABSENDER_KENNUNG", "ABSENDER_IDENTITÄT", "ABSENDER_TELEFON", "ABSENDER_EMAIL", "ABSENDER_FAX", "ABSENDER_URL", "BERICHTSEINHEIT_ID", "REGNAME")
def __init__(self):
content = reader.readCsv("testdaten/allgemein/absender.csv", "keys")
self.keys = content["tabelle"]["absender"]["_keys"]
class Vrfelder():
def __init__(self):
with open(r'src/config/vrFelder.yml') as file:
content = yaml.full_load(file)
self.conf = content
def getVrfeld(self, feld):
print("gerVrfeld "+feld)
if feld in self.conf["config"]:
print("gerVrfeld treffer"+feld)
return self.conf["config"][feld]
pass
class Anschrift():
FELDER = ("GEMSCHL", "HNR", "HNR_BUCHST", "HNR_TEILNR", "HNR_UNST", "ORTSTEIL", "ORTSTEIL_POST", "PLZ", "STOCKNR_WHG", "STR", "STRSCHL", "WOHNORT", "ZUSATZ")
class Person():
FELDER = ("IDNR", "DR", "FAMNAME", "FAMNAME_AE", "FAMNAME_BEST", "FAMNAME_ZUS", "FAMNAME_VOR", "FAMNAME_UNST", "GEB_DAT", "GEBNAME", "GEBNAME_BEST1", "GEBNAME_BEST2", "GEBNAME_UNST", "GEBSTAAT", "GEB_ORT", "GESCHLECHT", "STAATSANG", "VORNAME", "VORNAME_AE")
"""
Werkzeuge, um Register-Merkmale zu erstellen, setzen, ...
"""
class fcts:
#+MULTI_NODE = ("mm", "hmm", "omm", "erhebung")
MULTI_NODE = ("erhebung")
#+ARRAY_NODE = ("satz", "ressource")
ARRAY_NODE = ("mm", "hmm", "omm", "satz", "ressource")
def get
def addMerkmal(baum, pfad, index, wert, ids=[]):
"""
Eingabe:
baum inklusive Elternknoten
pfad kompletter Pfad von Wurzel bis Blatt
index auf den Pfad
wert einzutragender Wert
Verarbeitung
basis der zu bearbeitende Knotenname extrahiert von Attributen
elem zu bearbeitender Knoten
zu betrachtende Fälle:
a: basis intendiert ein Array:
b: basis intendiert ein Dictionary:
c: basis ist ein einfacher Knoten
Weiter zu betrachten hinsichtlich Komplexität des Knotens
.a einfacher Knoten
.b Knoten mit Attribut als Blatt (anode+bnode) oder mit Unterknoten (nur anode)
Weiter zu betrachten hinsichtlich Pfadposition
..a: Blatt - Wert eintragen und
..b: im Pfad - weiterer rekursiver Aufruf
Ausgabe:
baum inklusive Elternknoten
"""
logdebug = 3 # Schalter zur Debug-Einstellung innerhalb der Funktion
if logdebug >= 1: print("--- addMerkmal "+pfad+" "+str(index)+" "+wert)
if logdebug >= 3: print("--- addMerkmal "+str(baum))
# # # # # # # # # # # # # # # # # # # #
# basis der zu bearbeitende Knotenname extrahiert von Attributen
# elem zu bearbeitender Knoten
elem = {}
a = pfad.strip('/').split('/')
anode = None # optional fuer Attribute
bnode = None # optional fuer direkte Wertzuordnung #text
basis = "" # Knotenname - ohne Attribute
if ("[@" in a[index]):
if logdebug >= 1: print("0b")
basis = a[index][0:a[index].find("[@")]
attribs = a[index][a[index].find("[@")+1:a[index].find("]")]
aname = attribs[0:attribs.find("=")]
aval = attribs[attribs.find("=")+2:-1]
if logdebug >= 3 : print("0b "+basis+" : "+aname+" = "+aval)
aarr = [aname , aval]
anode = {}
anode[aname] = aval
if (index == len(a)-1):
barr = ["#text", wert]
bnode = {}
bnode["#text"] = wert
if logdebug >= 3 : print("0b "+basis+" "+str(anode)+" "+str(bnode))
else:
basis = a[index]
if logdebug >= 3 : print("0b "+basis+" "+str(anode)+" "+str(bnode))
if (basis in baum):
elem[basis] = baum[basis]
nindex = index+1
if logdebug >= 1: print("1 "+a[index]+" "+wert+" "+str(baum))
# # # # # # # # # # # # # # # # # # # #
# zu betrachtende Fälle:
# a: basis intendiert ein Array:
# b: basis intendiert ein Dictionary:
# c: basis ist ein einfacher Knoten
# Weiter zu betrachten
# .a: Blatt - Wert eintragen und
# .b: im Pfad - weiterer rekursiver Aufruf
# # # Fall a : Array
if (basis in fcts.ARRAY_NODE):
if logdebug >= 1: print("a: ARRAY "+a[index-1]+"-"+basis+" "+str(baum))
if (basis in baum):
if (isinstance(baum[basis], dict)):
print ("200 dict vorhanden")
if (len(baum[basis].keys()) <1):
elem[basis] = []
elem[basis].append({})
if (isinstance(baum[basis], list)):
print ("201 array vorhanden")
elem[basis] = baum[basis]
else:
print ("203 neues array")
elem[basis] = []
elem[basis].append({})
xnode = {}
if (anode ):
xnode = {} # Komplexitaetsfaelle in xnode setzen
if (anode):
if logdebug >= 3: print("a-b "+basis+" "+str(anode))
##elem[basis].append(anode)
if (not aarr[0] in elem[basis]): #+
if logdebug >= 3: print("a-b "+basis+" anode ergaenzt "+str(anode))
#-elem[basis].append(anode)
elem[basis][0][aarr[0]] = aarr[1]
xnode = anode
if (bnode):
if logdebug >= 3: print("-->a-b-a> kompl. Blatt "+basis+" "+str(bnode))
#-elem[basis].append(bnode)
elem[basis][0][barr[0]] = barr[1]
baum[basis] = elem[basis]
if logdebug >= 3: print("<--a-b-a< kompl. Blatt "+basis+" "+str(elem))
return baum
if logdebug >= 3: print("-->a-b-b> kompl. Pfad "+basis+" "+str(nindex))
if ("[@" in a[nindex]):
subnode = a[nindex][0:a[nindex].find("[@")]
else:
subnode = a[nindex]
if not subnode in elem[basis][0]:
elem[basis][0][subnode] = {}
#+elem[basis].append(fcts.addMerkmal(xnode, pfad, nindex, wert))
elem[basis][0][subnode] = fcts.addMerkmal(elem[basis][0][subnode], pfad, nindex, wert)
baum[basis] = elem[basis]
if logdebug >= 3: print("<--a-b-b< kompl. Pfad "+basis+""+str(baum))
return baum ##
##### Fall 3ab ## Dict
elif (basis in fcts.MULTI_NODE):
if logdebug >= 1: print("b: DICT "+basis)
if (not basis in baum):
print ("elem angelegt")
elem[basis] = {}
if (anode ):
if logdebug >= 3: print("b-b "+basis+" "+str(anode))
elem[basis][aarr[0]] = aarr[1]
if (bnode):
elem[basis][barr[0]] = barr[1]
baum[basis] = elem[basis]
#elem[basis].append(bnode)
if logdebug >= 3: print("<<--b-b-a< kompl. Blatt "+basis+" "+str(index)+" "+str(elem))
return baum
if ("[@" in a[index+1]):
subbasis = a[index+1][0:a[index+1].find("[@")]
if logdebug >= 3: print("-->b-b-b> kompl. Pfad "+basis+" "+str(index)+" "+str(elem))
elem[basis] = fcts.addMerkmal(elem[basis], pfad, nindex, wert)
baum[basis] = elem[basis]
if logdebug >= 3: print("<--b-b-b< kompl. Pfad "+basis+" "+str(index)+" "+str(baum))
return baum
else:
subbasis = a[index+1]
if logdebug >= 3: print("-->b-a-b> einf. Pfad "+basis+" "+str(index)+" "+str(elem))
elem[basis][subbasis] = fcts.addMerkmal({}, pfad, nindex, wert)
baum[basis] = elem[basis]
if logdebug >= 3: print("<--b-a-b< einf. Pfad "+basis+" "+str(index)+" "+str(elem))
return baum
# baum[basis] = elem[basis]
return baum[basis]
###### Fall einfach + vorhanden
else:
if (basis in baum):
elem[basis] = baum[basis]
else:
elem[basis] = {}
if logdebug >= 1: print("c: EINFACH "+a[index-1]+"-"+basis+" "+str(baum))
if (1 == len(a)-index):
if logdebug >= 3 : print("3b "+basis+" "+str(anode)+" "+str(bnode))
if (anode):
elem[basis] = {}
elem[basis][aarr[0]] = aarr[1]
if logdebug >= 3 : print("3b1")
if (bnode):
elem[basis][barr[0]] = barr[1]
baum[basis] = elem[basis]
if logdebug >= 3 : print("<<-- c-b-a kompl. Blatt "+basis+" "+str(baum))
return baum
elem[basis] = {}
elem[basis] = wert
baum[basis] = elem[basis]
if logdebug >= 3 : print("<<-- c-a-a einf. Blatt "+basis+" "+str(anode)+" "+str(baum))
return baum
if (anode):
if logdebug >= 3: print("-->c-b-b> kompl. Pfad "+basis+" "+nindex)
elem[basis] = fcts.addMerkmal(elem[basis], pfad, nindex, wert)
baum[basis] = elem[basis]
if logdebug >= 3: print("<--c-b-b< kompl. Pfad "+basis+" "+nindex+" "+str(baum))
return baum[basis][basis] ##
if logdebug >= 3: print("-->c-a-b> einf. Pfad "+basis)
baum[basis] = fcts.addMerkmal(elem[basis], pfad, nindex, wert)
if logdebug >= 3: print("<--c-a-b< einf. Pfad "+basis+" "+str(index)+" "+str(baum))
return baum
# -----------------------------------------
def setMerkmal(baum, pfad, wert):
logdebug = 3 # Schalter zur Debug-Einstellung innerhalb der Funktion
if logdebug >= 1: print("----------------------------------------- setMerkmal ---------------------")
if logdebug >= 1: print("setMerkmal "+pfad+" "+wert)
#MULTI_NODE = ("mm", "hmm", "omm", "ressource")
elem = baum
if logdebug >= 2 : print ("setMerkmal "+pfad+" "+wert)
a = pfad.strip('/').split('/')
if ("[@" in a[0]):
basis = a[0][0:a[0].find("[@")]
else:
basis = a[0]
if (1 == len(a)):
baum[basis] = wert
return baum
if (basis in baum):
elem = baum[basis]
else:
elem = {}
# for i in range(0, len(a)):
if logdebug >= 1: print("setMerkmal "+basis+" "+str(elem))
baum = fcts.addMerkmal(baum, pfad, 0, wert)
if logdebug >= 1: print("setMerkmal "+basis+" "+str(baum))
if True:
return baum
if (1 == len(a)):
elem[a[0]] = wert
for i in range(0, len(a)):
if logdebug >= 3 : print("for "+str(i)+" "+a[i])
if ("[@" in a[i]):
basis = a[i][0:a[i].find("[@")]
else:
basis = a[i]
print ("basis "+basis )
#if (a[i] in elem):
if ((not basis in fcts.MULTI_NODE) and (basis in elem)):
if logdebug >= 3 : print("a "+basis)
elem = elem[basis]
else:
if ("[@" in a[i]):
if logdebug >= 3 : print("b "+a[i])
node = a[i][0:a[i].find("[@")]
attribs = a[i][a[i].find("[@")+1:a[i].find("]")]
aname = attribs[0:attribs.find("=")]
aval = attribs[attribs.find("=")+2:-1]
if logdebug >= 3 : print("b "+node+" : "+aname+" = "+aval)
if (node in elem):
if logdebug >= 3 : print("b "+node+" in elem ")
#if logdebug >= 3 : print(json.dumps(elem[node]))
if ("dict" in str(type(elem[node]))):
if logdebug >= 3 : print(node +" elem[node] ist dict")
xnode = elem[node]
elem[node] = []
elem[node].append(xnode)
else:
if logdebug >= 3 : print(node+" elem[node] ist nicht dict"+str(type(elem[node])))
pnode = {}
pnode[aname] = aval
print (str(i)+" ?= "+str(len(a)))
if (i == len(a)-1):
pnode["#text"] = wert
elem[node].append(pnode)
else:
elem[node] = {}
elem = elem[node]
elem[aname] = aval
if (i == len(a)-1):
elem["#text"] = wert
elif (i == len(a)-1):
if logdebug >= 3 : print("c "+a[i])
elem[a[i]] = wert
else:
if logdebug >= 3 : print("d "+a[i])
elem[str(a[i])] = {}
elem = elem[str(a[i])]
return baum
def setMerkmalContent(baum, content, index, feld):
logdebug = 1 # Schalter zur Debug-Einstellung innerhalb der Funktion
if logdebug >= 1 : print ("setMerkmalContent "+feld+" "+content["absender"]["REGNAME"])
vrfeld = content["vrfelder"].getVrfeld(feld)
if logdebug >= 2 : print(type(vrfeld))
if logdebug >= 2 : print(json.dumps(vrfeld))
tabelle = "#"
#print (type(vrfeld)+" "+str(vrfeld))
if (("tabelle" in vrfeld.keys()) and (vrfeld["tabelle"] == "absender")):
if logdebug >= 2 : print ("absender")
pass
elif (not "Register" in vrfeld):
if logdebug >= 2 : print ("kein Register")
return baum
elif (vrfeld["Register"].find(content["absender"]["REGNAME"])<0):
if logdebug >= 2 : print("verboten ")
return baum
if (not ("tabelle" in vrfeld.keys())):
if logdebug >= 2 : print ("keine tabelle")
return baum
elif (vrfeld["tabelle"][0:1] == "#"):
if logdebug >= 2 : print ("# tabelle")
return baum
xpath = vrfeld["Xpath"]
tabelle = vrfeld["tabelle"]
if logdebug >= 3 : print ("args "+tabelle+" "+xpath)
# pfadkennung ersetzen
if (xpath.find('${')>0):
ersatz = xpath[xpath.find('${')+2:xpath.find('}')]
npath = xpath[0:xpath.find('${')]+content["absender"][ersatz]+xpath[xpath.find('}')+1:]
else:
npath = xpath
if logdebug >= 2 : print ("npath "+npath)
npath = npath.replace('nachricht/', '')
if logdebug >= 2 : print ("npath "+npath)
if (tabelle == "absender"):
baum = fcts.setMerkmal(baum, npath, content["absender"][feld])
return baum
# Wert ermitteln aus Content
if logdebug >= 3 : print (str(index) +" tabelle "+str(content))
if (not content["tabelle"][tabelle]["_data"][int(index)-1]):
if logdebug >= 2 : print ("Zeile fehlt")
return baum
if logdebug >= 3 : print (content["tabelle"][tabelle]["_data"][int(index)-1])
if (not feld in content["tabelle"][tabelle]["_data"][int(index)-1].keys()):
if logdebug >= 2 : print ("Feld fehlt")
return baum
wert = content["tabelle"][tabelle]["_data"][int(index)-1][feld]
baum = fcts.setMerkmal(baum, npath, wert)
#print("ausgabe "+str(baum))
return baum
def test():
print("test in fcts")
class merkmal:
def __init__(self):
self.merkmale = "loadyaml"