from test.helpers.int_registers import get_die_id from test.helpers.int_registers import get_internal_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.tests.qethernet import Qethernet from test.tests.qram import Qram from test.tests.qusb import Qusb from test.tests.qeeprom import Qeeprom from test.tests.qserial import Qserial from test.tests.qusbdual import Qusbdual 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.tests.qaudio import Qaudio from test.tests.qdmesg import Qdmesg from test.helpers.globalVariables import globalVar import socket from test.helpers.iseelogger import ISEE_Logger import logging from test.tasks.flasheeprom import flash_eeprom from test.tasks.flashmemory import flash_memory from test.helpers.qrreader import QRReader from test.helpers.cmdline import LinuxKernel from test.helpers.amper import Amper from test.enums.StationStates import StationStates import time import sh # global variables psdbObj = TestSrv_Database() xmlObj = None loggerObj = None test_abspath = None tests_manually_executed = [] # define clear function def clear(): # check and make call for specific operating system _ = call('clear' if os.name == 'posix' else 'cls') def execute_station_error(text): print("Error: {}".format(text)) loggerObj.getlogger().info("Station error: #{}#".format(text)) while True: currentstate = psdbObj.read_station_state(globalVar.station) if currentstate == StationStates.STATION_REBOOT.name: reboot_board() time.sleep(1) def check_first_state(): currentstate = psdbObj.read_station_state(globalVar.station) if currentstate == StationStates.STATION_ERROR.name: while True: currentstate = psdbObj.read_station_state(globalVar.station) if currentstate == StationStates.STATION_REBOOT.name: reboot_board() time.sleep(1) elif currentstate == StationStates.STATION_REBOOT.name: reboot_board() else: starttime = time.time() while currentstate != StationStates.WAIT_TEST_START.name: if time.time() - starttime >= 30.0: # Error execute_station_error("Timeout while waiting for WAIT_TEST_START state") else: time.sleep(5) currentstate = psdbObj.read_station_state(globalVar.station) if currentstate == StationStates.STATION_ERROR.name: while True: currentstate = psdbObj.read_station_state(globalVar.station) if currentstate == StationStates.STATION_REBOOT.name: reboot_board() time.sleep(1) elif currentstate == StationStates.STATION_REBOOT.name: reboot_board() def reboot_board(): sh.reboot() exit(0) def set_next_state(newstate): statewritten = psdbObj.change_station_state(globalVar.station, newstate) if statewritten == StationStates.STATION_REBOOT.name: reboot_board() elif statewritten == StationStates.STATION_ERROR.name: while True: currentstate = psdbObj.read_station_state(globalVar.station) if currentstate == StationStates.STATION_REBOOT.name: reboot_board() time.sleep(1) def reboot_if_autotest(): # reset board if AUTOTEST is enabled autotest = psdbObj.get_setup_variable("AUTOTEST_" + globalVar.station) if int(autotest) == 1: reboot_board() def create_paramslist(params): paramlist = {} for row in params: varname, varvalue = row paramlist[varname] = varvalue return paramlist def add_test(suite, testdefname, paramlist): if testdefname == "RAM": suite.addTest(Qram(testdefname, "execute", paramlist)) return 0 elif testdefname == "SERIALDUAL": suite.addTest(Qduplex(testdefname, "execute", paramlist)) return 0 elif testdefname == "EEPROM": suite.addTest(Qeeprom(testdefname, "execute", paramlist)) return 0 elif testdefname == "SERIAL": suite.addTest(Qserial(testdefname, "execute", paramlist)) return 0 elif testdefname == "RTC": suite.addTest(Qrtc(testdefname, "execute", paramlist)) return 0 elif testdefname == "CONSUMPTION": suite.addTest(Qamper(testdefname, "execute", paramlist)) return 0 elif testdefname == "ETHERNET": suite.addTest(Qethernet(testdefname, "execute", paramlist)) return 0 elif testdefname == "NAND": suite.addTest(Qnand(testdefname, "execute", paramlist)) return 0 elif testdefname == "WIFI": suite.addTest(Qwifi(testdefname, "execute", paramlist)) return 0 elif testdefname == "USB": suite.addTest(Qusb(testdefname, "execute", paramlist)) return 0 elif testdefname == "AUDIO": suite.addTest(Qaudio(testdefname, "execute", paramlist)) return 0 elif testdefname == "USBLOOP": suite.addTest(Qusbdual(testdefname, "execute", paramlist)) return 0 elif testdefname == "DMESG": tests_manually_executed.append(Qdmesg(testdefname, "execute", paramlist)) return 0 else: return 1 def create_testsuite(): # create an object TestSuite suite1 = 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 paramlist["xml"] = xmlObj paramlist["db"] = psdbObj # add test to TestSuite add_test(suite1, testdefname, paramlist) return suite1 def create_board(): cmd = LinuxKernel() model_id = cmd.getkvar("bmodel", "none") variant = cmd.getkvar("bvariant", "none") if model_id == "none" or variant == "none": # Error execute_station_error("Cannot get model and variant information") # get model id globalVar.g_mid = model_id + "-" + variant # get station if "Station" not in globalVar.station: # Error execute_station_error("Wrong station name") 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_internal_mac(globalVar.g_mid)) print(globalVar.g_uuid) 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) # Change state to "TESTS_RUNNING" set_next_state(StationStates.TESTS_RUNNING.name) # generate suits suite1 = create_testsuite() # Execute tests (round 1) runner1 = SimpleTestRunner(psdbObj) testresult = runner1.run(suite1) # Execute manually tests for test in tests_manually_executed: test.execute() # execute aditional tasks, only if the test was successful if testresult.wasSuccessful(): # Change state to "TESTS_OK" set_next_state(StationStates.TESTS_OK.name) # Change state to "EXTRATASKS_RUNNING" set_next_state(StationStates.EXTRATASKS_RUNNING.name) # 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 "flashimagepath" in varlist and len(varlist["flashimagepath"]) > 0: resultmemory = flash_memory(varlist["flashimagepath"]) psdbObj.create_task_result(globalVar.taskid_ctl, "FLASHMEMORY", "TASK_OK" if resultmemory == 0 else "TASK_FAIL", varlist["flashimagepath"]) # update status with the result if resulteeprom == 0 and resultmemory == 0: # finish tasks psdbObj.update_taskctl_status(globalVar.taskid_ctl, "TASK_BOARD_OK") # Change state to "WAITING_FOR_SCANNER" set_next_state(StationStates.WAITING_FOR_SCANNER.name) # get barcode using the scanner, only if autotest is disabled autotest = psdbObj.get_setup_variable("AUTOTEST_" + globalVar.station) if int(autotest) == 1: psdbObj.set_factorycode(globalVar.g_uuid, "1234567890") # fake factory code else: 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() # Change state to "FINISHED" set_next_state(StationStates.FINISHED.name) else: psdbObj.update_taskctl_status(globalVar.taskid_ctl, "TASK_BOARD_FAIL") loggerObj.getlogger().info("Station error: #Unable to complete extra tasks#") else: # Change state to "TESTS_FAILED" set_next_state(StationStates.TESTS_FAILED.name) # reset board if AUTOTEST is enabled reboot_if_autotest() if __name__ == "__main__": # Clear the shell screen clear() test_abspath = os.path.dirname(os.path.abspath(__file__)) # create logger loggerObj = ISEE_Logger(logging.INFO) # Try to parse the setup.xml file try: xmlObj = XMLSetup(os.path.join(test_abspath, "setup.xml")) except: # Error execute_station_error("Cannot parse setup.xml file") # Try to connect to the DB, according to setup.xml configuration if not psdbObj.open(xmlObj): # Error execute_station_error("Cannot open DB connection") # get station name globalVar.station = socket.gethostname() # Check if current state is "WAIT_TEST_START". if not, waits here check_first_state() # Change state to "TESTS_CHECKING_ENV" set_next_state(StationStates.TESTS_CHECKING_ENV.name) # Wait before beginning time.sleep(2) # Look for a barcode scanner qr = QRReader() if qr.openQR(): qr.closeQR() else: # Error execute_station_error("Cannot find a barcode scanner") # Look for an amperimeter amp = Amper() if not amp.open(): # Error execute_station_error("Cannot open an amperimeter COM port") if not amp.hello(): if not amp.hello(): # Error execute_station_error("Cannot open find an amperimeter") # Execute main main() exit(0)