summaryrefslogtreecommitdiff
path: root/test-cli/test
diff options
context:
space:
mode:
Diffstat (limited to 'test-cli/test')
-rw-r--r--test-cli/test/helpers/psqldb.py11
-rw-r--r--test-cli/test/helpers/setup_xml.py22
-rw-r--r--test-cli/test/helpers/testsrv_db.py172
3 files changed, 189 insertions, 16 deletions
diff --git a/test-cli/test/helpers/psqldb.py b/test-cli/test/helpers/psqldb.py
index 26dd03d..ad40618 100644
--- a/test-cli/test/helpers/psqldb.py
+++ b/test-cli/test/helpers/psqldb.py
@@ -1,13 +1,14 @@
import psycopg2
+
class PgSQLConnection(object):
- """aaaaaaa"""
+ """Database Connection Object"""
__conection_object = None
__db_config = {'dbname': 'testsrv', 'host': '192.168.2.171',
'password': 'Idkfa2009', 'port': 5432, 'user': 'admin'}
- def __init__ (self):
+ def __init__(self):
# self.__conection_object = None
# if connect_str is not None:
# self.__db_config = connect_str
@@ -15,10 +16,10 @@ class PgSQLConnection(object):
self.__db_config = {'dbname': 'testsrv', 'host': '192.168.2.171',
'password': 'Idkfa2009', 'port': 5432, 'user': 'admin'}
- def db_connect (self, connect_str):
+ def db_connect(self, connect_str):
result = False
try:
- if connect_str == None:
+ if connect_str is None:
self.__conection_object = psycopg2.connect(**self.__db_config)
else:
self.__db_config = connect_str;
@@ -39,14 +40,12 @@ class PgSQLConnection(object):
def commit(self):
self.__conection_object.commit()
-
def db_close(self):
if self.__conection_object is not None:
self.__conection_object.close()
del self.__conection_object
self.__conection_object = None
-
def __del__(self):
if self.__conection_object is not None:
self.__conection_object.close()
diff --git a/test-cli/test/helpers/setup_xml.py b/test-cli/test/helpers/setup_xml.py
index 3fd9fd5..c1dc21e 100644
--- a/test-cli/test/helpers/setup_xml.py
+++ b/test-cli/test/helpers/setup_xml.py
@@ -1,22 +1,23 @@
import xml.etree.ElementTree as XMLParser
+
class XMLSetup (object):
- """aaaaa"""
+ """XML Setup Parser"""
__tree = None # Parser
__dbType = None # database connection required: PgSQLConnection
__dbConnectionRaw = None # Connection string in raw
__dbConnectionStr = None # Connection string to use in sql object connection
def __init__(self, filename):
- """aaaaa"""
+ """Parse the file in the constructor"""
self.__tree = XMLParser.parse(filename)
def __del__(self):
- """aaaaa"""
+ """Destructor do nothing"""
pass
- def getdbConnectionStr (self):
- """aaaaa"""
+ def getdbConnectionStr(self):
+ """XML to database connection string"""
if self.__dbConnectionRaw is not None:
return self.__dbConnectionRaw
@@ -29,7 +30,7 @@ class XMLSetup (object):
return None
- def getPostgresConnectionStr (self):
+ def getPostgresConnectionStr(self):
"""aaaaa"""
str = self.__dbConnectionRaw
del str['type']
@@ -37,4 +38,11 @@ class XMLSetup (object):
def getMysqlConnectionStr (self):
"""aaaaa"""
- pass \ No newline at end of file
+ pass
+
+ def getBoard(self, key, default):
+ for element in self.__tree.iter('board'):
+ if key in element.attrib:
+ return element.attrib[key]
+ return default
+
diff --git a/test-cli/test/helpers/testsrv_db.py b/test-cli/test/helpers/testsrv_db.py
index 94181f9..0de7739 100644
--- a/test-cli/test/helpers/testsrv_db.py
+++ b/test-cli/test/helpers/testsrv_db.py
@@ -11,7 +11,7 @@ def find_between( s, first, last ):
class TestSrv_Database(object):
- ''' TestSrv Database Helper '''
+ '''TestSrv Database Helper'''
__sqlObject = None
__xml_setup = None
@@ -19,7 +19,7 @@ class TestSrv_Database(object):
def __init__(self):
pass
- def open (self, filename):
+ def open(self, filename):
'''Open database connection'''
self.__xml_setup = XMLSetup(filename)
self.__sqlObject = PgSQLConnection()
@@ -173,4 +173,170 @@ class TestSrv_Database(object):
r = find_between(str(err), '#', '#')
#print(r)
return None
- \ No newline at end of file
+
+
+class testServerDb (object):
+ '''TestSrv Database Helper'''
+
+ __sqlObject = None
+ __xml_setup = None
+
+ def __init__(self):
+ pass
+
+ def open(self, xmlfile):
+ '''Open database connection'''
+ self.__xml_setup = xmlfile
+ self.__sqlObject = PgSQLConnection()
+ return self.__sqlObject.db_connect(self.__xml_setup.getdbConnectionStr())
+
+ def create_board(self, processor_id, model_id, variant, bmac=None):
+ '''create a new board'''
+ if bmac is None:
+ sql = "SELECT isee.create_board('{}', '{}', '{}', NULL);".format(processor_id, model_id, variant)
+ else:
+ sql = "SELECT isee.create_board('{}', '{}', '{}', '{}');".format(processor_id, model_id, variant, bmac)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res[0][0];
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def create_model(self, modid, variant, descr, tgid):
+ '''create new model'''
+ sql = "SELECT isee.create_model('{}', '{}', '{}', '{}')".format(modid, variant, descr, tgid)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res[0][0];
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def create_test_definition(self, testname, testdesc, testfunc):
+ '''Create a new definition and return definition id on fail (testname already exist) return -1'''
+ sql = "SELECT isee.define_test('{}', '{}', '{}')".format(testname, testdesc, testfunc)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res[0][0];
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def add_testdef_to_group(self, testgroupid, testname, testparam):
+ '''Assign definition to group test return true on success or false if it fails'''
+ sql = "SELECT isee.add_test_to_group('{}', '{}', '{}')".format(testgroupid, testname, testparam)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res[0][0];
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def getboard_test_list(self, board_uuid):
+ '''get the board test list'''
+ sql = "SELECT isee.gettestlist('{}')".format(board_uuid)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def getboard_comp_test_list(self, board_uuid):
+ '''get the board test list'''
+ sql = "SELECT isee.gettestcompletelist('{}')".format(board_uuid)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def open_testbatch(self, board_uuid):
+ '''get the board test list'''
+ sql = "SELECT isee.open_testbatch('{}')".format(board_uuid)
+ # print('>>>' + sql)
+ try:
+ res = str(self.__sqlObject.db_execute_query(sql)[0])
+ res = res.replace('(', '')
+ res = res.replace(')', '')
+ res = res.replace(',', '')
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def add_test_to_batch(self, board_uuid, testid, testid_ctl, result, groupid, data):
+ '''get the board test list'''
+ sql = "SELECT isee.add_test_to_batch_c('{}','{}','{}','{}','{}','{}')".format(board_uuid, testid, testid_ctl,
+ result, groupid, data)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def close_testbatch(self, board_uuid, testid_ctl):
+ '''get the board test list'''
+ sql = "SELECT isee.close_testbatch('{}','{}')".format(board_uuid, testid_ctl)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def update_set_test_row(self, nstation, testid_ctl, board_uuid, ctest, cstatus):
+ '''get the board test list'''
+ sql = "SELECT isee.update_set_test_row('{}','{}','{}','{}','{}')".format(nstation, testid_ctl, board_uuid,
+ ctest, cstatus)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
+
+ def getboard_eeprom(self, board_uuid):
+ '''get the board eeprom struct '''
+ sql = "SELECT isee.getboard_eeprom('{}')".format(board_uuid)
+ # print('>>>' + sql)
+ try:
+ res = self.__sqlObject.db_execute_query(sql)
+ # print(res)
+ return res;
+ except Exception as err:
+ r = find_between(str(err), '#', '#')
+ # print(r)
+ return None
generated by cgit v1.1 at 2024-11-10 03:51:48 +0000