Skip to content

[202511] [xcvrd] Support mixed application codes in CMIS decommission logic (#794) #860

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vmittal-msft merged 1 commit into from
Jul 22, 2026
+445 −32
Merged

[202511] [xcvrd] Support mixed application codes in CMIS decommission logic (#794) #860

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
324 changes: 308 additions & 16 deletions sonic-xcvrd/tests/test_xcvrd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2724,7 +2724,8 @@ def test_CmisManagerTask_process_single_lport_invalid_host_lanes_mask(self, mock
info = task.port_dict['Ethernet0']

# Process the port - should fail due to invalid host_lanes_mask
task.process_single_lport('Ethernet0', info, {})
task._gearbox_lanes_dict = {}
task.process_single_lport('Ethernet0', info)

# Verify state transitioned to FAILED
assert common.get_cmis_state_from_state_db('Ethernet0', mock_get_status_sw_tbl) == CMIS_STATE_FAILED
Expand Down Expand Up @@ -2797,7 +2798,8 @@ def test_CmisManagerTask_process_single_lport_tx_power_config_failure(self, mock
info = task.port_dict['Ethernet0']

# Process the port - should log error when tx power config fails
task.process_single_lport('Ethernet0', info, {})
task._gearbox_lanes_dict = {}
task.process_single_lport('Ethernet0', info)

# Verify configure_tx_output_power was called
assert task.configure_tx_output_power.called
Expand All @@ -2821,20 +2823,307 @@ def test_CmisManagerTask_task_run_stop(self, mock_chassis):
cmis_manager.join()
assert not cmis_manager.is_alive()

@pytest.mark.parametrize("app_new, lane_appl_code, expected", [
(2, {0 : 1, 1 : 1, 2 : 1, 3 : 1, 4 : 2, 5 : 2, 6 : 2, 7 : 2}, True),
(0, {0 : 1, 1 : 1, 2 : 1, 3 : 1}, True),
(1, {0 : 0, 1 : 0, 2 : 0, 3 : 0, 4 : 0, 5 : 0, 6 : 0, 7 : 0}, False)
@pytest.mark.parametrize("active_map, desired_map, expected", [
# Lanes 0-3 have app 1 but desired is 2 for all → decommission
([1,1,1,1,2,2,2,2], [2,2,2,2,2,2,2,2], True),
# Lanes 0-3 active with app 1, desired is 0 → decommission
([1,1,1,1,0,0,0,0], [0,0,0,0,0,0,0,0], True),
# All lanes unused, desired has app 1 → no decommission
([0,0,0,0,0,0,0,0], [1,1,1,1,0,0,0,0], False),
# Mixed mode: adding new DPs on unused lanes → no decommission
([3,3,3,3,0,0,0,0], [3,3,3,3,1,1,1,1], False),
# Mixed mode steady state: all lanes match → no decommission
([3,3,3,3,1,1,1,1], [3,3,3,3,1,1,1,1], False),
])
def test_CmisManagerTask_is_decommission_required(self, app_new, lane_appl_code, expected):
def test_CmisManagerTask_is_decommission_required(self, active_map, desired_map, expected):
mock_xcvr_api = MagicMock()
def get_application(lane):
return lane_appl_code.get(lane, 0)
mock_xcvr_api.get_application = MagicMock(side_effect=get_application)
mock_xcvr_api.get_active_apsel_hostlane = MagicMock(return_value={
'ActiveAppSelLane{}'.format(lane + 1): active_map[lane]
for lane in range(CmisManagerTask.CMIS_MAX_HOST_LANES)
})
mock_xcvr_api.get_application = MagicMock(side_effect=lambda lane: active_map[lane])
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}
task.get_desired_app_map = MagicMock(return_value=desired_map)
assert task.is_decommission_required(mock_xcvr_api, 'Ethernet0') == expected

def test_CmisManagerTask_is_decommission_required_uses_active_appsel_not_staged(self):
mock_xcvr_api = MagicMock()
# Active app map still old layout, staged app map already matches desired.
active_map = [1,1,1,1,1,1,1,1]
staged_map = [3,3,3,3,1,1,1,1]
desired_map = [3,3,3,3,1,1,1,1]

mock_xcvr_api.get_active_apsel_hostlane = MagicMock(return_value={
'ActiveAppSelLane{}'.format(lane + 1): active_map[lane]
for lane in range(CmisManagerTask.CMIS_MAX_HOST_LANES)
})
mock_xcvr_api.get_application = MagicMock(side_effect=lambda lane: staged_map[lane])

port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}
task.get_desired_app_map = MagicMock(return_value=desired_map)

assert task.is_decommission_required(mock_xcvr_api, 'Ethernet0') is True

def test_CmisManagerTask_is_decommission_required_invalid_active_appsel(self):
mock_xcvr_api = MagicMock()
desired_map = [2,2,2,2,2,2,2,2]

mock_xcvr_api.get_active_apsel_hostlane = MagicMock(return_value={
'ActiveAppSelLane1': 'N/A',
'ActiveAppSelLane2': 1,
'ActiveAppSelLane3': 1,
'ActiveAppSelLane4': 1,
'ActiveAppSelLane5': 2,
'ActiveAppSelLane6': 2,
'ActiveAppSelLane7': 2,
'ActiveAppSelLane8': 2,
})

port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}
task.get_desired_app_map = MagicMock(return_value=desired_map)

assert task.is_decommission_required(mock_xcvr_api, 'Ethernet0') is True

def test_CmisManagerTask_is_decommission_required_missing_active_appsel_lane(self):
mock_xcvr_api = MagicMock()
desired_map = [2,2,2,2,2,2,2,2]

# Missing ActiveAppSelLane8 should trigger fail-safe decommission.
mock_xcvr_api.get_active_apsel_hostlane = MagicMock(return_value={
'ActiveAppSelLane1': 1,
'ActiveAppSelLane2': 1,
'ActiveAppSelLane3': 1,
'ActiveAppSelLane4': 1,
'ActiveAppSelLane5': 2,
'ActiveAppSelLane6': 2,
'ActiveAppSelLane7': 2,
})

port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}
task.get_desired_app_map = MagicMock(return_value=desired_map)

assert task.is_decommission_required(mock_xcvr_api, 'Ethernet0') is True

def test_CmisManagerTask_get_desired_app_map(self):
"""Test get_desired_app_map with mixed mode, skipped siblings, and edge cases"""
mock_xcvr_api = MagicMock()
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}

# Mock cfg_port_tbl
cfg_port_tbl = MagicMock()

# CONFIG_DB has 5 keys:
# Ethernet0 (index=1, 400G, 8 lanes, subport=0) — non-breakout
# Ethernet8 (index=2, different pport — should be skipped)
# Ethernet16 (index=1, but missing speed — should be skipped)
# Ethernet24 (missing index in full hash — should be skipped)
# Ethernet32 (get returns not found — should be skipped)
cfg_port_tbl.getKeys = MagicMock(return_value=[
'Ethernet0', 'Ethernet8', 'Ethernet16', 'Ethernet24', 'Ethernet32'
])

def get_side_effect(key):
data = {
'Ethernet0': (True, [('speed', '400000'), ('lanes', '1,2,3,4,5,6,7,8'), ('subport', '0'), ('index', '1')]),
'Ethernet8': (True, [('speed', '100000'), ('lanes', '9'), ('subport', '1'), ('index', '2')]),
'Ethernet16': (True, [('lanes', '1,2,3,4'), ('index', '1')]), # missing speed
'Ethernet24': (True, [('speed', '100000'), ('lanes', '1')]), # missing index
'Ethernet32': (False, []),
}
return data.get(key, (False, []))
cfg_port_tbl.get = MagicMock(side_effect=get_side_effect)

task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=cfg_port_tbl)
task._gearbox_lanes_dict = {}

# get_cmis_application_desired returns app code 3 for 8-lane 400G
mock_xcvr_api.get_host_lane_assignment_option = MagicMock(return_value=0xFF)

with patch('xcvrd.xcvrd_utilities.common.get_cmis_application_desired', return_value=3):
result = task.get_desired_app_map(mock_xcvr_api, 'Ethernet0')

assert result == [3, 3, 3, 3, 3, 3, 3, 3]

def test_CmisManagerTask_get_desired_app_map_mixed_mode(self):
"""Test get_desired_app_map with mixed application codes (1x400G + 4x100G)"""
mock_xcvr_api = MagicMock()
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}

cfg_port_tbl = MagicMock()
cfg_port_tbl.getKeys = MagicMock(return_value=['Ethernet0', 'Ethernet4', 'Ethernet5', 'Ethernet6', 'Ethernet7'])

def get_side_effect(key):
data = {
'Ethernet0': (True, [('speed', '400000'), ('lanes', '1,2,3,4'), ('subport', '1'), ('index', '1')]),
'Ethernet4': (True, [('speed', '100000'), ('lanes', '5'), ('subport', '5'), ('index', '1')]),
'Ethernet5': (True, [('speed', '100000'), ('lanes', '6'), ('subport', '6'), ('index', '1')]),
'Ethernet6': (True, [('speed', '100000'), ('lanes', '7'), ('subport', '7'), ('index', '1')]),
'Ethernet7': (True, [('speed', '100000'), ('lanes', '8'), ('subport', '8'), ('index', '1')]),
}
return data.get(key, (False, []))
cfg_port_tbl.get = MagicMock(side_effect=get_side_effect)

task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=cfg_port_tbl)
task._gearbox_lanes_dict = {}

# app=3 for 4-lane 400G, app=1 for 1-lane 100G
def get_app_desired(api, lane_count, speed):
if lane_count == 4 and speed == 400000:
return 3
if lane_count == 1 and speed == 100000:
return 1
return None
# host_lane_assignment_option: all lanes assignable
mock_xcvr_api.get_host_lane_assignment_option = MagicMock(return_value=0xFF)

with patch('xcvrd.xcvrd_utilities.common.get_cmis_application_desired', side_effect=get_app_desired):
result = task.get_desired_app_map(mock_xcvr_api, 'Ethernet0')

assert result == [3, 3, 3, 3, 1, 1, 1, 1]

def test_CmisManagerTask_get_desired_app_map_no_matching_app(self):
"""Test get_desired_app_map when get_cmis_application_desired returns None"""
mock_xcvr_api = MagicMock()
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}

cfg_port_tbl = MagicMock()
cfg_port_tbl.getKeys = MagicMock(return_value=['Ethernet0'])
cfg_port_tbl.get = MagicMock(return_value=(True, [('speed', '999999'), ('lanes', '1,2,3,4'), ('subport', '0'), ('index', '1')]))
task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=cfg_port_tbl)
task._gearbox_lanes_dict = {}

with patch('xcvrd.xcvrd_utilities.common.get_cmis_application_desired', return_value=None):
result = task.get_desired_app_map(mock_xcvr_api, 'Ethernet0')

assert result == [0, 0, 0, 0, 0, 0, 0, 0]

def test_CmisManagerTask_get_desired_app_map_uses_gearbox_lane_count(self):
"""Test get_desired_app_map uses gearbox lane count over PORT table lanes"""
mock_xcvr_api = MagicMock()
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}

cfg_port_tbl = MagicMock()
cfg_port_tbl.getKeys = MagicMock(return_value=['Ethernet0'])
# PORT table lanes suggest 4 lanes, but gearbox says 2 for this lport.
cfg_port_tbl.get = MagicMock(return_value=(True, [('speed', '200000'), ('lanes', '1,2,3,4'), ('subport', '1'), ('index', '1')]))
task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=cfg_port_tbl)
task._gearbox_lanes_dict = {'Ethernet0': 2}

# Return app only when lane_count comes from gearbox (2 lanes) for speed 200G.
def get_app_desired(api, lane_count, speed):
if lane_count == 2 and speed == 200000:
return 2
return None

# app=2 with host_lane_count=2 and subport=1 should map lanes 0 and 1.
mock_xcvr_api.get_host_lane_assignment_option = MagicMock(return_value=0xFF)

with patch('xcvrd.xcvrd_utilities.common.get_cmis_application_desired', side_effect=get_app_desired):
result = task.get_desired_app_map(mock_xcvr_api, 'Ethernet0')

assert result == [2, 2, 0, 0, 0, 0, 0, 0]

def test_CmisManagerTask_get_sibling_port_configs(self):
"""get_sibling_port_configs returns one entry per sibling sharing pport,
skipping rows with mismatched pport, missing index/speed/lanes, invalid
numeric fields, or get() returning not-found."""
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
assert task.is_decommission_required(mock_xcvr_api, app_new) == expected
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}

cfg_port_tbl = MagicMock()
cfg_port_tbl.getKeys = MagicMock(return_value=[
'Ethernet0', # match
'Ethernet4', # match
'Ethernet8', # different pport — skip
'Ethernet12', # missing index — skip
'Ethernet16', # missing speed — skip
'Ethernet20', # missing lanes — skip
'Ethernet24', # invalid speed — skip
'Ethernet28', # invalid index — skip
'Ethernet32', # not found — skip
])

def get_side_effect(key):
data = {
'Ethernet0': (True, [('speed', '400000'), ('lanes', '1,2,3,4'), ('subport', '1'), ('index', '1')]),
'Ethernet4': (True, [('speed', '100000'), ('lanes', '5'), ('subport', '5'), ('index', '1')]),
'Ethernet8': (True, [('speed', '100000'), ('lanes', '9'), ('subport', '1'), ('index', '2')]),
'Ethernet12': (True, [('speed', '100000'), ('lanes', '1')]),
'Ethernet16': (True, [('lanes', '1,2,3,4'), ('index', '1')]),
'Ethernet20': (True, [('speed', '100000'), ('subport', '1'), ('index', '1')]),
'Ethernet24': (True, [('speed', 'foo'), ('lanes', '1'), ('subport', '1'), ('index', '1')]),
'Ethernet28': (True, [('speed', '100000'), ('lanes', '1'), ('subport', '1'), ('index', 'bar')]),
'Ethernet32': (False, []),
}
return data.get(key, (False, []))
cfg_port_tbl.get = MagicMock(side_effect=get_side_effect)

task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=cfg_port_tbl)
task._gearbox_lanes_dict = {}

siblings = task.get_sibling_port_configs('Ethernet0')

assert siblings == [
{'lport': 'Ethernet0', 'subport': 1, 'speed': 400000, 'host_lane_count': 4},
{'lport': 'Ethernet4', 'subport': 5, 'speed': 100000, 'host_lane_count': 1},
]

def test_CmisManagerTask_get_sibling_port_configs_no_cfg_port_tbl(self):
"""get_sibling_port_configs returns [] when cfg_port_tbl is None."""
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}
task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=None)
task._gearbox_lanes_dict = {}

assert task.get_sibling_port_configs('Ethernet0') == []

def test_CmisManagerTask_get_sibling_port_configs_uses_gearbox_lane_count(self):
"""get_sibling_port_configs uses gearbox lane count when present."""
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task.port_dict['Ethernet0'] = {'index': 1, 'asic_id': 0}

cfg_port_tbl = MagicMock()
cfg_port_tbl.getKeys = MagicMock(return_value=['Ethernet0'])
cfg_port_tbl.get = MagicMock(return_value=(
True, [('speed', '200000'), ('lanes', '1,2,3,4'), ('subport', '1'), ('index', '1')]))
task.xcvr_table_helper.get_cfg_port_tbl = MagicMock(return_value=cfg_port_tbl)
task._gearbox_lanes_dict = {'Ethernet0': 2}

siblings = task.get_sibling_port_configs('Ethernet0')

assert siblings == [
{'lport': 'Ethernet0', 'subport': 1, 'speed': 200000, 'host_lane_count': 2},
]

DEFAULT_DP_STATE = {
'DP1State': 'DataPathActivated',
Expand Down Expand Up @@ -3067,8 +3356,9 @@ def test_CmisManagerTask_get_host_lane_count(self, gearbox_lanes_dict, lport, po
port_mapping = PortMapping()
stop_event = threading.Event()
task = CmisManagerTask(DEFAULT_NAMESPACE, port_mapping, stop_event, platform_chassis=MagicMock())
task._gearbox_lanes_dict = gearbox_lanes_dict

result = task.get_host_lane_count(lport, port_config_lanes, gearbox_lanes_dict)
result = task.get_host_lane_count(lport, port_config_lanes)
assert result == expected_count

def test_CmisManagerTask_gearbox_integration_end_to_end(self):
Expand Down Expand Up @@ -3103,7 +3393,8 @@ def test_CmisManagerTask_gearbox_integration_end_to_end(self):
}

# Test the integration: should use gearbox line lanes (2) not port config lanes (4)
host_lane_count = task.get_host_lane_count("Ethernet0", port_config_lanes, gearbox_lanes_dict)
task._gearbox_lanes_dict = gearbox_lanes_dict
host_lane_count = task.get_host_lane_count("Ethernet0", port_config_lanes)
assert host_lane_count == 2 # Should use gearbox line lanes, not port config

# Test that this leads to correct CMIS application selection
Expand All @@ -3123,13 +3414,14 @@ def test_CmisManagerTask_gearbox_caching_integration(self):
task.xcvr_table_helper.get_gearbox_line_lanes_dict.return_value = mock_gearbox_lanes_dict

# Test that get_host_lane_count uses the cached dictionary correctly
result1 = task.get_host_lane_count("Ethernet0", "25,26,27,28", mock_gearbox_lanes_dict)
task._gearbox_lanes_dict = mock_gearbox_lanes_dict
result1 = task.get_host_lane_count("Ethernet0", "25,26,27,28")
assert result1 == 2 # Should use gearbox count

result2 = task.get_host_lane_count("Ethernet4", "29,30", mock_gearbox_lanes_dict)
result2 = task.get_host_lane_count("Ethernet4", "29,30")
assert result2 == 4 # Should use gearbox count

result3 = task.get_host_lane_count("Ethernet8", "33,34,35", mock_gearbox_lanes_dict)
result3 = task.get_host_lane_count("Ethernet8", "33,34,35")
assert result3 == 3 # Should fall back to port config count

@patch('swsscommon.swsscommon.FieldValuePairs')
Expand Down
Loading
Loading