[Mellanox] Implement auto_firmware_update platform API for to support fwutil auto-update (#7721)

Why I did it
The Mellanox platform is required to support the fwutil auto-update feature defined here

This is to allow switches, when performing SONiC upgrades to choose whether to perform firmware upgrades that may interrupt the data plane through a cold boot.

How I did it
Two methods were added to the component implementations for mellanox.

In the base Component class we add a default function that chooses to skip the installation of any firmware unless the cold boot option is provided. This is because the Mellanox platform, by default, does not support installing firmware on ONIE, the CPLD, or the BIOS "on-the-fly".

In the ComponentSSD class we add a function that behaves similarly but uses the Mellanox specific SSD firmware upgrade tool to check if the current SSD supports being upgraded on the fly in order to decide whether to skip or perform the installation.

How to verify it
Unit tests are included with this PR. These test will run on build of target sonic-mellanox.bin

You may also perform fwutil auto-update ... commands after Azure/sonic-utilities#1242 is merged in.

platform/mellanox/mlnx-platform-api/sonic_platform/component.py

@@ -18,7 +18,11 @@ try:
     else:
         import ConfigParser as configparser
 
-    from sonic_platform_base.component_base import ComponentBase
+    from sonic_platform_base.component_base import ComponentBase,           \
+                                                    FW_AUTO_INSTALLED,      \
+                                                    FW_AUTO_ERR_BOOT_TYPE,  \
+                                                    FW_AUTO_ERR_IMAGE,      \
+                                                    FW_AUTO_ERR_UKNOWN
 except ImportError as e:
     raise ImportError(str(e) + "- required module not found")
 
@@ -332,6 +336,30 @@ class Component(ComponentBase):
     def get_description(self):
         return self.description
 
+    def auto_update_firmware(self, image_path, boot_action):
+        """
+        Default handling of attempted automatic update for a component of a Mellanox switch.
+        Will skip the installation if the boot_action is 'warm' or 'fast' and will call update_firmware()
+        if boot_action is fast.
+        """
+
+        default_supported_boot = ['cold']
+
+        # Verify image path exists
+        if not os.path.exists(image_path):
+            # Invalid image path
+            return FW_AUTO_ERR_IMAGE
+
+        if boot_action in default_supported_boot:
+            if self.install_firmware(image_path):
+                # Successful update
+                return FW_AUTO_INSTALLED
+            # Failed update (unknown reason)
+            return FW_AUTO_ERR_UKNOWN
+
+        # boot_type did not match (skip)
+        return FW_AUTO_ERR_BOOT_TYPE
+
     @staticmethod
     def _read_generic_file(filename, len, ignore_errors=False):
         """
@@ -467,6 +495,40 @@ class ComponentSSD(Component):
 
         return True
 
+    def auto_update_firmware(self, image_path, boot_action):
+        """
+        Handling of attempted automatic update for a SSD of a Mellanox switch.
+        Will first check the image_path to determine if a post-install reboot is required, 
+        then compares it against boot_action to determine whether to proceed with install.
+        """
+
+        # All devices support cold boot
+        supported_boot = ['cold']
+
+        # Verify image path exists
+        if not os.path.exists(image_path):
+            # Invalid image path
+            return FW_AUTO_ERR_IMAGE
+
+        # Check if post_install reboot is required
+        try:
+            if self.get_firmware_update_notification(image_path) is None:
+                # No power cycle required
+                supported_boot += ['warm', 'fast', 'none', 'any']
+        except RuntimeError:
+            # Unknown error from firmware probe
+            return FW_AUTO_ERR_UKNOWN
+
+        if boot_action in supported_boot:
+            if self.install_firmware(image_path):
+                # Successful update
+                return FW_AUTO_INSTALLED
+            # Failed update (unknown reason)
+            return FW_AUTO_ERR_UKNOWN
+
+        # boot_type did not match (skip)
+        return FW_AUTO_ERR_BOOT_TYPE
+
     def get_firmware_version(self):
         cmd = self.SSD_INFO_COMMAND
 

platform/mellanox/mlnx-platform-api/tests/test_firmware.py

@@ -0,0 +1,82 @@
+import os
+import sys
+import pytest
+from mock import MagicMock
+from .mock_platform import MockFan
+
+test_path = os.path.dirname(os.path.abspath(__file__))
+modules_path = os.path.dirname(test_path)
+sys.path.insert(0, modules_path)
+
+from sonic_platform.component import Component, ComponentSSD
+
+from sonic_platform_base.component_base import ComponentBase,           \
+                                                FW_AUTO_INSTALLED,      \
+                                                FW_AUTO_ERR_BOOT_TYPE,  \
+                                                FW_AUTO_ERR_IMAGE,      \
+                                                FW_AUTO_ERR_UKNOWN
+
+def mock_install_firmware_success(image_path):
+    return True
+
+def mock_install_firmware_fail(image_path):
+    return False
+
+def mock_update_notification_cold_boot(image_path):
+    return "Immediate power cycle is required to complete NAME firmware update"
+
+def mock_update_notification_warm_boot(image_path):
+    return None
+
+def mock_update_notification_error(image_path):
+    raise RuntimeError("Failed to parse NAME firmware upgrade status")
+
+test_data_default = [
+        (None, False, None, FW_AUTO_ERR_IMAGE),
+        (None, True, 'warm', FW_AUTO_ERR_BOOT_TYPE),
+        (mock_install_firmware_fail, True, 'cold', FW_AUTO_ERR_UKNOWN),
+        (mock_install_firmware_success, True, 'cold', FW_AUTO_INSTALLED)
+        ]
+
+test_data_ssd = [
+        (None, None, False, None, FW_AUTO_ERR_IMAGE),
+        (None, mock_update_notification_error, True, None, FW_AUTO_ERR_UKNOWN),
+        (mock_install_firmware_fail,    mock_update_notification_cold_boot, True, 'cold', FW_AUTO_ERR_UKNOWN),
+        (mock_install_firmware_success, mock_update_notification_cold_boot, True, 'warm', FW_AUTO_ERR_BOOT_TYPE),
+        (mock_install_firmware_success, mock_update_notification_cold_boot, True, 'cold', FW_AUTO_INSTALLED),
+        (mock_install_firmware_success, mock_update_notification_warm_boot, True, 'warm', FW_AUTO_INSTALLED),
+        (mock_install_firmware_success, mock_update_notification_warm_boot, True, 'cold', FW_AUTO_INSTALLED)
+        ]
+
+@pytest.mark.parametrize('install_func, image_found, boot_type, expect', test_data_default)
+def test_auto_update_firmware_default(monkeypatch, install_func, image_found, boot_type, expect):
+
+    def mock_path_exists(path):
+        return image_found
+
+    test_component = Component()
+
+    monkeypatch.setattr(test_component, 'install_firmware', install_func)
+    monkeypatch.setattr(os.path, 'exists', mock_path_exists)
+
+    result = test_component.auto_update_firmware(None, boot_type)
+
+    assert result == expect
+
+
+@pytest.mark.parametrize('install_func, notify, image_found, boot_type, expect', test_data_ssd)
+def test_auto_update_firmware_default(monkeypatch, install_func, notify, image_found, boot_type, expect):
+
+    def mock_path_exists(path):
+        return image_found
+
+    test_component_ssd = ComponentSSD()
+
+    monkeypatch.setattr(test_component_ssd, 'install_firmware', install_func)
+    monkeypatch.setattr(test_component_ssd, 'get_firmware_update_notification', notify)
+    monkeypatch.setattr(os.path, 'exists', mock_path_exists)
+
+    result = test_component_ssd.auto_update_firmware(None, boot_type)
+
+    assert result == expect
+