summaryrefslogtreecommitdiff
path: root/test-cli
diff options
context:
space:
mode:
authorHector Fernandez <hector@iatec.biz>2020-03-05 18:50:47 +0100
committerHector Fernandez <hector@iatec.biz>2020-03-05 18:50:47 +0100
commitdd9ffc52507c391271d0821636c683fd7562b46a (patch)
treeaf9cffa6c0be8606ea0bb3eda032a170d2b1b6d8 /test-cli
parenta615dac03a9ea7897bf3947ba8d31278b2cea121 (diff)
downloadboard-dd9ffc52507c391271d0821636c683fd7562b46a.zip
board-dd9ffc52507c391271d0821636c683fd7562b46a.tar.gz
board-dd9ffc52507c391271d0821636c683fd7562b46a.tar.bz2
Modified postgre functions.
Diffstat (limited to 'test-cli')
-rw-r--r--test-cli/test/helpers/testsrv_db.py155
-rw-r--r--test-cli/test/runners/simple.py6
-rw-r--r--test-cli/test/tests/qeeprom.py17
-rw-r--r--test-cli/test_main.py106
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__":