scottk/wlan-testing
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-testing.git synced 2025-11-02 11:57:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/Telecominfraproject/wlan-testing

Browse Source
This commit is contained in:
Sushant Bawiskar
2021-12-02 01:59:49 +05:30
parent 07ec4c5d82 0ce7c55ff6
commit 818155ac3f
4 changed files with 184 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
libs/apnos/apnos.py
View File

@@ -582,6 +582,24 @@ class APNOS:
pass
return vlan_list
def get_ap_uci_show_ucentral(self):
try:
client = self.ssh_cli_connect()
cmd = "uci show ucentral"
if self.mode:
cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {self.owrt_args} -t {self.tty} --action " \
f"cmd --value \"{cmd}\" "
stdin, stdout, stderr = client.exec_command(cmd)
output = stdout.read()
status = output.decode('utf-8').splitlines()
for i in status:
if i.startswith("ucentral.config.server="):
status = i.split("=")[1]
client.close()
except Exception as e:
print(e)
status = "Error"
return status
if __name__ == '__main__':
obj = {

11
tests/conftest.py
View File

@@ -264,6 +264,11 @@ def get_sdk_version(fixtures_ver):
version = fixtures_ver.get_sdk_version()
yield version
@pytest.fixture(scope="session")
def get_uci_show(fixtures_ver, get_apnos, get_configuration):
uci_show = fixtures_ver.get_uci_show(get_apnos, get_configuration)
yield uci_show
@pytest.fixture(scope="session")
def skip_lf(request):
@@ -522,6 +527,12 @@ def test_access_point(fixtures_ver, request, get_configuration, get_apnos):
yield status
@pytest.fixture(scope="session")
def test_ap_connection_status(fixtures_ver, request, get_configuration, get_apnos):
"""used to check the manager status of AP, should be used as a setup to verify if ap can reach cloud"""
connection, redirector_value = fixtures_ver.get_ap_status_logs(get_configuration, get_apnos)
yield connection, redirector_value
@pytest.fixture(scope="session")
def traffic_generator_connectivity(testbed, get_configuration):
"""Verify if traffic generator is reachable"""

33
tests/fixtures_2x.py
View File

@@ -318,6 +318,39 @@ class Fixtures_2x:
version_list.append(version)
return version_list
def get_uci_show(self, get_apnos, get_configuration):
version_list = []
for access_point_info in get_configuration['access_point']:
ap_ssh = get_apnos(access_point_info)
connectivity_data = ap_ssh.run_generic_command(cmd="uci show ucentral.config.server")
# connectivity_data.pop(0)
# connectivity_data.pop(1)
break
# version_list.append(connectivity_data)
return connectivity_data[1]
def get_ap_status_logs(self, get_configuration, get_apnos):
connected = 0
redirector_data = None
for access_point_info in get_configuration['access_point']:
ap_ssh = get_apnos(access_point_info, sdk="2.x")
# for i in range(15):
connectivity_data = ap_ssh.run_generic_command(cmd="ubus call ucentral status")
if "disconnected" in str(connectivity_data):
print("AP in disconnected state, sleeping for 30 sec")
# time.sleep(30)
connected = 0
# # if i == 10:
# print("rebooting AP")
# ap_ssh.reboot()
# print("sleep for 300 sec")
# time.sleep(300)
else:
connected = 1
redirector_data = ap_ssh.run_generic_command(cmd="cat /etc/ucentral/redirector.json")
return connected, redirector_data
def get_sdk_version(self):
version = self.controller_obj.get_sdk_version()
return version

132
tests/test_connectivity.py
View File

@@ -1,10 +1,12 @@
"""
Test Case Module: Testing Basic Connectivity with Resources
"""
import time
import allure
import pytest
import requests
import json
pytestmark = [pytest.mark.test_resources, pytest.mark.sanity, pytest.mark.uc_sanity,
pytest.mark.sanity_55]
@@ -17,26 +19,133 @@ class TestResources(object):
@pytest.mark.test_cloud_controller
@pytest.mark.uc_sanity
@allure.testcase(name="test_controller_connectivity", url="")
def test_controller_connectivity(self, setup_controller):
def test_controller_connectivity(self, setup_controller, get_configuration):
"""Test case to verify cloud Controller Connectivity"""
login_response_json = setup_controller.login_resp.json()
response_code = setup_controller.login_resp.status_code
allure.attach(name="Login Response Code", body=str(response_code))
allure.attach(name="Login Response JSON",
body=str(login_response_json),
attachment_type=allure.attachment_type.JSON)
version = setup_controller.get_sdk_version()
print(version)
assert response_code == 200
request_url = setup_controller.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(setup_controller.login_resp.json()))
if response_code != 200:
pytest.exit(
"exiting from pytest, login response is no 200: " + str(setup_controller.login_resp.status_code))
gw_system_info = setup_controller.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 = setup_controller.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))
# if gw_status_check != 200:
# for i in range(10):
# if setup_controller.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 setup_controller.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 setup_controller.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 setup_controller.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 = setup_controller.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 = setup_controller.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_points_connectivity", url="")
def test_access_points_connectivity(self, test_access_point, fixtures_ver):
def test_access_points_connectivity(self, setup_controller, get_uci_show, test_access_point, get_configuration,
test_ap_connection_status, fixtures_ver):
"""Test case to verify Access Points Connectivity"""
data = []
for status in test_access_point:
data.append(status[0])
connection, redirector = test_ap_connection_status
allure.attach(name="AP - Cloud connectivity info", body=str(fixtures_ver.ubus_connection))
print("test_ap_connection_status: ", connection, redirector)
expected_sdk = str(get_configuration["controller"]['url'].replace("https://sec", "\'gw").replace(":16001", "\'"))
print("Expected SDK: ", expected_sdk)
allure.attach(name="Exoected SDK: ", body=str(expected_sdk))
print("SDK On AP: ", str(get_uci_show.split("=")[1]))
allure.attach(name="SDK Pointed by AP: ", body=str(get_uci_show.split("=")[1]))
if expected_sdk not in get_uci_show:
pytest.exit("AP has invalid Redirector")
if test_ap_connection_status[0] == 0:
pytest.exit("AP in Disconnected State")
assert False not in data
@pytest.mark.traffic_generator_connectivity
@@ -135,9 +244,12 @@ class TestFMS(object):
assert False not in PASS
@pytest.mark.firmware_upgrade
def test_firmware_upgrade_request(self, firmware_upgrade, get_ap_logs):
def test_firmware_upgrade_request(self, firmware_upgrade, get_ap_logs, test_ap_connection_status):
for update in firmware_upgrade:
allure.attach(name='serial: ' + update[0], body="")
if test_ap_connection_status[0] == 0:
assert False
pytest.exit("AP in Disconnected State")
assert True
@pytest.mark.test_firmware_ap