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