Merge branch 'WIFI-1321-create-a-lan-forge-pip-module' of https://github.com/Telecominfraproject/wlan-lanforge-scripts into WIFI-1321-create-a-lan-forge-pip-module

lf_libs/lf_libs.py

@@ -1517,6 +1517,43 @@ class lf_libs:
         atten_obj = Attenuator_modify(self.manager_ip, self.manager_http_port, serno, idx, val)
         atten_obj.build()
 
+    def attenuator_serial_radio(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", mode="BRIDGE", atn_val=400,
+                                   vlan_id=100, sta_mode=0, station_name=[], lf_tools_obj=None, radio='1.1.wiphy0'):
+        radio = radio
+        # index 0 of atten_serial_radio will ser no of 1st 2g radio and index 1 will ser no of 2nd and 3rd 2g radio
+        atten_serial_radio = []
+        atten_serial = self.attenuator_serial()
+        self.Client_Connect_Using_Radio(ssid=ssid, passkey=passkey, security=security, mode=mode,
+                                        vlan_id=vlan_id, radio=radio, sta_mode=sta_mode, station_name=station_name)
+        signal1 = self.json_get(_req_url=f'/port/1/1/{station_name[0]}?fields=signal')['interface']['signal']
+        atten_sr = atten_serial[0].split(".")
+        for i in range(4):
+            self.attenuator_modify(int(atten_sr[2]), i, atn_val)
+            time.sleep(0.5)
+        signal2 = self.json_get(_req_url=f'/port/1/1/{station_name[0]}?fields=signal')['interface']['signal']
+        if abs(int(signal2.split(" ")[0])) - abs(int(signal1.split(" ")[0])) >= 5:
+            atten_serial_radio = atten_serial
+        else:
+            atten_serial_radio = atten_serial[::-1]
+        self.client_disconnect(station_name=station_name)
+        return atten_serial_radio
+
+    def read_kpi_file(self, column_name, dir_name):
+        if column_name == None:
+            df = pd.read_csv("../reports/" + str(dir_name) + "/kpi.csv", sep=r'\t', engine='python')
+            if df.empty == True:
+                return "empty"
+            else:
+                return df
+        else:
+            df = pd.read_csv("../reports/" + str(dir_name) + "/kpi.csv", sep=r'\t', usecols=column_name,
+                             engine='python')
+            if df.empty == True:
+                return "empty"
+            else:
+                result = df[column_name].values.tolist()
+                return result
+
     def read_csv_individual_station_throughput(self, dir_name, option, individual_station_throughput=True,
                                                kpi_csv=False,
                                                file_name="/csv-data/data-Combined_bps__60_second_running_average-1.csv",
@@ -1614,8 +1651,7 @@ class lf_libs:
             allure.attach(name=name, body=str(data_table))
 
     def client_connect_using_radio(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", mode="BRIDGE",
-                                   vlan_id=100, radio=None, sta_mode=0,
+                                   vlan_id=100, radio=None, sta_mode=0, station_name=[]):
-                                   station_name=[]):
         self.client_connect = CreateStation(_host=self.manager_ip, _port=self.manager_http_port, _mode=sta_mode,
                                             _sta_list=station_name, _password=passkey, _ssid=ssid, _security=security)
 

lf_libs/lf_tests.py

@@ -52,6 +52,8 @@ lf_dataplane_test = importlib.import_module("py-scripts.lf_dataplane_test")
 DataplaneTest = lf_dataplane_test.DataplaneTest
 ttlstest = importlib.import_module("py-scripts.test_ipv4_ttls")
 TTLSTest = ttlstest.TTLSTest
+tr398v2test = importlib.import_module("py-scripts.lf_tr398v2_test")
+TR398v2Test = tr398v2test.TR398v2Test
 
 
 class lf_tests(lf_libs):
@@ -1385,6 +1387,7 @@ class lf_tests(lf_libs):
                 self.client_disconnect(clear_all_sta=True, clean_l3_traffic=True)
             except Exception as e:
                 logging.error(f"{e}")
+                return False, f"{e}"
 
     def country_code_channel_division(self, ssid="[BLANK]", passkey='[BLANK]', security="wpa2", mode="BRIDGE",
                                       band='twog', num_sta=1, vlan_id=100, channel='1', channel_width=20,
@@ -1427,6 +1430,128 @@ class lf_tests(lf_libs):
                 logging.error(f"{e}")
                 return False
 
+    def tr398(self,radios_2g=[], radios_5g=[], radios_ax=[], dut_name="TIP", dut_5g="", dut_2g="", mode="BRIDGE",
+              vlan_id=1, skip_2g=True, skip_5g=False, instance_name="", test=None, move_to_influx=False,dut_data={},
+              ssid_name='', security_key='[BLANK]', security="open"):
+        #User can select one or more TR398 tests
+        try:
+            if type(test) == str:
+                test = test.split(",")
+            self.client_disconnect(clean_l3_traffic=True)
+            raw_line = []
+            skip_twog, skip_fiveg = '1' if skip_2g else '0', '1' if skip_5g else '0'
+            if mode == "BRIDGE" or mode == "NAT":
+                upstream_port = list(self.lanforge_data['wan_ports'].keys())[0]
+            else:
+                upstream_port = list(self.lanforge_data['wan_ports'].keys())[0] + "." + str(vlan_id)
+            atten_serial = self.attenuator_serial_radio(ssid=ssid_name, passkey=security_key, security=security, sta_mode=0,
+                                                        station_name=['sta0000'])
+
+            enable_tests = [['rxsens: 0'], ['max_cx: 0'], ['max_tput: 0'], ['peak_perf: 0'], ['max_tput_bi: 0'],
+                            ['dual_band_tput: 0'],
+                            ['multi_band_tput: 0'], ['atf: 0'], ['atf3: 0'], ['qos3: 0'], ['lat3: 0'], ['mcast3: 0'],
+                            ['rvr: 0'],
+                            ['spatial: 0'], ['multi_sta: 0'], ['reset: 0'], ['mu_mimo: 0'], ['stability: 0'],
+                            ['ap_coex: 0'], ['acs: 0']]
+
+            rad_atten = [[f'atten-0: {atten_serial[0]}.0'], [f'atten-1: {atten_serial[0]}.1'], [f'atten-2: {atten_serial[0]}.2'],
+                         [f'atten-3: {atten_serial[0]}.3'], [f'atten-4: {atten_serial[1]}.0'], [f'atten-5: {atten_serial[1]}.1'],
+                         [f'atten-8: {atten_serial[1]}.2'], [f'atten-9: {atten_serial[1]}.3']]
+
+            skip_band = [['Skip 2.4Ghz Tests', f'{skip_twog}'], ['Skip 5Ghz Tests', f'{skip_fiveg}'],
+                         ['2.4Ghz Channel', 'AUTO'], ['5Ghz Channel', 'AUTO'], ['Skip N/AC Tests', '1'],
+                         ['Skip AX Tests', '1']]
+            for t in test:
+                if [f"{t}: 0"] in enable_tests:
+                    enable_tests[enable_tests.index([f"{t}: 0"])] = [f"{t}: 1"]
+            if len(radios_2g) >= 3 and len(radios_5g) >= 3:
+                for i in range(6):
+                    if i == 0 or i == 2:
+                        raw_line.append([f'radio-{i}: {radios_5g[0] if i == 0 else radios_5g[1]}'])
+                    if i == 1 or i == 3:
+                        raw_line.append([f'radio-{i}: {radios_2g[0] if i == 1 else radios_2g[1]}'])
+                    if i == 4 or i == 5:
+                        raw_line.append([f'radio-{i}: {radios_5g[2] if i == 4 else radios_2g[2]}'])
+            elif len(radios_2g) >= 2 and len(radios_5g) >= 2 and len(radios_ax) >= 2:
+                for i in range(6):
+                    if i == 0 or i == 2:
+                        raw_line.append([f'radio-{i}: {radios_5g[0] if i == 0 else radios_5g[1]}'])
+                    if i == 1 or i == 3:
+                        raw_line.append([f'radio-{i}: {radios_2g[0] if i == 1 else radios_2g[1]}'])
+                    if i == 4 or i == 5:
+                        raw_line.append([f'radio-{i}: {radios_ax[0] if i == 4 else radios_ax[1]}'])
+
+            if len(raw_line) != 6:
+                raw_line = [['radio-0: 1.1.5 wiphy1'], ['radio-1: 1.1.4 wiphy0'], ['radio-2: 1.1.7 wiphy3'],
+                            ['radio-3: 1.1.6 wiphy2'], ['radio-4: 1.1.8 wiphy4'], ['radio-5: 1.1.9 wiphy5']]
+            raw_line.extend(enable_tests + rad_atten)
+            self.update_dut_ssid(dut_data=dut_data)
+            instance_name = "tr398-instance-{}".format(str(random.randint(0, 100000)))
+
+            # if not os.path.exists("tr398-test-config.txt"):
+            with open("tr398-test-config.txt", "wt") as f:
+                for i in raw_line:
+                    f.write(str(i[0]) + "\n")
+                f.close()
+
+            self.cvtest_obj = TR398v2Test(lf_host=self.manager_ip,
+                                        lf_port=self.manager_http_port,
+                                        lf_user="lanforge",
+                                        lf_password="lanforge",
+                                        instance_name=instance_name,
+                                        config_name="cv_dflt_cfg",
+                                        upstream=upstream_port,
+                                        pull_report=True,
+                                        local_lf_report_dir=self.local_report_path,
+                                        load_old_cfg=False,
+                                        dut2=dut_2g,
+                                        dut5=dut_5g,
+                                        raw_lines_file="tr398-test-config.txt",
+                                        enables=[],
+                                        disables=[],
+                                        raw_lines=[],
+                                        sets=skip_band,
+                                        test_rig=dut_name
+                                        )
+            self.cvtest_obj.test_name = "TR-398 Issue 2"
+            self.cvtest_obj.result = True
+            self.cvtest_obj.setup()
+            self.cvtest_obj.run()
+            if os.path.exists("tr398-test-config.txt"):
+                os.remove("tr398-test-config.txt")
+
+            if move_to_influx:
+                try:
+                    report_name = "../reports/" + self.cvtest_obj.report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1]
+                    influx = CSVtoInflux(influx_host=self.influx_params["influx_host"],
+                                         influx_port=self.influx_params["influx_port"],
+                                         influx_org=self.influx_params["influx_org"],
+                                         influx_token=self.influx_params["influx_token"],
+                                         influx_bucket=self.influx_params["influx_bucket"],
+                                         path=report_name)
+
+                    influx.glob()
+                except Exception as e:
+                    print(e)
+                    pass
+            report_name = self.cvtest_obj[0].report_name[0]['LAST']["response"].split(":::")[1].split("/")[-1] + "/"
+            self.attach_report_graphs(report_name=report_name)
+            result = self.read_kpi_file(column_name=["pass/fail"], dir_name=report_name)
+            allure.attach.file(source="../reports/" + report_name + "/kpi.csv",
+                               name=f"{test}_CSV", attachment_type=allure.attachment_type.CSV)
+            if result[0][0] == "PASS":
+                return True, "Test Passed"
+            else:
+                return False, "Test Failed"
+        except Exception as e:
+            logging.error(f"{e}")
+            return False, f"{e}"
+        finally:
+            try:
+                self.client_disconnect(clear_all_sta=True, clean_l3_traffic=True)
+            except Exception as e:
+                logging.error(f"{e}")
+                return False, f"{e}"
 
 if __name__ == '__main__':
     basic = {