mirror of
				https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
				synced 2025-11-03 20:27:54 +00:00 
			
		
		
		
	Added client type functionality
Signed-off-by: jitendracandela <jitendra.kushavah@candelatech.com>
This commit is contained in:
		@@ -1541,12 +1541,12 @@ class lf_libs:
 | 
			
		||||
            return signal
 | 
			
		||||
 | 
			
		||||
    def attenuator_serial_radio(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", mode="BRIDGE", atn_val=400,
 | 
			
		||||
                                   vlan_id=100, sta_mode=0, station_name=[], radio='1.1.wiphy0', timeout=20):
 | 
			
		||||
                                   vlan_id=100, client_type=0, station_name=[], radio='1.1.wiphy0', timeout=20):
 | 
			
		||||
        # index 0 of atten_serial_radio will ser no of 1st 2g/5g radio and index 1 will ser no of 2nd and 3rd 2g/5g 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)
 | 
			
		||||
                                        vlan_id=vlan_id, radio=radio, client_type=client_type, station_name=station_name)
 | 
			
		||||
        signal1 = self.get_station_signal(station_name[0],timeout)
 | 
			
		||||
        atten_sr = atten_serial[0].split(".")
 | 
			
		||||
        self.attenuator_modify(int(atten_sr[2]), "all", atn_val)
 | 
			
		||||
@@ -1692,7 +1692,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", band=None,
 | 
			
		||||
                                   vlan_id=[None], radio=None, sta_mode=0, station_name=[], dut_data=None,
 | 
			
		||||
                                   vlan_id=[None], radio=None, client_type=0, station_name=[], dut_data=None,
 | 
			
		||||
                                   sniff_radio=False, create_vlan=True):
 | 
			
		||||
        # pre cleanup
 | 
			
		||||
        # if pre_cleanup:
 | 
			
		||||
@@ -1716,7 +1716,7 @@ class lf_libs:
 | 
			
		||||
                    self.add_vlan(vlan_ids=vlan_id, build=False)
 | 
			
		||||
        print("upstream_port1:", upstream_port)
 | 
			
		||||
 | 
			
		||||
        client_connect = CreateStation(_host=self.manager_ip, _port=self.manager_http_port, _mode=sta_mode,
 | 
			
		||||
        client_connect = CreateStation(_host=self.manager_ip, _port=self.manager_http_port, _mode=client_type,
 | 
			
		||||
                                       _sta_list=station_name, _password=passkey, _ssid=ssid, _security=security)
 | 
			
		||||
        client_connect.upstream_port = upstream_port
 | 
			
		||||
        client_connect.upstream_resource = 1
 | 
			
		||||
 
 | 
			
		||||
@@ -77,7 +77,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
        self.influx_params = influx_params
 | 
			
		||||
 | 
			
		||||
    def client_connectivity_test(self, ssid="[BLANK]", passkey="[BLANK]", bssid="[BLANK]", dut_data={},
 | 
			
		||||
                                 security="open", extra_securities=[], sta_mode=0,
 | 
			
		||||
                                 security="open", extra_securities=[], client_type=0,
 | 
			
		||||
                                 num_sta=1, mode="BRIDGE", vlan_id=[None], band="twog",
 | 
			
		||||
                                 allure_attach=True, runtime_secs=40):
 | 
			
		||||
        self.check_band_ap(band=band)
 | 
			
		||||
@@ -101,7 +101,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
                obj_sta_connect = StaConnect2(self.manager_ip, self.manager_http_port, outfile="shivam",
 | 
			
		||||
                                              _cleanup_on_exit=False)
 | 
			
		||||
 | 
			
		||||
                obj_sta_connect.sta_mode = sta_mode
 | 
			
		||||
                obj_sta_connect.sta_mode = client_type
 | 
			
		||||
                obj_sta_connect.upstream_resource = data[dut]["upstream_resource"]
 | 
			
		||||
                obj_sta_connect.upstream_port = data[dut]["upstream"]
 | 
			
		||||
                self.enable_verbose_debug(radio=radio, enable=True)
 | 
			
		||||
@@ -222,7 +222,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
                    time.sleep(runtime_secs)
 | 
			
		||||
        pass_fail_result = []
 | 
			
		||||
        for obj in sta_connect_obj:
 | 
			
		||||
            sta_rows = ["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip", "signal", "mac"]
 | 
			
		||||
            sta_rows = ["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip", "signal", "mac", "mode"]
 | 
			
		||||
            station_data = self.get_station_data(sta_name=obj.station_names, rows=sta_rows,
 | 
			
		||||
                                                 allure_attach=False)
 | 
			
		||||
            sta_table_dict = {}
 | 
			
		||||
@@ -321,7 +321,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
        return result, description
 | 
			
		||||
 | 
			
		||||
    def enterprise_client_connectivity_test(self, ssid="[BLANK]", passkey="[BLANK]", bssid="[BLANK]", dut_data={},
 | 
			
		||||
                                            security="open", extra_securities=[], sta_mode=0, key_mgmt="WPA-EAP",
 | 
			
		||||
                                            security="open", extra_securities=[], client_type=0, key_mgmt="WPA-EAP",
 | 
			
		||||
                                            pairwise="NA", group="NA", wpa_psk="DEFAULT", ttls_passwd="nolastart",
 | 
			
		||||
                                            ieee80211w=1, wep_key="NA", ca_cert="NA", eap="TTLS", identity="nolaradius",
 | 
			
		||||
                                            d_vlan=False, cleanup=True,
 | 
			
		||||
@@ -348,7 +348,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
            for radio in data[dut]["station_data"]:
 | 
			
		||||
                obj_eap_connect = TTLSTest(host=self.manager_ip, port=self.manager_http_port,
 | 
			
		||||
                                           sta_list=data[dut]["station_data"][radio], vap=False, _debug_on=True)
 | 
			
		||||
                obj_eap_connect.station_profile.sta_mode = sta_mode
 | 
			
		||||
                obj_eap_connect.station_profile.sta_mode = client_type
 | 
			
		||||
                obj_eap_connect.upstream_resource = data[dut]["upstream_resource"]
 | 
			
		||||
                obj_eap_connect.l3_cx_obj_udp.upstream_resource = data[dut]["upstream_resource"]
 | 
			
		||||
                obj_eap_connect.l3_cx_obj_tcp.upstream_resource = data[dut]["upstream_resource"]
 | 
			
		||||
@@ -478,7 +478,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
                    time.sleep(runtime_secs)
 | 
			
		||||
        pass_fail_result = []
 | 
			
		||||
        for obj in eap_connect_objs:
 | 
			
		||||
            sta_rows = ["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip", "signal", "mac"]
 | 
			
		||||
            sta_rows = ["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip", "signal", "mac", "mode"]
 | 
			
		||||
            self.station_data = self.get_station_data(sta_name=obj.sta_list, rows=sta_rows,
 | 
			
		||||
                                                      allure_attach=False)
 | 
			
		||||
            sta_table_dict = {}
 | 
			
		||||
@@ -714,9 +714,9 @@ class lf_tests(lf_libs):
 | 
			
		||||
            logging.info("ALL Stations got IP as Expected")
 | 
			
		||||
 | 
			
		||||
    def client_connect(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", mode="BRIDGE", band="twog",
 | 
			
		||||
                       vlan_id=[None], num_sta=None, scan_ssid=True, sta_mode=0, pre_cleanup=True,
 | 
			
		||||
                       station_data=["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip", "signal"],
 | 
			
		||||
                       allure_attach=True, identifier=None, allure_name="station data", client_type=None, dut_data={}):
 | 
			
		||||
                       vlan_id=[None], num_sta=None, scan_ssid=True, client_type=0, pre_cleanup=True,
 | 
			
		||||
                       station_data=["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip", "signal", "mode"],
 | 
			
		||||
                       allure_attach=True, identifier=None, allure_name="station data", dut_data={}):
 | 
			
		||||
        # pre cleanup
 | 
			
		||||
        if pre_cleanup:
 | 
			
		||||
            self.pre_cleanup()
 | 
			
		||||
@@ -781,7 +781,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
                                           _password=data[identifier]["passkey"],
 | 
			
		||||
                                           _ssid=data[identifier]["ssid"],
 | 
			
		||||
                                           _security=data[identifier]["encryption"])
 | 
			
		||||
            client_connect.station_profile.sta_mode = sta_mode
 | 
			
		||||
            client_connect.station_profile.sta_mode = client_type
 | 
			
		||||
            client_connect.upstream_resource = data[identifier]["upstream_resource"]
 | 
			
		||||
            client_connect.upstream_port = data[identifier]["upstream"]
 | 
			
		||||
            client_connect.radio = radio
 | 
			
		||||
@@ -1243,13 +1243,13 @@ class lf_tests(lf_libs):
 | 
			
		||||
 | 
			
		||||
    def dataplane_throughput_test(self, ssid="[BLANK]", passkey="[BLANK]", security="wpa2", num_sta=1, mode="BRIDGE",
 | 
			
		||||
                                  vlan_id=[None],
 | 
			
		||||
                                  download_rate="85%", band="twog", scan_ssid=True, sta_mode=0,
 | 
			
		||||
                                  download_rate="85%", band="twog", scan_ssid=True,
 | 
			
		||||
                                  upload_rate="0", duration="15s", instance_name="test_demo", raw_lines=None,
 | 
			
		||||
                                  influx_tags="",
 | 
			
		||||
                                  move_to_influx=False,
 | 
			
		||||
                                  station_data=["4way time (us)", "channel", "cx time (us)", "dhcp (ms)", "ip",
 | 
			
		||||
                                                "signal"],
 | 
			
		||||
                                  allure_attach=True, allure_name="station data", client_type=None, dut_data={}):
 | 
			
		||||
                                                "signal", "mode"],
 | 
			
		||||
                                  allure_attach=True, allure_name="station data", client_type=0, dut_data={}):
 | 
			
		||||
        instance_name = ''.join(random.choices(string.ascii_uppercase + string.digits, k=12))
 | 
			
		||||
        dataplane_obj_list = []
 | 
			
		||||
        for dut in self.dut_data:
 | 
			
		||||
@@ -1275,7 +1275,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
                    upstream_port = ret[identifier] + "." + str(vlan_id[0])
 | 
			
		||||
            logging.info("Upstream data: " + str(upstream_port))
 | 
			
		||||
            station_data = self.client_connect(ssid=ssid, passkey=passkey, security=security, mode=mode, band=band,
 | 
			
		||||
                                               vlan_id=vlan_id, num_sta=num_sta, scan_ssid=scan_ssid, sta_mode=sta_mode,
 | 
			
		||||
                                               vlan_id=vlan_id, num_sta=num_sta, scan_ssid=scan_ssid,
 | 
			
		||||
                                               station_data=station_data,
 | 
			
		||||
                                               allure_attach=allure_attach, identifier=identifier,
 | 
			
		||||
                                               allure_name=allure_name, client_type=client_type, dut_data=dut_data)
 | 
			
		||||
@@ -1742,12 +1742,12 @@ class lf_tests(lf_libs):
 | 
			
		||||
                    # mode = 2/1 will create legacy client
 | 
			
		||||
                    create_sta = self.client_connect_using_radio(ssid=ssid, passkey=passkey, security=security,
 | 
			
		||||
                                                                 radio=radios[i], station_name=[sta[i]],
 | 
			
		||||
                                                                 sta_mode=sta_mode[0])
 | 
			
		||||
                                                                 client_type=sta_mode[0])
 | 
			
		||||
                else:
 | 
			
		||||
                    # mode = 11/9 will create bgn-AC/an-AC client
 | 
			
		||||
                    create_sta = self.client_connect_using_radio(ssid=ssid, passkey=passkey, security=security,
 | 
			
		||||
                                                                 radio=radios[i], station_name=[sta[i]],
 | 
			
		||||
                                                                 sta_mode=sta_mode[1])
 | 
			
		||||
                                                                 client_type=sta_mode[1])
 | 
			
		||||
                if create_sta == False:
 | 
			
		||||
                    logging.info(f"Test failed due to no IP for {sta[i]}")
 | 
			
		||||
                    assert False, f"Test failed due to no IP for {sta[i]}"
 | 
			
		||||
@@ -2400,7 +2400,7 @@ class lf_tests(lf_libs):
 | 
			
		||||
        logging.info("DUT DATA: " + str(dut_data))
 | 
			
		||||
        for dut in self.dut_data:
 | 
			
		||||
            station_result = self.client_connect_using_radio(ssid=ssid, passkey=passkey, security=security, mode=mode,
 | 
			
		||||
                                                             band=band, vlan_id=vlan_id, radio="1.1.wiphy0", sta_mode=0,
 | 
			
		||||
                                                             band=band, vlan_id=vlan_id, radio="1.1.wiphy0", client_type=0,
 | 
			
		||||
                                                             station_name=["sta0000"],
 | 
			
		||||
                                                             dut_data=dut_data)
 | 
			
		||||
            sta = "sta0000"
 | 
			
		||||
 
 | 
			
		||||
		Reference in New Issue
	
	Block a user