From b1591818bd47a3f892a7052c86ba7ca520666394 Mon Sep 17 00:00:00 2001 From: Junchao-Mellanox Date: Tue, 5 Dec 2023 13:09:19 +0200 Subject: [PATCH] User click.IntRange to avoid duplicate validation --- sfputil/main.py | 47 +++++-------- tests/sfputil_test.py | 156 +++++++++++++++++++++++++----------------- 2 files changed, 114 insertions(+), 89 deletions(-) diff --git a/sfputil/main.py b/sfputil/main.py index 29b0ee1e7e..1f018a8027 100644 --- a/sfputil/main.py +++ b/sfputil/main.py @@ -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 @@ -1606,14 +1605,16 @@ def target(port_name, target): # 'read-eeprom' subcommand @cli.command() -@click.argument('port_name', metavar='', required=True) -@click.argument('page', metavar='', type=click.INT, required=True) -@click.argument('offset', metavar='', type=click.INT, required=True) -@click.argument('size', metavar='', type=click.INT, required=True) +@click.argument('port_name', metavar='', required=True) +@click.argument('page', metavar='', type=click.IntRange(0, MAX_EEPROM_PAGE), required=True) +@click.argument('offset', metavar='', type=click.IntRange(0, MAX_EEPROM_OFFSET), required=True) +@click.argument('size', metavar='', 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 + : + """ try: if platform_sfputil.is_logical_port(port_name) == 0: click.echo("Error: invalid port {}".format(port_name)) @@ -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='', required=True) -@click.argument('page', metavar='', type=click.INT, required=True) -@click.argument('offset', metavar='', type=click.INT, required=True) +@click.argument('port_name', metavar='', required=True) +@click.argument('page', metavar='', type=click.IntRange(0, MAX_EEPROM_PAGE), required=True) +@click.argument('offset', metavar='', type=click.IntRange(0, MAX_EEPROM_OFFSET), required=True) @click.argument('data', metavar='', required=True) @click.option('--wire-addr', help="Wire address of sff8472") @click.option('--verify', is_flag=True, help="Verify the data by reading back") @@ -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 @@ -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 diff --git a/tests/sfputil_test.py b/tests/sfputil_test.py index 77664c8a3a..86ed9f23f2 100644 --- a/tests/sfputil_test.py +++ b/tests/sfputil_test.py @@ -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')