from test.helpers.int_registers import get_die_id from test.helpers.int_registers import get_mac from subprocess import call import os import unittest from test.helpers.testsrv_db import TestSrv_Database from test.helpers.setup_xml import XMLSetup from test.runners.simple import SimpleTestRunner from test.helpers.syscmd import TestSysCommand from test.tests.qethernet import Qethernet from test.tests.qaudio import Qaudio from test.tests.qram import Qram from test.tests.qusb import Qusb from test.tests.qi2c import Qi2c from test.tests.qeeprom import Qeeprom from test.tests.qserial import Qserial from test.tests.qwifi import Qwifi from test.tests.qrtc import Qrtc from test.tests.qduplex_ser import Qduplex from test.tests.qamper import Qamper from test.tests.qnand import Qnand from test.helpers.globalVariables import globalVar import socket from test.helpers.iseelogger import ISEE_Logger import logging from test.flashers.flasheeprom import flash_eeprom from test.flashers.flashmemory import flash_memory from test.helpers.qrreader import QRReader # global variables psdbObj = TestSrv_Database() xmlObj = None loggerObj = None test_abspath = None # define clear function def clear(): # check and make call for specific operating system _ = 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): 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 == "RTC": suite.addTest(Qrtc(testdefname, "execute", paramlist)) elif testdefname == "CONSUMPTION": suite.addTest(Qamper(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 the testid as parameter paramlist["testid"] = testid paramlist["boarduuid"] = globalVar.g_uuid paramlist["testidctl"] = globalVar.testid_ctl # add test to TestSuite add_test_task(suite, testdefname, paramlist) return suite def create_board(): model_id = xmlObj.gettagKey('board', 'model') variant = xmlObj.gettagKey('board', 'variant') # get model id globalVar.g_mid = model_id + "-" + variant # get station globalVar.station = socket.gethostname() processor_id = get_die_id(globalVar.g_mid) print(globalVar.g_mid) print(processor_id) globalVar.g_uuid = psdbObj.create_board(processor_id, model_id, variant, globalVar.station, get_mac(globalVar.g_mid)) print(globalVar.g_uuid) psdbObj.bond_to_station(globalVar.g_uuid, globalVar.station) def get_taskvars_list(uuid): varlist = {} for row in psdbObj.get_task_variables_list(uuid): varname, varvalue = row varlist[varname] = varvalue return varlist def main(): # initialize the board create_board() # create a process globalVar.testid_ctl = psdbObj.open_test(globalVar.g_uuid) # create and run tests according to the board type runner = SimpleTestRunner(psdbObj) loggerObj.getlogger().info("Tests running") testresult = runner.run(create_testsuite()) # execute aditional tasks, only if the test was succesfull if testresult.wasSuccessful(): loggerObj.getlogger().info("Extra tasks running") # create task control globalVar.taskid_ctl = psdbObj.open_task(globalVar.g_uuid) # get extra variables varlist = get_taskvars_list(globalVar.g_uuid) alltasksok = False # flash eeprom resulteeprom = 0 if "eeprompath" in varlist and len(varlist["eeprompath"]) > 0: mac0 = psdbObj.get_board_macaddr(globalVar.g_uuid) resulteeprom, eepromdata = flash_eeprom(varlist["eeprompath"], globalVar.g_uuid, mac0) psdbObj.create_task_result(globalVar.taskid_ctl, "FLASHEEPROM", "TASK_OK" if resulteeprom == 0 else "TASK_FAIL", eepromdata) # flash non-volatile memory resultmemory = 0 if "image" in varlist and len(varlist["image"]) > 0: resultmemory = flash_memory(varlist["image"]) psdbObj.create_task_result(globalVar.taskid_ctl, "FLASHMEMORY", "TASK_OK" if resultmemory == 0 else "TASK_FAIL", varlist["image"]) # update status with the result if resulteeprom == 0 and resultmemory == 0: alltasksok = True psdbObj.update_taskctl_status(globalVar.taskid_ctl, "TASK_BOARD_OK") else: psdbObj.update_taskctl_status(globalVar.taskid_ctl, "TASK_BOARD_FAIL") if alltasksok: # get barcode using the scanner loggerObj.getlogger().info("Waiting for barcode scanner") qrreceived = False while not qrreceived: qr = QRReader() if qr.openQR(): # waits 5s to receive a valid code if qr.readQRasync(5): qrreceived = True factorycode = qr.getQRNumber() psdbObj.set_factorycode(globalVar.g_uuid, factorycode) qr.closeQR() loggerObj.getlogger().info("Python program finished") if __name__ == "__main__": # Clear the shell screen clear() test_abspath = os.path.dirname(os.path.abspath(__file__)) print(test_abspath) # create logger loggerObj = ISEE_Logger(logging.INFO) loggerObj.getlogger().info("Python program started") # Try to parse the setup.xml file try: xmlObj = XMLSetup(os.path.join(test_abspath, "setup.xml")) except: print("Error: Cannot parse setup.xml file") exit(1) # Try to connect to the DB, according to setup.xml configuration if psdbObj.open(xmlObj): main() else: print("Error: Cannot open DB connection") exit(2)