scottk/wlan-testing
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-testing.git synced 2026-01-10 22:31:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

unit-tests: Move sw update into libraries.

Browse Source 
Add sdk_upgrade_fw.py to call the upgrade (only).

This needs more testing, but it at least partially
works.

Signed-off-by: Ben Greear <greearb@candelatech.com>
This commit is contained in:
Ben Greear
2021-02-01 17:12:12 -08:00
parent 83a64afe69
commit 855cfc42c6
6 changed files with 484 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
unit_tests/JfrogHelper.py Normal file
View File

@@ -0,0 +1,35 @@
import ssl
import base64
import urllib.request
from bs4 import BeautifulSoup
import re
###Class for jfrog Interaction
class GetBuild:
def __init__(self, jfrog_user, jfrog_passwd):
self.user = jfrog_user
self.password = jfrog_passwd
ssl._create_default_https_context = ssl._create_unverified_context
def get_latest_image(self, url, build):
auth = str(
base64.b64encode(
bytes('%s:%s' % (self.user, self.password), 'utf-8')
),
'ascii'
).strip()
headers = {'Authorization': 'Basic ' + auth}
''' FIND THE LATEST FILE NAME'''
# print(url)
req = urllib.request.Request(url, headers=headers)
response = urllib.request.urlopen(req)
html = response.read()
soup = BeautifulSoup(html, features="html.parser")
##find the last pending link on dev
last_link = soup.find_all('a', href=re.compile(build))[-1]
latest_file = last_link['href']
latest_fw = latest_file.replace('.tar.gz', '')
return latest_fw

245
unit_tests/Nightly_Sanity.py
View File

@@ -1,7 +1,9 @@
#!/usr/bin/python3
from JfrogHelper import *
from UnitTestBase import *
from cloudsdk import CreateAPProfiles
parser = argparse.ArgumentParser(description="Nightly Combined Tests", add_help=False)
parser.add_argument("--default_ap_profile", type=str,
help="Default AP profile to use as basis for creating new ones, typically: TipWlan-2-Radios or TipWlan-3-Radios",
@@ -59,35 +61,6 @@ if command_line_args.testbed == None:
client: TestRail_Client = TestRail_Client(command_line_args)
###Class for jfrog Interaction
class GetBuild:
def __init__(self):
self.user = jfrog_user
self.password = jfrog_pwd
ssl._create_default_https_context = ssl._create_unverified_context
def get_latest_image(self, url, build):
auth = str(
base64.b64encode(
bytes('%s:%s' % (self.user, self.password), 'utf-8')
),
'ascii'
).strip()
headers = {'Authorization': 'Basic ' + auth}
''' FIND THE LATEST FILE NAME'''
# print(url)
req = urllib.request.Request(url, headers=headers)
response = urllib.request.urlopen(req)
html = response.read()
soup = BeautifulSoup(html, features="html.parser")
##find the last pending link on dev
last_link = soup.find_all('a', href=re.compile(build))[-1]
latest_file = last_link['href']
latest_fw = latest_file.replace('.tar.gz', '')
return latest_fw
###Class for Tests
class RunTest:
def Single_Client_Connectivity(self, port, radio, ssid_name, ssid_psk, security, station, test_case, rid):
@@ -282,7 +255,7 @@ for model in ap_models:
###Check Latest FW on jFrog
jfrog_url = 'https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/'
url = jfrog_url + apModel + "/dev/"
Build: GetBuild = GetBuild()
Build: GetBuild = GetBuild(jfrog_user, jfrog_pwd)
latest_image = Build.get_latest_image(url, build)
print(model, "Latest FW on jFrog:", latest_image)
ap_latest_dict[model] = latest_image
@@ -303,7 +276,7 @@ for key in equipment_ids:
##Get Bearer Token to make sure its valid (long tests can require re-auth)
bearer = cloud.get_bearer(cloudSDK_url, cloud_type)
###Get Current AP Firmware and upgrade
print("AP MODEL UNDER TEST IS", key)
try:
ap_cli_info = ssh_cli_active_fw(command_line_args)
@@ -321,34 +294,10 @@ for key in equipment_ids:
##Compare Latest and Current AP FW and Upgrade
latest_ap_image = ap_latest_dict[fw_model]
do_upgrade = False
if ap_cli_fw == latest_ap_image and command_line_args.force_upgrade != True:
print('FW does not require updating')
report_data['fw_available'][key] = "No"
logger.info(fw_model + " does not require upgrade.")
cloudsdk_cluster_info = {
"date": "N/A",
"commitId": "N/A",
"projectVersion": "N/A"
}
report_data['cloud_sdk'][key] = cloudsdk_cluster_info
if ap_cli_fw != latest_ap_image and command_line_args.skip_upgrade == True:
print('FW needs updating, but skip_upgrade is True, so skipping upgrade')
report_data['fw_available'][key] = "No"
logger.info(fw_model + " firmware upgrade skipped, running with " + ap_cli_fw)
cloudsdk_cluster_info = {
"date": "N/A",
"commitId": "N/A",
"projectVersion": "N/A"
}
report_data['cloud_sdk'][key] = cloudsdk_cluster_info
if (ap_cli_fw != latest_ap_image or command_line_args.force_upgrade == True) and not command_line_args.skip_upgrade:
print('Updating firmware, old: %s new: %s'%(ap_cli_fw, latest_ap_image))
do_upgrade = True
report_data['fw_available'][key] = "Yes"
report_data['fw_under_test'][key] = latest_ap_image
do_upgrade = cloud.should_upgrade_ap_fw(bearer, command_line_args, report_data, latest_ap_image, fw_model, ap_cli_fw)
###Create Test Run
today = str(date.today())
@@ -395,186 +344,14 @@ for key in equipment_ids:
json.dump(report_data, report_json_file)
if do_upgrade:
###Test Create Firmware Version
test_id_fw = test_cases["create_fw"]
latest_image = ap_latest_dict[key]
cloudModel = cloud_sdk_models[key]
print(cloudModel)
firmware_list_by_model = cloud.CloudSDK_images(cloudModel, cloudSDK_url, bearer)
print("Available", cloudModel, "Firmware on CloudSDK:", firmware_list_by_model)
if latest_image in firmware_list_by_model:
print("Latest Firmware", latest_image, "is already on CloudSDK, need to delete to test create FW API")
old_fw_id = cloud.get_firmware_id(latest_image, cloudSDK_url, bearer)
delete_fw = cloud.delete_firmware(str(old_fw_id), cloudSDK_url, bearer)
fw_url = "https://" + jfrog_user + ":" + jfrog_pwd + "@tip.jfrog.io/artifactory/tip-wlan-ap-firmware/" + key + "/dev/" + latest_image + ".tar.gz"
commit = latest_image.split("-")[-1]
try:
fw_upload_status = cloud.firwmare_upload(commit, cloudModel, latest_image, fw_url, cloudSDK_url,
bearer)
fw_id = fw_upload_status['id']
print("Upload Complete.", latest_image, "FW ID is", fw_id)
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
report_data['tests'][key][test_id_fw] = "passed"
except:
fw_upload_status = 'error'
print("Unable to upload new FW version. Skipping Sanity on AP Model")
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
report_data['tests'][key][test_id_fw] = "failed"
continue
else:
print("Latest Firmware is not on CloudSDK! Uploading...")
fw_url = "https://" + jfrog_user + ":" + jfrog_pwd + "@tip.jfrog.io/artifactory/tip-wlan-ap-firmware/" + key + "/dev/" + latest_image + ".tar.gz"
commit = latest_image.split("-")[-1]
try:
fw_upload_status = cloud.firwmare_upload(commit, cloudModel, latest_image, fw_url, cloudSDK_url,
bearer)
fw_id = fw_upload_status['id']
print("Upload Complete.", latest_image, "FW ID is", fw_id)
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
report_data['tests'][key][test_id_fw] = "passed"
except:
fw_upload_status = 'error'
print("Unable to upload new FW version. Skipping Sanity on AP Model")
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
report_data['tests'][key][test_id_fw] = "failed"
continue
# Upgrade AP firmware
print("Upgrading...firmware ID is: ", fw_id)
upgrade_fw = cloud.update_firmware(equipment_id, str(fw_id), cloudSDK_url, bearer)
logger.info("Lab " + fw_model + " Requires FW update")
print(upgrade_fw)
if "success" in upgrade_fw:
if upgrade_fw["success"] == True:
print("CloudSDK Upgrade Request Success")
report_data['tests'][key][test_cases["upgrade_api"]] = "passed"
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=1, msg='Upgrade request using API successful')
logger.info('Firmware upgrade API successfully sent')
else:
print("Cloud SDK Upgrade Request Error!")
# mark upgrade test case as failed with CloudSDK error
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5, msg='Error requesting upgrade via API')
report_data['tests'][key][test_cases["upgrade_api"]] = "failed"
logger.warning('Firmware upgrade API failed to send')
continue
else:
print("Cloud SDK Upgrade Request Error!")
# mark upgrade test case as failed with CloudSDK error
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5,msg='Error requesting upgrade via API')
report_data['tests'][key][test_cases["upgrade_api"]] = "failed"
logger.warning('Firmware upgrade API failed to send')
continue
sdk_ok = False
for i in range(10):
time.sleep(30)
# Check if upgrade success is displayed on CloudSDK
test_id_cloud = test_cases["cloud_fw"]
cloud_ap_fw = cloud.ap_firmware(customer_id, equipment_id, cloudSDK_url, bearer)
print('Current AP Firmware from CloudSDK: %s latest-image: %s'%(cloud_ap_fw, latest_ap_image))
logger.info('AP Firmware from CloudSDK: ' + cloud_ap_fw)
if cloud_ap_fw == "ERROR":
print("AP FW Could not be read from CloudSDK")
elif cloud_ap_fw == latest_ap_image:
print("CloudSDK status shows upgrade successful!")
sdk_ok = True
break
else:
print("AP FW from CloudSDK status is not latest build. Will try again in 30 seconds.")
cli_ok = False
if sdk_ok:
for i in range(10):
# Check if upgrade successful on AP CLI
test_id_cli = test_cases["ap_upgrade"]
try:
ap_cli_info = ssh_cli_active_fw(command_line_args)
ap_cli_fw = ap_cli_info['active_fw']
print("CLI reporting AP Active FW as:", ap_cli_fw)
logger.info('Firmware from CLI: ' + ap_cli_fw)
if ap_cli_fw == latest_image:
cli_ok = True
break
else:
print("probed api-cli-fw: %s != latest-image: %s"%(ap_cli_fw, latest_image))
continue
except Exception as ex:
ap_cli_info = "ERROR"
print(ex)
logging.error(logging.traceback.format_exc())
print("Cannot Reach AP CLI to confirm upgrade!")
logger.warning('Cannot Reach AP CLI to confirm upgrade!')
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot reach AP after upgrade to check CLI - re-test required')
continue
time.sleep(30)
else:
print("ERROR: Cloud did not report firmware upgrade within expiration time.")
if not sdk_ok and cli_ok:
continue
if cloud_ap_fw == latest_ap_image and ap_cli_fw == latest_ap_image:
print("CloudSDK and AP CLI both show upgrade success, passing upgrade test case")
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + latest_ap_image + ' successful')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=1,
msg='CLOUDSDK reporting correct firmware version.')
report_data['tests'][key][test_id_cli] = "passed"
report_data['tests'][key][test_id_cloud] = "passed"
print(report_data['tests'][key])
elif cloud_ap_fw != latest_ap_image and ap_cli_fw == latest_ap_image:
print("AP CLI shows upgrade success - CloudSDK reporting error!")
##Raise CloudSDK error but continue testing
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + latest_ap_image + ' successful.')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
report_data['tests'][key][test_id_cli] = "passed"
report_data['tests'][key][test_id_cloud] = "failed"
print(report_data['tests'][key])
elif cloud_ap_fw == latest_ap_image and ap_cli_fw != latest_ap_image:
print("AP CLI shows upgrade failed - CloudSDK reporting error!")
# Testrail TC fail
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + latest_ap_image + ' Failed')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
report_data['tests'][key][test_id_cli] = "failed"
report_data['tests'][key][test_id_cloud] = "failed"
print(report_data['tests'][key])
continue
elif cloud_ap_fw != latest_ap_image and ap_cli_fw != latest_ap_image:
print("Upgrade Failed! Confirmed on CloudSDK and AP CLI. Upgrade test case failed.")
##fail TR testcase and exit
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + latest_ap_image + ' Failed')
report_data['tests'][key][test_id_cli] = "failed"
print(report_data['tests'][key])
continue
else:
print("Unable to determine upgrade status. Skipping AP variant")
# update TR testcase as error
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot determine upgrade status - re-test required')
report_data['tests'][key][test_id_cli] = "error"
print(report_data['tests'][key])
continue
pf = cloud.do_upgrade_ap_fw(bearer, command_line_args, report_data, test_cases, client,
latest_image, cloudModel, key, jfrog_user, jfrog_pwd, rid,
customer_id, equipment_id, logger, latest_ap_image)
print(report_data)
if not pf:
continue # Try next model
# TODO: Fix indentation, break all this up into small test cases anyway.
if True:

2
unit_tests/UnitTestBase.py
View File

@@ -90,7 +90,7 @@ class UnitTestBase:
default = False)
self.parser.add_argument("--force-upgrade", type=bool, help="Force upgrading firmware even if it is already current version",
default = False)
self.parser.add_argument("-m", "--model", type=str, choices=['ea8300', 'ecw5410', 'ecw5211', 'ec420'],
self.parser.add_argument("-m", "--model", type=str, choices=['ea8300', 'ecw5410', 'ecw5211', 'ec420', 'wf188n'],
help="AP model to be run", required=True)
self.parser.add_argument("--equipment_id", type=str,
help="AP model ID, as exists in the cloud-sdk. -1 to auto-detect.",

287
unit_tests/cloudsdk.py
View File

@@ -63,8 +63,9 @@ class CloudSDK:
bearer_token = token_data['access_token']
return(bearer_token)
def check_response(self, response, headers, data_str):
def check_response(self, response, headers, data_str, url):
if response.status_code >= 500:
print("check-response: ERROR, url: ", url)
print("response-status: ", response.status_code)
print("response-headers: ", response.headers)
print("headers: ", headers)
@@ -74,6 +75,242 @@ class CloudSDK:
return False
return True
def should_upgrade_ap_fw(self, bearer, command_line_args, report_data, latest_ap_image, fw_model, ap_cli_fw,
logger):
do_upgrade = False
if ap_cli_fw == latest_ap_image and command_line_args.force_upgrade != True:
print('FW does not require updating')
if report_data:
report_data['fw_available'][key] = "No"
logger.info(fw_model + " does not require upgrade.")
cloudsdk_cluster_info = {
"date": "N/A",
"commitId": "N/A",
"projectVersion": "N/A"
}
if report_data:
report_data['cloud_sdk'][key] = cloudsdk_cluster_info
if ap_cli_fw != latest_ap_image and command_line_args.skip_upgrade == True:
print('FW needs updating, but skip_upgrade is True, so skipping upgrade')
if report_data:
report_data['fw_available'][key] = "No"
logger.info(fw_model + " firmware upgrade skipped, running with " + ap_cli_fw)
cloudsdk_cluster_info = {
"date": "N/A",
"commitId": "N/A",
"projectVersion": "N/A"
}
if report_data:
report_data['cloud_sdk'][key] = cloudsdk_cluster_info
if (ap_cli_fw != latest_ap_image or command_line_args.force_upgrade == True) and not command_line_args.skip_upgrade:
print('Updating firmware, old: %s new: %s'%(ap_cli_fw, latest_ap_image))
do_upgrade = True
if report_data:
report_data['fw_available'][key] = "Yes"
report_data['fw_under_test'][key] = latest_ap_image
return do_upgrade
# client is testrail client
def do_upgrade_ap_fw(self, bearer, command_line_args, report_data, test_cases, client, latest_image, cloudModel, model,
jfrog_user, jfrog_pwd, testrails_rid, customer_id, equipment_id, logger, latest_ap_image):
###Test Create Firmware Version
key = model
rid = testrails_rid
cloudSDK_url = command_line_args.sdk_base_url
test_id_fw = test_cases["create_fw"]
print(cloudModel)
firmware_list_by_model = self.CloudSDK_images(cloudModel, cloudSDK_url, bearer)
print("Available", cloudModel, "Firmware on CloudSDK:", firmware_list_by_model)
if latest_image in firmware_list_by_model:
print("Latest Firmware", latest_image, "is already on CloudSDK, need to delete to test create FW API")
old_fw_id = self.get_firmware_id(latest_image, cloudSDK_url, bearer)
delete_fw = self.delete_firmware(str(old_fw_id), cloudSDK_url, bearer)
fw_url = "https://" + jfrog_user + ":" + jfrog_pwd + "@tip.jfrog.io/artifactory/tip-wlan-ap-firmware/" + key + "/dev/" + latest_image + ".tar.gz"
commit = latest_image.split("-")[-1]
try:
fw_upload_status = self.firwmare_upload(commit, cloudModel, latest_image, fw_url, cloudSDK_url,
bearer)
fw_id = fw_upload_status['id']
print("Upload Complete.", latest_image, "FW ID is", fw_id)
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
if report_data:
report_data['tests'][key][test_id_fw] = "passed"
except:
fw_upload_status = 'error'
print("Unable to upload new FW version. Skipping Sanity on AP Model")
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
if report_data:
report_data['tests'][key][test_id_fw] = "failed"
return False
else:
print("Latest Firmware is not on CloudSDK! Uploading...")
fw_url = "https://" + jfrog_user + ":" + jfrog_pwd + "@tip.jfrog.io/artifactory/tip-wlan-ap-firmware/" + key + "/dev/" + latest_image + ".tar.gz"
commit = latest_image.split("-")[-1]
try:
fw_upload_status = self.firwmare_upload(commit, cloudModel, latest_image, fw_url, cloudSDK_url,
bearer)
fw_id = fw_upload_status['id']
print("Upload Complete.", latest_image, "FW ID is", fw_id)
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=1,
msg='Create new FW version by API successful')
if report_data:
report_data['tests'][key][test_id_fw] = "passed"
except:
fw_upload_status = 'error'
print("Unable to upload new FW version. Skipping Sanity on AP Model")
client.update_testrail(case_id=test_id_fw, run_id=rid, status_id=5,
msg='Error creating new FW version by API')
if report_data:
report_data['tests'][key][test_id_fw] = "failed"
return False
# Upgrade AP firmware
print("Upgrading...firmware ID is: ", fw_id)
upgrade_fw = self.update_firmware(equipment_id, str(fw_id), cloudSDK_url, bearer)
logger.info("Lab " + model + " Requires FW update")
print(upgrade_fw)
if "success" in upgrade_fw:
if upgrade_fw["success"] == True:
print("CloudSDK Upgrade Request Success")
if report_data:
report_data['tests'][key][test_cases["upgrade_api"]] = "passed"
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=1, msg='Upgrade request using API successful')
logger.info('Firmware upgrade API successfully sent')
else:
print("Cloud SDK Upgrade Request Error!")
# mark upgrade test case as failed with CloudSDK error
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5, msg='Error requesting upgrade via API')
if report_data:
report_data['tests'][key][test_cases["upgrade_api"]] = "failed"
logger.warning('Firmware upgrade API failed to send')
return False
else:
print("Cloud SDK Upgrade Request Error!")
# mark upgrade test case as failed with CloudSDK error
client.update_testrail(case_id=test_cases["upgrade_api"], run_id=rid, status_id=5,msg='Error requesting upgrade via API')
if report_data:
report_data['tests'][key][test_cases["upgrade_api"]] = "failed"
logger.warning('Firmware upgrade API failed to send')
return False
sdk_ok = False
for i in range(10):
time.sleep(30)
# Check if upgrade success is displayed on CloudSDK
test_id_cloud = test_cases["cloud_fw"]
cloud_ap_fw = self.ap_firmware(customer_id, equipment_id, cloudSDK_url, bearer)
print('Current AP Firmware from CloudSDK: %s latest-image: %s'%(cloud_ap_fw, latest_ap_image))
logger.info('AP Firmware from CloudSDK: ' + cloud_ap_fw)
if cloud_ap_fw == "ERROR":
print("AP FW Could not be read from CloudSDK")
elif cloud_ap_fw == latest_ap_image:
print("CloudSDK status shows upgrade successful!")
sdk_ok = True
break
else:
print("AP FW from CloudSDK status is not latest build. Will try again in 30 seconds.")
cli_ok = False
if sdk_ok:
for i in range(10):
# Check if upgrade successful on AP CLI
test_id_cli = test_cases["ap_upgrade"]
try:
ap_cli_info = ssh_cli_active_fw(command_line_args)
ap_cli_fw = ap_cli_info['active_fw']
print("CLI reporting AP Active FW as:", ap_cli_fw)
logger.info('Firmware from CLI: ' + ap_cli_fw)
if ap_cli_fw == latest_image:
cli_ok = True
break
else:
print("probed api-cli-fw: %s != latest-image: %s"%(ap_cli_fw, latest_image))
continue
except Exception as ex:
ap_cli_info = "ERROR"
print(ex)
logging.error(logging.traceback.format_exc())
print("Cannot Reach AP CLI to confirm upgrade!")
logger.warning('Cannot Reach AP CLI to confirm upgrade!')
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot reach AP after upgrade to check CLI - re-test required')
continue
time.sleep(30)
else:
print("ERROR: Cloud did not report firmware upgrade within expiration time.")
if not (sdk_ok and cli_ok):
return False # Cannot talk to AP/Cloud, cannot make intelligent decision on pass/fail
# Check status
if cloud_ap_fw == latest_ap_image and ap_cli_fw == latest_ap_image:
print("CloudSDK and AP CLI both show upgrade success, passing upgrade test case")
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + latest_ap_image + ' successful')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=1,
msg='CLOUDSDK reporting correct firmware version.')
if report_data:
report_data['tests'][key][test_id_cli] = "passed"
report_data['tests'][key][test_id_cloud] = "passed"
print(report_data['tests'][key])
return True
elif cloud_ap_fw != latest_ap_image and ap_cli_fw == latest_ap_image:
print("AP CLI shows upgrade success - CloudSDK reporting error!")
##Raise CloudSDK error but continue testing
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=1,
msg='Upgrade to ' + latest_ap_image + ' successful.')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
if report_data:
report_data['tests'][key][test_id_cli] = "passed"
report_data['tests'][key][test_id_cloud] = "failed"
print(report_data['tests'][key])
return True
elif cloud_ap_fw == latest_ap_image and ap_cli_fw != latest_ap_image:
print("AP CLI shows upgrade failed - CloudSDK reporting error!")
# Testrail TC fail
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + latest_ap_image + ' Failed')
client.update_testrail(case_id=test_id_cloud, run_id=rid, status_id=5,
msg='CLOUDSDK reporting incorrect firmware version.')
if report_data:
report_data['tests'][key][test_id_cli] = "failed"
report_data['tests'][key][test_id_cloud] = "failed"
print(report_data['tests'][key])
return False
elif cloud_ap_fw != latest_ap_image and ap_cli_fw != latest_ap_image:
print("Upgrade Failed! Confirmed on CloudSDK and AP CLI. Upgrade test case failed.")
##fail TR testcase and exit
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=5,
msg='AP failed to download or apply new FW. Upgrade to ' + latest_ap_image + ' Failed')
if report_data:
report_data['tests'][key][test_id_cli] = "failed"
print(report_data['tests'][key])
return False
else:
print("Unable to determine upgrade status. Skipping AP variant")
# update TR testcase as error
client.update_testrail(case_id=test_id_cli, run_id=rid, status_id=4,
msg='Cannot determine upgrade status - re-test required')
if report_data:
report_data['tests'][key][test_id_cli] = "error"
print(report_data['tests'][key])
return False
def ap_firmware(self, customer_id, equipment_id, cloudSDK_url, bearer):
equip_fw_url = cloudSDK_url+"/portal/status/forEquipment?customerId="+customer_id+"&equipmentId="+equipment_id
@@ -89,7 +326,7 @@ class CloudSDK:
current_ap_fw = status_data[2]['details']['reportedSwVersion']
return current_ap_fw
else:
self.check_response(status_response, headers, payload)
self.check_response(status_response, headers, payload, equip_fw_url)
return("ERROR")
def CloudSDK_images(self, apModel, cloudSDK_url, bearer):
@@ -99,7 +336,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", getFW_url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, getFW_url)
ap_fw_details = response.json()
###return ap_fw_details
fwlist = []
@@ -118,7 +355,7 @@ class CloudSDK:
}
response = requests.request("POST", fw_upload_url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, fw_upload_url)
#print(response)
upload_result = response.json()
return(upload_result)
@@ -132,7 +369,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", fw_id_url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, fw_id_url)
fw_data = response.json()
latest_fw_id = fw_data['id']
return latest_fw_id
@@ -150,7 +387,7 @@ class CloudSDK:
while True:
print("Request %i: in get-paged-url, url: %s"%(req, url))
response = requests.request("GET", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
rjson = response.json()
rv.append(rjson)
if not 'context' in rjson:
@@ -176,7 +413,7 @@ class CloudSDK:
print("Get-url, url: %s"%(url))
response = requests.request("GET", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
return response.json()
def get_customer_profiles(self, cloudSDK_url, bearer, customer_id, object_id):
@@ -210,7 +447,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("DELETE", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
return(response)
def delete_equipment(self, cloudSDK_url, bearer, eq_id):
@@ -221,7 +458,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("DELETE", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
return(response)
def get_customer_locations(self, cloudSDK_url, bearer, customer_id):
@@ -268,7 +505,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", fw_id_url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, fw_id_url)
return response.json()
def delete_firmware(self, fw_id, cloudSDK_url, bearer):
@@ -278,7 +515,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("DELETE", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
return(response)
def update_firmware(self, equipment_id, latest_firmware_id, cloudSDK_url, bearer):
@@ -290,7 +527,7 @@ class CloudSDK:
}
response = requests.request("POST", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
#print(response.text)
return response.json()
@@ -304,7 +541,7 @@ class CloudSDK:
}
response = requests.request("POST", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
#print(response.text)
return response.json()
@@ -318,7 +555,7 @@ class CloudSDK:
}
response = requests.request("POST", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
#print(response.text)
return response.json()
@@ -332,7 +569,7 @@ class CloudSDK:
}
response = requests.request("POST", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
#print(response.text)
return response.json()
@@ -346,7 +583,7 @@ class CloudSDK:
}
response = requests.request("POST", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
#print(response.text)
return response.json()
@@ -359,7 +596,7 @@ class CloudSDK:
}
response = requests.request("GET", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
print(response)
###Add Lab Profile ID to Equipment
@@ -376,7 +613,7 @@ class CloudSDK:
}
response = requests.request("PUT", url, headers=headers, data=json.dumps(equipment_info))
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
print(response)
def get_cloudsdk_version(self, cloudSDK_url, bearer):
@@ -388,7 +625,7 @@ class CloudSDK:
'Authorization': 'Bearer ' + bearer
}
response = requests.request("GET", url, headers=headers, data=payload)
self.check_response(response, headers, payload)
self.check_response(response, headers, payload, url)
cloud_sdk_version = response.json()
return cloud_sdk_version
@@ -408,7 +645,7 @@ class CloudSDK:
data_str = json.dumps(profile)
print("Creating new ap-profile, data: %s"%(data_str))
response = requests.request("POST", url, headers=headers, data=data_str)
self.check_response(response, headers, data_str)
self.check_response(response, headers, data_str, url)
ap_profile = response.json()
print("New AP profile: ", ap_profile)
ap_profile_id = ap_profile['id']
@@ -431,7 +668,7 @@ class CloudSDK:
data_str = json.dumps(profile)
print("Updating ap-profile, data: %s"%(data_str))
response = requests.request("PUT", url, headers=headers, data=data_str)
self.check_response(response, headers, data_str)
self.check_response(response, headers, data_str, url)
print(response)
return profile['id']
@@ -455,7 +692,7 @@ class CloudSDK:
}
data_str = json.dumps(profile)
response = requests.request("POST", url, headers=headers, data=data_str)
self.check_response(response, headers, data_str)
self.check_response(response, headers, data_str, url)
ssid_profile = response.json()
return ssid_profile['id']
@@ -487,7 +724,7 @@ class CloudSDK:
}
data_str = json.dumps(profile)
response = requests.request("PUT", url, headers=headers, data=data_str)
self.check_response(response, headers, data_str)
self.check_response(response, headers, data_str, url)
return profile['id']
def create_radius_profile(self, cloudSDK_url, bearer, customer_id, template, name, subnet_name, subnet, subnet_mask,
@@ -528,7 +765,7 @@ class CloudSDK:
data_str = json.dumps(profile)
print("Sending json to create radius: %s"%(data_str))
response = requests.request("POST", url, headers=headers, data=data_str)
self.check_response(response, headers, data_str)
self.check_response(response, headers, data_str, url)
radius_profile = response.json()
print(radius_profile)
radius_profile_id = radius_profile['id']
@@ -572,7 +809,7 @@ class CloudSDK:
}
data_str = json.dumps(profile)
response = requests.request("PUT", url, headers=headers, data=data_str)
self.check_response(response, headers, data_str)
self.check_response(response, headers, data_str, url)
# TODO: Commented code below is wrong, obviously in hindsight. But, need to parse response
# and throw exception or something if it is an error code.
#response = requests.request("PUT", url, headers=headers, data=profile)

3
unit_tests/lab_ap_info.py
View File

@@ -24,7 +24,8 @@ cloud_sdk_models = {
"ec420": "EC420-G1",
"ea8300": "EA8300-CA",
"ecw5211": "ECW5211",
"ecw5410": "ECW5410"
"ecw5410": "ECW5410",
"wf188n": "WF188N"
}
mimo_5g = {

173
unit_tests/sdk_upgrade_fw.py Executable file
View File

@@ -0,0 +1,173 @@
#!/usr/bin/python3
# Example to upgrade firmware on NOLA-12 testbed:
"""
./sdk_upgrade_fw.py --testrail-user-id NONE --model wf188n --ap-jumphost-address localhost --ap-jumphost-port 8823 \
--ap-jumphost-password pumpkin77 --ap-jumphost-tty /dev/ttyAP1 --testbed \"NOLA-12\" \
--sdk-base-url https://wlan-portal-svc-ben-testbed.cicd.lab.wlan.tip.build --force-upgrade true
"""
# Example to upgrade fw on NOLA-01 testbed
from JfrogHelper import *
from UnitTestBase import *
from cloudsdk import CreateAPProfiles
parser = argparse.ArgumentParser(description="SDK Upgrade Firmware", add_help=False)
base = UnitTestBase("sdk-upgrade-fw", parser)
command_line_args = base.command_line_args
# cmd line takes precedence over env-vars.
cloudSDK_url = command_line_args.sdk_base_url # was os.getenv('CLOUD_SDK_URL')
local_dir = command_line_args.local_dir # was os.getenv('SANITY_LOG_DIR')
report_path = command_line_args.report_path # was os.getenv('SANITY_REPORT_DIR')
report_template = command_line_args.report_template # was os.getenv('REPORT_TEMPLATE')
## TestRail Information
tr_user = command_line_args.testrail_user_id # was os.getenv('TR_USER')
tr_pw = command_line_args.testrail_user_password # was os.getenv('TR_PWD')
milestoneId = command_line_args.milestone # was os.getenv('MILESTONE')
projectId = command_line_args.testrail_project # was os.getenv('PROJECT_ID')
testRunPrefix = command_line_args.testrail_run_prefix # os.getenv('TEST_RUN_PREFIX')
##Jfrog credentials
jfrog_user = command_line_args.jfrog_user_id # was os.getenv('JFROG_USER')
jfrog_pwd = command_line_args.jfrog_user_password # was os.getenv('JFROG_PWD')
##EAP Credentials
identity = command_line_args.eap_id # was os.getenv('EAP_IDENTITY')
ttls_password = command_line_args.ttls_password # was os.getenv('EAP_PWD')
## AP Credentials
ap_username = command_line_args.ap_username # was os.getenv('AP_USER')
##LANForge Information
lanforge_ip = command_line_args.lanforge_ip_address
lanforge_port = command_line_args.lanforge_port_number
lanforge_prefix = command_line_args.lanforge_prefix
build = command_line_args.build_id
logger = base.logger
hdlr = base.hdlr
client: TestRail_Client = TestRail_Client(command_line_args)
rid = 0 # testrails run-id, not actually supported at the moment.
###Get Cloud Bearer Token
cloud: CloudSDK = CloudSDK(command_line_args)
bearer = cloud.get_bearer(cloudSDK_url, cloud_type)
cloud.assert_bad_response = True
model_id = command_line_args.model
equipment_id = command_line_args.equipment_id
print("equipment-id: %s"%(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 = cloud.get_customer_equipment(cloudSDK_url, bearer, 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'])
if equipment_id == -1:
print("ERROR: Could not find equipment-id.")
sys.exit(1)
###Get Current AP Firmware and upgrade
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)
fw_model = ap_cli_fw.partition("-")[0]
print('Current Active AP FW from CLI:', ap_cli_fw)
###Find Latest FW for Current AP Model and Get FW ID
############################################################################
#################### Create Report #########################################
############################################################################
# Create Report Folder for Today
today = str(date.today())
try:
os.mkdir(report_path + today)
except OSError:
print("Creation of the directory %s failed" % report_path)
else:
print("Successfully created the directory %s " % report_path)
logger.info('Report data can be found here: ' + report_path + today)
###Dictionaries
ap_latest_dict = {}
############################################################################
#################### Jfrog Firmware Check ##################################
############################################################################
apModel = model_id
cloudModel = cloud_sdk_models[apModel]
# print(cloudModel)
###Check Latest FW on jFrog
jfrog_url = 'https://tip.jfrog.io/artifactory/tip-wlan-ap-firmware/'
url = jfrog_url + apModel + "/dev/"
Build: GetBuild = GetBuild(jfrog_user, jfrog_pwd)
latest_image = Build.get_latest_image(url, build)
print(apModel, "Latest FW on jFrog:", latest_image)
ap_latest_dict[model_id] = latest_image
##Get Bearer Token to make sure its valid (long tests can require re-auth)
bearer = cloud.get_bearer(cloudSDK_url, cloud_type)
print("AP MODEL UNDER TEST IS", model_id)
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("Cannot Reach AP CLI, will not test this variant");
sys.exit(1)
fw_model = ap_cli_fw.partition("-")[0]
print('Current Active AP FW from CLI:', ap_cli_fw)
###Find Latest FW for Current AP Model and Get FW ID
##Compare Latest and Current AP FW and Upgrade
latest_ap_image = ap_latest_dict[fw_model]
report_data = None
do_upgrade = cloud.should_upgrade_ap_fw(bearer, command_line_args, report_data, latest_ap_image, fw_model, ap_cli_fw,
logger)
latest_image = ap_latest_dict[model_id]
cloudModel = cloud_sdk_models[model_id]
pf = cloud.do_upgrade_ap_fw(bearer, command_line_args, report_data, test_cases, client,
latest_image, cloudModel, model_id, jfrog_user, jfrog_pwd, rid,
customer_id, equipment_id, logger, latest_ap_image)
if pf:
sys.exit(0)
sys.exit(1)