mirror of
https://github.com/Telecominfraproject/wlan-testing.git
synced 2025-11-04 04:48:01 +00:00
roam ota twog and fiveg wpa3psk test (#954)
* add roam ota twog and fiveg wpa2 psk test Signed-off-by: anil-tegala <anil.tegala@candelatech.com> * add roam ota twog and fiveg wpa3 psk test Signed-off-by: anil-tegala <anil.tegala@candelatech.com> * set missing band in 2G config Signed-off-by: anil-tegala <anil.tegala@candelatech.com> --------- Signed-off-by: anil-tegala <anil.tegala@candelatech.com>
This commit is contained in:
Anil Kumar Tegala
committed by
GitHub
GitHub
parent
016f766466
commit
bcf55b6f45
@@ -1689,3 +1689,248 @@ class TestRoamOTA(object):
|
||||
pytest.fail(f"Test failed with the following reasons: \n{message}")
|
||||
else:
|
||||
assert True
|
||||
|
||||
@pytest.mark.roam
|
||||
@pytest.mark.fiveg
|
||||
@pytest.mark.twog
|
||||
@pytest.mark.wpa2_personal
|
||||
def test_roam_5g_and_2g_wpa2psk(self, get_target_object, get_test_library, get_lab_info, selected_testbed):
|
||||
"""
|
||||
Test Roaming between two APs, 2G & 5G, WPA2 Personal
|
||||
pytest -m "roam and fiveg and twog and wpa2_personal and ota"
|
||||
"""
|
||||
ap_data = dict()
|
||||
dut_list = [str(selected_testbed)]
|
||||
dut_names = list()
|
||||
bssid_list = list()
|
||||
freqs_ = ""
|
||||
testbed_info = get_lab_info.CONFIGURATION
|
||||
config = copy.deepcopy(config_data)
|
||||
temp_list = list()
|
||||
for key, val in testbed_info.items():
|
||||
tb_type, tb_name = selected_testbed.split("-")
|
||||
if tb_type in key and tb_name[0] in key:
|
||||
temp_list.append(key)
|
||||
temp_list.sort()
|
||||
dut_list = [temp_list[idx] for idx in range(len(temp_list)) if idx <= 1]
|
||||
config['radios'] = [
|
||||
{"band": "5G", "channel": 36, "channel-mode": "HE", "channel-width": 80, "country": "CA"}]
|
||||
config['interfaces'][0]["ssids"][0]["wifi-bands"] = ["5G"]
|
||||
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'])
|
||||
if ap == 1:
|
||||
config['interfaces'][0]["ssids"][0]["wifi-bands"] = ["2G"]
|
||||
config['radios'] = [
|
||||
{"band": "2G", "channel": 11, "channel-mode": "HE", "channel-width": 20, "country": "CA"}]
|
||||
logging.info(config)
|
||||
payload = {"configuration": json.dumps(config), "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=True)
|
||||
if str(ap_iwinfo) != "Error: pop from empty list":
|
||||
include_essid = config['interfaces'][0]["ssids"][0]["name"]
|
||||
re_obj = re.compile(
|
||||
rf'(wlan\d(?:-\d)?)\s+ESSID: "{re.escape(include_essid)}".*?\s+Access Point:\s+([0-9A-Fa-f:]+).*?Channel:\s+('
|
||||
r'\d+)\s+\(([\d.]+) GHz\)',
|
||||
re.DOTALL
|
||||
)
|
||||
# find all matches
|
||||
interface_matches = re_obj.finditer(ap_iwinfo)
|
||||
if interface_matches:
|
||||
for match in interface_matches:
|
||||
interface_name = match.group(1)
|
||||
access_point = match.group(2)
|
||||
channel = match.group(3)
|
||||
frequency = match.group(4).replace('.', '')
|
||||
ap_data.update(
|
||||
{serial_number: {'Access Point': access_point, 'Channel': channel, 'frequency': frequency}})
|
||||
logging.info(f"AP Data from iwinfo: {ap_data}")
|
||||
else:
|
||||
logging.error("Failed to get iwinfo")
|
||||
pytest.exit("Failed to get iwinfo")
|
||||
elif ap_iwinfo == {}:
|
||||
pytest.fail("Empty iwinfo reponse from AP through minicom")
|
||||
else:
|
||||
pytest.fail("Failed to get iwinfo from minicom")
|
||||
for serial in ap_data:
|
||||
bssid_list.append(ap_data[serial]['Access Point'])
|
||||
if not ap_data[serial]['frequency'].endswith(","):
|
||||
freqs_ = freqs_ + ap_data[serial]['frequency'] + ","
|
||||
else:
|
||||
freqs_ = freqs_ + ap_data[serial]['frequency']
|
||||
ssid = config['interfaces'][0]["ssids"][0]["name"]
|
||||
key = config['interfaces'][0]["ssids"][0]["encryption"]["key"]
|
||||
try:
|
||||
pass_fail, message = get_test_library.roam_test(ap1_bssid=bssid_list[0], ap2_bssid=bssid_list[1],
|
||||
scan_freq=freqs_,
|
||||
band="both", num_sta=2, security="wpa2", security_key=key,
|
||||
ssid=ssid, upstream="1.1.eth1", duration=None,
|
||||
iteration=1, channel="36", option="ota", dut_name=dut_names,
|
||||
traffic_type="lf_udp", sta_type="11r")
|
||||
except Exception as e:
|
||||
logging.error(f"Exception in roam test : {e}")
|
||||
pass_fail, message = False, e
|
||||
finally:
|
||||
get_target_object.dut_library_object.get_dut_logs(print_log=False)
|
||||
if not pass_fail:
|
||||
pytest.fail(f"Test failed with the following reasons: \n{message}")
|
||||
else:
|
||||
assert True
|
||||
|
||||
@pytest.mark.roam
|
||||
@pytest.mark.fiveg
|
||||
@pytest.mark.twog
|
||||
@pytest.mark.sae
|
||||
@pytest.mark.wpa3_personal
|
||||
def test_roam_5g_and_2g_wpa3psk(self, get_target_object, get_test_library, get_lab_info, selected_testbed):
|
||||
"""
|
||||
Test Roaming between two APs, 2G & 5G, WPA3 Personal
|
||||
pytest -m "roam and fiveg and twog and wpa3_personal and ota"
|
||||
"""
|
||||
ap_data = dict()
|
||||
dut_list = [str(selected_testbed)]
|
||||
dut_names = list()
|
||||
bssid_list = list()
|
||||
freqs_ = ""
|
||||
testbed_info = get_lab_info.CONFIGURATION
|
||||
config = copy.deepcopy(config_data)
|
||||
temp_list = list()
|
||||
for key, val in testbed_info.items():
|
||||
tb_type, tb_name = selected_testbed.split("-")
|
||||
if tb_type in key and tb_name[0] in key:
|
||||
temp_list.append(key)
|
||||
temp_list.sort()
|
||||
dut_list = [temp_list[idx] for idx in range(len(temp_list)) if idx <= 1]
|
||||
config['radios'] = [
|
||||
{"band": "5G", "channel": 36, "channel-mode": "HE", "channel-width": 80, "country": "CA"}]
|
||||
config['interfaces'][0]["ssids"][0]["wifi-bands"] = ["5G"]
|
||||
# change ssid config data to sae
|
||||
config['interfaces'][0]["ssids"][0]["encryption"]["proto"] = "sae"
|
||||
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'])
|
||||
if ap == 1:
|
||||
config['radios'] = [
|
||||
{"band": "2G", "channel": 11, "channel-mode": "HE", "channel-width": 20, "country": "CA"}]
|
||||
config['interfaces'][0]["ssids"][0]["wifi-bands"] = ["2G"]
|
||||
logging.info(config)
|
||||
payload = {"configuration": json.dumps(config), "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=True)
|
||||
if str(ap_iwinfo) != "Error: pop from empty list":
|
||||
include_essid = config['interfaces'][0]["ssids"][0]["name"]
|
||||
re_obj = re.compile(
|
||||
rf'(wlan\d(?:-\d)?)\s+ESSID: "{re.escape(include_essid)}".*?\s+Access Point:\s+([0-9A-Fa-f:]+).*?Channel:\s+('
|
||||
r'\d+)\s+\(([\d.]+) GHz\)',
|
||||
re.DOTALL
|
||||
)
|
||||
# find all matches
|
||||
interface_matches = re_obj.finditer(ap_iwinfo)
|
||||
if interface_matches:
|
||||
for match in interface_matches:
|
||||
interface_name = match.group(1)
|
||||
access_point = match.group(2)
|
||||
channel = match.group(3)
|
||||
frequency = match.group(4).replace('.', '')
|
||||
ap_data.update(
|
||||
{serial_number: {'Access Point': access_point, 'Channel': channel, 'frequency': frequency}})
|
||||
logging.info(f"AP Data from iwinfo: {ap_data}")
|
||||
else:
|
||||
logging.error("Failed to get iwinfo")
|
||||
pytest.exit("Failed to get iwinfo")
|
||||
elif ap_iwinfo == {}:
|
||||
pytest.fail("Empty iwinfo reponse from AP through minicom")
|
||||
else:
|
||||
pytest.fail("Failed to get iwinfo from minicom")
|
||||
for serial in ap_data:
|
||||
bssid_list.append(ap_data[serial]['Access Point'])
|
||||
if not ap_data[serial]['frequency'].endswith(","):
|
||||
freqs_ = freqs_ + ap_data[serial]['frequency'] + ","
|
||||
else:
|
||||
freqs_ = freqs_ + ap_data[serial]['frequency']
|
||||
ssid = config['interfaces'][0]["ssids"][0]["name"]
|
||||
key = config['interfaces'][0]["ssids"][0]["encryption"]["key"]
|
||||
try:
|
||||
pass_fail, message = get_test_library.roam_test(ap1_bssid=bssid_list[0], ap2_bssid=bssid_list[1],
|
||||
scan_freq=freqs_,
|
||||
band="both", num_sta=2, security="wpa2", security_key=key,
|
||||
ssid=ssid, upstream="1.1.eth1", duration=None,
|
||||
iteration=1, channel="36", option="ota", dut_name=dut_names,
|
||||
traffic_type="lf_udp", sta_type="11r")
|
||||
except Exception as e:
|
||||
logging.error(f"Exception in roam test : {e}")
|
||||
pass_fail, message = False, e
|
||||
finally:
|
||||
get_target_object.dut_library_object.get_dut_logs(print_log=False)
|
||||
if not pass_fail:
|
||||
pytest.fail(f"Test failed with the following reasons: \n{message}")
|
||||
else:
|
||||
assert True
|
||||
|
||||
Reference in New Issue
Block a user