mirror of
https://github.com/Telecominfraproject/wlan-testing.git
synced 2025-11-02 03:48:09 +00:00
Improved firmware upgrade logic
Signed-off-by: shivam <shivam.thakur@candelatech.com>
This commit is contained in:
shivam
@@ -253,6 +253,8 @@ class APLIBS:
|
||||
"identifier"] + " -s serial --log stdout --user root --passwd openwifi"
|
||||
cmd = f"cd ~/cicd-git/ && ./openwrt_ctl.py {owrt_args} -t {self.device_under_tests_data[idx]['serial_tty']} --action " \
|
||||
f"cmd --value \"{cmd}\" "
|
||||
if print_log:
|
||||
logging.info(cmd)
|
||||
stdin, stdout, stderr = client.exec_command(cmd)
|
||||
output = stdout.read()
|
||||
final_output = str(output)
|
||||
@@ -261,6 +263,7 @@ class APLIBS:
|
||||
status.pop(0)
|
||||
final_output = '\n'.join(status)
|
||||
if print_log:
|
||||
logging.info(cmd)
|
||||
logging.info("Output for command: " + input_command + "\n" + final_output)
|
||||
if attach_allure:
|
||||
allure.attach(name=input_command, body=output, attachment_type=expected_attachment_type)
|
||||
@@ -330,7 +333,7 @@ class APLIBS:
|
||||
ret_val = data
|
||||
return ret_val
|
||||
|
||||
def get_ap_version(self, idx=0, print_log=False, attach_allure=False):
|
||||
def get_ap_version(self, idx=0, print_log=True, attach_allure=False):
|
||||
output = self.run_generic_command(cmd="cat /tmp/ucentral.version", idx=idx,
|
||||
print_log=print_log,
|
||||
attach_allure=attach_allure,
|
||||
@@ -347,31 +350,17 @@ if __name__ == '__main__':
|
||||
"password": "OpenWifi%123"
|
||||
},
|
||||
"device_under_tests": [{
|
||||
"model": "cig_wf188n",
|
||||
"model": "edgecore_eap101",
|
||||
"supported_bands": ["2G", "5G"],
|
||||
"supported_modes": ["BRIDGE", "NAT", "VLAN"],
|
||||
"ssid": {
|
||||
"2g-ssid": "OpenWifi",
|
||||
"5g-ssid": "OpenWifi",
|
||||
"6g-ssid": "candela6ghz",
|
||||
"2g-password": "OpenWifi",
|
||||
"5g-password": "OpenWifi",
|
||||
"6g-password": "hello123",
|
||||
"2g-encryption": "WPA2",
|
||||
"5g-encryption": "open",
|
||||
"6g-encryption": "WPA3",
|
||||
"2g-bssid": "68:7d:b4:5f:5c:31 ",
|
||||
"5g-bssid": "68:7d:b4:5f:5c:3c",
|
||||
"6g-bssid": "68:7d:b4:5f:5c:38"
|
||||
},
|
||||
"mode": "wifi6",
|
||||
"identifier": "0000c1018812",
|
||||
"identifier": "903cb36c44f0",
|
||||
"method": "serial",
|
||||
"host_ip": "10.28.3.103",
|
||||
"host_ip": "192.168.200.101",
|
||||
"host_username": "lanforge",
|
||||
"host_password": "pumpkin77",
|
||||
"host_password": "Endurance@123",
|
||||
"host_ssh_port": 22,
|
||||
"serial_tty": "/dev/ttyAP1",
|
||||
"serial_tty": "/dev/ttyUSB0",
|
||||
"firmware_version": "next-latest"
|
||||
}],
|
||||
"traffic_generator": {
|
||||
@@ -401,7 +390,7 @@ if __name__ == '__main__':
|
||||
}
|
||||
logging.basicConfig(format='%(asctime)s - %(message)s', level=logging.NOTSET)
|
||||
obj = APLIBS(dut_data=basic_1["device_under_tests"])
|
||||
obj.check_serial_connection()
|
||||
# obj.exit_from_uboot()
|
||||
obj.setup_serial_environment()
|
||||
# obj.run_generic_command("uci show ucentral")
|
||||
# obj.verify_certificates()
|
||||
@@ -410,4 +399,4 @@ if __name__ == '__main__':
|
||||
# a = obj.get_active_config()
|
||||
# if a == l:
|
||||
# print("a = l")
|
||||
print(obj.get_ap_version())
|
||||
# print(obj.get_ap_version())
|
||||
|
||||
@@ -9,6 +9,7 @@ import time
|
||||
|
||||
import allure
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
logging = importlib.import_module("logging")
|
||||
|
||||
@@ -482,6 +483,256 @@ class tip_2x:
|
||||
add_allure_environment_property('DUT-Model/s', ", ".join(models))
|
||||
add_allure_environment_property('Serial-Number/s', ", ".join(identifiers))
|
||||
|
||||
def setup_firmware(self):
|
||||
# Query AP Firmware
|
||||
upgrade_status = []
|
||||
for ap in range(len(self.device_under_tests_info)):
|
||||
|
||||
# If specified as URL
|
||||
try:
|
||||
response = requests.get(self.device_under_tests_info[ap]['firmware_version'])
|
||||
logging.info("URL is valid and exists on the internet")
|
||||
allure.attach(name="firmware url: ", body=str(self.device_under_tests_info[ap]['firmware_version']))
|
||||
target_revision_commit = self.device_under_tests_info[ap]['firmware_version'].split("-")[-2]
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version_commit = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
|
||||
# if AP is already in target Version then skip upgrade unless force upgrade is specified
|
||||
if target_revision_commit in current_version_commit:
|
||||
continue
|
||||
self.firmware_library_object.upgrade_firmware(serial=self.device_under_tests_info[ap]['identifier'],
|
||||
url=str(self.device_under_tests_info[ap]['firmware_version']))
|
||||
|
||||
items = list(range(0, 300))
|
||||
l = len(items)
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version_commit = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
if target_revision_commit in current_version_commit:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision_commit, current_version_commit])
|
||||
logging.info("Firmware Upgraded to :", ap_version)
|
||||
else:
|
||||
logging.info("firmware upgraded failed: ", target_revision)
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision_commit, current_version_commit])
|
||||
break
|
||||
except Exception as e:
|
||||
logging.error("URL does not exist on Internet")
|
||||
# else Specified as "branch-commit_id" / "branch-latest"
|
||||
firmware_url = ""
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
response = self.firmware_library_object.get_latest_fw(model=self.device_under_tests_info[ap]['model'])
|
||||
# if the target version specified is "branch-latest"
|
||||
if self.device_under_tests_info[ap]['firmware_version'].split('-')[1] == "latest":
|
||||
# get the latest branch
|
||||
firmware_list = self.firmware_library_object.get_firmwares(model=self.device_under_tests_info[ap]['model'], branch="", commit_id='')
|
||||
firmware_list.reverse()
|
||||
|
||||
for firmware in firmware_list:
|
||||
if firmware['image'] == "":
|
||||
continue
|
||||
if str(firmware['image']).__contains__("upgrade.bin"):
|
||||
temp = firmware['image'].split("-")
|
||||
temp.pop(-1)
|
||||
temp = "-".join(temp)
|
||||
firmware['image'] = temp
|
||||
if self.device_under_tests_info[ap]['firmware_version'].split('-')[0] == 'release':
|
||||
if firmware['revision'].split("/")[1].replace(" ", "").split('-')[1].__contains__('v2.'):
|
||||
logging.info("Target Firmware: \n", firmware)
|
||||
allure.attach(name="Target firmware : ", body=str(firmware))
|
||||
target_revision = firmware['revision'].split("/")[1].replace(" ", "")
|
||||
|
||||
# check the current AP Revision before upgrade
|
||||
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
|
||||
# print and report the firmware versions before upgrade
|
||||
allure.attach(name="Before Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
|
||||
# if AP is already in target Version then skip upgrade unless force upgrade is specified
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision,
|
||||
current_version, 'skip'])
|
||||
logging.info("Skipping Upgrade! AP is already in target version")
|
||||
allure.attach(name="Skipping Upgrade because AP is already in the target Version",
|
||||
body="")
|
||||
break
|
||||
|
||||
self.firmware_library_object.upgrade_firmware(serial=self.device_under_tests_info[ap]['identifier'],
|
||||
url=str(firmware['uri']))
|
||||
# wait for 300 seconds after firmware upgrade
|
||||
logging.info("waiting for 300 Sec for Firmware Upgrade")
|
||||
time.sleep(300)
|
||||
|
||||
# check the current AP Revision again
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
# print and report the Firmware versions after upgrade
|
||||
allure.attach(name="After Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: ", current_version, "\ntarget revision: ", target_revision)
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version])
|
||||
logging.info("firmware upgraded successfully: ", target_revision)
|
||||
else:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version])
|
||||
logging.info("firmware upgraded failed: ", target_revision)
|
||||
break
|
||||
if firmware['image'].split("-")[-2] == self.device_under_tests_info[ap]['firmware_version'].split('-')[0]:
|
||||
logging.info("Target Firmware: \n", firmware)
|
||||
allure.attach(name="Target firmware : ", body=str(firmware))
|
||||
|
||||
target_revision = firmware['revision'].split("/")[1].replace(" ", "")
|
||||
|
||||
# check the current AP Revision before upgrade
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
|
||||
# print and report the firmware versions before upgrade
|
||||
allure.attach(name="Before Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
|
||||
# if AP is already in target Version then skip upgrade unless force upgrade is specified
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version, 'skip'])
|
||||
logging.info("Skipping Upgrade! AP is already in target version")
|
||||
allure.attach(name="Skipping Upgrade because AP is already in the target Version", body="")
|
||||
break
|
||||
|
||||
self.firmware_library_object.upgrade_firmware(serial=self.device_under_tests_info[ap]['identifier'], url=str(firmware['uri']))
|
||||
# wait for 300 seconds after firmware upgrade
|
||||
logging.info("waiting for 300 Sec for Firmware Upgrade")
|
||||
time.sleep(500)
|
||||
|
||||
# check the current AP Revision again
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
# print and report the Firmware versions after upgrade
|
||||
allure.attach(name="After Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: ", current_version, "\ntarget revision: ", target_revision)
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version])
|
||||
logging.info("firmware upgraded successfully: ", target_revision)
|
||||
else:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version])
|
||||
logging.info("firmware upgraded failed: ", target_revision)
|
||||
break
|
||||
# if branch-commit is specified
|
||||
else:
|
||||
firmware_list = self.firmware_library_object.get_firmwares(model=self.device_under_tests_info[ap]['model'],
|
||||
branch="", commit_id='')
|
||||
fw_list = []
|
||||
# getting the list of firmwares in fw_list that has the commit id specified as an input
|
||||
for firmware in firmware_list:
|
||||
if firmware['revision'].split("/")[1].replace(" ", "").split('-')[-1] == \
|
||||
self.device_under_tests_info[ap]['firmware_version'].split('-')[1]:
|
||||
fw_list.append(firmware)
|
||||
|
||||
# If there is only 1 commit ID in fw_list
|
||||
if len(fw_list) == 1:
|
||||
|
||||
logging.info("Target Firmware: \n", fw_list[0])
|
||||
allure.attach(name="Target firmware : ", body=str(fw_list[0]))
|
||||
|
||||
url = fw_list[0]['uri']
|
||||
target_revision = fw_list[0]['revision'].split("/")[1].replace(" ", "")
|
||||
|
||||
# check the current AP Revision before upgrade
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
|
||||
# print and report the firmware versions before upgrade
|
||||
allure.attach(name="Before Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: ", current_version, "\ntarget revision: ", target_revision)
|
||||
|
||||
# if AP is already in target Version then skip upgrade unless force upgrade is specified
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version, 'skip'])
|
||||
logging.info("Skipping Upgrade! AP is already in target version")
|
||||
allure.attach(name="Skipping Upgrade because AP is already in the target Version", body="")
|
||||
break
|
||||
|
||||
# upgrade the firmware in another condition
|
||||
else:
|
||||
self.firmware_library_object.upgrade_firmware(serial=self.device_under_tests_info[ap]['identifier'], url=str(url))
|
||||
|
||||
# wait for 300 seconds after firmware upgrade
|
||||
logging.info("waiting for 300 Sec for Firmware Upgrade")
|
||||
time.sleep(300)
|
||||
|
||||
# check the current AP Revision again
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
# print and report the Firmware versions after upgrade
|
||||
allure.attach(name="After Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: ", current_version, "\ntarget revision: ", target_revision)
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version])
|
||||
logging.info("firmware upgraded successfully: ", target_revision)
|
||||
else:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version])
|
||||
logging.info("firmware upgraded failed: ", target_revision)
|
||||
break
|
||||
|
||||
# if there are 1+ firmware images in fw_list then check for branch
|
||||
else:
|
||||
target_fw = ""
|
||||
for firmware in fw_list:
|
||||
if self.device_under_tests_info[ap]['firmware_version'].split('-')[0] == 'release':
|
||||
if firmware['revision'].split("/")[1].replace(" ", "").split('-')[1].__contains__('v2.'):
|
||||
target_fw = firmware
|
||||
break
|
||||
if firmware['image'].split("-")[-2] == self.device_under_tests_info[ap]['firmware_version'].split('-')[0]:
|
||||
target_fw = firmware
|
||||
break
|
||||
firmware = target_fw
|
||||
logging.info("Target Firmware: \n", firmware)
|
||||
allure.attach(name="Target firmware : ", body=str(firmware))
|
||||
|
||||
target_revision = firmware['revision'].split("/")[1].replace(" ", "")
|
||||
|
||||
# check the current AP Revision before upgrade
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
|
||||
# print and report the firmware versions before upgrade
|
||||
allure.attach(name="Before Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: ", current_version, "\ntarget revision: ", target_revision)
|
||||
|
||||
# if AP is already in target Version then skip upgrade unless force upgrade is specified
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([self.device_under_tests_info[ap]['identifier'], target_revision, current_version, 'skip'])
|
||||
logging.info("Skipping Upgrade! AP is already in target version")
|
||||
allure.attach(name="Skipping Upgrade because AP is already in the target Version", body="")
|
||||
break
|
||||
|
||||
self.firmware_library_object.upgrade_firmware(serial=self.device_under_tests_info[ap]['identifier'], url=str(firmware['uri']))
|
||||
# wait for 300 seconds after firmware upgrade
|
||||
|
||||
logging.info("waiting for 300 Sec for Firmware Upgrade")
|
||||
time.sleep(300)
|
||||
|
||||
# check the current AP Revision again
|
||||
ap_version = self.dut_library_object.get_ap_version(idx=ap)
|
||||
current_version = str(ap_version).split("/")[1].replace(" ", "").splitlines()[0]
|
||||
# print and report the Firmware versions after upgrade
|
||||
allure.attach(name="After Firmware Upgrade Request: ",
|
||||
body="current revision: " + current_version + "\ntarget revision: " + target_revision)
|
||||
logging.info("current revision: ", current_version, "\ntarget revision: ", target_revision)
|
||||
if current_version == target_revision:
|
||||
upgrade_status.append([target_revision, current_version])
|
||||
logging.info("firmware upgraded successfully: ", target_revision)
|
||||
else:
|
||||
upgrade_status.append([target_revision, current_version])
|
||||
logging.info("firmware upgraded failed: ", target_revision)
|
||||
break
|
||||
return upgrade_status
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
@@ -530,5 +781,5 @@ if __name__ == '__main__':
|
||||
}
|
||||
target = [['2G', 'wpa2_personal'], ['5G', 'wpa2_personal']]
|
||||
# var.setup_basic_configuration(configuration=setup_params_general, requested_combination=target)
|
||||
var.get_dut_version()
|
||||
var.setup_firmware()
|
||||
var.teardown_objects()
|
||||
|
||||
Reference in New Issue
Block a user