Update the fwutil function to support define the component for specific dut.(202012)

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"

tests/platform_tests/fwutil/conftest.py

@@ -2,65 +2,69 @@
 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:
         pytest.skip("No fw package specified.")
     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:
             fw_data = json.load(fw)
             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]
     components = fw_pkg["chassis"].get(chass, {}).get("component", []).keys()
 
     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)))

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