Updating xcvrd test to work with media_settings_parser and moved the …

…function check_port_in_range() from media_settings_parser back to xcvrd as optics_si_parser also uses it

sonic-xcvrd/tests/test_xcvrd.py

@@ -1,6 +1,7 @@
 #from unittest.mock import DEFAULT
 from xcvrd.xcvrd_utilities.port_mapping import *
 from xcvrd.xcvrd_utilities.sfp_status_helper import *
+from xcvrd.xcvrd_utilities.media_settings_parser import *
 from xcvrd.xcvrd_utilities.optics_si_parser import *
 from xcvrd.xcvrd import *
 import pytest
@@ -461,30 +462,30 @@ def test_get_media_settings_key(self, mock_chassis):
         }
 
         # Test a good 'specification_compliance' value
-        result = get_media_settings_key(0, xcvr_info_dict, 100000, 2)
-        assert result == ('MOLEX-1064141421', 'QSFP+-10GBase-SR-255M', 'speed:100GBASE-CR2')
+        result = media_settings_parser.get_media_settings_key(0, xcvr_info_dict, 100000, 2)
+        assert result == { 'vendor_key': 'MOLEX-1064141421', 'media_key': 'QSFP+-10GBase-SR-255M', 'lane_speed_key': 'speed:100GBASE-CR2' }
 
         # Test a bad 'specification_compliance' value
         xcvr_info_dict[0]['specification_compliance'] = 'N/A'
-        result = get_media_settings_key(0, xcvr_info_dict, 100000, 2)
-        assert result == ('MOLEX-1064141421', 'QSFP+-*', 'speed:100GBASE-CR2')
+        result = media_settings_parser.get_media_settings_key(0, xcvr_info_dict, 100000, 2)
+        assert result == { 'vendor_key': 'MOLEX-1064141421', 'media_key': 'QSFP+-*', 'lane_speed_key': 'speed:100GBASE-CR2' }
         # TODO: Ensure that error message was logged
 
     @patch('xcvrd.xcvrd.g_dict', media_settings_dict)
     @patch('xcvrd.xcvrd._wrapper_get_presence', MagicMock(return_value=True))
     @patch('xcvrd.xcvrd.XcvrTableHelper', MagicMock())
     @patch('xcvrd.xcvrd.XcvrTableHelper.get_cfg_port_tbl', MagicMock())
-    @patch('xcvrd.xcvrd.get_media_settings_key', MagicMock(return_value=('MOLEX-1064141421', 'QSFP+-10GBase-SR-255M', 'speed:100GBASE-CR2')))
-    @patch('xcvrd.xcvrd.get_speed_and_lane_count', MagicMock(return_value=(100000, 2)))
+    @patch('xcvrd.xcvrd_utilities.media_settings_parser.get_media_settings_key', MagicMock(return_value={ 'vendor_key': 'MOLEX-1064141421', 'media_key': 'QSFP+-10GBase-SR-255M', 'lane_speed_key': 'speed:100GBASE-CR2' }))
+    @patch('xcvrd.xcvrd_utilities.media_settings_parser.get_speed_and_lane_count', MagicMock(return_value=(100000, 2)))
     def test_notify_media_setting(self):
         self._check_notify_media_setting(1)
 
     @patch('xcvrd.xcvrd.g_dict', media_settings_with_comma_dict)
     @patch('xcvrd.xcvrd._wrapper_get_presence', MagicMock(return_value=True))
     @patch('xcvrd.xcvrd.XcvrTableHelper', MagicMock())
     @patch('xcvrd.xcvrd.XcvrTableHelper.get_cfg_port_tbl', MagicMock())
-    @patch('xcvrd.xcvrd.get_media_settings_key', MagicMock(return_value=('MOLEX-1064141421', 'QSFP+-10GBase-SR-255M', 'speed:100GBASE-CR2')))
-    @patch('xcvrd.xcvrd.get_speed_and_lane_count', MagicMock(return_value=(100000, 2)))
+    @patch('xcvrd.xcvrd_utilities.media_settings_parser.get_media_settings_key', MagicMock(return_value={ 'vendor_key': 'MOLEX-1064141421', 'media_key': 'QSFP+-10GBase-SR-255M', 'lane_speed_key': 'speed:100GBASE-CR2' }))
+    @patch('xcvrd.xcvrd_utilities.media_settings_parser.get_speed_and_lane_count', MagicMock(return_value=(100000, 2)))
     def test_notify_media_setting_with_comma(self):
         self._check_notify_media_setting(1)
         self._check_notify_media_setting(6)
@@ -509,7 +510,7 @@ def _check_notify_media_setting(self, index):
         port_mapping = PortMapping()
         port_change_event = PortChangeEvent('Ethernet0', index, 0, PortChangeEvent.PORT_ADD)
         port_mapping.handle_port_change_event(port_change_event)
-        notify_media_setting(logical_port_name, xcvr_info_dict, app_port_tbl, mock_cfg_table, port_mapping)
+        media_settings_parser.notify_media_setting(logical_port_name, xcvr_info_dict, app_port_tbl, mock_cfg_table, port_mapping)
 
     @patch('xcvrd.xcvrd_utilities.optics_si_parser.g_optics_si_dict', optics_si_settings_dict)
     @patch('xcvrd.xcvrd._wrapper_get_presence', MagicMock(return_value=True))
@@ -792,6 +793,7 @@ def get_application(lane):
 
         assert task.is_cmis_application_update_required(mock_xcvr_api, app_new, host_lanes_mask) == expected
 
+
     @patch('xcvrd.xcvrd.is_cmis_api', MagicMock(return_value=True))
     @pytest.mark.parametrize("host_lane_count, speed, subport, expected", [
         (8, 400000, 0, 0xFF),
@@ -839,6 +841,7 @@ def get_host_lane_assignment_option_side_effect(app):
         appl = get_cmis_application_desired(mock_xcvr_api, host_lane_count, speed)
         assert task.get_cmis_host_lanes_mask(mock_xcvr_api, appl, host_lane_count, subport) == expected
 
+
     def test_CmisManagerTask_post_port_active_apsel_to_db(self):
         mock_xcvr_api = MagicMock()
         mock_xcvr_api.get_active_apsel_hostlane = MagicMock(side_effect=[
@@ -922,6 +925,7 @@ def test_CmisManagerTask_post_port_active_apsel_to_db(self):
         ret = task.post_port_active_apsel_to_db(mock_xcvr_api, lport, host_lanes_mask)
         assert int_tbl.getKeys() == []
 
+
     @patch('xcvrd.xcvrd.platform_chassis')
     @patch('xcvrd.xcvrd_utilities.port_mapping.subscribe_port_update_event', MagicMock(return_value=(None, None)))
     @patch('xcvrd.xcvrd_utilities.port_mapping.handle_port_update_event', MagicMock())
@@ -1263,7 +1267,7 @@ def test_SfpStateUpdateTask_mapping_event_from_change_event(self):
     @patch('xcvrd.xcvrd.SfpStateUpdateTask._mapping_event_from_change_event')
     @patch('xcvrd.xcvrd._wrapper_get_transceiver_change_event')
     @patch('xcvrd.xcvrd.del_port_sfp_dom_info_from_db')
-    @patch('xcvrd.xcvrd.notify_media_setting')
+    @patch('xcvrd.xcvrd_utilities.media_settings_parser.notify_media_setting')
     @patch('xcvrd.xcvrd.post_port_dom_threshold_info_to_db')
     @patch('xcvrd.xcvrd.post_port_sfp_info_to_db')
     @patch('xcvrd.xcvrd.update_port_transceiver_status_table_sw')
@@ -1359,7 +1363,7 @@ def test_SfpStateUpdateTask_task_worker(self, mock_del_status_hw,
 
     @patch('xcvrd.xcvrd.XcvrTableHelper')
     @patch('xcvrd.xcvrd._wrapper_get_presence')
-    @patch('xcvrd.xcvrd.notify_media_setting')
+    @patch('xcvrd.xcvrd_utilities.media_settings_parser.notify_media_setting')
     @patch('xcvrd.xcvrd.post_port_dom_threshold_info_to_db')
     @patch('xcvrd.xcvrd.post_port_sfp_info_to_db')
     @patch('xcvrd.xcvrd.update_port_transceiver_status_table_sw')
@@ -1632,7 +1636,7 @@ def test_check_port_in_range(self):
 
     def test_get_media_val_str_from_dict(self):
         media_dict = {'lane0': '1', 'lane1': '2'}
-        media_str = get_media_val_str_from_dict(media_dict)
+        media_str = media_settings_parser.get_media_val_str_from_dict(media_dict)
         assert media_str == '1,2'
 
     def test_get_media_val_str(self):

sonic-xcvrd/xcvrd/xcvrd.py

@@ -638,6 +638,17 @@ def del_port_sfp_dom_info_from_db(logical_port_name, port_mapping, int_tbl, dom_
             sys.exit(NOT_IMPLEMENTED_ERROR)
 
 
+def check_port_in_range(range_str, physical_port):
+    RANGE_SEPARATOR = '-'
+
+    range_list = range_str.split(RANGE_SEPARATOR)
+    start_num = int(range_list[0].strip())
+    end_num = int(range_list[1].strip())
+    if start_num <= physical_port <= end_num:
+        return True
+    return False
+
+
 def waiting_time_compensation_with_sleep(time_start, time_to_wait):
     time_now = time.time()
     time_diff = time_now - time_start

sonic-xcvrd/xcvrd/xcvrd_utilities/media_settings_parser.py

@@ -107,17 +107,6 @@ def is_si_per_speed_supported(media_dict):
     return LANE_SPEED_KEY_PREFIX in list(media_dict.keys())[0]
 
 
-def check_port_in_range(range_str, physical_port):
-    RANGE_SEPARATOR = '-'
-
-    range_list = range_str.split(RANGE_SEPARATOR)
-    start_num = int(range_list[0].strip())
-    end_num = int(range_list[1].strip())
-    if start_num <= physical_port <= end_num:
-        return True
-    return False
-
-
 def get_media_val_str_from_dict(media_dict):
     LANE_STR = 'lane'
     LANE_SEPARATOR = ','
@@ -183,15 +172,15 @@ def get_media_settings_value(physical_port, key):
                 port_list = keys.split(COMMA_SEPARATOR)
                 for port in port_list:
                     if RANGE_SEPARATOR in port:
-                        if check_port_in_range(port, physical_port):
+                        if xcvrd.check_port_in_range(port, physical_port):
                             media_dict = g_dict[GLOBAL_MEDIA_SETTINGS_KEY][keys]
                             break
                     elif str(physical_port) == port:
                         media_dict = g_dict[GLOBAL_MEDIA_SETTINGS_KEY][keys]
                         break
 
             elif RANGE_SEPARATOR in keys:
-                if check_port_in_range(keys, physical_port):
+                if xcvrd.check_port_in_range(keys, physical_port):
                     media_dict = g_dict[GLOBAL_MEDIA_SETTINGS_KEY][keys]
 
             # If there is a match in the global profile for a media type,