Enhance the fwutil test (#13814)

1. Remove the randomization in selecting component, instead, parameterize the test by the components type.
2. Add the FPGA component for Mellanox SN4280 platform.
3. Increase the ServerAliveCountMax in ansible config to keep the ssh connection when buring the FPGA(it takes 15~20 mins).
4. When testing the install test cases, even if we only have one version of the component, still run the test.

ansible/ansible.cfg

@@ -190,7 +190,7 @@ become_method='sudo'
 # ssh arguments to use
 # Leaving off ControlPersist will result in poor performance, so use
 # paramiko on older platforms rather than removing it
-ssh_args = -o ControlMaster=auto -o ControlPersist=120s -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no -o ServerAliveInterval=30 -o ServerAliveCountMax=4
+ssh_args = -o ControlMaster=auto -o ControlPersist=120s -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no -o ServerAliveInterval=30 -o ServerAliveCountMax=40
 
 
 # The path to use for the ControlPath sockets. This defaults to

tests/platform_tests/fwutil/conftest.py

@@ -3,7 +3,6 @@
 import pytest
 import logging
 import os
-from random import randrange
 from fwutil_common import show_firmware
 
 logger = logging.getLogger(__name__)
@@ -81,13 +80,17 @@ def extract_fw_data(fw_pkg_path):
     return fw_data
 
 
-@pytest.fixture(scope='function')
-def random_component(duthost, fw_pkg):
-    chass = list(show_firmware(duthost)["chassis"].keys())[0]
-    components = list(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))]
+@pytest.fixture(scope='function', params=["CPLD", "ONIE", "BIOS", "FPGA"])
+def component(request, duthost, fw_pkg):
+    component_type = request.param
+    chassis = list(show_firmware(duthost)["chassis"].keys())[0]
+    available_components = list(fw_pkg["chassis"].get(chassis, {}).get("component", {}).keys())
+    if len(available_components) > 0:
+        for component in available_components:
+            if component_type in component:
+                return component
+    pytest.skip(f"No suitable components found in config file for "
+                f"platform {duthost.facts['platform']}, firmware type {component_type}.")
 
 
 @pytest.fixture(scope='function')

tests/platform_tests/fwutil/fwutil_common.py

@@ -76,6 +76,8 @@ def complete_install(duthost, localhost, boot_type, res, pdu_ctrl, auto_reboot=F
         if not auto_reboot:
             logger.info("Waiting on install to finish.")
             res.get(timeout)
+            if res._value['failed']:
+                pytest.fail(f"The component installation is not successful: {res._value}")
             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)
@@ -132,7 +134,7 @@ def show_firmware(duthost):
     return output_data
 
 
-def get_install_paths(duthost, defined_fw, versions, chassis, target_component):
+def get_install_paths(request, duthost, defined_fw, versions, chassis, target_component):
     component = get_defined_components(duthost, defined_fw, chassis)
     ver = versions["chassis"].get(chassis, {})["component"]
 
@@ -147,12 +149,15 @@ def get_install_paths(duthost, defined_fw, versions, chassis, target_component):
                 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):
+            for rev in 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]:
+                if "install" in request.node.name and len(revs) == 1:
+                    paths[comp] = rev
+                    break
+                elif rev["version"] != ver[comp]:
                     paths[comp] = rev
                     break
                 elif rev.get("upgrade_only", False):
@@ -218,18 +223,16 @@ def upload_platform(duthost, paths, next_image=None):
                          dest=os.path.join(target, DEVICES_PATH, duthost.facts["platform"]))
 
 
-def validate_versions(init, final, config, chassis, boot):
+def validate_versions(final, config, chassis, boot):
     final = final["chassis"][chassis]["component"]
-    init = init["chassis"][chassis]["component"]
     for comp, dat in list(config.items()):
         logger.info("Validating {} is version {} (is {})".format(comp, dat["version"], final[comp]))
-        if (dat["version"] != final[comp] or init[comp] == final[comp]) and boot in dat["reboot"]:
+        if dat["version"] != final[comp] and boot in dat["reboot"]:
             pytest.fail("Failed to install FW verison {} on {}".format(dat["version"], comp))
-            return False
-    return True
 
 
-def call_fwutil(duthost, localhost, pdu_ctrl, fw_pkg, component=None, next_image=None, boot=None, basepath=None):
+def call_fwutil(request, 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,
@@ -238,7 +241,7 @@ def call_fwutil(duthost, localhost, pdu_ctrl, fw_pkg, component=None, next_image
     logger.info("Initial Versions: {}".format(init_versions))
     # Only one chassis
     chassis = list(init_versions["chassis"].keys())[0]
-    paths = get_install_paths(duthost, fw_pkg, init_versions, chassis, component)
+    paths = get_install_paths(request, duthost, fw_pkg, init_versions, chassis, component)
     if component not in paths:
         pytest.skip("No available firmware to install on {}. Skipping".format(component))
     boot_type = boot if boot else paths[component]["reboot"][0]
@@ -290,7 +293,7 @@ def call_fwutil(duthost, localhost, pdu_ctrl, fw_pkg, component=None, next_image
     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)
-    test_result = validate_versions(init_versions, final_versions, paths, chassis, boot_type)
+    validate_versions(final_versions, paths, chassis, boot_type)
 
     allure.step("Begin Switch Restoration")
     if next_image is None:
@@ -310,7 +313,5 @@ def call_fwutil(duthost, localhost, pdu_ctrl, fw_pkg, component=None, next_image
             update_needed["chassis"][chassis]["component"][comp] = defined_components[comp]
     if len(list(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,
+        call_fwutil(request, 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

tests/platform_tests/fwutil/test_fwutil.py

@@ -27,24 +27,26 @@ 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):
+def test_fwutil_install_file(request, duthost, localhost, pdu_controller, component, fw_pkg):
     """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']))
+    call_fwutil(request,
+                duthost,
+                localhost,
+                pdu_controller,
+                fw_pkg,
+                component=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):
+def test_fwutil_install_url(request, duthost, localhost, pdu_controller, component, fw_pkg, 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)
+    call_fwutil(request,
+                duthost,
+                localhost,
+                pdu_controller,
+                fw_pkg,
+                component=component,
+                basepath=host_firmware)
 
 
 def test_fwutil_install_bad_name(duthost):
@@ -54,34 +56,36 @@ def test_fwutil_install_bad_name(duthost):
     assert find_pattern(out['stderr_lines'], pattern)
 
 
-def test_fwutil_install_bad_path(duthost, random_component):
+def test_fwutil_install_bad_path(duthost, component):
     """Tests that fwutil install validates firmware paths correctly."""
-    out = duthost.command("fwutil install chassis component {} fw BAD.pkg".format(random_component),
+    out = duthost.command(f"fwutil install chassis component {component} fw BAD.pkg",
                           module_ignore_errors=True)
     pattern = re.compile(r'.*Error: Invalid value for "<fw_path>"*.')
     assert find_pattern(out['stderr_lines'], pattern)
 
 
-def test_fwutil_update_current(duthost, localhost, pdu_controller, fw_pkg, random_component):
+def test_fwutil_update_current(request, duthost, localhost, pdu_controller, component, fw_pkg):
     """Tests updating firmware from current image using fwutil update"""
-    assert call_fwutil(duthost,
-                       localhost,
-                       pdu_controller,
-                       fw_pkg,
-                       component=random_component)
+    call_fwutil(request,
+                duthost,
+                localhost,
+                pdu_controller,
+                fw_pkg,
+                component=component)
 
 
-def test_fwutil_update_next(duthost, localhost, pdu_controller, fw_pkg, random_component, next_image):
+def test_fwutil_update_next(request, duthost, localhost, pdu_controller, component, next_image, fw_pkg):
     """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)
+    call_fwutil(request,
+                duthost,
+                localhost,
+                pdu_controller,
+                fw_pkg,
+                component=component,
+                next_image=next_image)
 
 
-def test_fwutil_update_bad_config(duthost, random_component):
+def test_fwutil_update_bad_config(duthost, component):
     """Tests that fwutil update validates the platform_components.json schema correctly."""
     versions = show_firmware(duthost)
     chassis = list(versions["chassis"].keys())[0]  # Only one chassis
@@ -90,7 +94,7 @@ def test_fwutil_update_bad_config(duthost, random_component):
     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),
+    out_empty_json = duthost.command(f"fwutil update chassis component {component} fw -y",
                                      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)
@@ -100,19 +104,18 @@ def test_fwutil_update_bad_config(duthost, random_component):
     with open("platform_components.json", "w") as f:
         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(f"fwutil update chassis component {component} fw -y", 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"}}}}}},
+        json.dump({"chassis": {chassis: {"component": {component: {"version": {"version": "ver"}}}}}},
                   f,
                   indent=4)
     upload_platform(duthost, {})
-    out_bad_version = duthost.command("fwutil update chassis component {} fw -y".format(random_component),
+    out_bad_version = duthost.command("fwutil update chassis component {} fw -y".format(component),
                                       module_ignore_errors=True)
     pattern_bad_component = re.compile(r'.*Error: Failed to parse "platform_components.json": '
                                        r'invalid component schema*.')
@@ -121,10 +124,11 @@ def test_fwutil_update_bad_config(duthost, random_component):
 
 
 @pytest.mark.parametrize("reboot_type", ["none", "cold"])
-def test_fwutil_auto(duthost, localhost, pdu_controller, fw_pkg, reboot_type):
+def test_fwutil_auto(request, 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,
-                       boot=reboot_type)
+    call_fwutil(request,
+                duthost,
+                localhost,
+                pdu_controller,
+                fw_pkg,
+                boot=reboot_type)