Changes to adapt to new way to save results in DB. Created audio test. Added protections against unexpected status in DB.

1 files changed, 82 insertions, 47 deletions

diff --git a/test-cli/test_main.py b/test-cli/test_main.py
index 63d8dad..b32ad26 100644
--- a/test-cli/test_main.py
+++ b/test-cli/test_main.py
@@ -1,5 +1,5 @@
 from test.helpers.int_registers import get_die_id
-from test.helpers.int_registers import get_mac
+from test.helpers.int_registers import get_internal_mac
 from subprocess import call
 import os
 import unittest
@@ -9,7 +9,6 @@ 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.qi2c import Qi2c
 from test.tests.qeeprom import Qeeprom
 from test.tests.qserial import Qserial
 from test.tests.qwifi import Qwifi
@@ -17,6 +16,8 @@ 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
@@ -24,16 +25,17 @@ import logging
 from test.tasks.flasheeprom import flash_eeprom
 from test.tasks.flashmemory import flash_memory
 from test.helpers.qrreader import QRReader
-from test.tasks.generatedmesg import generate_dmesg
 from test.helpers.cmdline import LinuxKernel
 from test.helpers.amper import Amper
 from test.enums.StationStates import StationStates
+import time
 
 # global variables
 psdbObj = TestSrv_Database()
 xmlObj = None
 loggerObj = None
 test_abspath = None
+tests_manually_executed = []
 
 
 # define clear function
@@ -50,36 +52,50 @@ def create_paramslist(params):
     return paramlist
 
 
-def add_test_task(suite, testdefname, 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))
-    elif testdefname == "I2C":
-        suite.addTest(Qi2c(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 == "DMESG":
+        tests_manually_executed.append(Qdmesg(testdefname, "execute", paramlist))
+        return 0
     else:
-        raise Exception("Wrong testdefname")
+        return 1
 
 
 def create_testsuite():
     # create an object TestSuite
-    suite = unittest.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
@@ -93,10 +109,11 @@ def create_testsuite():
         paramlist["boarduuid"] = globalVar.g_uuid
         paramlist["testidctl"] = globalVar.testid_ctl
         paramlist["xml"] = xmlObj
+        paramlist["db"] = psdbObj
         # add test to TestSuite
-        add_test_task(suite, testdefname, paramlist)
+        rescode = add_test(suite1, testdefname, paramlist)
 
-    return suite
+    return suite1
 
 
 def create_board():
@@ -122,7 +139,7 @@ def create_board():
     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))
+                                            get_internal_mac(globalVar.g_mid))
     print(globalVar.g_uuid)
 
 
@@ -140,31 +157,38 @@ def main():
 
     # create a process
     globalVar.testid_ctl = psdbObj.open_test(globalVar.g_uuid)
-    # create and run tests according to the board type
-    runner = SimpleTestRunner(psdbObj)
     # Change state to "TESTS_RUNNING"
     if psdbObj.read_station_state(globalVar.station) == StationStates.STATION_ERROR.name:
         # Abort
-        print("Error: Wrong previous station state.")
+        print(
+            "Error: Wrong previous station state before changing to TESTS_RUNNING state. STATION_ERROR state unexpected.")
         exit(0)
     psdbObj.change_station_state(globalVar.station, StationStates.TESTS_RUNNING.name)
-    # Execute tests
-    testresult = runner.run(create_testsuite())
-    # save dmesg at the end of the tests
-    generate_dmesg(globalVar.testid_ctl, psdbObj)
 
-    # execute aditional tasks, only if the test was succesfull
+    # 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"
         if psdbObj.read_station_state(globalVar.station) == StationStates.STATION_ERROR.name:
             # Abort
-            print("Error: Wrong previous station state.")
+            print(
+                "Error: Wrong previous station state before changing to TESTS_OK state. STATION_ERROR state unexpected.")
             exit(0)
         psdbObj.change_station_state(globalVar.station, StationStates.TESTS_OK.name)
         # Change state to "EXTRATASKS_RUNNING"
         if psdbObj.read_station_state(globalVar.station) == StationStates.STATION_ERROR.name:
             # Abort
-            print("Error: Wrong previous station state.")
+            print(
+                "Error: Wrong previous station state before changing to EXTRATASKS_RUNNING state. STATION_ERROR state unexpected.")
             exit(0)
         psdbObj.change_station_state(globalVar.station, StationStates.EXTRATASKS_RUNNING.name)
         # create task control
@@ -192,28 +216,38 @@ def main():
 
         # 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"
             if psdbObj.read_station_state(globalVar.station) == StationStates.STATION_ERROR.name:
                 # Abort
-                print("Error: Wrong previous station state.")
+                print(
+                    "Error: Wrong previous station state before changing to WAITING_FOR_SCANNER state. STATION_ERROR state unexpected.")
                 exit(0)
             psdbObj.change_station_state(globalVar.station, StationStates.WAITING_FOR_SCANNER.name)
-            # get barcode using the 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()
+
+            # get barcode using the scanner, only if autotest is disabled
+            autotest = psdbObj.get_setup_variable("AUTOTEST_" + globalVar.station)
+            if autotest == "True":
+                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"
             if psdbObj.read_station_state(globalVar.station) == StationStates.STATION_ERROR.name:
                 # Abort
-                print("Error: Wrong previous station state.")
+                print(
+                    "Error: Wrong previous station state before changing to FINISHED state. STATION_ERROR state unexpected.")
                 exit(0)
             psdbObj.change_station_state(globalVar.station, StationStates.FINISHED.name)
         else:
@@ -224,10 +258,16 @@ def main():
         # Change state to "TESTS_FAILED"
         if psdbObj.read_station_state(globalVar.station) == StationStates.STATION_ERROR.name:
             # Abort
-            print("Error: Wrong previous station state.")
+            print(
+                "Error: Wrong previous station state before changing to TESTS_FAILED state. STATION_ERROR state unexpected.")
             exit(0)
         psdbObj.change_station_state(globalVar.station, StationStates.TESTS_FAILED.name)
 
+    # reset board if AUTOTEST is enabled
+    autotest = psdbObj.get_setup_variable("AUTOTEST_" + globalVar.station)
+    if autotest == "True":
+        os.system('reboot')
+
 
 if __name__ == "__main__":
     # Clear the shell screen
@@ -257,22 +297,17 @@ if __name__ == "__main__":
 
     # Check if current state is "WAIT_TEST_START"
     currentstate = psdbObj.read_station_state(globalVar.station)
-    if currentstate != StationStates.WAIT_TEST_START.name:
-        # Abort
-        print("Error: Wrong previous station state.")
-        exit(0)
+    while currentstate != StationStates.WAIT_TEST_START.name:
+        currentstate = psdbObj.read_station_state(globalVar.station)
+        if currentstate != StationStates.WAIT_TEST_START.name:
+            # Wait
+            print("Error: Wrong previous station state. WAIT_TEST_START state expected. Current state is: {}".format(currentstate))
+            print("Waiting for WAIT_TEST_START state")
+        time.sleep(5)
 
     # Change state to "TESTS_CHECKING_ENV"
     psdbObj.change_station_state(globalVar.station, StationStates.TESTS_CHECKING_ENV.name)
 
-    # Look for a sd or pendrive
-    # TODO
-    #if not disk_exists("/dev/mmcblk0") and not :
-        # abort
-        #print("error: cannot find a barcode scanner.")
-        #loggerObj.getlogger().info("station error: #cannot find a sd card or memory usb to load the test environment.#")
-        #exit(0)
-
     # Look for a barcode scanner
     qr = QRReader()
     if qr.openQR():