From 77355baca0f628f5e95cbf1d8bf3c2475a95150f Mon Sep 17 00:00:00 2001 From: Manel Caro Date: Fri, 29 Mar 2019 14:54:24 +0100 Subject: Modify database connection object and xml parser object --- test-cli/test/helpers/testsrv_db.py | 172 +++++++++++++++++++++++++++++++++++- 1 file changed, 169 insertions(+), 3 deletions(-) (limited to 'test-cli/test/helpers/testsrv_db.py') 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 -- cgit v1.1