Skip to content

Commit

Permalink
modify loopback functions for compatibility
Browse files Browse the repository at this point in the history
  • Loading branch information
@qinchuanares
qinchuanares committed Nov 18, 2021
1 parent 6e84eaa commit 8a4cb6c
Show file tree
Hide file tree
Showing 2 changed files with 111 additions and 49 deletions.
81 changes: 56 additions & 25 deletions sonic_platform_base/sonic_xcvr/api/public/cmis.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -852,6 +852,9 @@ def get_loopback_capability(self):
'''
This function returns the module loopback capability as advertised
'''
loopback_capability = dict()
if self.is_flat_memory():
return None
allowed_loopback_result = self.xcvr_eeprom.read(consts.LOOPBACK_CAPABILITY)
if allowed_loopback_result is None:
return None
Expand All @@ -877,32 +880,37 @@ def set_loopback_mode(self, loopback_mode):
The function will look at 13h:128 to check advertized loopback capabilities.
Return True if the provision succeeds, False if it fails
'''
loopback_support = not self.is_flat_memory()
if loopback_support is None:
return None
loopback_capability = self.get_loopback_capability()
if loopback_capability is None:
return None
return False
if loopback_mode == 'none':
status_host_input = self.xcvr_eeprom.write(consts.HOST_INPUT_LOOPBACK, 0)
status_host_output = self.xcvr_eeprom.write(consts.HOST_OUTPUT_LOOPBACK, 0)
status_media_input = self.xcvr_eeprom.write(consts.MEDIA_INPUT_LOOPBACK, 0)
status_media_output = self.xcvr_eeprom.write(consts.MEDIA_OUTPUT_LOOPBACK, 0)
return all([status_host_input, status_host_output, status_media_input, status_media_output])
elif loopback_mode == 'host-side-input':
assert loopback_capability['host_side_input_loopback_supported']
return self.xcvr_eeprom.write(consts.HOST_INPUT_LOOPBACK, 0xff)
if loopback_capability['host_side_input_loopback_supported']:
return self.xcvr_eeprom.write(consts.HOST_INPUT_LOOPBACK, 0xff)
else:
return False
elif loopback_mode == 'host-side-output':
assert loopback_capability['host_side_output_loopback_supported']
return self.xcvr_eeprom.write(consts.HOST_OUTPUT_LOOPBACK, 0xff)
if loopback_capability['host_side_output_loopback_supported']:
return self.xcvr_eeprom.write(consts.HOST_OUTPUT_LOOPBACK, 0xff)
else:
return False
elif loopback_mode == 'media-side-input':
assert loopback_capability['media_side_input_loopback_supported']
return self.xcvr_eeprom.write(consts.MEDIA_INPUT_LOOPBACK, 0xff)
if loopback_capability['media_side_input_loopback_supported']:
return self.xcvr_eeprom.write(consts.MEDIA_INPUT_LOOPBACK, 0xff)
else:
return False
elif loopback_mode == 'media-side-output':
assert loopback_capability['media_side_output_loopback_supported']
return self.xcvr_eeprom.write(consts.MEDIA_OUTPUT_LOOPBACK, 0xff)
if loopback_capability['media_side_output_loopback_supported']:
return self.xcvr_eeprom.write(consts.MEDIA_OUTPUT_LOOPBACK, 0xff)
else:
return False
else:
return 'N/A'
return False

def get_cdb_support(self):
return not self.is_flat_memory()
Expand Down Expand Up @@ -1659,20 +1667,43 @@ def get_transceiver_loopback(self):
host_input_loopback_lane8 = BOOLEAN ; host side input loopback enable lane8
========================================================================
"""
loopback_support = not self.is_flat_memory()
if loopback_support is None:
return None
trans_loopback = dict()
loopback_capability = self.get_loopback_capability()
if loopback_capability is None:
return None
trans_loopback = dict()
trans_loopback['simultaneous_host_media_loopback_supported'] = loopback_capability['simultaneous_host_media_loopback_supported']
trans_loopback['per_lane_media_loopback_supported'] = loopback_capability['per_lane_media_loopback_supported']
trans_loopback['per_lane_host_loopback_supported'] = loopback_capability['per_lane_host_loopback_supported']
trans_loopback['host_side_input_loopback_supported'] = loopback_capability['host_side_input_loopback_supported']
trans_loopback['host_side_output_loopback_supported'] = loopback_capability['host_side_output_loopback_supported']
trans_loopback['media_side_input_loopback_supported'] = loopback_capability['media_side_input_loopback_supported']
trans_loopback['media_side_output_loopback_supported'] = loopback_capability['media_side_output_loopback_supported']
trans_loopback['simultaneous_host_media_loopback_supported'] = 'N/A'
trans_loopback['per_lane_media_loopback_supported'] = 'N/A'
trans_loopback['per_lane_host_loopback_supported'] = 'N/A'
trans_loopback['host_side_input_loopback_supported'] = 'N/A'
trans_loopback['host_side_output_loopback_supported'] = 'N/A'
trans_loopback['media_side_input_loopback_supported'] = 'N/A'
trans_loopback['media_side_output_loopback_supported'] = 'N/A'
trans_loopback['media_output_loopback'] = 'N/A'
trans_loopback['media_input_loopback'] = 'N/A'
trans_loopback['host_output_loopback_lane1'] = 'N/A'
trans_loopback['host_output_loopback_lane2'] = 'N/A'
trans_loopback['host_output_loopback_lane3'] = 'N/A'
trans_loopback['host_output_loopback_lane4'] = 'N/A'
trans_loopback['host_output_loopback_lane5'] = 'N/A'
trans_loopback['host_output_loopback_lane6'] = 'N/A'
trans_loopback['host_output_loopback_lane7'] = 'N/A'
trans_loopback['host_output_loopback_lane8'] = 'N/A'
trans_loopback['host_input_loopback_lane1'] = 'N/A'
trans_loopback['host_input_loopback_lane2'] = 'N/A'
trans_loopback['host_input_loopback_lane3'] = 'N/A'
trans_loopback['host_input_loopback_lane4'] = 'N/A'
trans_loopback['host_input_loopback_lane5'] = 'N/A'
trans_loopback['host_input_loopback_lane6'] = 'N/A'
trans_loopback['host_input_loopback_lane7'] = 'N/A'
trans_loopback['host_input_loopback_lane8'] = 'N/A'
return trans_loopback
else:
trans_loopback['simultaneous_host_media_loopback_supported'] = loopback_capability['simultaneous_host_media_loopback_supported']
trans_loopback['per_lane_media_loopback_supported'] = loopback_capability['per_lane_media_loopback_supported']
trans_loopback['per_lane_host_loopback_supported'] = loopback_capability['per_lane_host_loopback_supported']
trans_loopback['host_side_input_loopback_supported'] = loopback_capability['host_side_input_loopback_supported']
trans_loopback['host_side_output_loopback_supported'] = loopback_capability['host_side_output_loopback_supported']
trans_loopback['media_side_input_loopback_supported'] = loopback_capability['media_side_input_loopback_supported']
trans_loopback['media_side_output_loopback_supported'] = loopback_capability['media_side_output_loopback_supported']
if loopback_capability['media_side_output_loopback_supported']:
trans_loopback['media_output_loopback'] = self.get_media_output_loopback()
else:
Expand Down
79 changes: 55 additions & 24 deletions tests/sonic_xcvr/test_cmis.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -801,26 +801,32 @@ def test_set_low_power(self, ):
self.api.set_low_power(True)

@pytest.mark.parametrize("mock_response, expected", [
(127,
{
'simultaneous_host_media_loopback_supported': True,
'per_lane_media_loopback_supported': True,
'per_lane_host_loopback_supported': True,
'host_side_input_loopback_supported': True,
'host_side_output_loopback_supported': True,
'media_side_input_loopback_supported': True,
'media_side_output_loopback_supported': True
}),
(None, None)
(
[False, 127],
{
'simultaneous_host_media_loopback_supported': True,
'per_lane_media_loopback_supported': True,
'per_lane_host_loopback_supported': True,
'host_side_input_loopback_supported': True,
'host_side_output_loopback_supported': True,
'media_side_input_loopback_supported': True,
'media_side_output_loopback_supported': True
}
),
([True, 0], None),
([False, None], None)
])
def test_get_loopback_capability(self, mock_response, expected):
self.api.is_flat_memory = MagicMock()
self.api.is_flat_memory.return_value = mock_response[0]
self.api.xcvr_eeprom.read = MagicMock()
self.api.xcvr_eeprom.read.return_value = mock_response
self.api.xcvr_eeprom.read.return_value = mock_response[1]
result = self.api.get_loopback_capability()
assert result == expected

@pytest.mark.parametrize("input_param, mock_response",[
('none', {
('none',{
'host_side_input_loopback_supported': True,
'host_side_output_loopback_supported': True,
'media_side_input_loopback_supported': True,
Expand Down Expand Up @@ -850,6 +856,9 @@ def test_get_loopback_capability(self, mock_response, expected):
'media_side_input_loopback_supported': True,
'media_side_output_loopback_supported': True
}),
(
'none', None
)
])
def test_set_loopback_mode(self, input_param, mock_response):
self.api.get_loopback_capability = MagicMock()
Expand Down Expand Up @@ -1644,7 +1653,6 @@ def test_get_transceiver_status(self, mock_response, expected):
@pytest.mark.parametrize("mock_response, expected",[
(
[
True,
{
'simultaneous_host_media_loopback_supported': True,
'per_lane_media_loopback_supported': True,
Expand Down Expand Up @@ -1688,12 +1696,37 @@ def test_get_transceiver_status(self, mock_response, expected):
}
),
(
[True, None, None, None, None, None],
None
[None, None, None, None, None],
{
'simultaneous_host_media_loopback_supported': 'N/A',
'per_lane_media_loopback_supported': 'N/A',
'per_lane_host_loopback_supported': 'N/A',
'host_side_input_loopback_supported': 'N/A',
'host_side_output_loopback_supported': 'N/A',
'media_side_input_loopback_supported': 'N/A',
'media_side_output_loopback_supported': 'N/A',
'media_output_loopback': 'N/A',
'media_input_loopback': 'N/A',
'host_output_loopback_lane1': 'N/A',
'host_output_loopback_lane2': 'N/A',
'host_output_loopback_lane3': 'N/A',
'host_output_loopback_lane4': 'N/A',
'host_output_loopback_lane5': 'N/A',
'host_output_loopback_lane6': 'N/A',
'host_output_loopback_lane7': 'N/A',
'host_output_loopback_lane8': 'N/A',
'host_input_loopback_lane1': 'N/A',
'host_input_loopback_lane2': 'N/A',
'host_input_loopback_lane3': 'N/A',
'host_input_loopback_lane4': 'N/A',
'host_input_loopback_lane5': 'N/A',
'host_input_loopback_lane6': 'N/A',
'host_input_loopback_lane7': 'N/A',
'host_input_loopback_lane8': 'N/A'
}
),
(
[
True,
{
'simultaneous_host_media_loopback_supported': False,
'per_lane_media_loopback_supported': False,
Expand Down Expand Up @@ -1738,18 +1771,16 @@ def test_get_transceiver_status(self, mock_response, expected):
)
])
def test_get_transceiver_loopback(self, mock_response, expected):
self.api.is_flat_memory = MagicMock()
self.api.is_flat_memory.return_value = mock_response[0]
self.api.get_loopback_capability = MagicMock()
self.api.get_loopback_capability.return_value = mock_response[1]
self.api.get_loopback_capability.return_value = mock_response[0]
self.api.get_media_output_loopback = MagicMock()
self.api.get_media_output_loopback.return_value = mock_response[2]
self.api.get_media_output_loopback.return_value = mock_response[1]
self.api.get_media_input_loopback = MagicMock()
self.api.get_media_input_loopback.return_value = mock_response[3]
self.api.get_media_input_loopback.return_value = mock_response[2]
self.api.get_host_output_loopback = MagicMock()
self.api.get_host_output_loopback.return_value = mock_response[4]
self.api.get_host_output_loopback.return_value = mock_response[3]
self.api.get_host_input_loopback = MagicMock()
self.api.get_host_input_loopback.return_value = mock_response[5]
self.api.get_host_input_loopback.return_value = mock_response[4]
result = self.api.get_transceiver_loopback()
assert result == expected

Expand Down

0 comments on commit 8a4cb6c

Please sign in to comment.