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
|
import unittest
import sh
import json
import time
import uuid
import iperf3
import netifaces
from test.helpers.utils import save_json_to_disk
from test.helpers.utils import station2Port
class Qethernet(unittest.TestCase):
__serverip = None
__numbytestx = None
__bwexpected = None
__port = None
params = None
__bwreal = None
__resultlist = None # resultlist is a python list of python dictionaries
timebetweenattempts = 5
# varlist content: serverip, bwexpected, port
def __init__(self, testname, testfunc, varlist):
self.params = varlist
super(Qethernet, self).__init__(testfunc)
self._testMethodDoc = testname
self.__xmlObj = varlist["xml"]
self.__QEthName = varlist.get('name', 'qeth')
self.__loops = varlist.get('loops', self.__xmlObj.getKeyVal(self.__QEthName, "loops", "1"))
self.__port = varlist.get('port', self.__xmlObj.getKeyVal(self.__QEthName, "port", "5000"))
self.__bw = varlist.get('bwexpected', self.__xmlObj.getKeyVal(self.__QEthName, "bwexpected", "40.0"))
self.__serverip = varlist.get('serverip', self.__xmlObj.getKeyVal(self.__QEthName, "serverip", "localhost"))
self.__duration = varlist.get('duration', self.__xmlObj.getKeyVal(self.__QEthName, "duration", "10"))
self.__interface = varlist.get('interface', self.__xmlObj.getKeyVal(self.__QEthName, "interface", "eth0"))
self.__toPath = varlist.get('to', self.__xmlObj.getKeyVal(self.__QEthName, "to", "/mnt/station_ramdisk"))
self.__retriesCount = varlist.get('retries', self.__xmlObj.getKeyVal(self.__QEthName, "retries", "5"))
self.__waitRetryTime = varlist.get('wait_retry', self.__xmlObj.getKeyVal(self.__QEthName, "wait_retry", "10"))
self.__wifi_res_file = varlist.get('eth_res_file', self.__xmlObj.getKeyVal(self.__QEthName, "eth_res_file", "eth_test_{}.json"))
self.__wifi_st_file = varlist.get('eth_st_file', self.__xmlObj.getKeyVal(self.__QEthName, "eth_st_file", "eth_st_{}.json"))
self.__file_uuid = uuid.uuid4()
self.__wifi_res_file = self.__wifi_res_file.format(self.__file_uuid)
self.__wifi_st_file = self.__wifi_st_file.format(self.__file_uuid)
self.__resultlist = []
def checkInterface(self, reqInterface):
ifaces = netifaces.interfaces()
for t in ifaces:
if t == reqInterface:
return True
return False
def execute(self):
self.__resultlist = []
# check interfaces
if not self.checkInterface(self.__interface):
self.fail('Interface not found: {}'.format(self.__interface))
# Start Iperf
client = iperf3.Client()
client.duration = int(self.__duration)
client.server_hostname = self.__serverip
client.port = station2Port(self.__port)
count = int(self.__retriesCount)
while count > 0:
result = client.run()
if result.error == None:
if result.received_Mbps >= float(self.__bw) and result.sent_Mbps >= float(self.__bw):
save_json_to_disk(filePath='{}/{}'.format(self.__toPath, self.__wifi_res_file),
description='eth {} iperf3'.format(self.__interface),
mime='application/json',
json_data=result.json,
result=self.__resultlist)
return
else:
self.fail('bw fail: rx:{} tx:{}'.format(result.received_Mbps, result.sent_Mbps))
else:
count = count - 1
time.sleep(int(self.__waitRetryTime))
self.fail('qEth (max tries) Execution error: {}'.format(result.error))
def execute2(self):
# execute iperf command against the server, but it implements attempts in case the server is busy
iperfdone = False
while not iperfdone:
try:
p = sh.iperf3("-c", self.__serverip, "-n", self.__numbytestx, "-f", "m", "-p", self.__port, "-J",
_timeout=20)
iperfdone = True
except sh.TimeoutException:
self.fail("failed: iperf timeout reached")
except sh.ErrorReturnCode:
time.sleep(self.timebetweenattempts)
# check if it was executed succesfully
if p.exit_code == 0:
if p.stdout == "":
self.fail("failed: error executing iperf command")
# analyze output string
data = json.loads(p.stdout.decode('ascii'))
self.__bwreal = float(data['end']['sum_received']['bits_per_second'])/1024/1024
# save result file
with open('/mnt/station_ramdisk/ethernet-iperf3.json', 'w') as outfile:
json.dump(data, outfile, indent=4)
outfile.close()
self.__resultlist.append(
{
"description": "iperf3 output",
"filepath": "/mnt/station_ramdisk/ethernet-iperf3.json",
"mimetype": "application/json"
}
)
# check if BW is in the expected range
if self.__bwreal < float(self.__bwexpected):
self.fail("failed: speed is lower than spected. Speed(Mbits/s): " + str(self.__bwreal))
else:
self.fail("failed: could not complete iperf command.")
def getresults(self):
return self.__resultlist
def gettextresult(self):
return ""
|
