Addressed review comments and added _validate_and_get_physical_port function

Author: mihirpat1

sonic-xcvrd/tests/test_xcvrd.py

@@ -586,7 +586,7 @@ def mock_get_transceiver_dom_flags(physical_port):
         dom_tbl = Table("STATE_DB", TRANSCEIVER_DOM_FLAG_TABLE)
         dom_db_utils.xcvr_table_helper.get_dom_flag_tbl = MagicMock(return_value=dom_tbl)
         dom_db_utils.dom_utils.get_transceiver_dom_flags = MagicMock(return_value=None)
-        dom_db_utils.update_flag_metadata_tables = MagicMock()
+        dom_db_utils._update_flag_metadata_tables = MagicMock()
         assert dom_tbl.get_size() == 0
 
         # Ensure table is empty asic_index is None
@@ -618,17 +618,17 @@ def mock_get_transceiver_dom_flags(physical_port):
         dom_db_utils.dom_utils.get_transceiver_dom_flags = MagicMock(side_effect=mock_get_transceiver_dom_flags)
         dom_db_utils.post_port_dom_flags_to_db(logical_port_name, db_cache=db_cache)
         assert dom_tbl.get_size_for_key(logical_port_name) == 21
-        assert dom_db_utils.update_flag_metadata_tables.call_count == 1
+        assert dom_db_utils._update_flag_metadata_tables.call_count == 1
 
         # Reset the mock to clear the call count
-        dom_db_utils.update_flag_metadata_tables.reset_mock()
+        dom_db_utils._update_flag_metadata_tables.reset_mock()
 
         # Ensure db_cache is populated correctly
         assert db_cache.get(0) is not None
         dom_db_utils.dom_utils.get_transceiver_dom_flags = MagicMock(return_value=None)
         dom_db_utils.post_port_dom_flags_to_db(logical_port_name, db_cache=db_cache)
         assert dom_tbl.get_size_for_key(logical_port_name) == 21
-        assert dom_db_utils.update_flag_metadata_tables.call_count == 0
+        assert dom_db_utils._update_flag_metadata_tables.call_count == 0
 
     @pytest.mark.parametrize("flag_value_table, flag_value_table_found, current_value, expected_change_count, expected_set_time, expected_clear_time", [
         (None, False, 'N/A', None, None, None),
@@ -665,7 +665,7 @@ def field_value_pairs_to_dict(fvp):
         stop_event = threading.Event()
         db_utils = DBUtils(mock_sfp_obj_dict, port_mapping, stop_event, mock_logger)
         # Call the function
-        db_utils.update_flag_metadata_tables(logical_port_name, mock_curr_flag_dict,
+        db_utils._update_flag_metadata_tables(logical_port_name, mock_curr_flag_dict,
                                             flag_values_dict_update_time, flag_value_table,
                                             flag_change_count_table, flag_last_set_time_table,
                                             flag_last_clear_time_table, table_name_for_logging)
@@ -966,7 +966,7 @@ def mock_get_transceiver_status(physical_port):
         status_db_utils.post_port_transceiver_hw_status_to_db(logical_port_name, db_cache=db_cache)
         assert status_tbl.get_size_for_key(logical_port_name) == 12
 
-    def testpost_port_transceiver_hw_status_flags_to_db(self):
+    def test_post_port_transceiver_hw_status_flags_to_db(self):
         def mock_get_transceiver_status_flags(physical_port):
             return {
                 "datapath_firmware_fault": "False",
@@ -996,7 +996,7 @@ def mock_get_transceiver_status_flags(physical_port):
         status_flag_tbl = Table("STATE_DB", TRANSCEIVER_STATUS_FLAG_TABLE)
         status_db_utils.xcvr_table_helper.get_status_flag_tbl = MagicMock(return_value=status_flag_tbl)
         status_db_utils.status_utils.get_transceiver_status_flags = MagicMock(return_value=None)
-        status_db_utils.update_flag_metadata_tables = MagicMock()
+        status_db_utils._update_flag_metadata_tables = MagicMock()
         assert status_flag_tbl.get_size() == 0
 
         # Ensure table is empty asic_index is None
@@ -1028,17 +1028,17 @@ def mock_get_transceiver_status_flags(physical_port):
         status_db_utils.status_utils.get_transceiver_status_flags = MagicMock(side_effect=mock_get_transceiver_status_flags)
         status_db_utils.post_port_transceiver_hw_status_flags_to_db(logical_port_name, db_cache=db_cache)
         assert status_flag_tbl.get_size_for_key(logical_port_name) == 12
-        assert status_db_utils.update_flag_metadata_tables.call_count == 1
+        assert status_db_utils._update_flag_metadata_tables.call_count == 1
 
         # Reset the mock to clear the call count
-        status_db_utils.update_flag_metadata_tables.reset_mock()
+        status_db_utils._update_flag_metadata_tables.reset_mock()
 
         # Ensure db_cache is populated correctly
         assert db_cache.get(0) is not None
         status_db_utils.status_utils.get_transceiver_status_flags = MagicMock(return_value=None)
         status_db_utils.post_port_transceiver_hw_status_flags_to_db(logical_port_name, db_cache=db_cache)
         assert status_flag_tbl.get_size_for_key(logical_port_name) == 12
-        assert status_db_utils.update_flag_metadata_tables.call_count == 0
+        assert status_db_utils._update_flag_metadata_tables.call_count == 0
 
     @patch('xcvrd.xcvrd_utilities.port_event_helper.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0]))
     @patch('xcvrd.xcvrd._wrapper_get_presence', MagicMock(return_value=True))
@@ -3301,11 +3301,16 @@ def test_beautify_dom_info_dict(self):
         port_mapping = PortMapping()
         xcvr_table_helper = XcvrTableHelper(DEFAULT_NAMESPACE)
         stop_event = threading.Event()
-        dom_db_utils = DOMDBUtils(mock_sfp_obj_dict, port_mapping, xcvr_table_helper, stop_event, helper_logger)
+        mock_logger = MagicMock()
+        dom_db_utils = DOMDBUtils(mock_sfp_obj_dict, port_mapping, xcvr_table_helper, stop_event, mock_logger)
 
         dom_db_utils._beautify_dom_info_dict(dom_info_dict)
         assert dom_info_dict == expected_dom_info_dict
 
+        # Ensure that the method handles None input gracefully and logs a warning
+        dom_db_utils._beautify_dom_info_dict(None)
+        mock_logger.log_warning.assert_called_once_with("DOM info dict is None while beautifying")
+
     def test_beautify_info_dict(self):
         dom_info_dict = {
             'eSNR' : 1.1,

sonic-xcvrd/xcvrd/dom/utilities/db/utils.py

@@ -29,25 +29,8 @@ def post_diagnostic_values_to_db(self, logical_port_name, table, get_values_func
             beautify_func (function, optional): Function to beautify the diagnostic values. Defaults to self.beautify_info_dict.
             enable_flat_memory_check (bool, optional): Flag to check for flat memory support. Defaults to False.
         """
-        if self.task_stopping_event.is_set():
-            return
-
-        pport_list = self.port_mapping.get_logical_to_physical(logical_port_name)
-        if not pport_list:
-            self.logger.log_error(f"Post port diagnostic values to db failed for {logical_port_name} "
-                                   "as no physical port found")
-            return
-        physical_port = pport_list[0]
-
-        if physical_port not in self.sfp_obj_dict:
-            self.logger.log_error(f"Post port diagnostic values to db failed for {logical_port_name} "
-                                   "as no sfp object found")
-            return
-
-        if not self.xcvrd_utils.get_transceiver_presence(physical_port):
-            return
-
-        if enable_flat_memory_check and self.xcvrd_utils.is_transceiver_flat_memory(physical_port):
+        physical_port = self._validate_and_get_physical_port(logical_port_name, enable_flat_memory_check)
+        if physical_port is None:
             return
 
         try:
@@ -78,7 +61,52 @@ def post_diagnostic_values_to_db(self, logical_port_name, table, get_values_func
                                          "as functionality is not implemented")
             return
 
-    def update_flag_metadata_tables(self, logical_port_name, curr_flag_dict,
+    def _validate_and_get_physical_port(self, logical_port_name, enable_flat_memory_check=False):
+        """
+        Validates the logical port and retrieves the corresponding physical port.
+
+        Validation Steps:
+        1. Ensures `task_stopping_event` is not set.
+        2. Checks if the logical port maps to a physical port.
+        3. Checks if the physical port has an associated SFP object.
+        4. Checks if the transceiver is present.
+        5. (Optional) Ensures the transceiver is not flat memory if `enable_flat_memory_check` is True.
+
+        If any of these checks fail, an error message is logged and `None` is returned.
+        If all checks pass, the physical port number is returned.
+
+        Args:
+            logical_port_name (str): Logical port name.
+            enable_flat_memory_check (bool): Flag to check for flat memory support.
+
+        Returns:
+            int: The physical port number if validation succeeds, or None if validation fails.
+        """
+        if self.task_stopping_event.is_set():
+            return None
+
+        pport_list = self.port_mapping.get_logical_to_physical(logical_port_name)
+        if not pport_list:
+            self.logger.log_error(f"Validate and get physical port failed for {logical_port_name} "
+                                   "as no physical port found")
+            return None
+
+        physical_port = pport_list[0]
+
+        if physical_port not in self.sfp_obj_dict:
+            self.logger.log_error(f"Validate and get physical port failed for {logical_port_name} "
+                                   "as no sfp object found")
+            return None
+
+        if not self.xcvrd_utils.get_transceiver_presence(physical_port):
+            return None
+
+        if enable_flat_memory_check and self.xcvrd_utils.is_transceiver_flat_memory(physical_port):
+            return None
+
+        return physical_port
+
+    def _update_flag_metadata_tables(self, logical_port_name, curr_flag_dict,
                                     flag_values_dict_update_time,
                                     flag_value_table,
                                     flag_change_count_table, flag_last_set_time_table, flag_last_clear_time_table,

sonic-xcvrd/xcvrd/dom/utilities/dom_sensor/db_utils.py

@@ -8,6 +8,10 @@ class DOMDBUtils(DBUtils):
     """
     This class provides utility functions for managing DB operations
     related to DOM on transceivers.
+    Handles data related to the following tables:
+        - TRANSCEIVER_DOM_SENSOR
+        - TRANSCEIVER_DOM_FLAG and its corresponding metadata tables (change count, set time, clear time)
+        - TRANSCEIVER_DOM_THRESHOLD
     """
     TEMP_UNIT = 'C'
     VOLT_UNIT = 'Volts'
@@ -35,31 +39,14 @@ def post_port_dom_sensor_info_to_db(self, logical_port_name, db_cache=None):
                                                  enable_flat_memory_check=True)
 
     def post_port_dom_flags_to_db(self, logical_port_name, db_cache=None):
-        if self.task_stopping_event.is_set():
-            return
-
         asic_index = self.port_mapping.get_asic_id_for_logical_port(logical_port_name)
         if asic_index is None:
             self.logger.log_error(f"Post port dom flags to db failed for {logical_port_name} "
                                   "as no asic index found")
             return
 
-        pport_list = self.port_mapping.get_logical_to_physical(logical_port_name)
-        if not pport_list:
-            self.logger.log_error(f"Post port dom flags to db failed for {logical_port_name} "
-                                    "as no physical port found")
-            return
-        physical_port = pport_list[0]
-
-        if physical_port not in self.sfp_obj_dict:
-            self.logger.log_error(f"Post port dom flags to db failed for {logical_port_name} "
-                                    "as no sfp object found")
-            return
-
-        if not self.xcvrd_utils.get_transceiver_presence(physical_port):
-            return
-
-        if self.xcvrd_utils.is_transceiver_flat_memory(physical_port):
+        physical_port = self._validate_and_get_physical_port(logical_port_name, enable_flat_memory_check=True)
+        if physical_port is None:
             return
 
         try:
@@ -75,7 +62,7 @@ def post_port_dom_flags_to_db(self, logical_port_name, db_cache=None):
                     return
                 if dom_flags_dict:
                     dom_flags_dict_update_time = self.get_current_time()
-                    self.update_flag_metadata_tables(logical_port_name, dom_flags_dict,
+                    self._update_flag_metadata_tables(logical_port_name, dom_flags_dict,
                                                      dom_flags_dict_update_time,
                                                      self.xcvr_table_helper.get_dom_flag_tbl(asic_index),
                                                      self.xcvr_table_helper.get_dom_flag_change_count_tbl(asic_index),
@@ -126,6 +113,10 @@ def _strip_unit(self, value, unit):
 
     # Remove unnecessary unit from the raw data
     def _beautify_dom_info_dict(self, dom_info_dict):
+        if dom_info_dict is None:
+            self.logger.log_warning("DOM info dict is None while beautifying")
+            return
+
         for k, v in dom_info_dict.items():
             if k == 'temperature':
                 dom_info_dict[k] = self._strip_unit(v, self.TEMP_UNIT)

sonic-xcvrd/xcvrd/dom/utilities/status/db_utils.py

@@ -6,7 +6,10 @@
 class StatusDBUtils(DBUtils):
     """
     This class provides utility functions for managing DB operations
-    related to transceiver status.
+    related to transceiver status (specifically, all the hardware related fields of transceiver status).
+    Handles data related to the following tables:
+        - TRANSCEIVER_STATUS
+        - TRANSCEIVER_STATUS_FLAG and its corresponding metadata tables (change count, set time, clear time)
     """
 
     def __init__(self, sfp_obj_dict, port_mapping, xcvr_table_helper, task_stopping_event, logger):
@@ -36,28 +39,14 @@ def post_port_transceiver_hw_status_to_db(self, logical_port_name, db_cache=None
                                                  db_cache=db_cache)
 
     def post_port_transceiver_hw_status_flags_to_db(self, logical_port_name, db_cache=None):
-        if self.task_stopping_event.is_set():
-            return
-
         asic_index = self.port_mapping.get_asic_id_for_logical_port(logical_port_name)
         if asic_index is None:
             self.logger.log_error(f"Post port transceiver hw status flags to db failed for {logical_port_name} "
                                   "as no asic index found")
             return
 
-        pport_list = self.port_mapping.get_logical_to_physical(logical_port_name)
-        if not pport_list:
-            self.logger.log_error(f"Post port transceiver hw status flags to db failed for {logical_port_name} "
-                                    "as no physical port found")
-            return
-        physical_port = pport_list[0]
-
-        if physical_port not in self.sfp_obj_dict:
-            self.logger.log_error(f"Post port transceiver hw status flags to db failed for {logical_port_name} "
-                                  "as no sfp object found")
-            return
-
-        if not self.xcvrd_utils.get_transceiver_presence(physical_port):
+        physical_port = self._validate_and_get_physical_port(logical_port_name)
+        if physical_port is None:
             return
 
         try:
@@ -72,7 +61,7 @@ def post_port_transceiver_hw_status_flags_to_db(self, logical_port_name, db_cach
                                             "as no status flags found")
                     return
                 if status_flags_dict:
-                    self.update_flag_metadata_tables(logical_port_name, status_flags_dict,
+                    self._update_flag_metadata_tables(logical_port_name, status_flags_dict,
                                                      self.get_current_time(),
                                                      self.xcvr_table_helper.get_status_flag_tbl(asic_index),
                                                      self.xcvr_table_helper.get_status_flag_change_count_tbl(asic_index),
@@ -97,6 +86,6 @@ def post_port_transceiver_hw_status_flags_to_db(self, logical_port_name, db_cach
                 return
 
         except NotImplementedError:
-            self.logger.log_error(f"Post port transceiver hw status flags to db failed for {logical_port_name} "
+            self.logger.log_notice(f"Post port transceiver hw status flags to db failed for {logical_port_name} "
                                    "as functionality is not implemented")
             return

sonic-xcvrd/xcvrd/dom/utilities/vdm/db_utils.py

@@ -9,6 +9,12 @@ class VDMDBUtils(DBUtils):
     """
     This class provides utility functions for managing
     DB operations related to VDM on transceivers.
+    Handles data related to the following tables:
+        - TRANSCEIVER_VDM_REAL_VALUE
+        - TRANSCEIVER_VDM_XXXX_FLAG and its corresponding metadata tables (change count, set time, clear time)
+            - XXXX refers to HALARM, LALARM, HWARN or LWARN
+        - TRANSCEIVER_VDM_XXXX_THRESHOLD
+            - XXXX refers to HALARM, LALARM, HWARN or LWARN
     """
     def __init__(self, sfp_obj_dict, port_mapping, xcvr_table_helper, task_stopping_event, logger):
         super().__init__(sfp_obj_dict, port_mapping, task_stopping_event, logger)
@@ -40,25 +46,8 @@ def post_port_vdm_thresholds_to_db(self, logical_port_name, db_cache=None):
     # Update transceiver VDM threshold or flag info to db
     def _post_port_vdm_thresholds_or_flags_to_db(self, logical_port_name, get_vdm_table_func,
                                                  get_vdm_values_func, flag_data=False, db_cache=None):
-        if self.task_stopping_event.is_set():
-            return
-
-        pport_list = self.port_mapping.get_logical_to_physical(logical_port_name)
-        if not pport_list:
-            self.logger.log_error(f"Post port vdm thresholds or flags to db failed for {logical_port_name} "
-                                         "as no physical port found with flag_data {flag_data}")
-            return
-        physical_port = pport_list[0]
-
-        if physical_port not in self.sfp_obj_dict:
-            self.logger.log_error(f"Post port vdm thresholds or flags to db failed for {logical_port_name} "
-                                         "as no sfp object found with flag_data {flag_data}")
-            return
-
-        if not self.xcvrd_utils.get_transceiver_presence(physical_port):
-            return
-
-        if self.xcvrd_utils.is_transceiver_flat_memory(physical_port):
+        physical_port = self._validate_and_get_physical_port(logical_port_name, enable_flat_memory_check=True)
+        if physical_port is None:
             return
 
         try:
@@ -89,7 +78,7 @@ def _post_port_vdm_thresholds_or_flags_to_db(self, logical_port_name, get_vdm_ta
                     # for the flags
                     if flag_data and threshold_value_dict:
                             asic_id = self.port_mapping.get_asic_id_for_logical_port(logical_port_name)
-                            self.update_flag_metadata_tables(logical_port_name, threshold_value_dict,
+                            self._update_flag_metadata_tables(logical_port_name, threshold_value_dict,
                                                              vdm_values_dict_update_time,
                                                              self.xcvr_table_helper.get_vdm_flag_tbl(asic_id, threshold_type),
                                                              self.xcvr_table_helper.get_vdm_flag_change_count_tbl(asic_id, threshold_type),