Check low power mode before vdm handling (#653)

* Check lp mode before vdm handling.

* Check lp mode before vdm handling.

* Add unit tests.

* Fix failing unit tests.

* Get platform independent method to get lpmode

* Update test.

Author: az-pz

sonic-xcvrd/tests/test_xcvrd.py

@@ -3673,6 +3673,7 @@ def test_DomInfoUpdateTask_task_worker(self, mock_post_pm_info,
         task.status_db_utils.post_port_transceiver_hw_status_to_db = MagicMock()
         task.status_db_utils.post_port_transceiver_hw_status_flags_to_db = MagicMock()
         task.vdm_utils.is_transceiver_vdm_supported = MagicMock(return_value=True)
+        task.xcvrd_utils.is_transceiver_lpmode_on = MagicMock(return_value=False)
         task.vdm_db_utils = MagicMock()
         task.vdm_db_utils.post_port_vdm_real_values_to_db = MagicMock()
         task.task_worker()
@@ -3737,6 +3738,7 @@ def test_DomInfoUpdateTask_task_worker_vdm_failure(self, mock_post_pm_info):
         task.vdm_utils._unfreeze_vdm_stats_and_confirm = MagicMock(return_value=True)
         task.vdm_db_utils.post_port_vdm_real_values_to_db = MagicMock()
         task.vdm_db_utils.post_port_vdm_flags_to_db = MagicMock()
+        task.xcvrd_utils.is_transceiver_lpmode_on = MagicMock(return_value=False)
         task.task_worker()
         assert task.vdm_utils._unfreeze_vdm_stats_and_confirm.call_count == 1
         assert task.vdm_db_utils.post_port_vdm_real_values_to_db.call_count == 0
@@ -4698,6 +4700,33 @@ def test_is_transceiver_flat_memory(self):
         mock_api.is_flat_memory = MagicMock(side_effect=NotImplementedError)
         assert xcvrd_util.is_transceiver_flat_memory(1)
 
+    def test_is_transceiver_lpmode_on(self):
+        from xcvrd.xcvrd_utilities.utils import XCVRDUtils
+        mock_sfp = MagicMock()
+        xcvrd_util = XCVRDUtils({1: mock_sfp}, MagicMock())
+
+        # Test case where get_xcvr_api returns None
+        mock_sfp.get_lpmode = MagicMock(return_value=None)
+        assert not xcvrd_util.is_transceiver_lpmode_on(1)
+
+        # Test case where get_lpmode returns True
+        mock_sfp.get_lpmode = MagicMock(return_value=True)
+        assert xcvrd_util.is_transceiver_lpmode_on(1)
+
+        # Test case where get_lpmode returns False
+
+        mock_sfp.get_lpmode = MagicMock(return_value=False)
+        assert not xcvrd_util.is_transceiver_lpmode_on(1)
+
+        # Test case where get_xcvr_api raises KeyError
+        xcvrd_util.sfp_obj_dict = {}
+        assert not xcvrd_util.is_transceiver_lpmode_on(1)
+
+        # Test case where is_flat_memory raises NotImplementedError
+        xcvrd_util.sfp_obj_dict = {1: mock_sfp}
+        mock_sfp.get_lpmode = MagicMock(side_effect=NotImplementedError)
+        assert not xcvrd_util.is_transceiver_lpmode_on(1)
+
     @patch('time.sleep', MagicMock())
     @patch('xcvrd.xcvrd.XcvrTableHelper', MagicMock())
     @patch('xcvrd.xcvrd._wrapper_soak_sfp_insert_event', MagicMock())

sonic-xcvrd/xcvrd/dom/dom_mgr.py

@@ -303,7 +303,7 @@ def task_worker(self):
                         self.log_warning("Got exception {} while processing transceiver status hw flags for "
                                          "port {}, ignored".format(repr(e), logical_port_name))
                         continue
-                    if self.vdm_utils.is_transceiver_vdm_supported(physical_port):
+                    if self.vdm_utils.is_transceiver_vdm_supported(physical_port) and (not self.xcvrd_utils.is_transceiver_lpmode_on(physical_port)):
                         # Freeze VDM stats before reading VDM values
                         with self.vdm_utils.vdm_freeze_context(physical_port) as vdm_frozen:
                             if not vdm_frozen:

sonic-xcvrd/xcvrd/xcvrd_utilities/utils.py

@@ -23,3 +23,10 @@ def is_transceiver_flat_memory(self, physical_port):
         except (KeyError, NotImplementedError):
             self.logger.log_error(f"Failed to check flat memory for port {physical_port}")
             return True
+
+    def is_transceiver_lpmode_on(self, physical_port):
+        try:
+            return self.sfp_obj_dict[physical_port].get_lpmode()
+        except Exception as e:
+            self.logger.log_error(f"Failed to get low power mode for port {physical_port}. Exception: {e}")
+            return False