22 changed files with 1044 additions and 38 deletions
|
|
@ -0,0 +1,238 @@ |
|||||
|
# --------------------------------------------------------------------------------------------------------- |
||||
|
# 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 = "project" |
||||
|
""" system-name for this entity """ |
||||
|
FIELD_ID = "prid" |
||||
|
FIELD_NAME = "name" |
||||
|
FIELD_DISCRIPTION = B.SUBJECT_REFERENCE |
||||
|
FIELD_REFERENCE = B.SUBJECT_REFERENCE |
||||
|
LIST_FIELDS = [FIELD_ID, FIELD_NAME, FIELD_DISCRIPTION, FIELD_REFERENCE] |
||||
|
""" list of object-attributes """ |
||||
|
|
||||
|
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 """ |
||||
|
|
||||
|
class Project(model.entity.Entity): |
||||
|
prid = 0 |
||||
|
name = "" |
||||
|
description = "" |
||||
|
reference = "" |
||||
|
|
||||
|
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 |
||||
|
""" |
||||
|
config = self.getConfig(job, B.SUBJECT_PROJECTS, "") |
||||
|
outList = list(config[B.SUBJECT_PROJECTS].keys()) |
||||
|
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 = self.getConfig(job, B.SUBJECT_PROJECTS, 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 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.SUBJECT_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.SUBJECT_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 getConfig(job, subject, 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_BASIC, subject) |
||||
|
if config is not None: |
||||
|
if len(name) == 0: |
||||
|
return config |
||||
|
elif name in config[subject]: |
||||
|
return config[subject][name] |
||||
|
raise Exception("keine Config zu "+name) |
||||
|
|
||||
|
@staticmethod |
||||
|
def getCurrentUser(job): |
||||
|
return os.environ.get("USERNAME") |
||||
|
|
||||
|
@ -0,0 +1,129 @@ |
|||||
|
import unittest |
||||
|
import os |
||||
|
import inspect |
||||
|
|
||||
|
import tools.path_tool |
||||
|
import basic.program |
||||
|
import test.testtools |
||||
|
import basic.constants as B |
||||
|
import test.constants as T |
||||
|
import model.project |
||||
|
import model.entity |
||||
|
|
||||
|
HOME_PATH = test.constants.HOME_PATH |
||||
|
PYTHON_CMD = "python" |
||||
|
TEST_FUNCTIONS = ["test_10getEntityNames", "test_11getEntities", "test_12getEntity", |
||||
|
"test_13writeEntity", "test_14insertEntity"] |
||||
|
TEST_FUNCTIONS = ["test_10getEntityNames", "test_12getEntity"] |
||||
|
PROGRAM_NAME = "clean_workspace" |
||||
|
|
||||
|
class MyTestCase(unittest.TestCase): |
||||
|
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() |
||||
|
project = model.project.Project() |
||||
|
entityNames = [] |
||||
|
entityNames = project.read_unique_names(job, "", "", "", {}) |
||||
|
self.assertEquals(type(entityNames), list) |
||||
|
#entityNames = project.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() |
||||
|
project = model.project.Project() |
||||
|
entityNames = [] |
||||
|
entityNames = project.get_entities(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertEqual(type(entityNames), list) |
||||
|
entityNames = project.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() |
||||
|
project = model.project.Project() |
||||
|
name = "TESTPROJ" |
||||
|
actproject = project.read_entity(job, name) |
||||
|
self.assertEqual(getattr(actproject, model.project.FIELD_NAME), name) |
||||
|
self.assertRaises(Exception, project.read_entity, job, "xyzxyz") |
||||
|
# |
||||
|
#actproject = project.select_entity(job, name) |
||||
|
#self.assertEqual(getattr(actproject, model.project.FIELD_USERNAME), name) |
||||
|
#self.assertRaises(Exception, project.select_entity, job, ["xyzxyz"]) |
||||
|
|
||||
|
def test_13writeEntity(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() |
||||
|
username = "hans_xyz" |
||||
|
project = model.project.Project() |
||||
|
entityNames = project.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
project.username = username |
||||
|
project.name = "Hans" |
||||
|
project.famname = "im Glueck" |
||||
|
project.project = "TESTPROJ" |
||||
|
project.write_entity(job, username) |
||||
|
entityNames = project.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertIn(username, entityNames) |
||||
|
actUser = project.read_entity(job, username) |
||||
|
self.assertEquals(getattr(actUser, model.project.FIELD_USERNAME), username) |
||||
|
actUser.remove_entity(job, username) |
||||
|
entityNames = project.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
|
||||
|
def test_14insertEntity(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() |
||||
|
username = "hans_xyz" |
||||
|
project = model.project.Project() |
||||
|
entityNames = collectInnerList(project.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
#self.assertNotIn(username, entityNames) |
||||
|
project.username = username |
||||
|
project.name = "Hans" |
||||
|
project.famname = "im Glueck" |
||||
|
project.project = "TESTPROJ" |
||||
|
#project.insert_entity(job, username, table="project") |
||||
|
entityNames = collectInnerList(project.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
self.assertIn(username, entityNames) |
||||
|
actUser = project.select_entity(job, username) |
||||
|
self.assertEquals(getattr(actUser, model.project.FIELD_USERNAME), username) |
||||
|
actUser.delete_entity(job, username, "project") |
||||
|
entityNames = collectInnerList(project.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
|
||||
|
def collectInnerList(inList): |
||||
|
outList = [] |
||||
|
for r in inList: |
||||
|
outList += r |
||||
|
return outList |
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
unittest.main() |
||||
@ -0,0 +1,129 @@ |
|||||
|
import unittest |
||||
|
import os |
||||
|
import inspect |
||||
|
|
||||
|
import tools.path_tool |
||||
|
import basic.program |
||||
|
import test.testtools |
||||
|
import basic.constants as B |
||||
|
import test.constants as T |
||||
|
import model.release |
||||
|
import model.entity |
||||
|
|
||||
|
HOME_PATH = test.constants.HOME_PATH |
||||
|
PYTHON_CMD = "python" |
||||
|
TEST_FUNCTIONS = ["test_10getEntityNames", "test_11getEntities", "test_12getEntity", |
||||
|
"test_13writeEntity", "test_14insertEntity"] |
||||
|
TEST_FUNCTIONS = ["test_10getEntityNames", "test_12getEntity"] |
||||
|
PROGRAM_NAME = "clean_workspace" |
||||
|
|
||||
|
class MyTestCase(unittest.TestCase): |
||||
|
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() |
||||
|
release = model.release.Release(job, "TESTPROJ") |
||||
|
entityNames = [] |
||||
|
entityNames = release.read_unique_names(job, "", "", "", {}) |
||||
|
self.assertEquals(type(entityNames), list) |
||||
|
#entityNames = project.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() |
||||
|
release = model.release.Release(job, "TESTPROJ") |
||||
|
entityNames = [] |
||||
|
entityNames = release.get_entities(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertEqual(type(entityNames), list) |
||||
|
#entityNames = release.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() |
||||
|
release = model.release.Release(job, "TESTPROJ") |
||||
|
name = "V-1.1.0" |
||||
|
actrelease = release.read_entity(job, name) |
||||
|
self.assertEqual(getattr(actrelease, model.project.FIELD_NAME), name) |
||||
|
self.assertRaises(Exception, release.read_entity, job, "xyzxyz") |
||||
|
# |
||||
|
#actproject = project.select_entity(job, name) |
||||
|
#self.assertEqual(getattr(actproject, model.project.FIELD_USERNAME), name) |
||||
|
#self.assertRaises(Exception, project.select_entity, job, ["xyzxyz"]) |
||||
|
|
||||
|
def test_13writeEntity(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() |
||||
|
username = "hans_xyz" |
||||
|
project = model.project.Project() |
||||
|
entityNames = project.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
project.username = username |
||||
|
project.name = "Hans" |
||||
|
project.famname = "im Glueck" |
||||
|
project.project = "TESTPROJ" |
||||
|
project.write_entity(job, username) |
||||
|
entityNames = project.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertIn(username, entityNames) |
||||
|
actUser = project.read_entity(job, username) |
||||
|
self.assertEquals(getattr(actUser, model.project.FIELD_USERNAME), username) |
||||
|
actUser.remove_entity(job, username) |
||||
|
entityNames = project.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
|
||||
|
def test_14insertEntity(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() |
||||
|
username = "hans_xyz" |
||||
|
project = model.project.Project() |
||||
|
entityNames = collectInnerList(project.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
#self.assertNotIn(username, entityNames) |
||||
|
project.username = username |
||||
|
project.name = "Hans" |
||||
|
project.famname = "im Glueck" |
||||
|
project.project = "TESTPROJ" |
||||
|
#project.insert_entity(job, username, table="project") |
||||
|
entityNames = collectInnerList(project.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
self.assertIn(username, entityNames) |
||||
|
actUser = project.select_entity(job, username) |
||||
|
self.assertEquals(getattr(actUser, model.project.FIELD_USERNAME), username) |
||||
|
actUser.delete_entity(job, username, "project") |
||||
|
entityNames = collectInnerList(project.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
|
||||
|
def collectInnerList(inList): |
||||
|
outList = [] |
||||
|
for r in inList: |
||||
|
outList += r |
||||
|
return outList |
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
unittest.main() |
||||
@ -0,0 +1,128 @@ |
|||||
|
import unittest |
||||
|
import os |
||||
|
import inspect |
||||
|
|
||||
|
import tools.path_tool |
||||
|
import basic.program |
||||
|
import test.testtools |
||||
|
import basic.constants as B |
||||
|
import test.constants as T |
||||
|
import model.component |
||||
|
import model.entity |
||||
|
|
||||
|
HOME_PATH = test.constants.HOME_PATH |
||||
|
PYTHON_CMD = "python" |
||||
|
TEST_FUNCTIONS = ["test_10getEntityNames", "test_11getEntities", "test_12getEntity", |
||||
|
"test_13writeEntity", "test_14insertEntity"] |
||||
|
TEST_FUNCTIONS = ["test_10getEntityNames", "test_12getEntity"] |
||||
|
PROGRAM_NAME = "clean_workspace" |
||||
|
|
||||
|
class MyTestCase(unittest.TestCase): |
||||
|
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() |
||||
|
component = model.component.Component(job) |
||||
|
entityNames = [] |
||||
|
entityNames = component.read_unique_names(job, "", "", "", {}) |
||||
|
self.assertEquals(type(entityNames), list) |
||||
|
#entityNames = component.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() |
||||
|
component = model.component.component() |
||||
|
entityNames = [] |
||||
|
entityNames = component.get_entities(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertEqual(type(entityNames), list) |
||||
|
entityNames = component.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() |
||||
|
component = model.component.Component(job) |
||||
|
name = "testprd" |
||||
|
actproject = component.read_entity(job, name) |
||||
|
self.assertEqual(getattr(actproject, model.component.FIELD_NAME), name) |
||||
|
self.assertRaises(Exception, component.read_entity, job, "xyzxyz") |
||||
|
# |
||||
|
#actproject = component.select_entity(job, name) |
||||
|
#self.assertEqual(getattr(actproject, model.component.FIELD_USERNAME), name) |
||||
|
#self.assertRaises(Exception, component.select_entity, job, ["xyzxyz"]) |
||||
|
|
||||
|
def test_13writeEntity(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() |
||||
|
username = "hans_xyz" |
||||
|
component = model.component.component() |
||||
|
entityNames = component.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
component.username = username |
||||
|
component.name = "Hans" |
||||
|
component.famname = "im Glueck" |
||||
|
component.component = "TESTPROJ" |
||||
|
component.write_entity(job, username) |
||||
|
entityNames = component.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertIn(username, entityNames) |
||||
|
actUser = component.read_entity(job, username) |
||||
|
self.assertEquals(getattr(actUser, model.component.FIELD_USERNAME), username) |
||||
|
actUser.remove_entity(job, username) |
||||
|
entityNames = component.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
|
||||
|
def test_14insertEntity(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() |
||||
|
username = "hans_xyz" |
||||
|
component = model.component.component() |
||||
|
entityNames = collectInnerList(component.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
#self.assertNotIn(username, entityNames) |
||||
|
component.username = username |
||||
|
component.name = "Hans" |
||||
|
component.famname = "im Glueck" |
||||
|
component.component = "TESTPROJ" |
||||
|
#component.insert_entity(job, username, table="component") |
||||
|
entityNames = collectInnerList(component.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
self.assertIn(username, entityNames) |
||||
|
actUser = component.select_entity(job, username) |
||||
|
self.assertEquals(getattr(actUser, model.component.FIELD_USERNAME), username) |
||||
|
actUser.delete_entity(job, username, "component") |
||||
|
entityNames = collectInnerList(component.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
||||
|
self.assertNotIn(username, entityNames) |
||||
|
|
||||
|
def collectInnerList(inList): |
||||
|
outList = [] |
||||
|
for r in inList: |
||||
|
outList += r |
||||
|
return outList |
||||
|
|
||||
|
if __name__ == '__main__': |
||||
|
unittest.main() |
||||
Loading…
Reference in new issue