scottk/wlan-testing
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-testing.git synced 2025-10-29 18:12:34 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
d7002b302443b61cb6ac66282fa4ca4276a7673e
wlan-testing/tests/test_connectivity.py
bhargavi-ct d7002b3024 Update pass fail criteria for firmware upgrade test (#1059) 
Signed-off-by: bhargavi-ct <bhargavimamidipaka@candelatech.com>
Co-authored-by: bhargavi-ct <bhargavimamidipaka@candelatech.com>
2025-07-14 16:16:13 +05:30

443 lines
24 KiB
Python
Raw Blame History

"""
Test Case Module: Testing Basic Connectivity with Resources
"""
import time
import allure
import pytest
import logging
import requests
import json
pytestmark = [pytest.mark.test_resources,
pytest.mark.sanity,
pytest.mark.ow_sanity_lf,
pytest.mark.ow_sanity_interop,
pytest.mark.uc_sanity,
pytest.mark.sanity_55,
pytest.mark.interop_uc_sanity,
pytest.mark.android,
pytest.mark.ios,
pytest.mark.client_connect]
state = True
sdk_expected = True
@allure.feature("Test Connectivity")
@allure.parent_suite("Test Connectivity")
# @allure.suite("Test Resources")
class TestResources(object):
"""Test Case Class: Test cases to cover resource Connectivity"""
@pytest.mark.test_cloud_controller
@allure.testcase(name="Test Cloud Controller", url="")
@allure.title("Cloud Controller Connectivity")
@allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-11615", name="11615")
def test_controller_connectivity(self, get_target_object, get_testbed_details, get_test_library):
"""Test case to verify cloud Controller Connectivity
Unique marker: pytest -m "test_cloud_controller"
"""
login_response_json = get_target_object.controller_library_object.login_resp.json()
response_code = get_target_object.controller_library_object.login_resp.status_code
request_url = get_target_object.controller_library_object.login_resp.request.url
print("Login_Request_URL: ", str(request_url))
allure.attach(name="Login_Request_URL: ", body=str(request_url))
print("response_code: ", response_code)
allure.attach(name="Login Response Code: ", body=str(response_code))
print("login_response_json: ", login_response_json)
allure.attach(name="Login Response JSON: ",
body=str(get_target_object.controller_library_object.login_resp.json()))
if response_code != 200:
pytest.exit(
"exiting from pytest, login response is no 200: " + str(
get_target_object.controller_library_object.login_resp.status_code))
gw_system_info = get_target_object.controller_library_object.get_system_gw()
request_url = gw_system_info.request.url
allure.attach(name="get_system_gw_request: ", body=str(request_url))
gw_system_status = gw_system_info.status_code
gw_system_status_json = gw_system_info.json()
print("gw_status_check response from gateway: ", gw_system_status)
allure.attach(name="gw_status_check response from gateway: ", body=str(gw_system_status) +
str(gw_system_status_json))
if gw_system_status != 200:
allure.attach(name="Login_Request_URL: ", body=str(request_url))
pytest.exit("gw_status_check response from gateway: " + str(gw_system_status))
fms_system_info = get_target_object.controller_library_object.get_system_fms()
request_url = fms_system_info.request.url
allure.attach(name="get_system_fms_request: ", body=str(request_url))
fms_system_status = fms_system_info.status_code
fms_system_status_json = fms_system_info.json()
print("fms_status_check response from fms: ", fms_system_status)
allure.attach(name="fms_status_check response from fms:", body=str(fms_system_status) +
str(fms_system_status_json))
if fms_system_status != 200:
pytest.exit("fms_status_check response from fms: " + str(fms_system_status))
# Provisioning system info
prov_system_info = get_target_object.controller_library_object.get_system_prov()
request_url = prov_system_info.request.url
allure.attach(name="get_system_prov_request: ", body=str(request_url))
prov_system_status = prov_system_info.status_code
prov_system_status_json = prov_system_info.json()
print("prov_status_check response from fms: ", prov_system_status)
allure.attach(name="prov_status_check response from Prov:", body=str(prov_system_status) +
str(prov_system_status_json))
if prov_system_status != 200:
pytest.exit("Prov_status_check response from Prov: " + str(prov_system_status))
# if gw_status_check != 200:
# for i in range(10):
# if get_target_object.get_system_gw().status_code != 200 and i < 9:
# print("sleeping for 30 sec, gw service is down with response not equals to 200")
# time.sleep(30)
# elif get_target_object.get_system_gw().status_code != 200 and i == 9:
# pytest.exit("GW service is not up yet, exiting from pytest")
# else:
# break
#
# if fms_status_check != 200:
# for i in range(10):
# if get_target_object.get_system_fms().status_code != 200 and i < 9:
# print("sleeping for 30 sec, fms service is down with response not equals to 200")
# time.sleep(30)
# elif get_target_object.get_system_fms().status_code != 200 and i == 9:
# pytest.exit("fms service is not up yet, exiting from pytest")
# else:
# break
#
# available_device_list = []
# devices = get_target_object.get_devices()
# number_devices = len(devices["devices"])
# for i in range(number_devices):
# available_device_list.append(devices["devices"][i]["serialNumber"])
# print("available_device_list: ", available_device_list)
#
# if get_configuration["access_point"][0]["serial"] not in available_device_list:
# for i in range(10):
# available_device_list = []
# devices = get_target_object.get_devices()
# number_devices = len(devices["devices"])
# for i in range(number_devices):
# available_device_list.append(devices["devices"][i]["serialNumber"])
# print(available_device_list)
#
# if get_configuration["access_point"][0]["serial"] not in available_device_list and i < 9:
# print("unable to find device on UI, Sleeping for 30 sec")
# time.sleep(30)
# elif get_configuration["access_point"][0]["serial"] not in available_device_list and i == 9:
# pytest.exit("Device" + get_configuration["access_point"][0]["serial"] + "not found on UI")
# else:
# break
#
# for ap in get_configuration['access_point']:
#
# ap_ssh = get_apnos(ap, pwd="../libs/apnos/", sdk="2.x")
# uci_show_ucentral = ap_ssh.run_generic_command("uci show ucentral")
# print(uci_show_ucentral)
# print("AP is pointing to: ", ap_ssh.get_ap_uci_show_ucentral())
# expected_sdk = str(get_configuration["controller"]['url'].replace("https://sec", "\'gw").replace(":16001","\'"))
# if ap_ssh.get_ap_uci_show_ucentral() != expected_sdk:
# for i in range(10):
# ucentral_show = str(ap_ssh.get_ap_uci_show_ucentral().strip())
# print("AP pointing to: ", ucentral_show)
# print("AP should point to: ", expected_sdk)
#
# if ucentral_show != expected_sdk and i < 9:
# print("AP is not pointing to right SDK, retry after 30 sec")
# time.sleep(30)
# elif ucentral_show != expected_sdk and i == 9:
# assert False
# pytest.exit("AP is not pointing to right SDK")
# else:
# break
assert True
@pytest.mark.test_access_points_connectivity
@allure.testcase(name="Test Access Point Connectivity", url="")
@allure.title("Cloud Access Point Connectivity")
@allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-11615", name="11615")
def test_access_points_connectivity(self, get_target_object, get_testbed_details):
"""Test case to verify Access Points Connectivity
Unique marker: pytest -m "test_access_points_connectivity"
"""
# Logic to Get ubus call ucentral status from AP
connected = True
ubus_data = []
for i in range(0, len(get_target_object.device_under_tests_info)):
ret_val = get_target_object.get_dut_library_object().ubus_call_ucentral_status(idx=i)
ubus_data.append(ret_val)
if not ret_val["connected"]:
connected = False
get_target_object.get_dut_library_object().verify_certificates(idx=i, print_log=True,
attach_allure=True)
get_target_object.get_dut_library_object().check_connectivity(idx=i, print_log=True, attach_allure=True)
# Logic to get uci show ucentral and check the target sdk
uci_data = []
for i in range(0, len(get_target_object.device_under_tests_info)):
ret_val = get_target_object.get_dut_library_object().get_uci_show(idx=i, param="ucentral.config.server")
logging.info(f"ret_val from get_uci_show:{ret_val}")
if "ucentral.config.server" in ret_val:
ret_val = str(ret_val).split("=")[1]
uci_data.append(ret_val)
else:
ret_val_dict = json.loads(ret_val)
if not ret_val_dict.get("valid") or "server" not in ret_val_dict:
logging.warning(
"First attempt to fetch 'server' from gateway.json failed. Retrying in 5 seconds...")
time.sleep(5)
# Retry
ret_val = get_target_object.get_dut_library_object().get_uci_show(idx=i,
param="ucentral.config.server")
ret_val_dict = json.loads(ret_val)
if not ret_val_dict.get("valid") or "server" not in ret_val_dict:
logging.error("Invalid or missing 'server' after retry. Exiting test.")
pytest.exit("No valid 'server' value found in gateway.json after retry")
uci_data.append(ret_val_dict["server"])
logging.info(f"uci_data::{uci_data}")
gw_host = get_target_object.controller_library_object.gw_host.hostname
expected_host = True
for j in uci_data:
if gw_host not in j:
expected_host = False
# If Connected but not with expected host
if connected:
if expected_host:
logging.info(("Connected With Expected HOST" + "\n Current GW HOST: " + str(uci_data[0]) +
"\n EXPECTED GW HOST: " + str(gw_host)))
assert True, "Connected With Expected HOST"
else:
logging.error("Connected With Unexpected HOST" + "\n Current GW HOST: " + str(uci_data[0]) +
"\n EXPECTED GW HOST: " + str(gw_host))
pytest.exit("Connected With Unexpected HOST" + "\n Current GW HOST: " + str(uci_data[0]) +
"\n EXPECTED GW HOST: " + str(gw_host))
else:
ap_logs = get_target_object.get_dut_library_object().get_dut_logs(idx=i, print_log=False, attach_allure=False)
allure.attach(body=ap_logs, name="Disconnected ap logs")
logging.error("AP is in disconnected state from Ucentral gateway!!!")
pytest.fail("AP is in disconnected state from Ucentral gateway!!!")
# @pytest.mark.traffic_generator_connectivity
# @allure.testcase(name="test_traffic_generator_connectivity", url="")
# @allure.title("Traffic Generator Connectivity")
# def test_traffic_generator_connectivity(self, get_test_library):
# """Test case to verify Traffic Generator Connectivity"""
# port_data = get_test_library.json_get(_req_url="/port?fields=alias,port+type,ip")['interfaces']
# logging.info("Port data: " + str(port_data))
# eth_table_data = {}
# port = []
# ip = []
# for i in port_data:
# for item in i:
# if i[item]['port type'] == 'Ethernet':
# port.append(item)
# ip.append(i[item]['ip'])
# # creating dict for eth table
# eth_table_data["Port"] = port
# eth_table_data["ip"] = ip
# # Attaching eth table to allure
# get_test_library.attach_table_allure(data=eth_table_data, allure_name="Ethernet Table")
# max_num_sta_table_data = {}
# col = ["max possible stations", "max 2g stations", "max 5g stations", "max 6g stations", "max ax stations",
# "max ac stations"]
# max_num_sta = [get_test_library.max_possible_stations, get_test_library.max_2g_stations,
# get_test_library.max_5g_stations, get_test_library.max_6g_stations,
# get_test_library.max_ax_stations, get_test_library.max_ac_stations]
# max_num_sta_table_data[""] = col
# max_num_sta_table_data["Max number of stations"] = max_num_sta
# get_test_library.attach_table_allure(data=max_num_sta_table_data, allure_name="Max number of stations Table")
#
# assert True
# def test_ap_conn_state(self):
# global state
# if state == False:
# pytest.exit("AP is in DISCONNECTED State")
# global sdk_expected
# if sdk_expected == False:
# pytest.exit("AP has invalid Redirector")
# assert True
@pytest.mark.test_ap_restrictions
@allure.testcase(name="Check AP is restricted", url="")
@allure.title("Check AP is restricted")
@allure.testcase(url="https://telecominfraproject.atlassian.net/browse/WIFI-12318", name="12318")
def test_check_restrictions(self, get_target_object, get_testbed_details):
device_name = get_testbed_details['device_under_tests'][0]['identifier']
resp = resp = get_target_object.controller_library_object.check_restrictions(device_name)
if not resp:
logging.info("AP is not restricted")
assert True, "AP is not in restricted mode we can continue testing"
else:
logging.info("AP is restricted, Removing Restrictions")
output = get_target_object.get_dut_library_object().remove_restrictions()
resp = resp = get_target_object.controller_library_object.check_restrictions(device_name)
if not resp:
logging.info("Removed Restrictions")
assert True, "Remove restrictions, Can continue testing"
else:
logging.error("Unable to remove restrictions")
assert False, "Unable to remove restrictions"
@allure.testcase(name="Firmware Management", url="")
@pytest.mark.firmware
@allure.feature("Test Connectivity")
@allure.parent_suite("Test Connectivity")
class TestFirmwareUpgrade(object):
@pytest.mark.get_firmware_list
def test_get_firmware_version_list(self, get_testbed_details, get_target_object):
PASS = []
for ap in range(len(get_target_object.device_under_tests_info)):
# get the latest branch
firmware_list = get_target_object.firmware_library_object.get_firmwares(
model=get_target_object.device_under_tests_info[ap]['model'],
branch="",
commit_id='',
limit='10000',
offset='3000')
# firmware_list.reverse()
# Firmware list from newer to older image
image_date = []
for i in firmware_list:
image_date.append(i["imageDate"])
image_date.sort(reverse=True)
ordered_list_firmware = []
for i in image_date:
for j in firmware_list:
if i == j["imageDate"]:
ordered_list_firmware.append(j)
break
firmware_list = ordered_list_firmware
release_list_data = []
for i in firmware_list:
release_list_data.append(str(i['release']))
allure.attach(name="firmware_list", body=str("\n".join(release_list_data)),
attachment_type=allure.attachment_type.JSON)
try:
response = requests.get(get_target_object.device_under_tests_info[ap]['firmware_version'])
print("URL is valid and exists on the internet")
allure.attach(name="firmware url: ",
body=str(get_target_object.device_under_tests_info[ap]['firmware_version']))
target_revision_commit = get_target_object.device_under_tests_info[ap]['firmware_version'].split("-")[
-2]
target_revision_branch = get_target_object.device_under_tests_info[ap]['firmware_version'].split("-")[
-3]
flag = True
for i in release_list_data:
if target_revision_commit == i.split('-')[-1] and target_revision_branch == i.split('-')[-2]:
print('target firmware : ' + get_target_object.device_under_tests_info[ap][
'version'] + " is available in FMS : " + i)
allure.attach(
name='target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is available in FMS : " + i,
body="")
PASS.append(True)
flag = False
if flag:
print('target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is not available in FMS : ")
allure.attach(name='target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is not available in FMS : ",
body="")
PASS.append(False)
break
except Exception as e:
pass
if get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[1] == "latest":
for firmware in firmware_list:
if get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[0] == 'release':
version = firmware['revision'].split("/")[1].replace(" ", "").split('-')[1]
if firmware['revision'].split("/")[1].replace(" ", "").split('-')[1].__contains__('v2.'):
print("Target Firmware: \n", firmware)
allure.attach(name="Target firmware : ", body=str(firmware['release']))
break
if firmware['release'].split("-")[-2] == \
get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[0]:
print("Target Firmware: \n", firmware)
allure.attach(name="Target firmware : ", body=str(firmware['release']))
break
else:
flag = True
for firmware in firmware_list:
if get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[0] == 'release':
branch = firmware['revision'].split("/")[1].replace(" ", "").split('-')[1]
commit = get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[1]
if branch.__contains__('v2.') and commit == firmware['release'].split('-')[-1]:
print("Target Firmware: \n", firmware)
allure.attach(name="Target firmware : ", body=str(firmware['release']))
PASS.append(True)
flag = False
break
if get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[1] == \
firmware['release'].split('-')[-1] and \
get_target_object.device_under_tests_info[ap]['firmware_version'].split('-')[
0] == \
firmware['release'].split('-')[-2]:
print('target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is available in FMS : " + firmware[
'release'])
allure.attach(
name='target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is available in FMS : " + firmware[
'release']
, body="")
PASS.append(True)
flag = False
if flag:
print('target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is not available in FMS : ")
allure.attach(name='target firmware : ' + get_target_object.device_under_tests_info[ap][
'firmware_version'] + " is not available in FMS : ",
body="")
PASS.append(False)
assert False not in PASS
@pytest.mark.firmware_upgrade
def test_firmware_upgrade_request(self, get_target_object, get_dut_logs_per_test_case, check_connectivity):
for update in get_target_object.setup_firmware():
logging.info(f"update from setup firmware::{update}")
allure.attach(name='serial: ' + update[0], body="")
if len(update) == 3:
if update[1] == update[2]:
logging.info(f"Yes, current and target firmware versions are same after firmware upgrade")
else:
logging.info(f"No, current and target firmware versions are not same")
pytest.fail(f"No, current and target firmware versions are not same")
else:
logging.info(f"upgrade_status list doesn't have firmware versions to compare: {update}")
pytest.fail(f"upgrade_status list doesn't have firmware versions to compare: {update})")
assert True
@pytest.mark.test_firmware_gw
def test_firmware_upgrade_status_gateway(self, get_testbed_details, get_target_object,
add_firmware_property_after_upgrade):
status = []
for ap in range(len(get_target_object.device_under_tests_info)):
ap_version = get_target_object.dut_library_object.get_ap_version(idx=ap)
current_version_ap = str(ap_version).split()
data = get_target_object.controller_library_object.get_device_by_serial_number(
serial_number=get_target_object.device_under_tests_info[ap]['identifier'])
data = data.json()
allure.attach(name=str(data['firmware']) + str(current_version_ap), body="")
status.append(current_version_ap == data['firmware'].split())
assert False not in status
Reference in New Issue View Git Blame Copy Permalink