summaryrefslogtreecommitdiff
path: root/test-cli/test/tests/qusb.py
blob: 3182685160daf4a4808f3fd7af4252331f20a051 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
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, varlist):
        self.params = varlist
        super(Qusb, self).__init__(testfunc)
        self._testMethodDoc = testname
        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 Exception as details:
            return False, 'USB Exception: {} tranfer failed'.format(details)
        return True, ''

    def execute(self):
        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 ""