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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
import unittest
import subprocess
import threading
class TestSysCommand(unittest.TestCase):
__str_cmd = None
__testname = None
__outfilename = None
__outdata = None
__outtofile = False
def __init__(self, testname, testfunc, str_cmd, outtofile=False):
""" init """
super(TestSysCommand, self).__init__(testfunc)
self.__str_cmd = str_cmd
self.__testname = testname
self.__outtofile = outtofile
self._testMethodDoc = testname
if self.__outtofile is True:
self.__outfilename = '/tmp/{}.txt'.format(testname)
def getName(self):
return self.__testname
def execute(self):
res = -1
try:
completed = subprocess.run(
self.__str_cmd,
check=True,
shell=True,
stdout=subprocess.PIPE,
)
self.assertTrue(completed.returncode is 0)
if completed.returncode is 0:
if self.__outtofile is True:
f = open(self.__outfilename, 'wb')
f.write(completed.stdout)
f.close()
res = 0
else:
res = -3
outdata = completed.stdout
self.longMessage = str(outdata).replace("'", "")
self.assertTrue(True)
except subprocess.CalledProcessError as err:
self.assertTrue(False)
res = -1
except Exception as t:
res = -2
return res
def remove_file(self):
pass
class SysCommand(object):
__str_cmd = None
__cmdname = None
__outdata = None
__errdata = None
__returnCode = None
__exception = None
def __init__(self, cmdname, str_cmd):
""" init """
self.__str_cmd = str_cmd
self.__cmdname = cmdname
def getName(self):
return self.__cmdname
def setName(self, cmdName):
self.__cmdname = cmdName
def getParams(self):
return self.__str_cmd
def setParam(self, params):
self.__str_cmd = params
def execute(self):
# try:
self.__outdata = None
self.__errdata = None
self.__exception = None
try:
completed = subprocess.run(
self.__str_cmd,
check=False,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
self.__returnCode = completed.returncode
self.__outdata = completed.stdout
self.__errdata = completed.stderr.decode('ascii')
except subprocess.CalledProcessError as err:
self.__exception = err
self.__returnCode = -1
return self.__returnCode
def getOutput(self):
return self.__outdata
def getOutErr(self):
return self.__errdata
def getException(self):
return self.__exception
def IsException(self):
return self.__exception is not None
def getOutputlines(self):
return self.__outdata.splitlines()
def save_file(self, fname):
f = open(fname, 'wb')
f.write(self.__outdata)
f.close()
class AsyncSys(threading.Thread):
sysObject = None
sysres = None
def __init__(self, threadID, name, counter):
threading.Thread.__init__(self)
self.threadID = threadID
self.name = name
self.counter = counter
self.sysObject = SysCommand(name, None)
def setParams(self, Params):
self.sysObject.setParam(Params)
def getRes(self):
return self.sysres
def getData(self):
return self.sysObject.getOutput()
def run(self):
self.sysres = self.sysObject.execute()
|
