[platform][mellanox] Implement auto_firmware_update for Mellanox platform to support fwutil auto-update

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

@@ -20,7 +20,11 @@
     else:
         import ConfigParser as configparser
 
-    from sonic_platform_base.component_base import ComponentBase
+    from sonic_platform_base.component_base import ComponentBase,           \
+                                                    FW_AUTO_UPDATED,        \
+                                                    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")
 
@@ -334,6 +338,30 @@ def get_name(self):
     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_UPDATED
+            # 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):
         """
@@ -469,6 +497,40 @@ def __install_firmware(self, image_path):
 
         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_UPDATED
+            # 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,76 @@
+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
+
+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, -2),
+        (None, True, 'warm', -1),
+        (mock_install_firmware_fail, True, 'cold', -3),
+        (mock_install_firmware_success, True, 'cold', 2)
+        ]
+
+test_data_ssd = [
+        (None, None, False, None, -2),
+        (None, mock_update_notification_error, True, None, -3),
+        (mock_install_firmware_fail,    mock_update_notification_cold_boot, True, 'cold', -3),
+        (mock_install_firmware_success, mock_update_notification_cold_boot, True, 'warm', -1),
+        (mock_install_firmware_success, mock_update_notification_cold_boot, True, 'cold', 2),
+        (mock_install_firmware_success, mock_update_notification_warm_boot, True, 'warm', 2),
+        (mock_install_firmware_success, mock_update_notification_warm_boot, True, 'cold', 2)
+        ]
+
+@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
+