diff options
Diffstat (limited to 'test-cli/test/helpers/testsrv_db.py')
| -rw-r--r-- | test-cli/test/helpers/testsrv_db.py | 262 |
1 files changed, 173 insertions, 89 deletions
diff --git a/test-cli/test/helpers/testsrv_db.py b/test-cli/test/helpers/testsrv_db.py index 94181f9..8e96e8c 100644 --- a/test-cli/test/helpers/testsrv_db.py +++ b/test-cli/test/helpers/testsrv_db.py @@ -1,10 +1,10 @@ from test.helpers.psqldb import PgSQLConnection -from test.helpers.setup_xml import XMLSetup -def find_between( s, first, last ): + +def find_between(s, first, last): try: - start = s.index( first ) + len( first ) - end = s.index( last, start ) + start = s.index(first) + len(first) + end = s.index(last, start) return s[start:end] except ValueError: return "" @@ -19,158 +19,242 @@ class TestSrv_Database(object): def __init__(self): pass - def open (self, filename): - '''Open database connection''' - self.__xml_setup = XMLSetup(filename) + def open(self, xmlObj): + self.__xml_setup = xmlObj self.__sqlObject = PgSQLConnection() return self.__sqlObject.db_connect(self.__xml_setup.getdbConnectionStr()) - def create_board(self, processor_id, model_id, variant, bmac = None): + def getConfig(self): + return self.__sqlObject.getConfig() + + def create_board(self, processor_id, model_id, variant, station, bmac=None): '''create a new board''' if bmac is None: - sql = "SELECT isee.create_board('{}', '{}', '{}', NULL);".format(processor_id, model_id, variant) + sql = "SELECT * FROM isee.f_create_board('{}', '{}', '{}', NULL, '{}');".format(processor_id, model_id, + variant, + station) else: - sql = "SELECT isee.create_board('{}', '{}', '{}', '{}');".format(processor_id, model_id, variant, bmac) - #print('>>>' + sql) + sql = "SELECT * FROM isee.f_create_board('{}', '{}', '{}', '{}', '{}');".format(processor_id, model_id, + variant, + bmac, station) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res[0][0]; + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + print(r) + print(str(err)) 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) + def get_tests_list(self, board_uuid): + '''get the board test list''' + sql = "SELECT * FROM isee.f_get_tests_list('{}')".format(board_uuid) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res[0][0]; + # print(res) + return res except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def get_test_params_list(self, testid): + sql = "SELECT * FROM isee.f_get_test_params_list({})".format(testid) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res[0][0]; + # print(res) + return res except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def open_test(self, board_uuid): + sql = "SELECT * FROM isee.f_open_test('{}')".format(board_uuid) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res[0][0]; + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def run_test(self, testid_ctl, testid): + sql = "SELECT isee.f_run_test({},{})".format(testid_ctl, testid) + # print('>>>' + sql) + try: + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + + def finish_test(self, testid_ctl, testid, newstatus, textresult): + sql = "SELECT isee.f_finish_test({},{},'{}','{}')".format(testid_ctl, testid, newstatus, textresult) + # print('>>>' + sql) + try: + print('finish_test => SQL {}'.format(sql)) + self.__sqlObject.db_execute_query(sql) + return True + except Exception as err: + r = find_between(str(err), '#', '#') + print('finish_test => {}'.format(err)) + return False + + def upload_result_file(self, testid_ctl, testid, desc, filepath, mimetype): + try: + # generate a new oid + fileoid = self.__sqlObject.db_upload_large_file(filepath) + # insert into a table + sql = "SELECT isee.f_upload_result_file({},{},'{}','{}','{}')".format(testid_ctl, testid, desc, fileoid, + mimetype) + # print('>>>' + sql) + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + + def get_task_variables_list(self, board_uuid): + sql = "SELECT * FROM isee.f_get_task_variables_list('{}')".format(board_uuid) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res; + # print(res) + return res except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + + def get_plc_macaddr(self, board_uuid): + sql = "SELECT * FROM isee.f_get_plcmac(\'{}\')".format(board_uuid) + # print('>>>' + sql) + try: + res = self.__sqlObject.db_execute_query(sql) + return res, '' + except Exception as err: + r = find_between(str(err), '#', '#') + print(r) + return None, r + + + def get_board_macaddr(self, board_uuid): + sql = "SELECT * FROM isee.f_get_board_macaddr('{}')".format(board_uuid) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res; + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def open_task(self, uuid): + sql = "SELECT * FROM isee.f_open_task('{}')".format(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; + res = self.__sqlObject.db_execute_query(sql) + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def create_task_result(self, taskid_ctl, name, newstatus, newinfo=None): + sql = "SELECT isee.f_create_task_result({},'{}','{}','{}')".format(taskid_ctl, name, newstatus, newinfo) + # print('>>>' + sql) + try: + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + + def update_taskctl_status(self, taskid_ctl, newstatus): + sql = "SELECT isee.f_update_taskctl_status({},'{}')".format(taskid_ctl, newstatus) + # print('>>>' + sql) + try: + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + + def set_factorycode(self, uuid, factorycode): + sql = "SELECT isee.f_set_factorycode('{}','{}')".format(uuid, factorycode) + # print('>>>' + sql) + try: + self.__sqlObject.db_execute_query(sql) + except Exception as err: + r = find_between(str(err), '#', '#') + # print(r) + return None + + def read_station_state(self, station): + sql = "SELECT * FROM station.getmystate('{}');".format(station) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res; + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def change_station_state(self, station, newstate): + sql = "SELECT station.setmystate('{}', '{}', NULL)".format(newstate, station) + print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res; + print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + 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) + def get_setup_variable(self, skey): + sql = "SELECT * FROM admin.get_setupvar('{}')".format(skey) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res; + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # 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) + def get_setup_variable(self, skey , default): + sql = "SELECT * FROM admin.get_setupvar('{}')".format(skey) + # 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 default + + def setDevelStationState(self, station, newState): + sql = "UPDATE station.station_state SET state = {} WHERE hostname={}".format(newState, station) try: res = self.__sqlObject.db_execute_query(sql) - #print(res) - return res; + # print(res) + return res[0][0] except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # print(r) return None -
\ No newline at end of file |