19 changed files with 661 additions and 361 deletions
@ -1,113 +0,0 @@ |
|||
# program to iinitialize the workspace |
|||
# pre: 1. program cloned |
|||
# 2. components are cloned, optional with path_const in config-folder |
|||
# post: 1. all folder are made with special name like components/const/path_const.py - otherwise with defaults |
|||
# 2. basic-configuration is created if it does not exist |
|||
# 3. necessary python-packages will be installed |
|||
# 4. git repos will be always updated |
|||
# ------------------------------------------------------------------------------------------------------------- |
|||
""" |
|||
|
|||
""" |
|||
import os |
|||
|
|||
import basic.program |
|||
import tools.path_tool |
|||
import tools.file_tool |
|||
import basic.constants as B |
|||
# look if the path-constanst are specific defined. |
|||
home = os.getcwd() |
|||
if os.path.exists(os.path.join(home, "components", "config", "path_const.py")): |
|||
import components.config.path_const as P |
|||
else: |
|||
import tools.path_const as P |
|||
# import always the keyword. |
|||
import tools.path_const as Q |
|||
|
|||
print("# ----------------------------------------------------------------------------------- ") |
|||
dirs = {} |
|||
dirs[Q.ATTR_PATH_HOME] = home |
|||
dirs[Q.ATTR_PATH_PROGRAM] = home |
|||
dirs[Q.ATTR_PATH_COMPONENTS] = os.path.join(home, "components") |
|||
pval = [P.ATTR_PATH_ARCHIV, P.ATTR_PATH_ENV, P.ATTR_PATH_DEBUG, P.ATTR_PATH_TEMP] |
|||
qkey = [Q.ATTR_PATH_ARCHIV, Q.ATTR_PATH_ENV, Q.ATTR_PATH_DEBUG, Q.ATTR_PATH_TEMP] |
|||
dirkey = [Q.ATTR_PATH_PROGRAM, Q.ATTR_PATH_COMPONENTS, Q.ATTR_PATH_HOME, Q.VAL_BASE_DATA] + qkey |
|||
home = tools.path_tool.getHome() |
|||
print("mkdirs in home " + home) |
|||
dirname = os.path.join(home, P.VAL_BASE_DATA) |
|||
dirs[Q.VAL_BASE_DATA] = dirname |
|||
if not os.path.exists(dirname): |
|||
os.makedirs(dirname, exist_ok=True) |
|||
print(" - "+dirname) |
|||
else: |
|||
print(" - " + dirname + " exists ") |
|||
dirname = os.path.join(home, P.VAL_CONFIG) |
|||
if not os.path.exists(dirname): |
|||
os.makedirs(dirname, exist_ok=True) |
|||
print(" - "+dirname) |
|||
else: |
|||
print(" - " + dirname + " exists ") |
|||
|
|||
for i in range(0, len(pval)): |
|||
dirname = os.path.join(home, P.VAL_BASE_DATA, pval[i]) |
|||
dirs[qkey[i]] = dirname |
|||
if not os.path.exists(dirname): |
|||
os.makedirs(dirname, exist_ok=True) |
|||
print(" - "+dirname) |
|||
else: |
|||
print(" - " + dirname + " exists ") |
|||
|
|||
print("\n# ----------------------------------------------------------------------------------- ") |
|||
for format in ["yml", "json", "xml"]: |
|||
filename = os.path.join(home, Q.VAL_CONFIG, B.BASIS_FILE+"."+format) |
|||
if os.path.isfile(filename): |
|||
break |
|||
filename = "" |
|||
if len(filename) < 1: |
|||
filename = os.path.join(home, Q.VAL_CONFIG, B.BASIS_FILE+".yml") |
|||
text = "basic:\n" |
|||
text += " language: en\n" |
|||
text += " paths:\n" |
|||
for d in dirkey: |
|||
text += " "+d+": "+dirs[d]+"\n" |
|||
print(text) |
|||
print(filename) |
|||
with open(filename, 'w') as file: |
|||
file.write(text) |
|||
file.close() |
|||
print("create basis-config in "+filename) |
|||
print(text) |
|||
else: |
|||
print("basis-config exists "+filename) |
|||
|
|||
print("\n# ----------------------------------------------------------------------------------- ") |
|||
print("please install requirements ") |
|||
# install requirements |
|||
# sudo apt install python3.10-venv |
|||
# python3 -m venv venv |
|||
# . venv/bin/activate |
|||
# pip install Flask |
|||
|
|||
job = basic.program.Job("unit") |
|||
print("\n# ----------------------------------------------------------------------------------- ") |
|||
import tools.git_tool |
|||
for repo in ["program", "components", "testdata"]: |
|||
tools.git_tool.gitPull(job, repo) |
|||
|
|||
print("\n# ----------------------------------------------------------------------------------- ") |
|||
if "db" in job.conf: |
|||
import basic.connection |
|||
entity = basic.connection.Connection(job) |
|||
entity.createSchema() |
|||
import model.testcase |
|||
entity = model.testcase.Testcase(job) |
|||
entity.createSchema() |
|||
import model.testsuite |
|||
entity = model.testsuite.Testsuite(job) |
|||
entity.createSchema() |
|||
import model.testplan |
|||
entity = model.testplan.Testplan(job) |
|||
entity.createSchema() |
|||
import basic.testexecution |
|||
entity = basic.testexecution.Testexecution(job) |
|||
entity.createSchema() |
|||
@ -1,143 +0,0 @@ |
|||
import getpass |
|||
import os |
|||
import re |
|||
import basic.toolHandling |
|||
import tools.data_const as D |
|||
import basic.constants as B |
|||
import tools.date_tool |
|||
import tools.file_tool |
|||
|
|||
class Entity: |
|||
def __int__(self, job): |
|||
self.job = job |
|||
self.table = "" |
|||
self.testserver = None |
|||
|
|||
def get_schema(self): |
|||
""" |
|||
gets schema/ddl-informations in order to create the database |
|||
""" |
|||
raise Exception(B.EXCEPT_NOT_IMPLEMENT) |
|||
|
|||
def read_entity(self, job, name): |
|||
""" |
|||
reads the entity from the file-system |
|||
:param job: |
|||
:param name: |
|||
:return: |
|||
""" |
|||
raise Exception(B.EXCEPT_NOT_IMPLEMENT) |
|||
|
|||
def select_entity(self, job, name): |
|||
""" |
|||
reads the entity from the database |
|||
it should get the same result like read_entity |
|||
:param job: |
|||
:param name: |
|||
:return: |
|||
""" |
|||
raise Exception(B.EXCEPT_NOT_IMPLEMENT) |
|||
|
|||
def write_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 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 getDbAttr(self, job): |
|||
out = {} |
|||
for attr in [B.ATTR_DB_HOST, B.ATTR_DB_USER, B.ATTR_DB_DATABASE, B.ATTR_DB_PASSWD]: |
|||
out[attr] = job.conf[B.TOPIC_NODE_DB][attr] |
|||
return out |
|||
|
|||
def getDdl(self, job, ddl): |
|||
out = {} |
|||
for t in ddl: |
|||
out[t] = {} |
|||
for f in ddl[t]: |
|||
out[t][f] = {} |
|||
for a in ddl[t][f]: |
|||
print("entity-23 "+f+", "+a+" "+str(ddl)) |
|||
out[t][f][a] = ddl[t][f][a] |
|||
out[t][f][D.DDL_FIELD] = f |
|||
out[t][B.DATA_NODE_HEADER] = list(ddl[t].keys()) |
|||
return out |
|||
|
|||
def createSchema(self, testserver): |
|||
if B.TOPIC_NODE_DB in self.job.conf: |
|||
dbi = basic.toolHandling.getDbTool(self.job, testserver, self.job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE]) |
|||
else: |
|||
return "No DB in job-config" |
|||
sql = self.get_schema() |
|||
print(sql) |
|||
for s in sql.split(";\n"): |
|||
if len(s) < 3: continue |
|||
try: |
|||
dbi.execStatement(s+";", self.job.conf[B.TOPIC_NODE_DB]) |
|||
print("SQL executed: "+s) |
|||
except Exception as e: |
|||
raise Exception("Fehler bei createSchema "+s) |
|||
|
|||
|
|||
|
|||
|
|||
def getHistoryFields(self): |
|||
dbtype = self.job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE] |
|||
dbi = basic.toolHandling.getDbTool(self.job, None, dbtype) |
|||
sql = dbi.getSchemaAttribut("inscommit", D.TYPE_STR)+"," |
|||
sql += dbi.getSchemaAttribut("insauthor", D.TYPE_STR)+"," |
|||
sql += dbi.getSchemaAttribut("instime", D.TYPE_TIME)+"," |
|||
sql += dbi.getSchemaAttribut("updcommit", D.TYPE_STR)+"," |
|||
sql += dbi.getSchemaAttribut("updauthor", D.TYPE_STR)+"," |
|||
sql += dbi.getSchemaAttribut("updtime", D.TYPE_TIME)+"," |
|||
sql += dbi.getSchemaAttribut("actual", D.TYPE_INT) |
|||
return sql |
|||
|
|||
def selectHistoryFields(self): |
|||
if B.TOPIC_NODE_DB in self.job.conf: |
|||
dbi = basic.toolHandling.getDbTool(self.job, self.testserver, self.job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE]) |
|||
else: |
|||
return "No DB in job-config" |
|||
dbi.selectRows |
|||
|
|||
def getHistoryIndex(self, table): |
|||
dbtype = self.job.conf[B.TOPIC_NODE_DB][B.ATTR_TYPE] |
|||
dbi = basic.toolHandling.getDbTool(self.job, None, dbtype) |
|||
sql = dbi.getSchemaIndex(table, "actual") + "\n" |
|||
return sql |
|||
|
|||
|
|||
def read_spec(job, testentity, testgran, specpath): |
|||
if not os.path.isfile(specpath): |
|||
return |
|||
text = tools.file_tool.read_file_text(job, specpath, job.m) |
|||
if re.match(r".*?depricated;[jJyY]", text): |
|||
return None |
|||
spec = {} |
|||
regex = re.compile(r".*\nhead:(.*?);(.+)") |
|||
for res in regex.finditer(text): |
|||
#res = re.search(r".*head:(.*?);(.+)\n", text) |
|||
key = res.group(1) |
|||
if key == B.SUBJECT_DESCRIPTION: |
|||
spec[B.SUBJECT_DESCRIPTION] = res.group(2).replace(";", "") |
|||
elif key in [B.SUBJECT_APPS, B.PAR_APP]: |
|||
apps = res.group(2).replace(";", ",").split(",") |
|||
spec[B.SUBJECT_APPS] = apps |
|||
else: |
|||
val = res.group(2).replace(";", "") |
|||
spec[key] = val |
|||
return spec |
|||
|
@ -0,0 +1,255 @@ |
|||
# --------------------------------------------------------------------------------------------------------- |
|||
# 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 = "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 """ |
|||
|
|||
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.SUBJECT_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 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 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") |
|||
|
|||
@ -0,0 +1,131 @@ |
|||
import unittest |
|||
import os |
|||
import inspect |
|||
import shutil |
|||
|
|||
import tools.path_tool |
|||
import basic.program |
|||
import test.testtools |
|||
import basic.constants as B |
|||
import test.constants as T |
|||
import tools.file_tool |
|||
import model.user |
|||
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_14insertEntity"] |
|||
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() |
|||
user = model.user.User() |
|||
entityNames = [] |
|||
entityNames = user.read_unique_names(job, "", "", "", {}) |
|||
self.assertEquals(type(entityNames), list) |
|||
entityNames = user.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() |
|||
user = model.user.User() |
|||
entityNames = [] |
|||
entityNames = user.get_entities(job, storage=model.entity.STORAGE_FILE) |
|||
self.assertEqual(type(entityNames), list) |
|||
entityNames = user.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() |
|||
user = model.user.User() |
|||
name = "ulrich" |
|||
actUser = user.read_entity(job, name) |
|||
self.assertEqual(getattr(actUser, model.user.FIELD_USERNAME), name) |
|||
self.assertRaises(Exception, user.read_entity, job, "xyzxyz") |
|||
# |
|||
actUser = user.select_entity(job, name) |
|||
self.assertEqual(getattr(actUser, model.user.FIELD_USERNAME), name) |
|||
self.assertRaises(Exception, user.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" |
|||
user = model.user.User() |
|||
entityNames = user.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
|||
self.assertNotIn(username, entityNames) |
|||
user.username = username |
|||
user.name = "Hans" |
|||
user.famname = "im Glueck" |
|||
user.project = "TESTPROJ" |
|||
user.write_entity(job, username) |
|||
entityNames = user.get_unique_names(job, storage=model.entity.STORAGE_FILE) |
|||
self.assertIn(username, entityNames) |
|||
actUser = user.read_entity(job, username) |
|||
self.assertEquals(getattr(actUser, model.user.FIELD_USERNAME), username) |
|||
actUser.remove_entity(job, username) |
|||
entityNames = user.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" |
|||
user = model.user.User() |
|||
entityNames = collectInnerList(user.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
|||
#self.assertNotIn(username, entityNames) |
|||
user.username = username |
|||
user.name = "Hans" |
|||
user.famname = "im Glueck" |
|||
user.project = "TESTPROJ" |
|||
#user.insert_entity(job, username, table="user") |
|||
entityNames = collectInnerList(user.get_unique_names(job, storage=model.entity.STORAGE_DB)) |
|||
self.assertIn(username, entityNames) |
|||
actUser = user.select_entity(job, username) |
|||
self.assertEquals(getattr(actUser, model.user.FIELD_USERNAME), username) |
|||
actUser.delete_entity(job, username, "user") |
|||
entityNames = collectInnerList(user.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