diff options
Diffstat (limited to 'test-cli/test/tests/qwifi.py')
| -rw-r--r-- | test-cli/test/tests/qwifi.py | 143 |
1 files changed, 70 insertions, 73 deletions
diff --git a/test-cli/test/tests/qwifi.py b/test-cli/test/tests/qwifi.py index 3dd9e38..4695041 100644 --- a/test-cli/test/tests/qwifi.py +++ b/test-cli/test/tests/qwifi.py @@ -1,9 +1,11 @@ import unittest -import sh -import re -import json import time - +import uuid +import iperf3 +import netifaces +from test.helpers.iw import iwScan +from test.helpers.utils import save_json_to_disk +from test.helpers.utils import station2Port class Qwifi(unittest.TestCase): __serverip = None @@ -19,81 +21,76 @@ class Qwifi(unittest.TestCase): 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.__xmlObj = varlist["xml"] + self.__QwifiName = varlist.get('name', 'qwifi') + self.__loops = varlist.get('loops', self.__xmlObj.getKeyVal(self.__QwifiName, "loops", "1")) + self.__port = varlist.get('port', self.__xmlObj.getKeyVal(self.__QwifiName, "port", "5000")) + self.__bw = varlist.get('bwexpected', self.__xmlObj.getKeyVal(self.__QwifiName, "bwexpected", "5.0")) + self.__serverip = varlist.get('serverip', self.__xmlObj.getKeyVal(self.__QwifiName, "serverip", "localhost")) + self.__duration = varlist.get('duration', self.__xmlObj.getKeyVal(self.__QwifiName, "duration", "10")) + self.__interface = varlist.get('interface', self.__xmlObj.getKeyVal(self.__QwifiName, "interface", "wlan0")) + self.__toPath = varlist.get('to', self.__xmlObj.getKeyVal(self.__QwifiName, "to", "/mnt/station_ramdisk")) + self.__retriesCount = varlist.get('retries', self.__xmlObj.getKeyVal(self.__QwifiName, "retries", "5")) + self.__waitRetryTime = varlist.get('wait_retry', self.__xmlObj.getKeyVal(self.__QwifiName, "wait_retry", "10")) + self.__wifi_res_file = varlist.get('wifi_res_file', self.__xmlObj.getKeyVal(self.__QwifiName, "wifi_res_file", "wifi_test_{}.json")) + self.__wifi_st_file = varlist.get('wifi_st_file', self.__xmlObj.getKeyVal(self.__QwifiName, "wifi_st_file", "wifi_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 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, 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/wifi-iperf3.json', 'w') as outfile: - json.dump(data, outfile, indent=4) - outfile.close() - self.__resultlist.append( - { - "description": "iperf3 output", - "filepath": "/mnt/station_ramdisk/wifi-iperf3.json", - "mimetype": "application/json" - } - ) + def checkInterface(self, reqInterface): + ifaces = netifaces.interfaces() + for t in ifaces: + if t == reqInterface: + return True + return False - # check if BW is in the expected range - if self.__bwreal < float(self.__bwexpected): - self.fail("failed: speed is lower than expected. Speed(Mbits/s): " + str(self.__bwreal)) - else: - self.fail("failed: could not complete iperf3 command") - else: - self.fail("failed: wlan0 interface doesn't have any ip address.") + def execute(self): + self.__resultlist = [] + stationList = {} + # check interfaces + if not self.checkInterface(self.__interface): + self.fail('Interface not found: {}'.format(self.__interface)) + # scan networks + scanner = iwScan(self.__interface) + if scanner.scan(): + stationList = scanner.getStations() + # check if we are connected + if not scanner.findConnected(): + self.fail('Not connected to test AP') + else: + strError = scanner.getLastError() + self.fail('qWifi scanner Execution error: {}'.format(strError)) + # 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_st_file), + description='wifi scan', + mime='application/json', + json_data=stationList, + result=self.__resultlist) + save_json_to_disk(filePath='{}/{}'.format(self.__toPath, self.__wifi_res_file), + description='wifi {} iperf3'.format(self.__interface), + mime='application/json', + json_data=result.json, + result=self.__resultlist) + return else: - self.fail("failed: could not complete ifconfig command.") + self.fail('bw fail: rx:{} tx:{}'.format(result.received_Mbps, result.sent_Mbps)) else: - self.fail("failed: wifi module is not connected to the router.") - else: - self.fail("failed: could not execute iw command") + count = count - 1 + time.sleep(int(self.__waitRetryTime)) + + self.fail('qWifi (max tries) Execution error: {}'.format(result.error)) def getresults(self): return self.__resultlist |