Merge branch 'sopa-test'rel.0.1sopa-test

1 files changed, 87 insertions, 33 deletions

diff --git a/test-cli/test/tests/qwifi.py b/test-cli/test/tests/qwifi.py
index d08149b..4695041 100644
--- a/test-cli/test/tests/qwifi.py
+++ b/test-cli/test/tests/qwifi.py
@@ -1,45 +1,99 @@
-from test.helpers.syscmd import SysCommand
 import unittest
-import subprocess
+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
+    __numbytestx = None
+    __bwexpected = None
+    __port = None
+    params = None
+    __bwreal = None
+    __resultlist = None  # resultlist is a python list of python dictionaries
+    timebetweenattempts = 5
 
-    def __init__(self, testname, testfunc, signal):
+    # varlist content: serverip, bwexpected, port
+    def __init__(self, testname, testfunc, varlist):
+        self.params = varlist
         super(Qwifi, self).__init__(testfunc)
-        self.__signal = signal
         self._testMethodDoc = testname
-        # WiFi SERVERIP fixed at the moment.
-        self._serverip = "192.168.5.1"
+        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 checkInterface(self, reqInterface):
+        ifaces = netifaces.interfaces()
+        for t in ifaces:
+            if t == reqInterface:
+                return True
+        return False
 
     def execute(self):
-        # First check connection with the wifi server using ping command
-        #str_cmd = "ping -c 1 {} > /dev/null".format(self._serverip)
-        #wifi_ping = SysCommand("wifi_ping", str_cmd)
-        #wifi_ping.execute()
-        #res = subprocess.call(['ping', '-c', '1', self._serverip])
-        p = subprocess.Popen(['ping','-c','1',self._serverip,'-W','2'],stdout=subprocess.PIPE)
-        p.wait()
-        res=p.poll()
-        if res == 0:
-            str_cmd= "iw wlan0 link"
-            wifi_stats = SysCommand("wifi_stats", str_cmd)
-            if wifi_stats.execute() == 0:
-                w_stats = wifi_stats.getOutput().decode('ascii').splitlines()
-                #self._longMessage = str(self.__raw_out).replace("'", "")
-                if not w_stats[0] == "Not connected.":
-                    signal_st = w_stats[5].split(" ")[1]
-                    try:
-                        int(signal_st)
-                        if -1*int(signal_st) > int(self.__signal):
-                            self.fail("failed: signal to server lower than expected")
-                    except ValueError:
-                        self.fail("failed: error output (Bad connection?)")
+        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: error output (Bad connection?)")
-                #tx_brate = float(w_stats[6].split(" ")[2])
+                    self.fail('bw fail: rx:{} tx:{}'.format(result.received_Mbps, result.sent_Mbps))
             else:
-                self.fail("failed: couldn't execute iw command")
-        else:
-            self.fail("failed: ping to server {}".format(self._serverip))
+                count = count - 1
+                time.sleep(int(self.__waitRetryTime))
+
+        self.fail('qWifi (max tries) Execution error: {}'.format(result.error))
 
+    def getresults(self):
+        return self.__resultlist
 
+    def gettextresult(self):
+        return ""