Merge branch 'sopa-test'rel.0.1sopa-test

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 ""