From 3e432e7e40223e6a4b3c679c655cc579d3a6fbb5 Mon Sep 17 00:00:00 2001 From: Ashwin Srinivasan <93744978+assrinivasan@users.noreply.github.com> Date: Fri, 24 Dec 2021 09:39:54 -0800 Subject: [PATCH] [Y-Cable] Increased unit test coverage of y_cable_helper.py (#229) Added unit tests for the following methods: y_cable_wrapper_get_transceiver_info y_cable_wrapper_get_presence y_cable_toggle_mux_torA y_cable_toggle_mux_torB format_mapping_identifier update_appdb_port_mux_cable_response_table update_table_mux_status_for_response_tbl update_tor_active_side read_y_cable_and_update_statedb_port_tbl delete_port_from_y_cable_table create_tables_and_insert_mux_unknown_entries check_identifier_presencead_update_mux_table_entry init_ports_status_for_y_cable change_ports_status_for_y_cable_change_event delete_ports_status_for_y_cable get_firmware_dict get_muxcable_info get_muxcable_static_info post_mux_info_to_db task_download_firmware_worker Motivation and Context The motivation for these changes stems from needing to increase the unit test coverage of the y_cable_helper methods to above 50%. How Has This Been Tested? This was tested by building and running the new unit tests and verifying that they contribute to and increase the code coverage as reflected in the xcvrd_xcvrd_utilities_y_cable_helper_py.html page. Co-authored-by: Ashwin Srinivasan --- sonic-xcvrd/tests/mock_swsscommon.py | 6 + sonic-xcvrd/tests/test_xcvrd.py | 213 +- sonic-xcvrd/tests/test_y_cable_helper.py | 2010 +++++++++++++++++ .../xcvrd/xcvrd_utilities/y_cable_helper.py | 2 +- 4 files changed, 2052 insertions(+), 179 deletions(-) create mode 100644 sonic-xcvrd/tests/test_y_cable_helper.py diff --git a/sonic-xcvrd/tests/mock_swsscommon.py b/sonic-xcvrd/tests/mock_swsscommon.py index 5794efcbcc51..819547082e48 100644 --- a/sonic-xcvrd/tests/mock_swsscommon.py +++ b/sonic-xcvrd/tests/mock_swsscommon.py @@ -6,14 +6,17 @@ class Table: def __init__(self, db, table_name): self.table_name = table_name self.mock_dict = {} + self.mock_keys = [] def _del(self, key): if key in self.mock_dict: del self.mock_dict[key] + self.mock_keys.remove(key) pass def set(self, key, fvs): self.mock_dict[key] = fvs + self.mock_keys.append(key) pass def get(self, key): @@ -23,6 +26,9 @@ def get(self, key): def get_size(self): return (len(self.mock_dict)) + + def getKeys(self): + return self.mock_keys class FieldValuePairs: diff --git a/sonic-xcvrd/tests/test_xcvrd.py b/sonic-xcvrd/tests/test_xcvrd.py index af2001f3475b..5c56730f354d 100644 --- a/sonic-xcvrd/tests/test_xcvrd.py +++ b/sonic-xcvrd/tests/test_xcvrd.py @@ -1,3 +1,7 @@ +from xcvrd.xcvrd_utilities.port_mapping import * +from xcvrd.xcvrd_utilities.sfp_status_helper import * +from xcvrd.xcvrd_utilities.y_cable_helper import * +from xcvrd.xcvrd import * import copy import os import sys @@ -19,10 +23,6 @@ swsscommon.ProducerStateTable = MagicMock() swsscommon.SubscriberStateTable = MagicMock() swsscommon.SonicDBConfig = MagicMock() -#swsscommon.Select = MagicMock() - -sys.modules['sonic_y_cable'] = MagicMock() -sys.modules['sonic_y_cable.y_cable'] = MagicMock() test_path = os.path.dirname(os.path.abspath(__file__)) modules_path = os.path.dirname(test_path) @@ -30,10 +30,6 @@ sys.path.insert(0, modules_path) os.environ["XCVRD_UNIT_TESTING"] = "1" -from xcvrd.xcvrd import * -from xcvrd.xcvrd_utilities.y_cable_helper import * -from xcvrd.xcvrd_utilities.sfp_status_helper import * -from xcvrd.xcvrd_utilities.port_mapping import * with open(os.path.join(test_path, 'media_settings.json'), 'r') as f: media_settings_dict = json.load(f) @@ -239,148 +235,6 @@ def test_init_port_sfp_status_tbl(self): stop_event = threading.Event() init_port_sfp_status_tbl(port_mapping, stop_event) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil', MagicMock(return_value=[0])) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) - @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.get_muxcable_info', MagicMock(return_value={'tor_active': 'self', - 'mux_direction': 'self', - 'manual_switch_count': '7', - 'auto_switch_count': '71', - 'link_status_self': 'up', - 'link_status_peer': 'up', - 'link_status_nic': 'up', - 'nic_lane1_active': 'True', - 'nic_lane2_active': 'True', - 'nic_lane3_active': 'True', - 'nic_lane4_active': 'True', - 'self_eye_height_lane1': '500', - 'self_eye_height_lane2': '510', - 'peer_eye_height_lane1': '520', - 'peer_eye_height_lane2': '530', - 'nic_eye_height_lane1': '742', - 'nic_eye_height_lane2': '750', - 'internal_temperature': '28', - 'internal_voltage': '3.3', - 'nic_temperature': '20', - 'nic_voltage': '2.7', - 'version_nic_active': '1.6MS', - 'version_nic_inactive': '1.7MS', - 'version_nic_next': '1.7MS', - 'version_self_active': '1.6MS', - 'version_self_inactive': '1.7MS', - 'version_self_next': '1.7MS', - 'version_peer_active': '1.6MS', - 'version_peer_inactive': '1.7MS', - 'version_peer_next': '1.7MS'})) - def test_post_port_mux_info_to_db(self): - logical_port_name = "Ethernet0" - port_mapping = PortMapping() - mux_tbl = Table("STATE_DB", y_cable_helper.MUX_CABLE_INFO_TABLE) - rc = post_port_mux_info_to_db(logical_port_name, port_mapping, mux_tbl) - assert(rc != -1) - - - @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil', MagicMock(return_value=[0])) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) - @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.get_muxcable_static_info', MagicMock(return_value={'read_side': 'self', - 'nic_lane1_precursor1': '1', - 'nic_lane1_precursor2': '-7', - 'nic_lane1_maincursor': '-1', - 'nic_lane1_postcursor1': '11', - 'nic_lane1_postcursor2': '11', - 'nic_lane2_precursor1': '12', - 'nic_lane2_precursor2': '7', - 'nic_lane2_maincursor': '7', - 'nic_lane2_postcursor1': '7', - 'nic_lane2_postcursor2': '7', - 'tor_self_lane1_precursor1': '17', - 'tor_self_lane1_precursor2': '17', - 'tor_self_lane1_maincursor': '17', - 'tor_self_lane1_postcursor1': '17', - 'tor_self_lane1_postcursor2': '17', - 'tor_self_lane2_precursor1': '7', - 'tor_self_lane2_precursor2': '7', - 'tor_self_lane2_maincursor': '7', - 'tor_self_lane2_postcursor1': '7', - 'tor_self_lane2_postcursor2': '7', - 'tor_peer_lane1_precursor1': '7', - 'tor_peer_lane1_precursor2': '7', - 'tor_peer_lane1_maincursor': '17', - 'tor_peer_lane1_postcursor1': '7', - 'tor_peer_lane1_postcursor2': '17', - 'tor_peer_lane2_precursor1': '7', - 'tor_peer_lane2_precursor2': '7', - 'tor_peer_lane2_maincursor': '17', - 'tor_peer_lane2_postcursor1': '7', - 'tor_peer_lane2_postcursor2': '17'})) - def test_post_port_mux_static_info_to_db(self): - logical_port_name = "Ethernet0" - port_mapping = PortMapping() - mux_tbl = Table("STATE_DB", y_cable_helper.MUX_CABLE_STATIC_INFO_TABLE) - rc = post_port_mux_static_info_to_db(logical_port_name, port_mapping, mux_tbl) - assert(rc != -1) - - def test_y_cable_helper_format_mapping_identifier1(self): - rc = format_mapping_identifier("ABC ") - assert(rc == "abc") - - def test_y_cable_wrapper_get_transceiver_info(self): - with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: - patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'Microsoft', - 'model': 'model1'} - - transceiver_dict = y_cable_wrapper_get_transceiver_info(1) - vendor = transceiver_dict.get('manufacturer') - model = transceiver_dict.get('model') - - assert(vendor == "Microsoft") - assert(model == "model1") - - def test_y_cable_wrapper_get_presence(self): - with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: - patched_util.get_presence.return_value = True - - presence = y_cable_wrapper_get_presence(1) - - assert(presence == True) - - @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) - def test_get_ycable_physical_port_from_logical_port(self): - port_mapping = PortMapping() - instance = get_ycable_physical_port_from_logical_port("Ethernet0", port_mapping) - - assert(instance == 0) - - @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) - @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) - def test_get_ycable_port_instance_from_logical_port(self): - - with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: - patched_util.get.return_value = 0 - port_mapping = PortMapping() - instance = get_ycable_port_instance_from_logical_port("Ethernet0", port_mapping) - - assert(instance == 0) - - def test_set_show_firmware_fields(self): - - mux_info_dict = {} - xcvrd_show_fw_res_tbl = Table("STATE_DB", "XCVRD_SHOW_FW_RES") - mux_info_dict['version_self_active'] = '0.8' - mux_info_dict['version_self_inactive'] = '0.7' - mux_info_dict['version_self_next'] = '0.7' - mux_info_dict['version_peer_active'] = '0.8' - mux_info_dict['version_peer_inactive'] = '0.7' - mux_info_dict['version_peer_next'] = '0.7' - mux_info_dict['version_nic_active'] = '0.8' - mux_info_dict['version_nic_inactive'] = '0.7' - mux_info_dict['version_nic_next'] = '0.7' - rc = set_show_firmware_fields("Ethernet0", mux_info_dict, xcvrd_show_fw_res_tbl) - - assert(rc == 0) - def test_get_media_settings_key(self): xcvr_info_dict = { 0: { @@ -432,7 +286,6 @@ def _check_notify_media_setting(self, index): port_mapping.handle_port_change_event(port_change_event) notify_media_setting(logical_port_name, xcvr_info_dict, app_port_tbl, port_mapping) - def test_detect_port_in_error_status(self): class MockTable: def get(self, key): @@ -458,7 +311,8 @@ def test_is_error_sfp_status(self): @patch('swsscommon.swsscommon.Select.select') def test_handle_port_config_change(self, mock_select, mock_sub_table): mock_selectable = MagicMock() - mock_selectable.pop = MagicMock(side_effect=[('Ethernet0', swsscommon.SET_COMMAND, (('index', '1'), )), (None, None, None)]) + mock_selectable.pop = MagicMock( + side_effect=[('Ethernet0', swsscommon.SET_COMMAND, (('index', '1'), )), (None, None, None)]) mock_select.return_value = (swsscommon.Select.OBJECT, mock_selectable) mock_sub_table.return_value = mock_selectable @@ -467,15 +321,18 @@ def test_handle_port_config_change(self, mock_select, mock_sub_table): stop_event = threading.Event() stop_event.is_set = MagicMock(return_value=False) logger = MagicMock() - handle_port_config_change(sel, asic_context, stop_event, port_mapping, logger, port_mapping.handle_port_change_event) + handle_port_config_change(sel, asic_context, stop_event, port_mapping, + logger, port_mapping.handle_port_change_event) assert port_mapping.logical_port_list.count('Ethernet0') assert port_mapping.get_asic_id_for_logical_port('Ethernet0') == 0 assert port_mapping.get_physical_to_logical(1) == ['Ethernet0'] assert port_mapping.get_logical_to_physical('Ethernet0') == [1] - mock_selectable.pop = MagicMock(side_effect=[('Ethernet0', swsscommon.DEL_COMMAND, (('index', '1'), )), (None, None, None)]) - handle_port_config_change(sel, asic_context, stop_event, port_mapping, logger, port_mapping.handle_port_change_event) + mock_selectable.pop = MagicMock( + side_effect=[('Ethernet0', swsscommon.DEL_COMMAND, (('index', '1'), )), (None, None, None)]) + handle_port_config_change(sel, asic_context, stop_event, port_mapping, + logger, port_mapping.handle_port_change_event) assert not port_mapping.logical_port_list assert not port_mapping.logical_to_physical assert not port_mapping.physical_to_logical @@ -498,13 +355,13 @@ def test_get_port_mapping(self, mock_swsscommon_table): assert port_mapping.get_physical_to_logical(2) == ['Ethernet4'] assert port_mapping.get_logical_to_physical('Ethernet4') == [2] - @patch('swsscommon.swsscommon.Select.addSelectable', MagicMock()) @patch('swsscommon.swsscommon.SubscriberStateTable') @patch('swsscommon.swsscommon.Select.select') def test_DaemonXcvrd_wait_for_port_config_done(self, mock_select, mock_sub_table): mock_selectable = MagicMock() - mock_selectable.pop = MagicMock(side_effect=[('Ethernet0', swsscommon.SET_COMMAND, (('index', '1'), )), ('PortConfigDone', None, None)]) + mock_selectable.pop = MagicMock( + side_effect=[('Ethernet0', swsscommon.SET_COMMAND, (('index', '1'), )), ('PortConfigDone', None, None)]) mock_select.return_value = (swsscommon.Select.OBJECT, mock_selectable) mock_sub_table.return_value = mock_selectable xcvrd = DaemonXcvrd(SYSLOG_IDENTIFIER) @@ -720,10 +577,11 @@ def test_DomInfoUpdateTask_task_run_stop(self): @patch('swsscommon.swsscommon.Select.select') def test_DomInfoUpdateTask_task_worker(self, mock_select, mock_sub_table, mock_post_dom_th, mock_post_dom_info, mock_detect_error): mock_selectable = MagicMock() - mock_selectable.pop = MagicMock(side_effect=[('Ethernet0', swsscommon.SET_COMMAND, (('index', '1'), )), (None, None, None), (None, None, None)]) + mock_selectable.pop = MagicMock( + side_effect=[('Ethernet0', swsscommon.SET_COMMAND, (('index', '1'), )), (None, None, None), (None, None, None)]) mock_select.return_value = (swsscommon.Select.OBJECT, mock_selectable) mock_sub_table.return_value = mock_selectable - + port_mapping = PortMapping() task = DomInfoUpdateTask(port_mapping) task.task_stopping_event.wait = MagicMock(side_effect=[False, True]) @@ -741,7 +599,6 @@ def test_DomInfoUpdateTask_task_worker(self, mock_select, mock_sub_table, mock_p assert mock_post_dom_th.call_count == 1 assert mock_post_dom_info.call_count == 1 - @patch('xcvrd.xcvrd._wrapper_get_presence', MagicMock(return_value=False)) @patch('xcvrd.xcvrd.xcvr_table_helper') def test_SfpStateUpdateTask_handle_port_change_event(self, mock_table_helper): @@ -852,7 +709,7 @@ def test_SfpStateUpdateTask_task_worker(self, mock_updata_status, mock_post_sfp_ task = SfpStateUpdateTask(port_mapping, retry_eeprom_set) stop_event = multiprocessing.Event() sfp_error_event = multiprocessing.Event() - mock_change_event.return_value = (True, {0:0}, {}) + mock_change_event.return_value = (True, {0: 0}, {}) mock_mapping_event.return_value = SYSTEM_NOT_READY # Test state machine: STATE_INIT + SYSTEM_NOT_READY event => STATE_INIT + SYSTEM_NOT_READY event ... => STATE_EXIT @@ -876,7 +733,8 @@ def test_SfpStateUpdateTask_task_worker(self, mock_updata_status, mock_post_sfp_ assert mock_os_kill.call_count == 1 assert not sfp_error_event.is_set() - mock_mapping_event.side_effect = [SYSTEM_BECOME_READY, SYSTEM_FAIL] + [SYSTEM_FAIL] * (RETRY_TIMES_FOR_SYSTEM_READY + 1) + mock_mapping_event.side_effect = [SYSTEM_BECOME_READY, SYSTEM_FAIL] + \ + [SYSTEM_FAIL] * (RETRY_TIMES_FOR_SYSTEM_READY + 1) mock_os_kill.reset_mock() sfp_error_event.clear() # Test state machine: STATE_INIT + SYSTEM_BECOME_READY event => STATE_NORMAL + SYSTEM_FAIL event ... => STATE_INIT @@ -886,7 +744,7 @@ def test_SfpStateUpdateTask_task_worker(self, mock_updata_status, mock_post_sfp_ assert sfp_error_event.is_set() task.port_mapping.handle_port_change_event(PortChangeEvent('Ethernet0', 1, 0, PortChangeEvent.PORT_ADD)) - mock_change_event.return_value = (True, {1:SFP_STATUS_INSERTED}, {}) + mock_change_event.return_value = (True, {1: SFP_STATUS_INSERTED}, {}) mock_mapping_event.side_effect = None mock_mapping_event.return_value = NORMAL_EVENT mock_post_sfp_info.return_value = SFP_EEPROM_NOT_READY @@ -894,7 +752,7 @@ def test_SfpStateUpdateTask_task_worker(self, mock_updata_status, mock_post_sfp_ # Test state machine: handle SFP insert event, but EEPROM read failure task.task_worker(stop_event, sfp_error_event, [False]) assert mock_updata_status.call_count == 1 - assert mock_post_sfp_info.call_count == 2 # first call and retry call + assert mock_post_sfp_info.call_count == 2 # first call and retry call assert mock_post_dom_info.call_count == 0 assert mock_post_dom_th.call_count == 0 assert mock_update_media_setting.call_count == 0 @@ -914,7 +772,7 @@ def test_SfpStateUpdateTask_task_worker(self, mock_updata_status, mock_post_sfp_ assert mock_update_media_setting.call_count == 1 stop_event.is_set = MagicMock(side_effect=[False, True]) - mock_change_event.return_value = (True, {1:SFP_STATUS_REMOVED}, {}) + mock_change_event.return_value = (True, {1: SFP_STATUS_REMOVED}, {}) mock_updata_status.reset_mock() # Test state machine: handle SFP remove event task.task_worker(stop_event, sfp_error_event, [False]) @@ -923,7 +781,7 @@ def test_SfpStateUpdateTask_task_worker(self, mock_updata_status, mock_post_sfp_ stop_event.is_set = MagicMock(side_effect=[False, True]) error = int(SFP_STATUS_INSERTED) | SfpBase.SFP_ERROR_BIT_BLOCKING | SfpBase.SFP_ERROR_BIT_POWER_BUDGET_EXCEEDED - mock_change_event.return_value = (True, {1:error}, {}) + mock_change_event.return_value = (True, {1: error}, {}) mock_updata_status.reset_mock() mock_del_dom.reset_mock() # Test state machine: handle SFP error event @@ -1012,12 +870,13 @@ class MockTable: # SFP information is not in DB, and SFP is not present, and SFP is in error status task.on_add_logical_port(port_change_event) assert mock_updata_status.call_count == 1 - mock_updata_status.assert_called_with('Ethernet0', status_tbl, task.sfp_error_dict[1][0], 'Blocking EEPROM from being read|Power budget exceeded') + mock_updata_status.assert_called_with( + 'Ethernet0', status_tbl, task.sfp_error_dict[1][0], 'Blocking EEPROM from being read|Power budget exceeded') def test_sfp_insert_events(self): from xcvrd.xcvrd import _wrapper_soak_sfp_insert_event sfp_insert_events = {} - insert = port_dict = {1:'1', 2:'1', 3:'1', 4:'1', 5:'1'} + insert = port_dict = {1: '1', 2: '1', 3: '1', 4: '1', 5: '1'} start = time.time() while True: _wrapper_soak_sfp_insert_event(sfp_insert_events, insert) @@ -1026,13 +885,12 @@ def test_sfp_insert_events(self): assert not bool(insert) assert insert == port_dict - def test_sfp_remove_events(self): from xcvrd.xcvrd import _wrapper_soak_sfp_insert_event sfp_insert_events = {} - insert = {1:'1', 2:'1', 3:'1', 4:'1', 5:'1'} - removal = {1:'0', 2:'0', 3:'0', 4:'0', 5:'0'} - port_dict = {1:'0', 2:'0', 3:'0', 4:'0', 5:'0'} + insert = {1: '1', 2: '1', 3: '1', 4: '1', 5: '1'} + removal = {1: '0', 2: '0', 3: '0', 4: '0', 5: '0'} + port_dict = {1: '0', 2: '0', 3: '0', 4: '0', 5: '0'} for x in range(5): _wrapper_soak_sfp_insert_event(sfp_insert_events, insert) time.sleep(1) @@ -1054,7 +912,7 @@ def test_wrapper_get_presence(self, mock_sfputil, mock_chassis): mock_chassis.get_sfp = MagicMock(side_effect=NotImplementedError) mock_sfputil.get_presence = MagicMock(return_value=True) - + assert _wrapper_get_presence(1) mock_sfputil.get_presence = MagicMock(return_value=False) @@ -1088,7 +946,7 @@ def test_wrapper_get_transceiver_info(self, mock_sfputil, mock_chassis): mock_chassis.get_sfp = MagicMock(side_effect=NotImplementedError) mock_sfputil.get_transceiver_info_dict = MagicMock(return_value=True) - + assert _wrapper_get_transceiver_info(1) mock_sfputil.get_transceiver_info_dict = MagicMock(return_value=False) @@ -1108,7 +966,7 @@ def test_wrapper_get_transceiver_dom_info(self, mock_sfputil, mock_chassis): mock_chassis.get_sfp = MagicMock(side_effect=NotImplementedError) mock_sfputil.get_transceiver_dom_info_dict = MagicMock(return_value=True) - + assert _wrapper_get_transceiver_dom_info(1) mock_sfputil.get_transceiver_dom_info_dict = MagicMock(return_value=False) @@ -1128,7 +986,7 @@ def test_wrapper_get_transceiver_dom_threshold_info(self, mock_sfputil, mock_cha mock_chassis.get_sfp = MagicMock(side_effect=NotImplementedError) mock_sfputil.get_transceiver_dom_threshold_info_dict = MagicMock(return_value=True) - + assert _wrapper_get_transceiver_dom_threshold_info(1) mock_sfputil.get_transceiver_dom_threshold_info_dict = MagicMock(return_value=False) @@ -1200,7 +1058,7 @@ def test_get_media_val_str(self): @patch('xcvrd.xcvrd.DaemonXcvrd.load_platform_util', MagicMock()) @patch('sonic_py_common.device_info.get_paths_to_platform_and_hwsku_dirs', MagicMock(return_value=('/tmp', None))) - @patch('swsscommon.swsscommon.WarmStart', MagicMock()) + @patch('swsscommon.swsscommon.WarmStart', MagicMock()) @patch('xcvrd.xcvrd.DaemonXcvrd.wait_for_port_config_done', MagicMock()) def test_DaemonXcvrd_init_deinit(self): xcvrd = DaemonXcvrd(SYSLOG_IDENTIFIER) @@ -1210,7 +1068,6 @@ def test_DaemonXcvrd_init_deinit(self): # xcvrd.init/deinit will not raise unexpected exception. In future, probably more check will be added - def wait_until(total_wait_time, interval, call_back, *args, **kwargs): wait_time = 0 while wait_time <= total_wait_time: diff --git a/sonic-xcvrd/tests/test_y_cable_helper.py b/sonic-xcvrd/tests/test_y_cable_helper.py new file mode 100644 index 000000000000..7848f4a41237 --- /dev/null +++ b/sonic-xcvrd/tests/test_y_cable_helper.py @@ -0,0 +1,2010 @@ +from xcvrd.xcvrd_utilities.port_mapping import * +from xcvrd.xcvrd_utilities.sfp_status_helper import * +from xcvrd.xcvrd_utilities.y_cable_helper import * +from xcvrd.xcvrd import * +import copy +import os +import sys +import time + +if sys.version_info >= (3, 3): + from unittest.mock import MagicMock, patch +else: + from mock import MagicMock, patch + +from sonic_py_common import daemon_base +from swsscommon import swsscommon +from sonic_platform_base.sfp_base import SfpBase +from .mock_swsscommon import Table + + +daemon_base.db_connect = MagicMock() +swsscommon.Table = MagicMock() + +sys.modules['sonic_y_cable'] = MagicMock() +sys.modules['sonic_y_cable.y_cable'] = MagicMock() + +os.environ["Y_CABLE_HELPER_UNIT_TESTING"] = "1" + + +class helper_logger: + mock_arg = MagicMock() + + def log_error(self, mock_arg): + return True + + def log_warning(self, mock_arg): + return True + + def log_debug(self, mock_arg): + return True + + +class TestYCableScript(object): + def test_xcvrd_helper_class_run(self): + Y_cable_task = YCableTableUpdateTask(None) + + def test_y_cable_helper_format_mapping_identifier1(self): + rc = format_mapping_identifier("ABC ") + assert(rc == "abc") + + def test_y_cable_helper_format_mapping_identifier_no_instance(self): + rc = format_mapping_identifier(None) + assert(rc == None) + + def test_y_cable_wrapper_get_transceiver_info(self): + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'Microsoft', + 'model': 'model1'} + + transceiver_dict = y_cable_wrapper_get_transceiver_info(1) + vendor = transceiver_dict.get('manufacturer') + model = transceiver_dict.get('model') + + assert(vendor == "Microsoft") + assert(model == "model1") + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_transceiver_info_with_platform_chassis(self, mock_chassis): + + mock_object = MagicMock() + mock_object.get_transceiver_info.return_value = {'type': '1000_BASE_SX_SFP', + 'hardware_rev': '5', + 'serial': 'PEP3L5D', + 'manufacturer': 'FINISAR', + 'model': 'ABC', + 'connector': 'LC', + 'encoding': '8B10B', + 'ext_identifier': 'SFP', + 'ext_rateselect_compliance': 'DEF', + 'cable_length': '850', + 'nominal_bit_rate': '100', + 'specification_compliance': 'GHI', + 'vendor_date': '2021-01-01', + 'vendor_oui': '00:90:65'} + + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + received_xcvr_info = y_cable_wrapper_get_transceiver_info(1) + + type = received_xcvr_info.get('type') + model = received_xcvr_info.get('model') + vendor_date = received_xcvr_info.get('vendor_date') + + assert(type == "1000_BASE_SX_SFP") + assert(model == "ABC") + assert(vendor_date == "2021-01-01") + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_transceiver_info_with_platform_chassis_not_implemented(self, mock_chassis): + + mock_object = MagicMock() + mock_object.get_transceiver_info.side_effect = NotImplementedError + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'microsoft', + 'model': 'simulated'} + + transceiver_dict = y_cable_wrapper_get_transceiver_info(1) + vendor = transceiver_dict.get('manufacturer') + model = transceiver_dict.get('model') + + assert(vendor == "microsoft") + assert(model == "simulated") + + def test_y_cable_wrapper_get_presence(self): + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_presence.return_value = True + + presence = y_cable_wrapper_get_presence(1) + + assert(presence == True) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_presence_with_platform_chassis(self, mock_chassis): + + mock_object = MagicMock() + mock_object.get_presence = MagicMock(return_value=True) + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + presence = y_cable_wrapper_get_presence(1) + + assert(presence == True) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_presence_with_platform_chassis_raise_exception(self, mock_chassis): + + mock_object = MagicMock(spec=SfpBase) + mock_object.get_presence = MagicMock(side_effect=NotImplementedError) + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_presence.return_value = True + + assert(y_cable_wrapper_get_presence(1) == True) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.get_muxcable_info', MagicMock(return_value={'tor_active': 'self', + 'mux_direction': 'self', + 'manual_switch_count': '7', + 'auto_switch_count': '71', + 'link_status_self': 'up', + 'link_status_peer': 'up', + 'link_status_nic': 'up', + 'nic_lane1_active': 'True', + 'nic_lane2_active': 'True', + 'nic_lane3_active': 'True', + 'nic_lane4_active': 'True', + 'self_eye_height_lane1': '500', + 'self_eye_height_lane2': '510', + 'peer_eye_height_lane1': '520', + 'peer_eye_height_lane2': '530', + 'nic_eye_height_lane1': '742', + 'nic_eye_height_lane2': '750', + 'internal_temperature': '28', + 'internal_voltage': '3.3', + 'nic_temperature': '20', + 'nic_voltage': '2.7', + 'version_nic_active': '1.6MS', + 'version_nic_inactive': '1.7MS', + 'version_nic_next': '1.7MS', + 'version_self_active': '1.6MS', + 'version_self_inactive': '1.7MS', + 'version_self_next': '1.7MS', + 'version_peer_active': '1.6MS', + 'version_peer_inactive': '1.7MS', + 'version_peer_next': '1.7MS'})) + def test_post_port_mux_info_to_db(self): + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + mux_tbl = Table("STATE_DB", y_cable_helper.MUX_CABLE_INFO_TABLE) + rc = post_port_mux_info_to_db(logical_port_name, port_mapping, mux_tbl) + assert(rc != -1) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.get_muxcable_static_info', MagicMock(return_value={'read_side': 'self', + 'nic_lane1_precursor1': '1', + 'nic_lane1_precursor2': '-7', + 'nic_lane1_maincursor': '-1', + 'nic_lane1_postcursor1': '11', + 'nic_lane1_postcursor2': '11', + 'nic_lane2_precursor1': '12', + 'nic_lane2_precursor2': '7', + 'nic_lane2_maincursor': '7', + 'nic_lane2_postcursor1': '7', + 'nic_lane2_postcursor2': '7', + 'tor_self_lane1_precursor1': '17', + 'tor_self_lane1_precursor2': '17', + 'tor_self_lane1_maincursor': '17', + 'tor_self_lane1_postcursor1': '17', + 'tor_self_lane1_postcursor2': '17', + 'tor_self_lane2_precursor1': '7', + 'tor_self_lane2_precursor2': '7', + 'tor_self_lane2_maincursor': '7', + 'tor_self_lane2_postcursor1': '7', + 'tor_self_lane2_postcursor2': '7', + 'tor_peer_lane1_precursor1': '7', + 'tor_peer_lane1_precursor2': '7', + 'tor_peer_lane1_maincursor': '17', + 'tor_peer_lane1_postcursor1': '7', + 'tor_peer_lane1_postcursor2': '17', + 'tor_peer_lane2_precursor1': '7', + 'tor_peer_lane2_precursor2': '7', + 'tor_peer_lane2_maincursor': '17', + 'tor_peer_lane2_postcursor1': '7', + 'tor_peer_lane2_postcursor2': '17'})) + def test_post_port_mux_static_info_to_db(self): + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + mux_tbl = Table("STATE_DB", y_cable_helper.MUX_CABLE_STATIC_INFO_TABLE) + rc = post_port_mux_static_info_to_db(logical_port_name, port_mapping, mux_tbl) + assert(rc != -1) + + def test_y_cable_helper_format_mapping_identifier1(self): + rc = format_mapping_identifier("ABC ") + assert(rc == "abc") + + def test_y_cable_helper_format_mapping_identifier_no_instance(self): + rc = format_mapping_identifier(None) + assert(rc == None) + + def test_y_cable_wrapper_get_transceiver_info(self): + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'Microsoft', + 'model': 'model1'} + + transceiver_dict = y_cable_wrapper_get_transceiver_info(1) + vendor = transceiver_dict.get('manufacturer') + model = transceiver_dict.get('model') + + assert(vendor == "Microsoft") + assert(model == "model1") + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_transceiver_info_with_platform_chassis(self, mock_chassis): + + mock_object = MagicMock() + mock_object.get_transceiver_info.return_value = {'type': '1000_BASE_SX_SFP', + 'hardware_rev': '5', + 'serial': 'PEP3L5D', + 'manufacturer': 'FINISAR', + 'model': 'ABC', + 'connector': 'LC', + 'encoding': '8B10B', + 'ext_identifier': 'SFP', + 'ext_rateselect_compliance': 'DEF', + 'cable_length': '850', + 'nominal_bit_rate': '100', + 'specification_compliance': 'GHI', + 'vendor_date': '2021-01-01', + 'vendor_oui': '00:90:65'} + + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + received_xcvr_info = y_cable_wrapper_get_transceiver_info(1) + + type = received_xcvr_info.get('type') + model = received_xcvr_info.get('model') + vendor_date = received_xcvr_info.get('vendor_date') + + assert(type == "1000_BASE_SX_SFP") + assert(model == "ABC") + assert(vendor_date == "2021-01-01") + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_transceiver_info_with_platform_chassis_not_implemented(self, mock_chassis): + + mock_object = MagicMock() + mock_object.get_transceiver_info.side_effect = NotImplementedError + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'microsoft', + 'model': 'simulated'} + + transceiver_dict = y_cable_wrapper_get_transceiver_info(1) + vendor = transceiver_dict.get('manufacturer') + model = transceiver_dict.get('model') + + assert(vendor == "microsoft") + assert(model == "simulated") + + def test_y_cable_wrapper_get_presence(self): + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_presence.return_value = True + + presence = y_cable_wrapper_get_presence(1) + + assert(presence == True) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + def test_y_cable_wrapper_get_presence_with_platform_chassis(self, mock_chassis): + + mock_object = MagicMock() + mock_object.get_presence = MagicMock(return_value=True) + mock_chassis.get_sfp = MagicMock(return_value=mock_object) + presence = y_cable_wrapper_get_presence(1) + + assert(presence == True) + + def test_y_cable_toggle_mux_torA_update_status_true(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_a.return_value = True + patched_util.get.return_value = mock_toggle_object + + rc = y_cable_toggle_mux_torA(1) + + assert(rc == 1) + + def test_y_cable_toggle_mux_torA_no_port_instance(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as port_instance: + + port_instance.get.return_value = None + rc = y_cable_toggle_mux_torA(1) + + assert(rc == -1) + + def test_y_cable_toggle_mux_torA_update_status_exception(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as port_instance: + + port_instance.get.return_value = "simulated_port" + port_instance.toggle_mux_to_tor_a.return_value = Exception(NotImplementedError) + + rc = y_cable_toggle_mux_torA(1) + + assert(rc == -1) + + def test_y_cable_toggle_mux_torA_update_status_true(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_a.return_value = True + patched_util.get.return_value = mock_toggle_object + + rc = y_cable_toggle_mux_torA(1) + + assert(rc == 1) + + def test_y_cable_toggle_mux_torB_no_port_instance(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as port_instance: + + port_instance.get.return_value = None + rc = y_cable_toggle_mux_torB(1) + + assert(rc == -1) + + def test_y_cable_toggle_mux_torB_update_status_exception(self): + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as port_instance: + + port_instance.get.return_value = "simulated_port" + port_instance.toggle_mux_to_tor_a.return_value = Exception(NotImplementedError) + + rc = y_cable_toggle_mux_torB(1) + + assert(rc == -1) + + def test_y_cable_toggle_mux_torB_update_status_true(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_b.return_value = True + patched_util.get.return_value = mock_toggle_object + + rc = y_cable_toggle_mux_torB(1) + + assert(rc == 2) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_tor_active_side_1_active(self): + read_side = 1 + state = "active" + logical_port_name = "Ethernet0" + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_a.return_value = True + patched_util.get.return_value = mock_toggle_object + port_mapping = PortMapping() + + rc = update_tor_active_side(read_side, state, logical_port_name, port_mapping) + + assert(rc == 1) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_tor_active_side_2_active(self): + read_side = 2 + state = "active" + logical_port_name = "Ethernet0" + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_b.return_value = True + patched_util.get.return_value = mock_toggle_object + port_mapping = PortMapping() + + rc = update_tor_active_side(read_side, state, logical_port_name, port_mapping) + + assert(rc == 2) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_tor_active_side_1_standby(self): + read_side = 1 + state = "standby" + logical_port_name = "Ethernet0" + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_b.return_value = True + patched_util.get.return_value = mock_toggle_object + port_mapping = PortMapping() + + rc = update_tor_active_side(read_side, state, logical_port_name, port_mapping) + + assert(rc == 2) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_tor_active_side_2_standby(self): + read_side = 2 + state = "standby" + logical_port_name = "Ethernet0" + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_a.return_value = True + patched_util.get.return_value = mock_toggle_object + port_mapping = PortMapping() + + rc = update_tor_active_side(read_side, state, logical_port_name, port_mapping) + + assert(rc == 1) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + def test_update_tor_active_side_no_cable_presence(self): + read_side = 1 + state = "active" + logical_port_name = "Ethernet0" + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_a.return_value = True + patched_util.get.return_value = mock_toggle_object + port_mapping = PortMapping() + + rc = update_tor_active_side(read_side, state, logical_port_name, port_mapping) + + assert(rc == -1) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0, 1, 2])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + def test_update_tor_active_side_multiple_mappings(self): + read_side = 1 + state = "active" + logical_port_name = "Ethernet0" + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + mock_toggle_object = MagicMock() + mock_toggle_object.toggle_mux_to_tor_a.return_value = True + patched_util.get.return_value = mock_toggle_object + port_mapping = PortMapping() + + rc = update_tor_active_side(read_side, state, logical_port_name, port_mapping) + + assert(rc == -1) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_get_ycable_physical_port_from_logical_port(self): + port_mapping = PortMapping() + instance = get_ycable_physical_port_from_logical_port("Ethernet0", port_mapping) + + assert(instance == 0) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + def test_get_ycable_physical_port_from_logical_port_physical_port_not_present(self): + port_mapping = PortMapping() + instance = get_ycable_physical_port_from_logical_port("Ethernet0", port_mapping) + + assert(instance == -1) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value={})) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + def test_get_ycable_physical_port_from_logical_port_physical_port_list_empty(self): + + port_mapping = PortMapping() + instance = get_ycable_physical_port_from_logical_port("Ethernet0", port_mapping) + + assert(instance == -1) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_get_ycable_port_instance_from_logical_port(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + patched_util.get.return_value = 0 + port_mapping = PortMapping() + instance = get_ycable_port_instance_from_logical_port("Ethernet0", port_mapping) + + assert(instance == 0) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + def test_get_ycable_port_instance_from_logical_port_no_presence(self): + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + patched_util.get.return_value = 0 + port_mapping = PortMapping() + instance = get_ycable_port_instance_from_logical_port("Ethernet0", port_mapping) + + assert(instance == PORT_INSTANCE_ERROR) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_get_ycable_port_instance_from_logical_port_no_port_instance(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + def mock_get(): + pass + + patched_util.get.return_value = mock_get() + port_mapping = PortMapping() + instance = get_ycable_port_instance_from_logical_port("E", port_mapping) + + assert(instance == PORT_INSTANCE_ERROR) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0, 1, 2])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_get_ycable_port_instance_from_logical_port_multiple_mapping(self): + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + patched_util.get.return_value = 0 + port_mapping = PortMapping() + instance = get_ycable_port_instance_from_logical_port("Ethernet0", port_mapping) + + assert(instance == -1) + + def test_update_table_mux_status_for_response_tbl(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + status = "standby" + + test_table = swsscommon.Table(appl_db[asic_index], "XCVRD_TEST_TABLE") + update_table_mux_status_for_response_tbl(test_table, status, logical_port_name) + + rc = test_table.get(logical_port_name) + + # Since the table class is mocked, the most we can test for is that get doesn't return None + assert(type(rc) != None) + + def test_set_result_and_delete_port(self): + + result = "result" + actual_result = "pass" + appl_db = "TEST_DB" + port = 0 + + command_table = swsscommon.Table(appl_db[0], "XCVRD_COMMAND_TABLE") + response_table = swsscommon.Table(appl_db[1], "XCVRD_RESPONSE_TABLE") + + rc = set_result_and_delete_port(result, actual_result, command_table, response_table, port) + assert(rc == None) + + def test_delete_port_from_y_cable_table(self): + logical_port_name = "Ethernet0" + appl_db = "TEST_DB" + y_cable_tbl = swsscommon.Table(appl_db[0], "XCVRD_Y_CBL_TABLE") + + rc = delete_port_from_y_cable_table(logical_port_name, y_cable_tbl) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_appdb_port_mux_cable_response_table_port_instance_none(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 1 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + def mock_get(): + pass + + patched_util.get.return_value = mock_get() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_appdb_port_mux_cable_response_table_read_side_none(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + def mock_read_side(): + pass + + read_side = mock_read_side() + + patched_util.get.return_value = 0 + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_appdb_port_mux_cable_response_table_active_side_none(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 1 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + # Defining function without self argument creates an exception, + # which is what we want for this test. + def get_mux_direction(): + pass + + patched_util.get.return_value = PortInstanceHelper() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_appdb_port_mux_cable_response_table_active_side_is_read_side(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 1 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_mux_direction(self): + return 1 + + patched_util.get.return_value = PortInstanceHelper() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_appdb_port_mux_cable_response_table_active_side_not_read_side(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 2 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_mux_direction(self): + return 1 + + patched_util.get.return_value = PortInstanceHelper() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_update_appdb_port_mux_cable_response_table_active_side_status_unknown(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 1 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_mux_direction(self): + return 4 + + patched_util.get.return_value = PortInstanceHelper() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + def test_update_appdb_port_mux_cable_response_table_no_presence_status_unknown(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 1 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_mux_direction(self): + return 4 + + patched_util.get.return_value = PortInstanceHelper() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0, 1, 2])) + def test_update_appdb_port_mux_cable_response_table_invalid_ycable_mapping(self): + asic_index = 0 + appl_db = "TEST_DB" + logical_port_name = "Ethernet0" + read_side = 1 + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_mux_direction(self): + return 4 + + patched_util.get.return_value = PortInstanceHelper() + port_mapping = PortMapping() + + rc = update_appdb_port_mux_cable_response_table( + logical_port_name, port_mapping, asic_index, appl_db, read_side) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0, 1, 2])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + def test_read_y_cable_and_update_statedb_port_tbl_invalid_ycable_mapping(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_port_instance_none(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + def mock_get(): + pass + + patched_util.get.return_value = mock_get() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_get_presence_false(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + def mock_get(): + pass + + patched_util.get.return_value = mock_get() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_port_instance_get_read_side_exception(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + # Defining function without self argument creates an exception, + # which is what we want for this test. + def get_read_side(): + pass + + patched_util.get.return_value = PortInstanceHelper() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_port_instance_get_mux_dir_exception(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_read_side(self): + return 1 + + # Defining function without self argument creates an exception, + # which is what we want for this test. + def get_mux_direction(): + pass + + patched_util.get.return_value = PortInstanceHelper() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_port_instance_status_active(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_read_side(self): + return 1 + + def get_mux_direction(self): + return 1 + + patched_util.get.return_value = PortInstanceHelper() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_port_instance_status_standby(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_read_side(self): + return 1 + + def get_mux_direction(self): + return 2 + + patched_util.get.return_value = PortInstanceHelper() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_read_y_cable_and_update_statedb_port_tbl_port_instance_status_unknown(self): + + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + statedb_port_tbl = {} + asic_index = 0 + appl_db = "TEST_DB" + + statedb_port_tbl[asic_index] = swsscommon.Table( + appl_db[asic_index], "STATEDB_PORT_TABLE") + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + + def get_read_side(self): + return 1 + + def get_mux_direction(self): + return 0 + + patched_util.get.return_value = PortInstanceHelper() + rc = read_y_cable_and_update_statedb_port_tbl(logical_port_name, port_mapping, statedb_port_tbl[asic_index]) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_create_tables_and_insert_mux_unknown_entries(self): + + state_db = {} + asic_index = 0 + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + + rc = create_tables_and_insert_mux_unknown_entries( + state_db, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_status_false(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = False + fvs = [('state', "manual")] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_state_absent(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('badstate', "auto")] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_bad_state_value(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "badvalue")] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=False)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_no_presence(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_no_port_info(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = None + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0, 1, 2])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_multiple_port_instances(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'Microsoft', 'model': 'simulated'} + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_no_vendor_port_info(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = { + 'bad_manufacturer': 'Microsoft', 'model': 'simulated'} + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_no_model_port_info(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = { + 'manufacturer': 'Microsoft', 'bad_model': 'simulated'} + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_invalid_vendor_port_info(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = { + 'manufacturer': 'not_Microsoft', 'model': 'simulated'} + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_invalid_model_port_info(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'Microsoft', 'model': 'bad_model1'} + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_update_mux_table_entry_module_dir_none(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = { + 'manufacturer': 'not_Microsoft', 'model': 'simulated'} + + with patch('sonic_y_cable.y_cable_vendor_mapping.mapping') as mock_mapping: + mock_mapping.get.return_value = None + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('sonic_y_cable.y_cable_vendor_mapping.mapping.get', MagicMock(return_value={"Microsoft": {"module": "test_module"}})) + def test_check_identifier_presence_and_update_mux_table_entry_module_none(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = True + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'not_Microsoft', 'model': 'model1'} + + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('sonic_y_cable.y_cable_vendor_mapping.mapping.get', MagicMock(return_value={"simulated": "microsoft.y_cable_simulated"})) + def test_check_identifier_presence_and_update_mux_table_entry_module_microsoft(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = [True] + + y_cable_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], swsscommon.STATE_HW_MUX_CABLE_TABLE_NAME) + static_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_STATIC_INFO_TABLE) + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'microsoft', 'model': 'simulated'} + + sys.modules['builtins.getattr'] = MagicMock() + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_wrapper_get_presence', MagicMock(return_value=True)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('sonic_y_cable.y_cable_vendor_mapping.mapping.get', MagicMock(return_value={"simulated": "microsoft.y_cable_simulated"})) + def test_check_identifier_presence_and_update_mux_table_entry_module_microsoft_y_cable_presence_false(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = [False] + + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_transceiver_info_dict.return_value = {'manufacturer': 'microsoft', 'model': 'simulated'} + + sys.modules['builtins.getattr'] = MagicMock() + rc = check_identifier_presence_and_update_mux_table_entry( + state_db, port_tbl, y_cable_tbl, static_tbl, mux_tbl, asic_index, logical_port_name, port_mapping, y_cable_presence) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_check_identifier_presence_and_delete_mux_table_entry(self): + + asic_index = 0 + logical_port_name = "Ethernet0" + status = True + fvs = [('state', "auto"), ('read_side', 1)] + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + y_cable_tbl = {} + static_tbl = {} + mux_tbl = {} + port_tbl = {} + y_cable_presence = [True] + delete_change_event = [True] + + port_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], "PORT_INFO_TABLE") + port_tbl[asic_index].get.return_value = (status, fvs) + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as port_instance: + rc = check_identifier_presence_and_delete_mux_table_entry( + state_db, port_tbl, asic_index, logical_port_name, y_cable_presence, port_mapping, delete_change_event) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_chassis') + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') + @patch('swsscommon.swsscommon.Table') + def test_init_ports_status_for_y_cable(self, platform_chassis, platform_sfp, mock_swsscommon_table): + + platform_sfp = MagicMock() + platform_chassis = MagicMock() + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + port_mapping.logical_port_list = (['Ethernet0', 'Ethernet4']) + + y_cable_presence = [True] + + mock_table = MagicMock() + mock_table.getKeys = MagicMock(return_value=['Ethernet0', 'Ethernet4']) + mock_swsscommon_table.return_value = mock_table + + rc = init_ports_status_for_y_cable(platform_sfp, platform_chassis, + y_cable_presence, port_mapping, stop_event=threading.Event()) + + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('swsscommon.swsscommon.Table') + def test_change_ports_status_for_y_cable_change_event(self, mock_swsscommon_table): + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + + y_cable_presence = [True] + logical_port_dict = {'Ethernet0': '1'} + + mock_table = MagicMock() + mock_table.getKeys = MagicMock(return_value=['Ethernet0', 'Ethernet4']) + mock_table.get = MagicMock(side_effect=[(True, (('index', 1), )), (True, (('index', 2), ))]) + mock_swsscommon_table.return_value = mock_table + + change_ports_status_for_y_cable_change_event( + logical_port_dict, port_mapping, y_cable_presence, stop_event=threading.Event()) + + mock_swsscommon_table.assert_called() + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('swsscommon.swsscommon.Table') + def test_change_ports_status_for_y_cable_change_event_sfp_removed(self, mock_swsscommon_table): + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + port_mapping.logical_port_list = (['Ethernet0', 'Ethernet4']) + + y_cable_presence = [True] + logical_port_dict = {'Ethernet0': '1'} + + mock_table = MagicMock() + mock_table.getKeys = MagicMock(return_value=['Ethernet0', 'Ethernet4']) + mock_table.get = MagicMock(side_effect=[(True, (('index', 1), )), (True, (('index', 2), ))]) + mock_swsscommon_table.return_value = mock_table + + change_ports_status_for_y_cable_change_event( + logical_port_dict, port_mapping, y_cable_presence, stop_event=threading.Event()) + + mock_swsscommon_table.assert_called() + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('swsscommon.swsscommon.Table') + def test_change_ports_status_for_y_cable_change_event_sfp_unknown(self, mock_swsscommon_table): + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + + y_cable_presence = [True] + logical_port_dict = {'Ethernet0': '2'} + + mock_table = MagicMock() + mock_table.getKeys = MagicMock(return_value=['Ethernet0', 'Ethernet4']) + mock_table.get = MagicMock(side_effect=[(True, (('index', 1), )), (True, (('index', 2), ))]) + mock_swsscommon_table.return_value = mock_table + + change_ports_status_for_y_cable_change_event( + logical_port_dict, port_mapping, y_cable_presence, stop_event=threading.Event()) + + mock_swsscommon_table.assert_called() + + @patch('swsscommon.swsscommon.Table') + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.logical_port_name_to_physical_port_list', MagicMock(return_value=[0])) + def test_delete_ports_status_for_y_cable(self, mock_swsscommon_table): + + mock_table = MagicMock() + mock_table.getKeys = MagicMock(return_value=['Ethernet0', 'Ethernet4']) + mock_table.get = MagicMock(side_effect=[(True, (('index', 1), )), (True, (('index', 2), ))]) + mock_swsscommon_table.return_value = mock_table + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + port_mapping.logical_port_list = (['Ethernet0', 'Ethernet4']) + + rc = delete_ports_status_for_y_cable(port_mapping) + + mock_swsscommon_table.assert_called() + + def test_check_identifier_presence_and_update_mux_info_entry(self): + asic_index = 0 + logical_port_name = "Ethernet0" + port_mapping = PortMapping() + state_db = {} + test_db = "TEST_DB" + mux_tbl = {} + + mux_tbl[asic_index] = swsscommon.Table( + test_db[asic_index], MUX_CABLE_INFO_TABLE) + + rc = check_identifier_presence_and_update_mux_info_entry( + state_db, mux_tbl, asic_index, logical_port_name, port_mapping) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') + def test_get_firmware_dict(self, port_instance): + + port_instance = MagicMock() + port_instance.FIRMWARE_DOWNLOAD_STATUS_INPROGRESS = 1 + port_instance.download_firmware_status = 1 + + physical_port = 1 + target = "simulated_target" + side = "a" + mux_info_dict = {} + logical_port_name = "Ethernet0" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + + patched_util.get_asic_id_for_logical_port.return_value = 0 + + status = True + fvs = [('state', "auto"), ('read_side', 1)] + Table = MagicMock() + Table.get.return_value = (status, fvs) + + rc = get_firmware_dict(physical_port, port_instance, target, side, mux_info_dict, logical_port_name) + + assert(mux_info_dict['version_a_active'] == None) + assert(mux_info_dict['version_a_inactive'] == None) + assert(mux_info_dict['version_a_next'] == None) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') + def test_get_firmware_dict_download_status_failed_exception(self, port_instance): + + port_instance = MagicMock() + port_instance.FIRMWARE_DOWNLOAD_STATUS_FAILED = -1 + port_instance.download_firmware_status = -1 + port_instance.get_firmware_version = MagicMock(side_effect=NotImplementedError) + + physical_port = 1 + target = "simulated_target" + side = "a" + mux_info_dict = {} + logical_port_name = "Ethernet0" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + + patched_util.get_asic_id_for_logical_port.return_value = 0 + + status = True + fvs = [('state', "auto"), ('read_side', 1)] + Table = MagicMock() + Table.get.return_value = (status, fvs) + + rc = get_firmware_dict(physical_port, port_instance, target, side, mux_info_dict, logical_port_name) + + assert(mux_info_dict['version_a_active'] == "N/A") + assert(mux_info_dict['version_a_inactive'] == "N/A") + assert(mux_info_dict['version_a_next'] == "N/A") + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') + def test_get_firmware_dict_download_status_failed(self, port_instance): + + port_instance = MagicMock() + port_instance.FIRMWARE_DOWNLOAD_STATUS_FAILED = -1 + port_instance.download_firmware_status = -1 + port_instance.get_firmware_version = MagicMock( + return_value={"version_active": "2021", "version_inactive": "2020", "version_next": "2022"}) + + physical_port = 1 + target = "simulated_target" + side = "a" + mux_info_dict = {} + logical_port_name = "Ethernet0" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + + patched_util.get_asic_id_for_logical_port.return_value = 0 + + status = True + fvs = [('state', "auto"), ('read_side', 1)] + Table = MagicMock() + Table.get.return_value = (status, fvs) + + rc = get_firmware_dict(physical_port, port_instance, target, side, mux_info_dict, logical_port_name) + + assert(mux_info_dict['version_a_active'] == "2021") + assert(mux_info_dict['version_a_inactive'] == "2020") + assert(mux_info_dict['version_a_next'] == "2022") + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.get_asic_id_for_logical_port', MagicMock(return_value=0)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_get_muxcable_info(self): + physical_port = 20 + port_mapping = PortMapping() + logical_port_name = "Ethernet20" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + self.TARGET_NIC = 1 + self.TARGET_TOR_A = 1 + self.TARGET_TOR_B = 1 + self.FIRMWARE_DOWNLOAD_STATUS_INPROGRESS = 1 + self.download_firmware_status = 1 + self.SWITCH_COUNT_MANUAL = "manual" + self.SWITCH_COUNT_AUTO = "auto" + + def get_active_linked_tor_side(self): + return 1 + + def get_mux_direction(self): + return 1 + + def get_switch_count_total(self, switch_count): + return 1 + + def get_eye_heights(self, tgt_tor): + return 500 + + def is_link_active(self, tgt_nic): + return True + + def get_local_temperature(self): + return 22.75 + + def get_local_voltage(self): + return 0.5 + + def get_nic_voltage(self): + return 2.7 + + def get_nic_temperature(self): + return 20 + + patched_util.get.return_value = PortInstanceHelper() + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_asic_id_for_logical_port.return_value = 0 + + rc = get_muxcable_info(physical_port, logical_port_name, port_mapping) + + assert(rc['tor_active'] == 'active') + assert(rc['mux_direction'] == 'self') + assert(rc['internal_voltage'] == 0.5) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.get_asic_id_for_logical_port', MagicMock(return_value=0)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_get_muxcable_info_exceptions(self): + physical_port = 20 + port_mapping = PortMapping() + logical_port_name = "Ethernet20" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + self.TARGET_NIC = 1 + self.TARGET_TOR_A = 1 + self.TARGET_TOR_B = 1 + self.FIRMWARE_DOWNLOAD_STATUS_INPROGRESS = 1 + self.download_firmware_status = 1 + self.SWITCH_COUNT_MANUAL = "manual" + self.SWITCH_COUNT_AUTO = "auto" + + def get_active_linked_tor_side(): + return 1 + + def get_mux_direction(): + return 1 + + def get_switch_count_total(self, switch_count): + return 1 + + def get_eye_heights(tgt_tor): + return 500 + + def is_link_active(self, tgt_nic): + return True + + def get_local_temperature(): + return 22.75 + + def get_local_voltage(): + return 0.5 + + def get_nic_voltage(): + return 2.7 + + def get_nic_temperature(): + return 20 + + patched_util.get.return_value = PortInstanceHelper() + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_platform_sfputil') as patched_util: + patched_util.get_asic_id_for_logical_port.return_value = 0 + + rc = get_muxcable_info(physical_port, logical_port_name, port_mapping) + + assert(rc['tor_active'] == 'unknown') + assert(rc['mux_direction'] == 'unknown') + assert(rc['self_eye_height_lane1'] == 'N/A') + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.get_asic_id_for_logical_port', MagicMock(return_value=0)) + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_locks', MagicMock(return_value=[0])) + def test_get_muxcable_static_info(self): + physical_port = 0 + port_mapping = PortMapping() + logical_port_name = "Ethernet0" + + with patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') as patched_util: + class PortInstanceHelper(): + def __init__(self): + self.EEPROM_ERROR = -1 + self.TARGET_NIC = 0 + self.TARGET_TOR_A = 1 + self.TARGET_TOR_B = 2 + self.FIRMWARE_DOWNLOAD_STATUS_INPROGRESS = 1 + self.download_firmware_status = 1 + self.SWITCH_COUNT_MANUAL = "manual" + self.SWITCH_COUNT_AUTO = "auto" + + def get_target_cursor_values(self, i, tgt): + if (tgt == self.TARGET_NIC): + return ([1, 7, 7, 1, 0]) + elif (tgt == self.TARGET_TOR_A): + return ([17, 17, 17, 17, 17]) + elif (tgt == self.TARGET_TOR_B): + return ([-17, -17, -17, -17, -17]) + + patched_util.get.return_value = PortInstanceHelper() + + rc = get_muxcable_static_info(physical_port, logical_port_name, port_mapping) + + assert (rc['read_side'] == 'tor1') + assert (rc['nic_lane1_precursor1'] == 1) + assert (rc['nic_lane1_precursor2'] == 7) + assert (rc['nic_lane1_maincursor'] == 7) + assert (rc['nic_lane1_postcursor1'] == 1) + assert (rc['nic_lane1_postcursor2'] == 0) + + assert (rc['nic_lane2_precursor1'] == 1) + assert (rc['nic_lane2_precursor2'] == 7) + assert (rc['nic_lane2_maincursor'] == 7) + assert (rc['nic_lane2_postcursor1'] == 1) + assert (rc['nic_lane2_postcursor2'] == 0) + + assert (rc['tor_self_lane1_precursor1'] == 17) + assert (rc['tor_self_lane1_precursor2'] == 17) + assert (rc['tor_self_lane1_maincursor'] == 17) + assert (rc['tor_self_lane1_postcursor1'] == 17) + assert (rc['tor_self_lane1_postcursor2'] == 17) + + assert (rc['tor_self_lane2_precursor1'] == 17) + assert (rc['tor_self_lane2_precursor2'] == 17) + assert (rc['tor_self_lane2_maincursor'] == 17) + assert (rc['tor_self_lane2_postcursor1'] == 17) + assert (rc['tor_self_lane2_postcursor2'] == 17) + + assert (rc['tor_peer_lane1_precursor1'] == -17) + assert (rc['tor_peer_lane1_precursor2'] == -17) + assert (rc['tor_peer_lane1_maincursor'] == -17) + assert (rc['tor_peer_lane1_postcursor1'] == -17) + assert (rc['tor_peer_lane1_postcursor2'] == -17) + + assert (rc['tor_peer_lane2_precursor1'] == -17) + assert (rc['tor_peer_lane2_precursor2'] == -17) + assert (rc['tor_peer_lane2_maincursor'] == -17) + assert (rc['tor_peer_lane2_postcursor1'] == -17) + assert (rc['tor_peer_lane2_postcursor2'] == -17) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.get_asic_id_for_logical_port', MagicMock(return_value=0)) + def test_post_mux_static_info_to_db(self): + is_warm_start = True + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + port_mapping.logical_port_list = (['Ethernet0', 'Ethernet4']) + + stop_event = threading.Event() + stop_event.is_set = MagicMock(return_value=False) + + rc = post_mux_static_info_to_db(is_warm_start, port_mapping, stop_event) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.port_mapping.PortMapping.get_asic_id_for_logical_port', MagicMock(return_value=0)) + def test_post_mux_info_to_db(self): + is_warm_start = True + + mock_logical_port_name = [""] + + def mock_get_asic_id(mock_logical_port_name): + return 0 + + port_mapping = PortMapping() + port_mapping.get_asic_id_for_logical_port = mock_get_asic_id + port_mapping.logical_port_list = (['Ethernet0', 'Ethernet4']) + + stop_event = threading.Event() + stop_event.is_set = MagicMock(return_value=False) + + rc = post_mux_info_to_db(is_warm_start, port_mapping, stop_event) + assert(rc == None) + + @patch('xcvrd.xcvrd_utilities.y_cable_helper.y_cable_port_instances') + @patch('swsscommon.swsscommon.Table') + def test_task_download_firmware_worker(self, port_instance, mock_swsscommon_table): + port = "Ethernet0" + physical_port = 0 + file_full_path = "/path/to/file" + + def mock_download_fw(filepath): + return 0 + + port_instance.download_firmware = mock_download_fw + + mock_table = MagicMock() + mock_swsscommon_table.return_value = mock_table + + xcvrd_down_fw_rsp_tbl = mock_swsscommon_table + xcvrd_down_fw_cmd_sts_tbl = mock_swsscommon_table + + rc = {} + + task_download_firmware_worker(port, physical_port, port_instance, file_full_path, + xcvrd_down_fw_rsp_tbl, xcvrd_down_fw_cmd_sts_tbl, rc) + + assert(rc[0] == 0) + + def test_set_show_firmware_fields(self): + + mux_info_dict = {} + xcvrd_show_fw_res_tbl = Table("STATE_DB", "XCVRD_SHOW_FW_RES") + mux_info_dict['version_self_active'] = '0.8' + mux_info_dict['version_self_inactive'] = '0.7' + mux_info_dict['version_self_next'] = '0.7' + mux_info_dict['version_peer_active'] = '0.8' + mux_info_dict['version_peer_inactive'] = '0.7' + mux_info_dict['version_peer_next'] = '0.7' + mux_info_dict['version_nic_active'] = '0.8' + mux_info_dict['version_nic_inactive'] = '0.7' + mux_info_dict['version_nic_next'] = '0.7' + rc = set_show_firmware_fields("Ethernet0", mux_info_dict, xcvrd_show_fw_res_tbl) + + assert(rc == 0) diff --git a/sonic-xcvrd/xcvrd/xcvrd_utilities/y_cable_helper.py b/sonic-xcvrd/xcvrd/xcvrd_utilities/y_cable_helper.py index a079c2468d7a..6534910ae18c 100644 --- a/sonic-xcvrd/xcvrd/xcvrd_utilities/y_cable_helper.py +++ b/sonic-xcvrd/xcvrd/xcvrd_utilities/y_cable_helper.py @@ -1426,7 +1426,7 @@ def post_mux_static_info_to_db(is_warm_start, port_mapping, stop_event=threading if asic_index is None: helper_logger.log_warning("Got invalid asic index for {}, ignored".format(logical_port_name)) continue - post_port_mux_static_info_to_db(logical_port_name, port_mapping, mux_tbl[asic_index]) + post_port_mux_static_info_to_db(logical_port_name, port_mapping, static_tbl[asic_index]) def post_mux_info_to_db(is_warm_start, port_mapping, stop_event=threading.Event()):