diff options
Diffstat (limited to 'test-cli')
| -rw-r--r-- | test-cli/test/helpers/testsrv_db.py | 155 | ||||
| -rw-r--r-- | test-cli/test/runners/simple.py | 6 | ||||
| -rw-r--r-- | test-cli/test/tests/qeeprom.py | 17 | ||||
| -rw-r--r-- | test-cli/test_main.py | 106 |
4 files changed, 119 insertions, 165 deletions
diff --git a/test-cli/test/helpers/testsrv_db.py b/test-cli/test/helpers/testsrv_db.py index 7ed02f3..b7b75b1 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,157 +19,90 @@ class TestSrv_Database(object): def __init__(self): pass - def open (self, xmlObj): - self.__xml_setup = xmlObj; + 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 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 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) - 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) + sql = "SELECT 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) return None - def getboard_test_list(self, board_uuid): + def get_tests_list(self, board_uuid): '''get the board test list''' - sql = "SELECT isee.gettestlist('{}')".format(board_uuid) - #print('>>>' + sql) + sql = "SELECT * FROM isee.f_get_tests_list('{}')".format(board_uuid) + # print('>>>' + sql) try: res = self.__sqlObject.db_execute_query(sql) - #print(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): + def get_test_params_list(self, testid): '''get the board test list''' - sql = "SELECT * FROM isee.gettestcompletelist('{}')".format(board_uuid) - #print('>>>' + sql) + 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; + # print(res) + return res except Exception as err: r = find_between(str(err), '#', '#') - #print(r) + # print(r) return None - def getboard_test_variables(self, board_uuid, testdefid): + def open_test(self, board_uuid): '''get the board test list''' - sql = "SELECT * FROM isee.getboardtestvariables('{}', '{}')".format(board_uuid, testdefid) - #print('>>>' + sql) + sql = "SELECT * FROM isee.f_open_test('{}')".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, station): - '''get the board test list''' - sql = "SELECT * FROM isee.open_testbatch('{}','{}')".format(board_uuid, station) - #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; + # print(res) + return res 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): + def run_test(self, testid_ctl, testid): '''get the board test list''' - sql = "SELECT isee.add_test_to_batch_c('{}','{}','{}','{}','{}','{}')".format(board_uuid, testid, testid_ctl, result, groupid, data) - #print('>>>' + sql) + sql = "SELECT * FROM isee.f_run_test('{}','{}')".format(testid_ctl, testid) + # 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 close_testbatch(self, board_uuid, testid_ctl): + def finish_test(self, testid_ctl, testid, newstatus): '''get the board test list''' - sql = "SELECT isee.close_testbatch('{}','{}')".format(board_uuid, testid_ctl) - #print('>>>' + sql) + sql = "SELECT * FROM isee.f_finish_test('{}','{}','{}')".format(testid_ctl, testid, newstatus) + # 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 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
\ No newline at end of file diff --git a/test-cli/test/runners/simple.py b/test-cli/test/runners/simple.py index 3537d0e..084a5b8 100644 --- a/test-cli/test/runners/simple.py +++ b/test-cli/test/runners/simple.py @@ -89,12 +89,6 @@ class TextTestResult(unittest.TestResult): dbdata['name'] = test.shortDescription() dbdata['result'] = self.result dbdata['msg'] = str(test.longMessage) - #DB: INSERT IN THE DATABASE - filename='test_results.dat' - testResult = open(filename, 'a') - testResult.write(str(dbdata)) - testResult.write("\n") - testResult.close() #CONVERT FANCY FAIL AND PASS if self.result==self.PASS: simple_result="TRUE" if self.result == self.FAIL: simple_result = "FALSE" diff --git a/test-cli/test/tests/qeeprom.py b/test-cli/test/tests/qeeprom.py index 5bb0755..c2293c2 100644 --- a/test-cli/test/tests/qeeprom.py +++ b/test-cli/test/tests/qeeprom.py @@ -5,10 +5,9 @@ import unittest class Qeeprom(unittest.TestCase): __position = None - __uuid = None __eeprompath = None - # varlist: position, uuid, eeprompath + # varlist: position, eeprompath def __init__(self, testname, testfunc, varlist): super(Qeeprom, self).__init__(testfunc) self._testMethodDoc = testname @@ -16,25 +15,23 @@ class Qeeprom(unittest.TestCase): self.__position = varlist["position"] else: raise Exception('position param inside Qeeprom must be defined') - if "uuid" in varlist: - self.__uuid = varlist["uuid"] - else: - raise Exception('uuid param inside Qeeprom must be defined') if "eeprompath" in varlist: self.__eeprompath = varlist["eeprompath"] else: raise Exception('eeprompath param inside Qeeprom must be defined') def execute(self): + # generate some data + data_tx = "isee_test" # write data into the eeprom - p = sh.dd(sh.echo(self._uuid), "of=" + self.__eeprompath, "bs=1", "seek=" + self.__position) + p = sh.dd(sh.echo(data_tx), "of=" + self.__eeprompath, "bs=1", "seek=" + self.__position) if p.exit_code == 0: # read data from the eeprom - p = sh.dd("if=" + self.__eeprompath, "bs=1", "skip=" + self.__position, "count=" + len(self.__uuid)) + p = sh.dd("if=" + self.__eeprompath, "bs=1", "skip=" + self.__position, "count=" + len(data_tx)) if p.exit_code == 0: - uuid_rx = p.stdout.decode('ascii') + data_rx = p.stdout.decode('ascii') # compare both values - if (uuid_rx != self.__uuid): + if data_rx != data_tx: self.fail("failed: mismatch between written and received values.") else: diff --git a/test-cli/test_main.py b/test-cli/test_main.py index b657f00..8648b6b 100644 --- a/test-cli/test_main.py +++ b/test-cli/test_main.py @@ -10,8 +10,6 @@ from test.helpers.setup_xml import XMLSetup from test.runners.simple import SimpleTestRunner from test.tests.qbutton import Qbutton from test.helpers.syscmd import TestSysCommand -from test.helpers.syscmd import SysCommand -from test.tests.qiperf import QIperf from test.tests.qethernet import Qethernet from test.tests.qaudio import Qaudio from test.tests.qram import Qram @@ -28,7 +26,6 @@ from test.tests.qflash import Qflasher from test.tests.qnand import Qnand from test.helpers.globalVariables import globalVar import socket -import re from test.helpers.iseelogger import ISEE_Logger import logging @@ -43,6 +40,72 @@ def clear(): _ = call('clear' if os.name == 'posix' else 'cls') +def create_paramslist(params): + paramlist = {} + for row in params: + varname, varvalue = row + paramlist[varname] = varvalue + return paramlist + + +def add_test_task(suite, testdefname, paramlist): + testfunc = None + + if testdefname == "AUDIO": + suite.addTest(Qaudio(testdefname, "execute", paramlist)) + elif testdefname == "RAM": + suite.addTest(Qram(testdefname, "execute", paramlist)) + elif testdefname == "SERIALDUAL": + suite.addTest(Qduplex(testdefname, "execute", paramlist)) + elif testdefname == "EEPROM": + suite.addTest(Qeeprom(testdefname, "execute", paramlist)) + elif testdefname == "SERIAL": + suite.addTest(Qserial(testdefname, "execute", paramlist)) + # elif testdefname == "HDMI": + # suite.addTest(Qhdmi(testdefname, "execute", paramlist)) + elif testdefname == "BUTTON": + suite.addTest(Qbutton(testdefname, "execute", paramlist)) + elif testdefname == "RTC": + suite.addTest(Qrtc(testdefname, "execute", paramlist)) + elif testdefname == "CONSUMPTION": + suite.addTest(Qamp(testdefname, "execute", paramlist)) + # elif testdefname == "SATA": + # suite.addTest(Qsata(testdefname, "execute", paramlist)) + elif testdefname == "DMESG": + suite.addTest(TestSysCommand(testdefname, "execute", paramlist)) + elif testdefname == "ETHERNET": + suite.addTest(Qethernet(testdefname, "execute", paramlist)) + elif testdefname == "NAND": + suite.addTest(Qnand(testdefname, "execute", paramlist)) + elif testdefname == "I2C": + suite.addTest(Qi2c(testdefname, "execute", paramlist)) + elif testdefname == "WIFI": + suite.addTest(Qwifi(testdefname, "execute", paramlist)) + elif testdefname == "USB": + suite.addTest(Qusb(testdefname, "execute", paramlist)) + else: + raise Exception("Wrong testdefname") + + +def create_testsuite(): + # create an object TestSuite + suite = unittest.TestSuite() + # get list of tests for this board + tests = psdbObj.get_tests_list(globalVar.g_uuid) + # loop in every test for this board + for row in tests: + testid, testdefname = row + # get params for this test + params = psdbObj.get_test_params_list(testid) + paramlist = create_paramslist(params) + # add test to TestSuite + add_test_task(suite, testdefname, paramlist) + + globalVar.testid_ctl = psdbObj.open_test(globalVar.g_uuid) + + return suite + + def create_board(): model_id = xmlObj.gettagKey('board', 'model') variant = xmlObj.gettagKey('board', 'variant') @@ -56,46 +119,13 @@ def create_board(): processor_id = genDieid(globalVar.g_mid) print(globalVar.g_mid) print(processor_id) - globalVar.g_uuid = psdbObj.create_board(processor_id, model_id, variant, bmac=None) - - -def createvarlist(testvars): - varlist = {} - for row in testvars: - varname, testparam = row - varlist[varname] = testparam - return varlist - - -def testsuite(): - suite = unittest.TestSuite() - tests = psdbObj.getboard_comp_test_list(globalVar.g_uuid) - for row in tests: - testdefname, testfunc = row - testvars = psdbObj.getboard_test_variables(globalVar.g_uuid, testdefname) - varlist = createvarlist(testvars) - command = "suite.addTest({}('{}','execute', varlist))".format(testfunc, testdefname) - exec(command) - - globalVar.testid_ctl = psdbObj.open_testbatch(globalVar.g_uuid, globalVar.station) - return suite - - -def finish_test(): - psdbObj.close_testbatch(globalVar.g_uuid, globalVar.testid_ctl) - # Update Set Test status with FINISH so that status column is not modified because close_testbatch already did it. - psdbObj.update_set_test_row(globalVar.station, globalVar.testid_ctl, globalVar.g_uuid, 'END', 'FINISH') + globalVar.g_uuid = psdbObj.create_board(processor_id, model_id, variant, globalVar.station, bmac=None) def main(): create_board() - try: - os.remove("test_results.dat") - except: - pass runner = SimpleTestRunner(psdbObj) - runner.run(testsuite()) - finish_test() + runner.run(create_testsuite()) if __name__ == "__main__": |