Add: 11r OTA 6g-6g wpa3 EAP on same channel test (#926)

Signed-off-by: anil-tegala <[email protected]>

tests/e2e/advanced/roam_test/hard_roam/OTA/test_roam_ota.py

@@ -1119,3 +1119,117 @@ def test_roam_5g_to_5g_sc_eap_wpa3(self, get_target_object, get_test_library, ge
               pk_passwd=radius_info["pk_password"], ca_cert='/home/lanforge/ca.pem', sta_type="11r-eap",
               iteration=1, channel="11", option="ota", dut_name=dut_names, traffic_type="lf_udp")
         assert pass_fail, message
+
+    @pytest.mark.same_channel
+    @pytest.mark.wpa3_enterprise
+    @pytest.mark.sixg
+    @pytest.mark.roam
+    def test_roam_6g_to_6g_sc_eap_wpa3(self, get_target_object, get_test_library, get_lab_info, selected_testbed, radius_info):
+        """
+            Test Roaming between two APs, Same channel, 6G, WPA3 Enterprise
+            pytest -m "roam and sixg and same_channel and wpa3_enterprise"
+        """
+        ap_data = dict()
+        dut_list = [str(selected_testbed)]
+        dut_names = list()
+        bssid_list = list()
+        testbed_info = get_lab_info.CONFIGURATION
+        if str(selected_testbed + 'a') in testbed_info:
+            dut_list.append(str(selected_testbed + 'a'))
+        logging.info(f"dut list: {dut_list}--")
+        config_data['interfaces'][0]["ssids"][0]["radius"] = {
+            "accounting": {
+                "host": radius_info["ip"],
+                "port": radius_info["port"],
+                "secret": radius_info["secret"]
+            },
+            "authentication": {
+                "host": radius_info["ip"],
+                "port": radius_info["port"],
+                "secret": radius_info["secret"]
+            }
+        }
+        config_data['radios'] = [
+            {
+                "band": "6G",
+                "channel": 161,
+                "channel-mode": "HE",
+                "channel-width": 80,
+                "country": "US"
+            }
+        ]
+        # change wifi-band and security type to wpa3
+        config_data['interfaces'][0]["ssids"][0]["wifi-bands"] = ["6G"]
+        config_data['interfaces'][0]["ssids"][0]["encryption"]["proto"] = "wpa3"
+        if "key" in config_data['interfaces'][0]["ssids"][0]["encryption"]:
+            config_data['interfaces'][0]["ssids"][0]["encryption"].pop("key")
+        if len(dut_list) < 2:
+            logging.error(f"This test need two AP's but number of DUT's available in the selected testbed is {dut_list}")
+            assert False, f"This test need two AP's but number of DUT's available in the selected testbed is {dut_list}"
+        for ap in range(len(dut_list)):
+            serial_number = testbed_info[dut_list[ap]]["device_under_tests"][0]['identifier']
+            dut_names.append(testbed_info[dut_list[ap]]["device_under_tests"][0]['model'])
+            logging.info(config_data)
+            payload = {"configuration": json.dumps(config_data), "serialNumber": serial_number, "UUID": 2}
+            uri = get_target_object.controller_library_object.build_uri(
+                "device/" + serial_number + "/configure")
+            logging.info("Sending Command: " + "\n" + str(uri) + "\n" +
+                         "TimeStamp: " + str(datetime.datetime.utcnow()) + "\n" +
+                         "Data: " + str(json.dumps(payload, indent=2)) + "\n" +
+                         "Headers: " + str(get_target_object.controller_library_object.make_headers()))
+            allure.attach(name=f"Push roam config on {serial_number}: ", body="Sending Command: " + str(uri) + "\n" +
+                                                                              "TimeStamp: " + str(
+                datetime.datetime.utcnow()) + "\n" +
+                                                                              "Data: " + str(payload) + "\n" +
+                                                                              "Headers: " + str(
+                get_target_object.controller_library_object.make_headers()))
+            resp = requests.post(uri, data=json.dumps(payload, indent=2),
+                                 headers=get_target_object.controller_library_object.make_headers(),
+                                 verify=False, timeout=120)
+            time.sleep(10)
+            logging.info(resp.json())
+            allure.attach(name=f"Response - {resp.status_code} {resp.reason}", body=str(resp.json()))
+            if resp.status_code != 200:
+                if resp.status_code == 400 and "Device is already executing a command. Please try later." in resp.json()["ErrorDescription"]:
+                    time.sleep(30)
+                    resp = requests.post(uri, data=json.dumps(payload, indent=2),
+                                         headers=get_target_object.controller_library_object.make_headers(),
+                                         verify=False, timeout=120)
+                    time.sleep(10)
+                    logging.info(resp.json())
+                else:
+                    assert False, f"push configuration to {serial_number} got failed"
+            get_target_object.dut_library_object.device_under_tests_data = testbed_info[dut_list[ap]]["device_under_tests"]
+            ap_iwinfo = get_target_object.dut_library_object.get_iwinfo(attach_allure=False)
+            if str(ap_iwinfo) != "Error: pop from empty list":
+                interfaces = {}
+                interface_matches = re.finditer(
+                    r'wlan\d\s+ESSID:\s+".*?"\s+Access Point:\s+([0-9A-Fa-f:]+).*?Channel:\s+([\d\s]+)', ap_iwinfo,
+                    re.DOTALL)
+                logging.info(str(interface_matches))
+                if interface_matches:
+                    for match in interface_matches:
+                        interface_name = f'wlan{match.group(0)[4]}'
+                        access_point = match.group(1)
+                        channel = match.group(2).strip()
+                        interfaces[interface_name] = {'Access Point': access_point, 'Channel': channel}
+                        ap_data.update({serial_number: {'Access Point': access_point, 'Channel': channel}})
+                    logging.info(interfaces)
+                else:
+                    logging.error("Failed to get iwinfo")
+                    pytest.exit("Failed to get iwinfo")
+            elif ap_iwinfo == {}:
+                assert False, "Empty iwinfo reponse from AP through minicom"
+            else:
+                assert False, "Failed to get iwinfo from minicom"
+        logging.info(f"AP Data from iwinfo: {ap_data}")
+        for serial in ap_data:
+            bssid_list.append(ap_data[serial]['Access Point'])
+        ssid = config_data['interfaces'][0]["ssids"][0]["name"]
+        pass_fail, message = get_test_library.roam_test(ap1_bssid=bssid_list[0], ap2_bssid=bssid_list[1],
+              band="sixg", num_sta=1, security="wpa3",  ssid=ssid, upstream="1.1.eth1", eap_method="TLS",
+              eap_identity=radius_info["user"], eap_password=radius_info["password"], private_key="1234567890",
+              pk_passwd=radius_info["pk_password"], ca_cert='/home/lanforge/ca.pem', sta_type="11r-eap",
+              iteration=1, channel="161", option="ota", dut_name=dut_names, traffic_type="lf_udp")
+        assert pass_fail, message
+