summaryrefslogtreecommitdiff
path: root/test-cli/test/tests/qusb.py
diff options
context:
space:
mode:
Diffstat (limited to 'test-cli/test/tests/qusb.py')
-rw-r--r--test-cli/test/tests/qusb.py139
1 files changed, 87 insertions, 52 deletions
diff --git a/test-cli/test/tests/qusb.py b/test-cli/test/tests/qusb.py
index 44490bc..4f76d0c 100644
--- a/test-cli/test/tests/qusb.py
+++ b/test-cli/test/tests/qusb.py
@@ -1,60 +1,95 @@
-from test.helpers.syscmd import SysCommand
+import shutil
import unittest
+import os.path
+import os
+import sh
+
+from test.helpers.md5 import md5_file
class Qusb(unittest.TestCase):
+ params = None
+ __resultlist = None # resultlist is a python list of python dictionaries
+ __QusbName = None
- def __init__(self, testname, testfunc, devLabel, numPorts):
+ def __init__(self, testname, testfunc, varlist):
+ self.params = varlist
super(Qusb, self).__init__(testfunc)
- self.__numPorts = numPorts
self._testMethodDoc = testname
- self.__devLabel = devLabel
- if testname=="USBOTG":
- self.__usbFileName = "/this_is_an_usb_otg"
- self.__usbtext = "USBOTG"
- elif testname=="SATA":
- self.__usbFileName = "/this_is_a_sata"
- self.__usbtext = "SATA"
- else:
- self.__usbFileName = "/this_is_an_usb_host"
- self.__usbtext = "USBHOST"
- self.__numUsbFail=[]
+ self.__xmlObj = varlist["xml"]
+
+ self.__QusbName = varlist.get('name', 'qusb')
+ self.__loops = varlist.get('loops', self.__xmlObj.getKeyVal(self.__QusbName, "loops", "3"))
+ self.__filepath_from = varlist.get('path', self.__xmlObj.getKeyVal(self.__QusbName, "path", "/root/hwtest-files"))
+ self.__file_source = varlist.get('file', self.__xmlObj.getKeyVal(self.__QusbName, "file", "usb-test.bin"))
+ self.__file_md5 = varlist.get('file_md5', self.__xmlObj.getKeyVal(self.__QusbName, "file_md5", "usb-test.bin.md5"))
+ self.__path_to = varlist.get('pathto', self.__xmlObj.getKeyVal(self.__QusbName, "pathto", "/mnt/test/disk2"))
+ self.__check_mounted = varlist.get('check_mounted', self.__xmlObj.getKeyVal(self.__QusbName, "check_mounted", ""))
+
+ self.__resultlist = []
+
+ def removefileIfExist(self, fname):
+ if os.path.exists(fname):
+ try:
+ os.remove(fname)
+ except OSError as error:
+ return False, str(error)
+ return True, ''
+
+ def checkfiles(self):
+ if not os.path.isfile('{}/{}'.format(self.__filepath_from, self.__file_source)):
+ return False, 'File test binary Not found {}/{}'.format(self.__filepath_from, self.__file_source)
+ if not os.path.isfile('{}/{}'.format(self.__filepath_from, self.__file_md5)):
+ return False, 'File test binary md5 Not found {}/{}'.format(self.__filepath_from, self.__file_md5)
+ if self.__check_mounted != '':
+ if not os.path.ismount('{}'.format(self.__check_mounted)):
+ return False, 'Required mounted failed: {}'.format(self.__path_to)
+ r, s = self.removefileIfExist('{}/{}'.format(self.__path_to, self.__file_source))
+ if not r:
+ return r, s
+ return True, ''
+
+ def getTestFile_md5(self, fname):
+ t = ''
+ with open('{}'.format(fname), 'r') as outfile:
+ t = outfile.read().rstrip("\n")
+ outfile.close()
+ return t
+
+ def ExecuteTranfer(self):
+ try:
+ originalMD5 = self.getTestFile_md5('{}/{}'.format(self.__filepath_from, self.__file_md5))
+ if originalMD5 == '':
+ self.fail('MD5 file {}/{} Not contain any md5'.format(self.__filepath_from, self.__file_md5))
+ # shutil.copy2('{}/{}'.format(self.__filepath_from, self.__file_source), '{}'.format(self.__path_to))
+ sh.cp('{}/{}'.format(self.__filepath_from, self.__file_source), '{}'.format(self.__path_to))
+ sh.sync()
+ md5val = md5_file('{}/{}'.format(self.__path_to, self.__file_source))
+ self.removefileIfExist('{}/{}'.format(self.__path_to, self.__file_source))
+ if originalMD5 != md5val:
+ return False, 'md5 check failed {}<>{}'.format(originalMD5, md5val)
+ except OSError as why:
+ return False,'Error: {} tranfer failed'.format(why.errno)
+ except sh.ErrorReturnCode as shError:
+ return False, 'USB Exception: tranfer failed'
+ except Exception as details:
+ return False, 'USB Exception: {} tranfer failed'.format(details)
+ return True, ''
def execute(self):
- str_cmd= "lsblk -o LABEL"
- lsblk_command = SysCommand("lsblk", str_cmd)
- if lsblk_command.execute() == 0:
- self.__raw_out = lsblk_command.getOutput()
- if self.__raw_out == "":
- return -1
- lines = lsblk_command.getOutput().splitlines()
- host_list=[]
- for i in range(len(lines)):
- if str(lines[i].decode('ascii'))==self.__devLabel:
- host_list.append(i)
- if len(host_list)==int(self.__numPorts):
- str_cmd = "lsblk -o MOUNTPOINT"
- lsblk_command = SysCommand("lsblk", str_cmd)
- if lsblk_command.execute() == 0:
- self.__raw_out = lsblk_command.getOutput()
- if self.__raw_out == "":
- print("failed: no command output")
- self.fail("failed: no command output")
- else:
- lines = lsblk_command.getOutput().splitlines()
- for i in range(len(host_list)):
- file_path=str(lines[host_list[i]].decode('ascii')) + self.__usbFileName
- usb_file = open(file_path, 'r')
- read=usb_file.read()
- if read.find(self.__usbtext)!=-1:
- print(file_path + " --> OK!")
- else:
- self.fail("failed: could not read from usb {}".format(lines[host_list[i]].decode('ascii')))
- self.__numUsbFail.append(host_list[i])
- usb_file.close()
- else:
- self.fail("failed: couldn't execute lsblk command")
-
- else:
- self.fail("failed: reference and real usb host devices number mismatch")
- else:
- self.fail("failed: couldn't execute lsblk command")
+ v, m = self.checkfiles()
+ if not v:
+ self.fail(m)
+ countLoops = int(self.__loops)
+ while countLoops > 0:
+ res, msg = self.ExecuteTranfer()
+ if not res:
+ res, msg = self.ExecuteTranfer()
+ if not res:
+ self.fail(msg)
+ countLoops = countLoops - 1
+
+ def getresults(self):
+ return self.__resultlist
+
+ def gettextresult(self):
+ return ""