Browse Source

one further classes, read-methods environment

refactor
Ulrich 1 year ago
parent
commit
f825776264
  1. 2
      basic/constants.py
  2. 51
      model/entity.py
  3. 65
      model/environment.py
  4. 52
      test/test_21environment.py
  5. 22
      test/test_31tdata.py
  6. 9
      tools/config_tool.py
  7. 2
      tools/path_const.py
  8. 113
      utils/data_const.py

2
basic/constants.py

@ -164,6 +164,8 @@ NODE_ATTRIBUTES = "attributes"
LIST_FILE_ATTR = [ATTR_FILE_OLD, ATTR_FILE_ROTATE] + LIST_ARTS_ATTR LIST_FILE_ATTR = [ATTR_FILE_OLD, ATTR_FILE_ROTATE] + LIST_ARTS_ATTR
LIST_TOPIC_NODES = [TOPIC_NODE_FILE, TOPIC_NODE_DB, TOPIC_NODE_CLI, TOPIC_NODE_API]
LIST_ATTR = { LIST_ATTR = {
TOPIC_NODE_DB: LIST_DB_ATTR, TOPIC_NODE_DB: LIST_DB_ATTR,
TOPIC_NODE_API: LIST_API_ATTR, TOPIC_NODE_API: LIST_API_ATTR,

51
model/entity.py

@ -4,6 +4,7 @@ import re
import basic.toolHandling import basic.toolHandling
import model.entity import model.entity
import tools.data_const as D import tools.data_const as D
import tools.path_const as P
import basic.constants as B import basic.constants as B
import tools.config_tool import tools.config_tool
import tools.date_tool import tools.date_tool
@ -247,7 +248,15 @@ class Entity:
:return: :return:
""" """
config = tools.config_tool.getConfig(job, module, subject) config = tools.config_tool.getConfig(job, module, subject)
oldConfig = config
if config is not None: if config is not None:
if subject not in config:
newConfig = {}
newConfig[subject] = {}
for k in config:
newConfig[subject][k] = config[k]
config = newConfig
pass
if len(name) == 0: if len(name) == 0:
return config return config
elif name in config[subject]: elif name in config[subject]:
@ -260,18 +269,58 @@ class Entity:
outConfig = {} outConfig = {}
outConfig[name] = config[subject][B.DATA_NODE_KEYS][name] outConfig[name] = config[subject][B.DATA_NODE_KEYS][name]
return outConfig return outConfig
raise Exception("keine Config zu "+name) elif name == subject:
return config
raise Exception("keine Config zu "+name)
@staticmethod
def getDirlist(job, path, ext):
outList = []
for k in os.listdir(path):
if k[:1] in [".", "_"]:
continue
if k in [P.KEY_CATALOG, P.KEY_TOOL, P.VAL_CONFIG, P.VAL_TEST, P.VAL_TOOLS]:
continue
if ext == "":
if not os.path.isdir(os.path.join(path, k)):
continue
outList.append(k)
continue
else:
if not os.path.isfile(os.path.join(path, k)):
continue
if len(k) < len(ext):
continue
xx = k[-len(ext):]
if ext != k[-len(ext):]:
continue
outList.append(k[:-len(ext)-1])
return outList
def setAttributes(self, config, rootname, fields, subjects): def setAttributes(self, config, rootname, fields, subjects):
""" 2023-05 """ """ 2023-05 """
for k in fields: for k in fields:
if k not in config[rootname]: if k not in config[rootname]:
for l in config[rootname]:
if l[1:] == k:
setattr(self, k, config[rootname][l])
break
continue continue
setattr(self, k, config[rootname][k]) setattr(self, k, config[rootname][k])
for k in subjects: for k in subjects:
if k not in config[rootname]: if k not in config[rootname]:
for l in config[rootname]:
if l[1:] == k:
setattr(self, k, config[rootname][l])
break
continue continue
setattr(self, k, config[rootname][k]) setattr(self, k, config[rootname][k])
topics = {}
for k in B.LIST_TOPIC_NODES:
if k in config[rootname]:
topics[k] = config[rootname][k]
if len(topics) > 0:
setattr(self, "topics", topics)
return self return self

65
model/environment.py

@ -9,6 +9,25 @@ import tools.path_const as P
import tools.config_tool import tools.config_tool
import tools.file_tool import tools.file_tool
import tools.git_tool import tools.git_tool
import tools.data_const as D
TABLE_NAME = "environment"
""" system-name for this entity """
FIELD_ID = "enid"
FIELD_NAME = "name"
FIELD_DESCRIPTION = B.SUBJECT_DESCRIPTION
FIELD_REFERENCE = B.SUBJECT_REFERENCE
FIELD_PROJECT = B.SUBJECT_PROJECT
FIELD_ATTRIBUTES = B.NODE_ATTRIBUTES
LIST_FIELDS = [FIELD_ID, FIELD_NAME, FIELD_DESCRIPTION, FIELD_REFERENCE, FIELD_PROJECT, FIELD_ATTRIBUTES]
""" list of object-attributes """
LIST_SUBTABLES = [B.SUBJECT_COMPS, B.SUBJECT_PROJECTS]
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 """
TABLE_NAMES = ["environment", "en_project", "en_component"] TABLE_NAMES = ["environment", "en_project", "en_component"]
DEFAULT_SYNC = model.entity.SYNC_FULL_GIT2DB DEFAULT_SYNC = model.entity.SYNC_FULL_GIT2DB
@ -41,3 +60,49 @@ def select_environments(job, projectList):
continue continue
return environments return environments
class Environment(model.entity.Entity):
name = ""
description = ""
reference = ""
attributes = ""
project = ""
component = ""
def __init__(self, job, project="", name=""):
"""
to be initialized by readSpec
project : optional
alternative parameter
name : name of variant or default - only from testspec
obj : object with main attributes
:param job:
"""
self.job = job
if project != "":
self.project = project
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
"""
path = os.path.join(job.conf[B.SUBJECT_PATH][B.ATTR_PATH_ENV])
outList = self.getDirlist(job, path, "")
return outList
def read_entity(self, job, name):
"""
reads the entity from the file-system
:param job:
:param name:
:return:
"""
config = self.getConfig(job, P.KEY_ENV, name, tools.config_tool.get_plain_filename(job, name))
return self.setAttributes(config, name, LIST_FIELDS, LIST_SUBTABLES)

52
test/test_21environment.py

@ -9,17 +9,67 @@ import test.testtools
import basic.constants as B import basic.constants as B
import test.constants as T import test.constants as T
import tools.file_tool import tools.file_tool
import model.entity
import model.environment import model.environment
HOME_PATH = test.constants.HOME_PATH HOME_PATH = test.constants.HOME_PATH
PYTHON_CMD = "python" PYTHON_CMD = "python"
TEST_FUNCTIONS = ["test_10getEnvironments"]
TEST_FUNCTIONS = ["test_10getEntityNames", "test_11getEntities", "test_12getEntity",
"test_13writeEntity", "test_14insertEntity"]
TEST_FUNCTIONS = ["test_10getEntityNames", "test_12getEntity"]
PROGRAM_NAME = "clean_workspace" PROGRAM_NAME = "clean_workspace"
class MyTestCase(unittest.TestCase): class MyTestCase(unittest.TestCase):
mymsg = "--------------------------------------------------------------" mymsg = "--------------------------------------------------------------"
def test_10getEntityNames(self):
global mymsg
global jobObject
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
environment = model.environment.Environment(job, "TESTPROJ")
entityNames = environment.read_unique_names(job, "", "", "", {})
self.assertEqual(type(entityNames), list)
#entityNames = environment.select_unique_names(job, "", "", "", {})
#self.assertEquals(type(entityNames), list)
def test_11getEntities(self):
global mymsg
global jobObject
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
environment = model.environment.Environment(job, "TESTPROJ")
entityNames = environment.get_entities(job, storage=model.entity.STORAGE_FILE)
self.assertEqual(type(entityNames), list)
#entityNames = environment.get_entities(job, storage=model.entity.STORAGE_DB)
#self.assertEqual(type(entityNames), list)
def test_12getEntity(self):
global mymsg
global jobObject
actfunction = str(inspect.currentframe().f_code.co_name)
cnttest = 0
if actfunction not in TEST_FUNCTIONS:
return
job = test.testtools.getJob()
environment = model.environment.Environment(job, "TESTPROJ")
name = "ENV01"
actrelease = environment.read_entity(job, name)
self.assertEqual(getattr(actrelease, model.environment.FIELD_NAME), name)
self.assertRaises(Exception, environment.read_entity, job, "xyzxyz")
#
#actenvironment = environment.select_entity(job, name)
#self.assertEqual(getattr(actenvironment, model.environment.FIELD_USERNAME), name)
#self.assertRaises(Exception, environment.select_entity, job, ["xyzxyz"])
def test_10getEnvironments(self): def test_10getEnvironments(self):
global mymsg global mymsg
global jobObject global jobObject

22
test/test_31tdata.py

@ -1,15 +1,16 @@
import unittest import unittest
import inspect import inspect
import utils.tdata_tool as t import model.testcase
# import utils.tdata_tool as t
import basic.constants as B import basic.constants as B
import utils.data_const as D import tools.data_const as D
import utils.path_const as P import tools.path_const as P
import utils.config_tool import tools.config_tool
import test.testtools import test.testtools
import test.constants import test.constants
import basic.program import basic.program
import utils.path_tool import tools.path_tool
import utils.file_tool import tools.file_tool
import os import os
HOME_PATH = test.constants.HOME_PATH HOME_PATH = test.constants.HOME_PATH
@ -20,7 +21,7 @@ OS_SYSTEM = test.constants.OS_SYSTEM
# if you minimize the list you can check the specific test-function # if you minimize the list you can check the specific test-function
TEST_FUNCTIONS = ["test_01tdata", "test_02getCsvSpec_data", "test_03getCsvSpec_tree", "test_04getCsvSpec_key", TEST_FUNCTIONS = ["test_01tdata", "test_02getCsvSpec_data", "test_03getCsvSpec_tree", "test_04getCsvSpec_key",
"test_05getCsvSpec_conf", "test_06parseCsv"] "test_05getCsvSpec_conf", "test_06parseCsv"]
TEST_FUNCTIONS = ["test_05getCsvSpec_conf"] TEST_FUNCTIONS = ["test_01tdata"]
# with this variable you can switch prints on and off # with this variable you can switch prints on and off
verbose = False verbose = False
@ -34,7 +35,11 @@ class MyTestCase(unittest.TestCase):
cnttest = 0 cnttest = 0
if actfunction not in TEST_FUNCTIONS: if actfunction not in TEST_FUNCTIONS:
return return
job = test.testtools.getJob() job = test.testtools.getTestJob()
print("project : "+job.par.project+", ")
T = model.testcase.Testcase(job, project=job.par.project, name="TC0002")
print("T: "+str(T))
"""
setattr(job.par, "tdtyp", "dir") setattr(job.par, "tdtyp", "dir")
setattr(job.par, "tdsrc", "TC0001") setattr(job.par, "tdsrc", "TC0001")
setattr(job.par, "tdname", "testspec") setattr(job.par, "tdname", "testspec")
@ -45,6 +50,7 @@ class MyTestCase(unittest.TestCase):
setattr(job.par, "tdsrc", "TST001") setattr(job.par, "tdsrc", "TST001")
#tdata = t.getTestdata() #tdata = t.getTestdata()
#self.assertEqual(("steps" in tdata), True) #self.assertEqual(("steps" in tdata), True)
"""
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest)
def test_02getCsvSpec_data(self): def test_02getCsvSpec_data(self):

9
tools/config_tool.py

@ -84,6 +84,8 @@ def select_config_path(job, modul, name, subname=""):
return getTestPath(job, name, B.SUBJECT_TESTSUITES, D.DFILE_TESTSUITE_NAME) return getTestPath(job, name, B.SUBJECT_TESTSUITES, D.DFILE_TESTSUITE_NAME)
elif modul == P.KEY_CATALOG: elif modul == P.KEY_CATALOG:
return getCatalogPath(job, name) return getCatalogPath(job, name)
elif modul == P.KEY_ENV:
return getEnvironmentPath(job, name)
elif modul == P.KEY_USER: elif modul == P.KEY_USER:
return getUserPath(job, name) return getUserPath(job, name)
else: else:
@ -178,7 +180,6 @@ def getCompPath(job, name, subname, filename):
return pathname return pathname
raise Exception(P.EXP_CONFIG_MISSING, filename + ", " + name) raise Exception(P.EXP_CONFIG_MISSING, filename + ", " + name)
def getBasicPath(job, name): def getBasicPath(job, name):
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS], path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_COMPONENTS],
P.VAL_CONFIG, name)) P.VAL_CONFIG, name))
@ -189,6 +190,12 @@ def getBasicPath(job, name):
if path is not None: if path is not None:
return path return path
def getEnvironmentPath(job, name):
path = getExistingPath(job, os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_ENV], name,
P.VAL_CONFIG, B.SUBJECT_ENV))
if path is not None:
return path
def getTestPath(job, name, subdir, filename): def getTestPath(job, name, subdir, filename):
for format in CONFIG_FORMAT: for format in CONFIG_FORMAT:
pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_TDATA], job.par.project, pathname = os.path.join(job.conf[B.SUBJECT_PATH][P.ATTR_PATH_TDATA], job.par.project,

2
tools/path_const.py

@ -23,6 +23,8 @@ KEY_COMP = "comp"
""" keyword for individual component """ """ keyword for individual component """
KEY_TOOL = "tool" KEY_TOOL = "tool"
""" keyword for technical tools """ """ keyword for technical tools """
KEY_ENV = B.SUBJECT_ENV
""" keyword for environment """
VAL_TOOLS = "tools" VAL_TOOLS = "tools"
""" subdir for any technical tools """ """ subdir for any technical tools """
VAL_TABLES = "tables" VAL_TABLES = "tables"

113
utils/data_const.py

@ -1,113 +0,0 @@
#!/usr/bin/python
"""
constants for used for api-functions
"""
import basic.constants as B
DDL_FILENAME = "DATASTRUCTURE"
DATA_NODE_TYPE = "type"
TYPE_STRING = "string"
TYPE_STR = "str"
TYPE_TEXT = "text"
TYPE_INT = "int"
TYPE_FLOAT = "float"
TYPE_DOUBLE = "double"
TYPE_DATE = "date"
TYPE_TIME = "time"
TYPE_PK = "pk"
# fields in DDL
DDL_FNULLABLE = "nullable"
DDL_FNAME = "field"
DDL_ACCEPTANCE = "acceptance"
DDL_KEY = "key"
DDL_TYPE = "type"
DDL_INDEX = "index"
DFILE_TYPE_YML = "yml"
DFILE_TYPE_JSON = "json"
DFILE_TYPE_CSV = "csv"
DFILE_TYPE_XML = "xml"
DFILE_TESTCASE_NAME = "testspec"
DFILE_TESTSUITE_NAME = "testsuite"
DFILE_TABLE_PREFIX = "table_"
LIST_DFNAME_ATTR = [DFILE_TESTCASE_NAME, DFILE_TESTSUITE_NAME, DFILE_TABLE_PREFIX]
LIST_DFNAME_CONST = ["DFILE_TESTCASE_NAME", "DFILE_TESTSUITE_NAME", "DFILE_TABLE_PREFIX"]
DATA_SRC_DIR = "dir"
DATA_SRC_CSV = "csv"
DATA_ATTR_COUNT = "_count"
""" statistical information of data-count """
DATA_ATTR_DATE = "_date"
""" reference-date for computing the actual date in relation to specification or expectation """
DATA_ATTR_COMP = "_comp"
""" reference to using componente with their object """
DATA_ATTR_CHAR = "_char"
""" character of the data in order to delete it ión initialization """
DATA_ATTR_KEY = "_key"
""" key for a data-specification of a catalog-list - default: the first field is the key """
DATA_ATTR_ALIAS = "_alias"
LIST_DATA_ATTR = [DATA_ATTR_COUNT, DATA_ATTR_DATE, DATA_ATTR_CHAR, DATA_ATTR_COMP, DATA_ATTR_ALIAS,DATA_ATTR_KEY]
LIST_ATTR_CONST = ["DATA_ATTR_COUNT", "DATA_ATTR_DATE", "DATA_ATTR_CHAR", "DATA_ATTR_COMP", "DATA_ATTR_ALIAS", "DATA_ATTR_KEY"]
HEAD_ATTR_DESCR = "decription"
HEAD_ATTR_TARGET = "target"
HEAD_ATTR_USECASE = "usecase"
HEAD_ATTR_UCID = "usecase-id"
HEAD_ATTR_STORY = "story"
HEAD_ATTR_STORYID = "storyid-id"
HEAD_ATTR_APPS = B.SUBJECT_APPS
HEAD_ATTR_DEPR = "deprecated"
LIST_HEAD_ATTR = [HEAD_ATTR_DESCR, HEAD_ATTR_TARGET, HEAD_ATTR_USECASE, HEAD_ATTR_UCID,
HEAD_ATTR_STORY, HEAD_ATTR_STORYID, HEAD_ATTR_APPS, HEAD_ATTR_DEPR]
LIST_HEAD_CONST = ["HEAD_ATTR_DESCR", "HEAD_ATTR_TARGET", "HEAD_ATTR_USECASE", "HEAD_ATTR_UCID",
"HEAD_ATTR_STORY", "HEAD_ATTR_STORYID", "HEAD_ATTR_APPS", "HEAD_ATTR_DEPR"]
CSV_HEADER_START = ["node", "table", "tabelle"]
CSV_DELIMITER = ";"
INTERNAL_DELIMITER = "||"
"""
internal structure of testdata
"""
CSV_SPECTYPE_DATA = "data"
CSV_SPECTYPE_TREE = "tree"
CSV_SPECTYPE_KEYS = "keys"
CSV_SPECTYPE_CONF = "conf"
CSV_NODETYPE_KEYS = "_keys"
CSV_BLOCK_HEAD = "_head"
CSV_BLOCK_OPTION = B.DATA_NODE_OPTION
CSV_BLOCK_STEP = B.DATA_NODE_STEPS
CSV_BLOCK_TABLES = B.DATA_NODE_TABLES
CSV_BLOCK_IMPORT = "_import"
LIST_CSV_BLOCKS = [CSV_BLOCK_HEAD, CSV_BLOCK_OPTION, CSV_BLOCK_STEP, CSV_BLOCK_TABLES, CSV_BLOCK_IMPORT]
LIST_BLOCK_CONST = ["CSV_BLOCK_HEAD", "CSV_BLOCK_OPTION", "CSV_BLOCK_STEP", "CSV_BLOCK_TABLES", "CSV_BLOCK_IMPORT"]
STEP_COMP_I = 1
STEP_EXECNR_I = 2
STEP_REFNR_I = 3
STEP_VARIANT_I = 4
STEP_ARGS_I = 5
STEP_LIST_I = 5
STEP_ATTR_COMP = "component"
STEP_ATTR_EXECNR = "exec-step"
STEP_ATTR_REFNR = "reference-nr"
STEP_ATTR_ARGS = "arguments"
LIST_STEP_ATTR = [STEP_ATTR_COMP, STEP_ATTR_EXECNR, STEP_ATTR_REFNR, STEP_ATTR_ARGS]
LIST_STEP_CONST = ["STEP_ATTR_COMP", "STEP_ATTR_EXECNR", "STEP_ATTR_REFNR", "STEP_ATTR_ARGS"]
EXCP_MALFORMAT = "malformated line: "
ATTR_SRC_TYPE = "tdtyp"
ATTR_SRC_DATA = "tdsrc"
ATTR_SRC_NAME = "tdname"
DEFAULT_DB_PARTITION = "n"
""" attribute if table is partitioned - partitions are parametrized """
DEFAULT_DB_CONN_JAR = "n"
""" attribute for connection-jar-file instead of connection by ip, port """
Loading…
Cancel
Save