Skip to content

Commit

Permalink
User click.IntRange to avoid duplicate validation
Browse files Browse the repository at this point in the history
  • Loading branch information
@Junchao-Mellanox
Junchao-Mellanox committed Dec 5, 2023
1 parent ff1052a commit b159181
Show file tree
Hide file tree
Showing 2 changed files with 114 additions and 89 deletions.
47 changes: 19 additions & 28 deletions sfputil/main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@
SFF8472_A0_SIZE = 256
MAX_EEPROM_PAGE = 255
MAX_EEPROM_OFFSET = 255
MIN_OFFSET_FOR_PAGE0 = 0
MIN_OFFSET_FOR_NON_PAGE0 = 128
MAX_OFFSET_FOR_A0H_UPPER_PAGE = 255
MAX_OFFSET_FOR_A0H_LOWER_PAGE = 127
Expand Down Expand Up @@ -1606,14 +1605,16 @@ def target(port_name, target):

# 'read-eeprom' subcommand
@cli.command()
@click.argument('port_name', metavar='<port_name>', required=True)
@click.argument('page', metavar='<page>', type=click.INT, required=True)
@click.argument('offset', metavar='<offset>', type=click.INT, required=True)
@click.argument('size', metavar='<size>', type=click.INT, required=True)
@click.argument('port_name', metavar='<logical_port_name>', required=True)
@click.argument('page', metavar='<page>', type=click.IntRange(0, MAX_EEPROM_PAGE), required=True)
@click.argument('offset', metavar='<offset>', type=click.IntRange(0, MAX_EEPROM_OFFSET), required=True)
@click.argument('size', metavar='<size>', type=click.IntRange(1, MAX_EEPROM_OFFSET + 1), required=True)
@click.option('--no-format', is_flag=True, help="Display non formatted data")
@click.option('--wire-addr', help="Wire address of sff8472")
def read_eeprom(port_name, page, offset, size, no_format, wire_addr):
"""Read SFP EEPROM data"""
"""Read SFP EEPROM data
<port_name>:
"""
try:
if platform_sfputil.is_logical_port(port_name) == 0:
click.echo("Error: invalid port {}".format(port_name))
Expand Down Expand Up @@ -1653,9 +1654,9 @@ def read_eeprom(port_name, page, offset, size, no_format, wire_addr):

# 'write-eeprom' subcommand
@cli.command()
@click.argument('port_name', metavar='<port_name>', required=True)
@click.argument('page', metavar='<page>', type=click.INT, required=True)
@click.argument('offset', metavar='<offset>', type=click.INT, required=True)
@click.argument('port_name', metavar='<logical_port_name>', required=True)
@click.argument('page', metavar='<page>', type=click.IntRange(0, MAX_EEPROM_PAGE), required=True)
@click.argument('offset', metavar='<offset>', type=click.IntRange(0, MAX_EEPROM_OFFSET), required=True)
@click.argument('data', metavar='<data>', required=True)
@click.option('--wire-addr', help="Wire address of sff8472")
@click.option('--verify', is_flag=True, help="Verify the data by reading back")
Expand Down Expand Up @@ -1722,21 +1723,15 @@ def get_overall_offset_general(api, page, offset, size):
Returns:
The overall offset
"""
max_page = 0 if api.is_flat_memory() else MAX_EEPROM_PAGE
if max_page == 0 and page != 0:
raise ValueError(f'Invalid page number {page}, only page 0 is supported')

if page < 0 or page > max_page:
raise ValueError(f'Invalid page number {page}, valid range: [0, {max_page}]')
if api.is_flat_memory():
if page != 0:
raise ValueError(f'Invalid page number {page}, only page 0 is supported')

if page == 0:
if offset < MIN_OFFSET_FOR_PAGE0 or offset > MAX_EEPROM_OFFSET:
raise ValueError(f'Invalid offset {offset} for page 0, valid range: [0, 255]')
else:
if offset < MIN_OFFSET_FOR_NON_PAGE0 or offset > MAX_EEPROM_OFFSET:
if page != 0:
if offset < MIN_OFFSET_FOR_NON_PAGE0:
raise ValueError(f'Invalid offset {offset} for page {page}, valid range: [128, 255]')

if size <= 0 or size + offset - 1 > MAX_EEPROM_OFFSET:
if size + offset - 1 > MAX_EEPROM_OFFSET:
raise ValueError(f'Invalid size {size}, valid range: [1, {255 - offset + 1}]')

return page * PAGE_SIZE + offset
Expand Down Expand Up @@ -1768,18 +1763,14 @@ def get_overall_offset_sff8472(api, page, offset, size, wire_addr):
if page != 0:
raise ValueError(f'Invalid page number {page} for wire address {wire_addr}, only page 0 is supported')
max_offset = MAX_OFFSET_FOR_A0H_UPPER_PAGE if is_active_cable else MAX_OFFSET_FOR_A0H_LOWER_PAGE
if offset < 0 or offset > max_offset:
if offset > max_offset:
raise ValueError(f'Invalid offset {offset} for wire address {wire_addr}, valid range: [0, {max_offset}]')
if size <= 0 or size + offset - 1 > max_offset:
if size + offset - 1 > max_offset:
raise ValueError(
f'Invalid size {size} for wire address {wire_addr}, valid range: [1, {max_offset - offset + 1}]')
return offset
else:
if page < 0 or page > MAX_EEPROM_PAGE:
raise ValueError(f'Invalid page number {page} for wire address {wire_addr}, valid range: [0, 255]')
if offset < 0 or offset > MAX_OFFSET_FOR_A2H:
raise ValueError(f'Invalid offset {offset} for wire address {wire_addr}, valid range: [0, 255]')
if size <= 0 or size + offset - 1 > MAX_OFFSET_FOR_A2H:
if size + offset - 1 > MAX_OFFSET_FOR_A2H:
raise ValueError(f'Invalid size {size} for wire address {wire_addr}, valid range: [1, {255 - offset + 1}]')
return page * PAGE_SIZE + offset + PAGE_SIZE_FOR_A0H

Expand Down
156 changes: 95 additions & 61 deletions tests/sfputil_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1085,78 +1085,112 @@ def test_write_eeprom_rj45(self):
result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '0', '0', '00'])
assert result.exit_code == EXIT_FAIL

def test_get_overall_offset_general(self):
@patch('sfputil.main.is_port_type_rj45', MagicMock(return_value=False))
@patch('sfputil.main.logical_port_to_physical_port_index', MagicMock(return_value=1))
@patch('sfputil.main.platform_sfputil', MagicMock(is_logical_port=MagicMock(return_value=1)))
@patch('sfputil.main.platform_chassis')
def test_get_overall_offset_general(self, mock_chassis):
api = MagicMock()
api.is_flat_memory = MagicMock(return_value=False)
mock_sfp = MagicMock()
mock_chassis.get_sfp = MagicMock(return_value=mock_sfp)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, -1, 0, 1)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 256, 0, 1)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 0, -1, 1)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 0, 256, 1)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 1, 127, 1)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 1, 256, 1)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 0, 0, 0)

with pytest.raises(ValueError):
sfputil.get_overall_offset_general(api, 0, 0, 257)

assert sfputil.get_overall_offset_general(api, 0, 1, 1) == 1
mock_sfp.get_presence = MagicMock(return_value=True)
mock_sfp.get_xcvr_api = MagicMock(return_value=api)

runner = CliRunner()
result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '-1', '0', '01'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '256', '0', '01'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '0', '-1', '01'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '0', '256', '01'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '1', '127', '01'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '1', '256', '01'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--', '0', '0', '0'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--', '0', '0', '257'])
assert result.exit_code != 0

result = runner.invoke(sfputil.cli.commands['write-eeprom'], ["Ethernet0", '--', '0', '1', '01'])
assert result.exit_code == 0

def test_get_overall_offset_sff8472(self):
@patch('sfputil.main.isinstance', MagicMock(return_value=True))
@patch('sfputil.main.is_port_type_rj45', MagicMock(return_value=False))
@patch('sfputil.main.logical_port_to_physical_port_index', MagicMock(return_value=1))
@patch('sfputil.main.platform_sfputil', MagicMock(is_logical_port=MagicMock(return_value=1)))
@patch('sfputil.main.platform_chassis')
def test_get_overall_offset_sff8472(self, mock_chassis):
api = MagicMock()
api.is_copper = MagicMock(return_value=False)
mock_sfp = MagicMock()
mock_chassis.get_sfp = MagicMock(return_value=mock_sfp)

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 0, 1, None)

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 0, 1, wire_addr='invalid')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 1, 0, 1, wire_addr='a0h')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, -1, 1, wire_addr='A0h')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 256, 1, wire_addr='A0h')
mock_sfp.get_presence = MagicMock(return_value=True)
mock_sfp.get_xcvr_api = MagicMock(return_value=api)

runner = CliRunner()
result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--', '0', '0', '1'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'invalid', '0', '0', '1'])
assert result.exit_code != 0
print(result.output)

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 0, 0, wire_addr='A0h')
result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'a0h', '1', '0', '1'])
assert result.exit_code != 0
print(result.output)

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 0, 257, wire_addr='A0h')
result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'A0h', '--', '0', '-1', '1'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'A0h', '--', '0', '256', '1'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'A0h', '--', '0', '0', '0'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'A0h', '--', '0', '0', '257'])
assert result.exit_code != 0
print(result.output)

assert sfputil.get_overall_offset_sff8472(api, 0, 2, 2, wire_addr='A0h') == 2

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, -1, 0, 1, wire_addr='a2h')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 256, 0, 1, wire_addr='a2h')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, -1, 1, wire_addr='a2h')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 0, 0, wire_addr='A2h')

with pytest.raises(ValueError):
sfputil.get_overall_offset_sff8472(api, 0, 0, 257, wire_addr='A2h')


result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'a2h', '--', '-1', '0', '1'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'a2h', '--', '256', '0', '1'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'a2h', '--', '0', '-1', '1'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'a2h', '--', '0', '0', '0'])
assert result.exit_code != 0
print(result.output)

result = runner.invoke(sfputil.cli.commands['read-eeprom'], ["Ethernet0", '--wire-addr', 'a2h', '--', '0', '0', '257'])
assert result.exit_code != 0
print(result.output)

assert sfputil.get_overall_offset_sff8472(api, 0, 2, 2, wire_addr='A2h') == 258

@patch('sfputil.main.platform_chassis')
Expand Down

0 comments on commit b159181

Please sign in to comment.