mirror of
https://github.com/Telecominfraproject/wlan-testing.git
synced 2025-11-02 03:48:09 +00:00
Added cloudsdk pytests and directory structuring
Signed-off-by: shivam <shivam.thakur@candelatech.com>
This commit is contained in:
shivam
@@ -1,287 +1,79 @@
|
||||
import pytest
|
||||
from time import sleep, gmtime, strftime
|
||||
|
||||
# import files in the current directory
|
||||
import sys
|
||||
import os
|
||||
sys.path.append(os.path.join(os.path.dirname(__file__), 'helpers'))
|
||||
|
||||
sys.path.append(f'..')
|
||||
|
||||
for folder in 'py-json', 'py-scripts':
|
||||
if folder not in sys.path:
|
||||
sys.path.append(f'../../lanforge/lanforge-scripts/{folder}')
|
||||
|
||||
sys.path.append(f'../../libs/lanforge/')
|
||||
sys.path.append(f'../../libs/cloudsdk/')
|
||||
sys.path.append(f'../../libs/apnos/')
|
||||
sys.path.append(f'../../libs/testrails/')
|
||||
sys.path.append(f'../../libs/')
|
||||
|
||||
sys.path.append(f'../test_utility/')
|
||||
|
||||
from utils import *
|
||||
from UnitTestBase import *
|
||||
from JfrogHelper import *
|
||||
from cloudsdk import *
|
||||
from testrail_api import TestRail_Client
|
||||
sys.path.append(
|
||||
os.path.dirname(
|
||||
os.path.realpath( __file__ )
|
||||
)
|
||||
)
|
||||
|
||||
import pytest
|
||||
from configuration_data import PROFILE_DATA
|
||||
|
||||
def pytest_addoption(parser):
|
||||
parser.addini("jfrog-base-url", "jfrog base url")
|
||||
parser.addini("jfrog-user-id", "jfrog username")
|
||||
parser.addini("jfrog-user-password", "jfrog password")
|
||||
|
||||
parser.addini("sdk-base-url", "cloud sdk base url")
|
||||
parser.addini("testbed-name", "cloud sdk base url")
|
||||
parser.addini("sdk-user-id", "cloud sdk username")
|
||||
parser.addini("sdk-user-password", "cloud sdk user password")
|
||||
parser.addini("customer-id", "cloud sdk customer id for the access points")
|
||||
parser.addini("equipment-id", "cloud sdk equipment id for the access point")
|
||||
parser.addini("default-ap-profile", "cloud sdk default AP profile name")
|
||||
|
||||
parser.addini("verbose", "Enable verbose logs?")
|
||||
|
||||
parser.addini("ap-ip", "AP IP address (or can use serial)")
|
||||
parser.addini("ap-username", "AP username")
|
||||
parser.addini("ap-password", "AP password")
|
||||
parser.addini("ap-jumphost-address", "AP jumphost IP address")
|
||||
parser.addini("ap-jumphost-username", "AP jumphost username")
|
||||
parser.addini("ap-jumphost-password", "AP jumphost password")
|
||||
parser.addini("ap-jumphost-port", "AP jumphost port")
|
||||
parser.addini("ap-jumphost-wlan-testing", "AP jumphost wlan-testing code directory")
|
||||
parser.addini("ap-jumphost-tty", "AP jumphost TTY")
|
||||
|
||||
parser.addini("build-id", "What build flavor to use, ie 'pending'")
|
||||
parser.addini("testbed", "Testbed name")
|
||||
parser.addini("mode", "AP Mode, bridge/vlan/nat")
|
||||
parser.addini("skip-wpa", "Should we skip setting up WPA?", default=False)
|
||||
parser.addini("skip-wpa2", "Should we skip setting up WPA2?", default=False)
|
||||
parser.addini("skip-radius", "Should we skip setting up EAP/Radius?", default=False)
|
||||
parser.addini("skip-profiles", "Should we skip setting up profiles")
|
||||
|
||||
parser.addini("ssid-2g-wpa", "Configure ssid-2g-wpa")
|
||||
parser.addini("psk-2g-wpa", "Configure psk-2g-wpa")
|
||||
parser.addini("ssid-5g-wpa", "Configure ssid-5g-wpa")
|
||||
parser.addini("psk-5g-wpa", "Configure psk-5g-wpa")
|
||||
parser.addini("ssid-2g-wpa2", "Configure ssid-2g-wpa2")
|
||||
parser.addini("psk-2g-wpa2", "Configure psk-2g-wpa2")
|
||||
parser.addini("ssid-5g-wpa2", "Configure ssid-5g-wpa2")
|
||||
parser.addini("psk-5g-wpa2", "Configure psk-5g-wpa2")
|
||||
|
||||
parser.addini("sdk-customer-id", "cloud sdk customer id for the access points")
|
||||
parser.addini("testrail-base-url", "testrail base url")
|
||||
parser.addini("testrail-project", "testrail project name to use to generate test reports")
|
||||
parser.addini("testrail-user-id", "testrail username")
|
||||
parser.addini("testrail-user-password", "testrail user password")
|
||||
parser.addini("lanforge-ip-address", "LANforge ip address to connect to")
|
||||
parser.addini("lanforge-port-number", "LANforge port number to connect to")
|
||||
parser.addini("lanforge-2g-radio", "LANforge radio to use")
|
||||
parser.addini("lanforge-5g-radio", "LANforge radio to use")
|
||||
parser.addini("lanforge-radio", "LANforge radio to use")
|
||||
parser.addini("lanforge-ethernet-port", "LANforge ethernet adapter to use")
|
||||
|
||||
add_base_parse_args_pytest(parser)
|
||||
|
||||
# change behaviour
|
||||
parser.addoption(
|
||||
"--skip-update-firmware",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help="skip updating firmware on the AP (useful for local testing)"
|
||||
)
|
||||
# this has to be the last argument
|
||||
# example: --access-points ECW5410 EA8300-EU
|
||||
parser.addoption(
|
||||
"--access-points",
|
||||
nargs="+",
|
||||
# nargs="+",
|
||||
default=[ "ECW5410" ],
|
||||
help="list of access points to test"
|
||||
)
|
||||
|
||||
def pytest_generate_tests(metafunc):
|
||||
metafunc.parametrize("access_points", metafunc.config.getoption('--access-points'), scope="session")
|
||||
if 'access_points' in metafunc.fixturenames:
|
||||
metafunc.parametrize("access_points", metafunc.config.getoption('--access-points'), scope="session")
|
||||
|
||||
# run something after all tests are done regardless of the outcome
|
||||
def pytest_unconfigure(config):
|
||||
print("Tests cleanup done")
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def setup_testrails(request, instantiate_testrail, access_points):
|
||||
if request.config.getoption("--testrail-user-id") == "NONE":
|
||||
yield -1
|
||||
return # needed to stop fixture execution
|
||||
|
||||
if request.config.getoption("--skip-update-firmware"):
|
||||
firmware_update_case = []
|
||||
else:
|
||||
firmware_update_case = [ 2831 ]
|
||||
seen = {None}
|
||||
test_data = []
|
||||
session = request.node
|
||||
for item in session.items:
|
||||
cls = item.getparent(pytest.Class)
|
||||
if cls not in seen:
|
||||
if hasattr(cls.obj, "get_test_data"):
|
||||
test_data.append(cls.obj.get_test_data())
|
||||
seen.add(cls)
|
||||
testrail_project_id = instantiate_testrail.get_project_id(request.config.getini("testrail-project"))
|
||||
runId = instantiate_testrail.create_testrun(
|
||||
name=f'Nightly_model_{access_points}_{strftime("%Y-%m-%d", gmtime())}',
|
||||
case_ids=( [*test_data] + firmware_update_case ),
|
||||
project_id=testrail_project_id
|
||||
)
|
||||
yield runId
|
||||
|
||||
# TODO: Should not be session wide I think, you will want to run different
|
||||
# configurations (bridge, nat, vlan, wpa/wpa2/eap, etc
|
||||
@pytest.fixture(scope="session")
|
||||
def setup_cloudsdk(request, instantiate_cloudsdk, instantiate_testrail):
|
||||
# snippet to do cleanup after all the tests are done
|
||||
@pytest.fixture(scope="function")
|
||||
def setup_cloudsdk(request, instantiate_cloudsdk):
|
||||
def fin():
|
||||
print("Cloud SDK cleanup done")
|
||||
print(f"Cloud SDK cleanup for {request.node.originalname}")
|
||||
request.addfinalizer(fin)
|
||||
|
||||
# Set up bridged setup by default.
|
||||
|
||||
command_line_args = create_command_line_args(request)
|
||||
|
||||
cloud = instantiate_cloudsdk
|
||||
|
||||
cloud.assert_bad_response = True
|
||||
|
||||
equipment_id = instantiate_cloudsdk.equipment_id
|
||||
|
||||
print("equipment-id: %s" % (equipment_id))
|
||||
if equipment_id == "-1":
|
||||
print("ERROR: Could not find equipment-id.")
|
||||
sys.exit(1)
|
||||
|
||||
###Get Current AP info
|
||||
try:
|
||||
ap_cli_info = ssh_cli_active_fw(command_line_args)
|
||||
ap_cli_fw = ap_cli_info['active_fw']
|
||||
except Exception as ex:
|
||||
print(ex)
|
||||
logging.error(logging.traceback.format_exc())
|
||||
ap_cli_info = "ERROR"
|
||||
print("FAILED: Cannot Reach AP CLI.");
|
||||
sys.exit(1)
|
||||
|
||||
# LANForge Information
|
||||
lanforge = {
|
||||
"ip": "localhost",
|
||||
"port": 8806,
|
||||
# "prefix": command_line_args.lanforge_prefix,
|
||||
"2g_radio": "wiphy4",
|
||||
"5g_radio": "wiphy5",
|
||||
"eth_port": "eth2"
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
fw_model = ap_cli_fw.partition("-")[0]
|
||||
|
||||
print('Current Active AP FW from CLI:', ap_cli_fw)
|
||||
|
||||
radius_name = "%s-%s-%s" % (command_line_args.testbed, fw_model, "Radius")
|
||||
|
||||
print("Create profiles")
|
||||
ap_object = CreateAPProfiles(command_line_args, cloud=cloud, client=instantiate_testrail, fw_model=fw_model)
|
||||
|
||||
# Logic to create AP Profiles (Bridge Mode)
|
||||
|
||||
# ap_object.set_ssid_psk_data(ssid_2g_wpa="Pytest-run-2g-wpa",
|
||||
# ssid_5g_wpa="Pytest-run-2g-wpa",
|
||||
# psk_2g_wpa="Pytest-run-2g-wpa",
|
||||
# psk_5g_wpa="Pytest-run-2g-wpa",
|
||||
# ssid_2g_wpa2="Pytest-run-2g-wpa",
|
||||
# ssid_5g_wpa2="Pytest-run-2g-wpa",
|
||||
# psk_2g_wpa2="Pytest-run-2g-wpa",
|
||||
# psk_5g_wpa2="Pytest-run-2g-wpa")
|
||||
|
||||
print(ap_object)
|
||||
today = str(date.today())
|
||||
rid = instantiate_testrail.get_run_id(
|
||||
test_run_name=command_line_args.testrail_run_prefix + fw_model + "_" + today + "_" + "ecw5410-2021-02-12-pending-e8bb466")
|
||||
print("creating Profiles")
|
||||
ssid_template = "TipWlan-Cloud-Wifi"
|
||||
|
||||
if not command_line_args.skip_profiles:
|
||||
if not command_line_args.skip_radius:
|
||||
# Radius Profile needs to be set here
|
||||
radius_name = "Test-Radius-" + str(time.time()).split(".")[0]
|
||||
radius_template = "templates/radius_profile_template.json"
|
||||
ap_object.create_radius_profile(radius_name=radius_name, radius_template=radius_template, rid=rid,
|
||||
key=fw_model)
|
||||
ap_object.create_ssid_profiles(ssid_template=ssid_template, skip_eap=True, skip_wpa=True,
|
||||
skip_wpa2=False, mode="bridge")
|
||||
|
||||
print("Create AP with equipment-id: ", equipment_id)
|
||||
ap_object.create_ap_profile(eq_id=equipment_id, fw_model=fw_model, mode=command_line_args.mode)
|
||||
ap_object.validate_changes(mode=command_line_args.mode)
|
||||
|
||||
print("Profiles Created")
|
||||
data = {"lanforge": lanforge, "ap_object": ap_object}
|
||||
|
||||
yield data
|
||||
yield PROFILE_DATA[request.node.originalname]
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def update_firmware(request, setup_testrails, instantiate_jFrog, instantiate_cloudsdk, access_points):
|
||||
def update_firmware(request, instantiate_jFrog, instantiate_cloudsdk, retrieve_latest_image, access_points):
|
||||
if request.config.getoption("--skip-update-firmware"):
|
||||
return True
|
||||
return
|
||||
yield "update_firmware"
|
||||
|
||||
#access_points is really a single AP.
|
||||
ap = access_points
|
||||
|
||||
if True:
|
||||
latest_image = instantiate_jFrog.get_latest_image(ap)
|
||||
if latest_image is None:
|
||||
print("AP Model: %s doesn't match the available Models"%(ap))
|
||||
sys.exit(1) # TODO: How to return error properly here?
|
||||
|
||||
cloudModel = cloud_sdk_models[ap]
|
||||
logger = None
|
||||
report_data = None
|
||||
test_cases = None
|
||||
testrail_client = None
|
||||
jfrog_user = instantiate_jFrog.get_user()
|
||||
jfrog_pwd = instantiate_jFrog.get_passwd()
|
||||
testrail_rid = 0
|
||||
customer_id = request.config.getoption("--customer-id")
|
||||
equipment_id = instantiate_cloudsdk.equipment_id
|
||||
pf = instantiate_cloudsdk.do_upgrade_ap_fw(request.config, report_data, test_cases, testrail_client,
|
||||
latest_image, cloudModel, ap, jfrog_user, jfrog_pwd, testrail_rid,
|
||||
customer_id, equipment_id, logger)
|
||||
|
||||
return pf
|
||||
@pytest.fixture(scope="session")
|
||||
def retrieve_latest_image(request, access_points):
|
||||
if request.config.getoption("--skip-update-firmware"):
|
||||
return
|
||||
yield "retrieve_latest_image"
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def instantiate_cloudsdk(request):
|
||||
command_line_args = create_command_line_args(request)
|
||||
rv = CloudSDK(command_line_args)
|
||||
|
||||
equipment_id = request.config.getoption("--equipment-id")
|
||||
if equipment_id == "-1":
|
||||
eq_id = ap_ssh_ovsh_nodec(command_line_args, 'id')
|
||||
print("EQ Id: %s" % (eq_id))
|
||||
|
||||
# Now, query equipment to find something that matches.
|
||||
eq = rv.get_customer_equipment(customer_id)
|
||||
for item in eq:
|
||||
for e in item['items']:
|
||||
print(e['id'], " ", e['inventoryId'])
|
||||
if e['inventoryId'].endswith("_%s" % (eq_id)):
|
||||
print("Found equipment ID: %s inventoryId: %s" % (e['id'], e['inventoryId']))
|
||||
equipment_id = str(e['id'])
|
||||
|
||||
rv.equipment_id = equipment_id
|
||||
|
||||
if equipment_id == "-1":
|
||||
print("EQ ID invalid: ", equipment_id)
|
||||
sys.exit(1)
|
||||
|
||||
yield rv
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def instantiate_testrail(request):
|
||||
yield TestRail_Client(create_command_line_args(request))
|
||||
yield "instantiate_cloudsdk"
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def instantiate_jFrog(request):
|
||||
yield GetBuild(
|
||||
request.config.getini("jfrog-user-id"),
|
||||
request.config.getini("jfrog-user-password"),
|
||||
"pending", # TODO make this optional
|
||||
url=request.config.getini("jfrog-base-url")
|
||||
)
|
||||
yield "instantiate_jFrog"
|
||||
Reference in New Issue
Block a user