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
|
import unittest
import sh
import re
import json
class Qwifi(unittest.TestCase):
__serverip = None
__numbytestx = None
__bwexpected = None
__port = None
params = None
__bwreal = None
__resultlist = None # resultlist is a python list of python dictionaries
# varlist content: serverip, bwexpected, port
def __init__(self, testname, testfunc, varlist):
self.params = varlist
super(Qwifi, self).__init__(testfunc)
if "serverip" in varlist:
self.__serverip = varlist["serverip"]
else:
raise Exception('serverip param inside Qwifi have been be defined')
if "bwexpected" in varlist:
self.__bwexpected = varlist["bwexpected"]
else:
raise Exception('OKBW param inside Qwifi must be defined')
if "port" in varlist:
self.__port = varlist["port"]
else:
raise Exception('port param inside Qwifi must be defined')
self.__numbytestx = "10M"
self._testMethodDoc = testname
self.__resultlist = []
def execute(self):
# check if the board is connected to the router by wifi
p = sh.iw("wlan0", "link")
if p.exit_code == 0:
# get the first line of the output stream
out1 = p.stdout.decode('ascii').splitlines()[0]
if out1 != "Not connected.":
# check if the board has ip in the wlan0 interface
p = sh.ifconfig("wlan0")
if p.exit_code == 0:
# check if wlan0 has an IP
result = re.search(
'inet addr:(?!127\.0{1,3}\.0{1,3}\.0{0,2}1$)((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)',
p.stdout.decode('ascii'))
if result:
# execute iperf command against the server
try:
p = sh.iperf3("-c", self.__serverip, "-n", self.__numbytestx, "-f", "m", "-p", self.__port,
"-J", _timeout=20)
except sh.TimeoutException:
self.fail("failed: iperf timeout reached")
# 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('/tmp/wifi-iperf.json', 'w') as outfile:
json.dump(data, outfile, indent=4)
self.__resultlist.append(
{
"desc": "iperf3 output",
"data": "/tmp/wifi-iperf.json",
"type": "file"
}
)
# check if BW is in the expected range
if self.__bwreal < float(self.__bwexpected):
self.__resultlist.append(
{
"desc": "Test result",
"data": "FAILED: speed is lower than expected. Speed(Mbits/s): " + str(self.__bwreal),
"type": "string"
}
)
self.fail("failed: speed is lower than expected. Speed(Mbits/s): " + str(self.__bwreal))
else:
self.__resultlist.append(
{
"desc": "Test result",
"data": "FAILED: could not complete iperf3 command",
"type": "string"
}
)
self.fail("failed: could not complete iperf3 command")
else:
self.__resultlist.append(
{
"desc": "Test result",
"data": "FAILED: wlan0 interface doesn't have any ip address",
"type": "string"
}
)
self.fail("failed: wlan0 interface doesn't have any ip address.")
else:
self.__resultlist.append(
{
"desc": "Test result",
"data": "FAILED: could not complete ifconfig command",
"type": "string"
}
)
self.fail("failed: could not complete ifconfig command.")
else:
self.__resultlist.append(
{
"desc": "Test result",
"data": "FAILED: wifi module is not connected to the router",
"type": "string"
}
)
self.fail("failed: wifi module is not connected to the router.")
else:
self.__resultlist.append(
{
"desc": "Test result",
"data": "FAILED: could not execute iw command",
"type": "string"
}
)
self.fail("failed: could not execute iw command")
# Test successful
self.__resultlist.append(
{
"desc": "Test result",
"data": "OK",
"type": "string"
}
)
def getresults(self):
return self.__resultlist
|
