diff --git a/tests/common/plugins/conditional_mark/tests_mark_conditions_platform_tests.yaml b/tests/common/plugins/conditional_mark/tests_mark_conditions_platform_tests.yaml index c9292d3affd..36247fc301b 100644 --- a/tests/common/plugins/conditional_mark/tests_mark_conditions_platform_tests.yaml +++ b/tests/common/plugins/conditional_mark/tests_mark_conditions_platform_tests.yaml @@ -576,12 +576,6 @@ platform_tests/daemon/test_ledd.py::test_pmon_ledd_kill_and_start_status: ####################################### ##### fwutil/test_fwutil.py ##### ####################################### -platform_tests/fwutil/test_fwutil.py: - skip: - reason: "There is a fixture issue to block this script, skip for now" - conditions: - - https://github.com/sonic-net/sonic-mgmt/issues/6489 - platform_tests/fwutil/test_fwutil.py::test_fwutil_auto: skip: reason: "Command not yet merged into sonic-utilites" diff --git a/tests/platform_tests/fwutil/conftest.py b/tests/platform_tests/fwutil/conftest.py index 8e1f82ecc29..a85a2ab1296 100644 --- a/tests/platform_tests/fwutil/conftest.py +++ b/tests/platform_tests/fwutil/conftest.py @@ -2,29 +2,32 @@ import json import pytest import logging - +import os from random import randrange - +import subprocess from fwutil_common import show_firmware logger = logging.getLogger(__name__) -DUT_HOME="/home/admin" -DEVICES_PATH="/usr/share/sonic/device" +DUT_HOME = "/home/admin" +DEVICES_PATH = "/usr/share/sonic/device" FS_PATH_TEMPLATE = "/host/image-{}/fs.squashfs" FS_RW_TEMPLATE = "/host/image-{}/rw" FS_WORK_TEMPLATE = "/host/image-{}/work" FS_MOUNTPOINT_TEMPLATE = "/tmp/image-{}-fs" OVERLAY_MOUNTPOINT_TEMPLATE = "/tmp/image-{}-overlay" -def check_path_exists(path): - return duthost.stat(path = path)["stat"]["exists"] + +def check_path_exists(duthost, path): + return duthost.stat(path=path)["stat"]["exists"] + def pytest_generate_tests(metafunc): val = metafunc.config.getoption('--fw-pkg') if 'fw_pkg_name' in metafunc.fixturenames and val is not None: metafunc.parametrize('fw_pkg_name', [val], scope="module") + @pytest.fixture(scope='module') def fw_pkg(fw_pkg_name): if fw_pkg_name is None: @@ -32,8 +35,8 @@ def fw_pkg(fw_pkg_name): logger.info("Unpacking firmware package to ./firmware") try: os.mkdir("firmware") - except Exception as e: - pass # Already exists, thats fine + except OSError: + pass # Already exists, thats fine with tarfile.open(fw_pkg_name, "r:gz") as f: f.extractall("./firmware/") with open('./firmware/firmware.json', 'r') as fw: @@ -41,6 +44,7 @@ def fw_pkg(fw_pkg_name): yield fw_data subprocess.call("rm -rf firmware", shell=True) + @pytest.fixture(scope='function') def random_component(duthost, fw_pkg): chass = show_firmware(duthost)["chassis"].keys()[0] @@ -48,19 +52,19 @@ def random_component(duthost, fw_pkg): if len(components) == 0: pytest.skip("No suitable components found in config file for platform {}.".format(duthost.facts['platform'])) + return components[randrange(len(components))] - return components[randrange(len(components))] @pytest.fixture(scope='function') def host_firmware(localhost, duthost): logger.info("Starting local python server to test URL firmware update....") - comm = "python3 -m http.server --directory {}".format(os.path.join(DEVICES_PATH, - duthost.facts['platform'])) + comm = "python3 -m http.server --directory {}".format(os.path.join(DEVICES_PATH, duthost.facts['platform'])) duthost.command(comm, module_ignore_errors=True, module_async=True) yield "http://localhost:8000/" logger.info("Stopping local python server.") duthost.command('pkill -f "{}"'.format(comm), module_ignore_errors=True) + @pytest.fixture(scope='function') def next_image(duthost, fw_pkg): @@ -90,6 +94,7 @@ def next_image(duthost, fw_pkg): overlay_mountpoint = OVERLAY_MOUNTPOINT_TEMPLATE.format(target) logger.info("Attempting to stage test firware onto newly-installed image.") + # noinspection PyBroadException try: wait_until(10, 1, 0, check_path_exists, fs_rw) @@ -105,12 +110,11 @@ def next_image(duthost, fw_pkg): overlay_mountpoint ) duthost.command(cmd) - except Exception as e: + except Exception: pytest.fail("Failed to setup next-image.") duthost.command("sonic_installer set-default {}".format(current)) yield overlay_mountpoint logger.info("Ensuring correct image is set to default boot.") - duthost.command("sonic_installer set-default {}".format(current)) - + duthost.command("sonic-installer remove {} -y".format("SONiC-OS-{}".format(target))) diff --git a/tests/platform_tests/fwutil/fwutil_common.py b/tests/platform_tests/fwutil/fwutil_common.py index 8f6088e41fb..f3497de284c 100644 --- a/tests/platform_tests/fwutil/fwutil_common.py +++ b/tests/platform_tests/fwutil/fwutil_common.py @@ -1,7 +1,10 @@ +import time import pytest import os import json import logging +import allure +import re from copy import deepcopy @@ -14,20 +17,28 @@ POWER_CYCLE = "power off" FAST_REBOOT = "fast" -DEVICES_PATH="usr/share/sonic/device" -TIMEOUT=3600 +DEVICES_PATH = "usr/share/sonic/device" +TIMEOUT = 3600 REBOOT_TYPES = { COLD_REBOOT: "reboot", WARM_REBOOT: "warm-reboot", FAST_REBOOT: "fast-reboot" } + def find_pattern(lines, pattern): for line in lines: if pattern.match(line): return True return False + +def get_hw_revision(duthost): + out = duthost.command("show platform summary") + rev_line = out["stdout"].splitlines()[6] + return rev_line.split(": ")[1] + + def power_cycle(duthost=None, pdu_ctrl=None, delay_time=60): if pdu_ctrl is None: pytest.skip("No PSU controller for %s, skipping" % duthost.hostname) @@ -42,17 +53,21 @@ def power_cycle(duthost=None, pdu_ctrl=None, delay_time=60): for outlet in all_outlets: pdu_ctrl.turn_on_outlet(outlet) + def reboot(duthost, pdu_ctrl, reboot_type, pdu_delay=60): - if reboot_type == POWER_CYCLE: + if reboot_type == POWER_CYCLE: power_cycle(duthost, pdu_ctrl, pdu_delay) return - if reboot_type not in REBOOT_TYPES: pytest.fail("Invalid reboot type {}".format(reboot_type)) + if reboot_type not in REBOOT_TYPES: + pytest.fail("Invalid reboot type {}".format(reboot_type)) logger.info("Rebooting using {}".format(reboot_type)) duthost.command(REBOOT_TYPES[reboot_type], module_ignore_errors=True, module_async=True) -def complete_install(duthost, localhost, boot_type, res, pdu_ctrl, auto_reboot=False, current=None, next_image=None, timeout=TIMEOUT, pdu_delay=60): + +def complete_install(duthost, localhost, boot_type, res, pdu_ctrl, auto_reboot=False, current=None, next_image=None, + timeout=TIMEOUT, pdu_delay=60): hn = duthost.mgmt_ip if boot_type != "none": @@ -62,7 +77,7 @@ def complete_install(duthost, localhost, boot_type, res, pdu_ctrl, auto_reboot=F logger.info("Rebooting switch using {} boot".format(boot_type)) duthost.command("sonic-installer set-default {}".format(current)) reboot(duthost, pdu_ctrl, boot_type, pdu_delay) - + logger.info("Waiting on switch to shutdown...") # Wait for ssh flap localhost.wait_for(host=hn, port=22, state='stopped', delay=10, timeout=timeout) @@ -87,12 +102,12 @@ def complete_install(duthost, localhost, boot_type, res, pdu_ctrl, auto_reboot=F wait_until(300, 30, 0, duthost.critical_services_fully_started) time.sleep(60) + def show_firmware(duthost): out = duthost.command("fwutil show status") - num_spaces = 2 curr_chassis = "" - output_data = {"chassis":{}} + output_data = {"chassis": {}} status_output = out['stdout'] separators = re.split(r'\s{2,}', status_output.splitlines()[1]) # get separators output_lines = status_output.splitlines()[2:] @@ -114,26 +129,48 @@ def show_firmware(duthost): return output_data -def get_install_paths(duthost, fw, versions, chassis): - component = fw["chassis"].get(chassis, {})["component"] + + +def get_install_paths(duthost, defined_fw, versions, chassis, target_component): + component = get_defined_components(duthost, defined_fw, chassis) ver = versions["chassis"].get(chassis, {})["component"] - + paths = {} for comp, revs in component.items(): if comp in ver: if revs[0].get("upgrade_only", False) and ver[comp] not in [r["version"] for r in revs]: - log.warning("Firmware is upgrade only and existing firmware {} is not present in version list. Skipping {}".format(ver[comp], comp)) + logger.warning("Firmware is upgrade only and existing firmware {} is not present in version list. " + "Skipping {}".format(ver[comp], comp)) continue for i, rev in enumerate(revs): + if "hw_revision" in rev and rev["hw_revision"] != get_hw_revision(duthost): + logger.warning("Firmware {} only supports HW Revision {} and this chassis is {}. Skipping". + format(rev["version"], rev["hw_revision"], get_hw_revision(duthost))) + continue if rev["version"] != ver[comp]: paths[comp] = rev break elif rev.get("upgrade_only", False): - log.warning("Firmware is upgrade only and newer version than {} is not available. Skipping {}".format(ver[comp], comp)) + logger.warning("Firmware is upgrade only and newer version than {} is not available. Skipping {}". + format(ver[comp], comp)) break return paths + +def get_defined_components(duthost, defined_fw, chassis): + """ + Update the component content, in case there is a pre-definition for a specific host. + Sometimes, if there is some DUTs has specific component(for example a respined board which requires + a different CPLD) - it can be defined in the firmware.json file + """ + component = defined_fw["chassis"].get(chassis, {})["component"] + if "host" in defined_fw and duthost.hostname in defined_fw["host"]: + for component_type in defined_fw["host"][duthost.hostname]["component"].keys(): + component[component_type] = defined_fw["host"][duthost.hostname]["component"][component_type] + return component + + def generate_config(duthost, cfg, versions): valid_keys = ["firmware", "version"] chassis = versions["chassis"].keys()[0] @@ -143,40 +180,36 @@ def generate_config(duthost, cfg, versions): for comp in versions["chassis"][chassis]["component"].keys(): paths[comp] = paths.get(comp, {}) if "firmware" in paths[comp]: - paths[comp]["firmware"] = os.path.join("/", DEVICES_PATH, - duthost.facts["platform"], - os.path.basename(paths[comp]["firmware"])) + paths[comp]["firmware"] = os.path.join("/", DEVICES_PATH, duthost.facts["platform"], + os.path.basename(paths[comp]["firmware"])) # Populate items we are installing with open("platform_components.json", "w") as f: - json.dump({"chassis":{chassis:{"component":{comp:{k: v - for k, v in dat.items() - if k in valid_keys} - for comp, dat in paths.items()}}}}, f, indent=4) + json.dump({"chassis": {chassis: {"component": {comp: {k: v for k, v in dat.items() if k in valid_keys} + for comp, dat in paths.items()}}}}, f, indent=4) + def upload_platform(duthost, paths, next_image=None): target = next_image if next_image else "/" # Backup the original platform_components.json file - duthost.fetch(dest=os.path.join("firmware", "platform_components_backup.json"), - src=os.path.join(target, DEVICES_PATH, duthost.facts["platform"], "platform_components.json"), - flat=True) + duthost.fetch(dest=os.path.join("firmware", "platform_components_backup.json"), + src=os.path.join(target, DEVICES_PATH, duthost.facts["platform"], "platform_components.json"), + flat=True) logger.info("Backing up platform_components.json") # Copy over the platform_components.json file - duthost.copy(src="platform_components.json", - dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"])) + duthost.copy(src="platform_components.json", dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"])) logger.info("Copying platform_components.json to {}".format( os.path.join(target, DEVICES_PATH, duthost.facts["platform"]))) for comp, dat in paths.items(): - duthost.copy(src=os.path.join("firmware", dat["firmware"]), - dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"])) - if "install" in dat: - duthost.copy(src=os.path.join("firmware", dat["install"]["firmware"]), - dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"])) - logger.info("Copying {} to {}".format(os.path.join("firmware", dat["install"]["firmware"]), - os.path.join(target, DEVICES_PATH, duthost.facts["platform"]))) + if dat["firmware"].startswith("http"): + duthost.get_url(url=dat["firmware"], dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"])) + else: + duthost.copy(src=os.path.join("firmware", dat["firmware"]), + dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"])) + def validate_versions(init, final, config, chassis, boot): final = final["chassis"][chassis]["component"] @@ -188,12 +221,18 @@ def validate_versions(init, final, config, chassis, boot): return False return True -def call_fwutil(duthost, localhost, pdu_ctrl, fw, component=None, next_image=None, boot=None, basepath=None): - logger.info("Calling fwutil with component: {} | next_image: {} | boot: {} | basepath: {}".format(component, next_image, boot, basepath)) + + +def call_fwutil(duthost, localhost, pdu_ctrl, fw_pkg, component=None, next_image=None, boot=None, basepath=None): + allure.step("Collect firmware versions") + logger.info("Calling fwutil with component: {} | next_image: {} | boot: {} | basepath: {}".format(component, + next_image, + boot, basepath)) init_versions = show_firmware(duthost) logger.info("Initial Versions: {}".format(init_versions)) - chassis = init_versions["chassis"].keys()[0] # Only one chassis - paths = get_install_paths(duthost, fw, init_versions, chassis) + # Only one chassis + chassis = init_versions["chassis"].keys()[0] + paths = get_install_paths(duthost, fw_pkg, init_versions, chassis, component) current = duthost.shell('sonic_installer list | grep Current | cut -f2 -d " "')['stdout'] if component not in paths: pytest.skip("No available firmware to install on {}. Skipping".format(component)) @@ -237,24 +276,33 @@ def call_fwutil(duthost, localhost, pdu_ctrl, fw, component=None, next_image=Non if paths[comp].get("auto_reboot", False): auto_reboot = True timeout = max([v.get("timeout", TIMEOUT) for k, v in paths.items()]) - pdu_delay = fw["chassis"][chassis].get("power_cycle_delay", 60) + pdu_delay = fw_pkg["chassis"][chassis].get("power_cycle_delay", 60) complete_install(duthost, localhost, boot_type, res, pdu_ctrl, auto_reboot, current, next_image, timeout, pdu_delay) + allure.step("Collect Updated Firmware Versions") + time.sleep(2) # Give a little bit of time in case of no-op install for mounts to complete final_versions = show_firmware(duthost) - assert validate_versions(init_versions, final_versions, paths, chassis, boot_type) - - duthost.copy(src=os.path.join("firmware", "platform_components_backup.json"), - dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"], "platform_components.json")) - logger.info("Restoring backup platform_components.json to {}".format( - os.path.join(DEVICES_PATH, duthost.facts["platform"]))) - - update_needed = copy(fw) + test_result = validate_versions(init_versions, final_versions, paths, chassis, boot_type) + + allure.step("Begin Switch Restoration") + if next_image is None: + duthost.copy(src=os.path.join("firmware", "platform_components_backup.json"), + dest=os.path.join("/", DEVICES_PATH, duthost.facts["platform"], "platform_components.json")) + logger.info("Restoring backup platform_components.json to {}".format( + os.path.join(DEVICES_PATH, duthost.facts["platform"]))) + + update_needed = deepcopy(fw_pkg) + update_needed["chassis"][chassis]["component"] = {} + defined_components = get_defined_components(duthost, fw_pkg, chassis) + final_components = final_versions["chassis"][chassis]["component"] for comp in paths.keys(): - if fw["chassis"][chassis]["component"][comp][0]["version"] == final_versions[comp] or paths[comp]["upgrade_only"]: - del update_needed["chassis"][chassis]["component"][comp] + if defined_components[comp][0]["version"] != final_components[comp] and \ + boot in defined_components[comp][0]["reboot"] + [None] and \ + not paths[comp].get("upgrade_only", False): + update_needed["chassis"][chassis]["component"][comp] = defined_components[comp] if len(update_needed["chassis"][chassis]["component"].keys()) > 0: logger.info("Latest firmware not installed after test. Installing....") - call_fwutil(duthost, localhost, pdu_ctrl, update_needed, component, None, boot, basepath) - - return True + call_fwutil(duthost, localhost, pdu_ctrl, update_needed, component, None, boot, + os.path.join("/", DEVICES_PATH, duthost.facts['platform']) if basepath is not None else None) + return test_result diff --git a/tests/platform_tests/fwutil/test_fwutil.py b/tests/platform_tests/fwutil/test_fwutil.py index a8150bef0e3..cc0f6746092 100644 --- a/tests/platform_tests/fwutil/test_fwutil.py +++ b/tests/platform_tests/fwutil/test_fwutil.py @@ -2,21 +2,20 @@ import pytest import os import json - from fwutil_common import call_fwutil, show_firmware, upload_platform, find_pattern pytestmark = [ pytest.mark.topology("any") ] -DEVICES_PATH="/usr/share/sonic/device" +DEVICES_PATH = "/usr/share/sonic/device" + def test_fwutil_show(duthost): """Tests that fwutil show has all components defined for platform""" - platform_comp = {} duthost.fetch(dest=os.path.join("firmware", "platform_components_backup.json"), - src=os.path.join(DEVICES_PATH, duthost.facts["platform"], "platform_components.json"), - flat=True) + src=os.path.join(DEVICES_PATH, duthost.facts["platform"], "platform_components.json"), + flat=True) with open(os.path.join("firmware", "platform_components_backup.json")) as f: platform_comp = json.load(f) @@ -28,85 +27,96 @@ def test_fwutil_show(duthost): assert show_fw_comp_set == platform_comp_set + def test_fwutil_install_file(duthost, localhost, pdu_controller, fw_pkg, random_component): """Tests manually installing firmware to a component from a file.""" assert call_fwutil(duthost, - localhost, - pdu_controller, - fw_pkg, - component=random_component, - basepath=os.path.join(DEVICES_PATH, duthost.facts['platform'])) + localhost, + pdu_controller, + fw_pkg, + component=random_component, + basepath=os.path.join(DEVICES_PATH, duthost.facts['platform'])) + def test_fwutil_install_url(duthost, localhost, pdu_controller, fw_pkg, random_component, host_firmware): """Tests manually installing firmware to a component from a URL.""" assert call_fwutil(duthost, - localhost, - pdu_controller, - fw_pkg, - component=random_component, - basepath=host_firmware) + localhost, + pdu_controller, + fw_pkg, + component=random_component, + basepath=host_firmware) + -def test_fwutil_install_bad_name(duthost, fw_pkg): +def test_fwutil_install_bad_name(duthost): """Tests that fwutil install validates component names correctly.""" out = duthost.command("fwutil install chassis component BAD fw BAD.pkg", module_ignore_errors=True) pattern = re.compile(r'.*Invalid value for ""*.') - found = find_pattern(out['stderr_lines'], pattern) - assert found + assert find_pattern(out['stderr_lines'], pattern) -def test_fwutil_install_bad_path(duthost, fw_pkg, random_component): + +def test_fwutil_install_bad_path(duthost, random_component): """Tests that fwutil install validates firmware paths correctly.""" - out = duthost.command("fwutil install chassis component {} fw BAD.pkg".format(random_component), module_ignore_errors=True) + out = duthost.command("fwutil install chassis component {} fw BAD.pkg".format(random_component), + module_ignore_errors=True) pattern = re.compile(r'.*Error: Invalid value for ""*.') - found = find_pattern(out['stderr_lines'], pattern) - assert found + assert find_pattern(out['stderr_lines'], pattern) + def test_fwutil_update_current(duthost, localhost, pdu_controller, fw_pkg, random_component): """Tests updating firmware from current image using fwutil update""" assert call_fwutil(duthost, - localhost, - pdu_controller, - fw_pkg, - component=random_component) + localhost, + pdu_controller, + fw_pkg, + component=random_component) + def test_fwutil_update_next(duthost, localhost, pdu_controller, fw_pkg, random_component, next_image): """Tests updating firmware from the "next" image using fwutil update""" assert call_fwutil(duthost, - localhost, - pdu_controller, - fw_pkg, - component=random_component, - next_image=next_image) + localhost, + pdu_controller, + fw_pkg, + component=random_component, + next_image=next_image) -def test_fwutil_update_bad_config(duthost, fw_pkg, random_component): + +def test_fwutil_update_bad_config(duthost, random_component): """Tests that fwutil update validates the platform_components.json schema correctly.""" versions = show_firmware(duthost) - chassis = versions["chassis"].keys()[0] # Only one chassis + chassis = versions["chassis"].keys()[0] # Only one chassis # Test fwutil update with config file without chassis section with open("platform_components.json", "w") as f: json.dump({}, f, indent=4) upload_platform(duthost, {}) - out_empty_json = duthost.command("fwutil update chassis component {} fw -y".format(random_component), module_ignore_errors=True) + out_empty_json = duthost.command("fwutil update chassis component {} fw -y".format(random_component), + module_ignore_errors=True) pattern_bad_platform = re.compile(r'.*Error: Failed to parse "platform_components.json": invalid platform schema*.') found_bad_platform = find_pattern(out_empty_json['stdout_lines'], pattern_bad_platform) assert found_bad_platform # Test fwutil update with config file without component section with open("platform_components.json", "w") as f: - json.dump({"chassis":{chassis:{}}}, f, indent=4) + json.dump({"chassis": {chassis: {}}}, f, indent=4) upload_platform(duthost, {}) - out_empty_chassis = duthost.command("fwutil update chassis component {} fw -y".format(random_component), module_ignore_errors=True) + out_empty_chassis = duthost.command("fwutil update chassis component {} fw -y". + format(random_component), module_ignore_errors=True) pattern_bad_chassis = re.compile(r'.*Error: Failed to parse "platform_components.json": invalid chassis schema*.') found_bad_chassis = find_pattern(out_empty_chassis['stdout_lines'], pattern_bad_chassis) assert found_bad_chassis # Test fwutil update with config file with version of type dict with open("platform_components.json", "w") as f: - json.dump({"chassis":{chassis:{"component":{random_component:{"version":{"version":"ver"}}}}}} - , f, indent=4) + json.dump({"chassis": {chassis: {"component": {random_component: {"version": {"version": "ver"}}}}}}, + f, + indent=4) upload_platform(duthost, {}) - out_bad_version = duthost.command("fwutil update chassis component {} fw -y".format(random_component), module_ignore_errors=True) - pattern_bad_component = re.compile(r'.*Error: Failed to parse "platform_components.json": invalid component schema*.') + out_bad_version = duthost.command("fwutil update chassis component {} fw -y".format(random_component), + module_ignore_errors=True) + pattern_bad_component = re.compile(r'.*Error: Failed to parse "platform_components.json": ' + r'invalid component schema*.') found_bad_component = find_pattern(out_bad_version['stdout_lines'], pattern_bad_component) assert found_bad_component @@ -115,8 +125,7 @@ def test_fwutil_update_bad_config(duthost, fw_pkg, random_component): def test_fwutil_auto(duthost, localhost, pdu_controller, fw_pkg, reboot_type): """Tests fwutil update all command ability to properly select firmware for install based on boot type.""" assert call_fwutil(duthost, - localhost, - pdu_controller, - fw_pkg, - reboot=reboot_type) - + localhost, + pdu_controller, + fw_pkg, + boot=reboot_type)