diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py b/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py index 5f0b981ccae..7c4a13925c2 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py @@ -9,12 +9,57 @@ def mock_cfgdb(): 'VLAN': { 'Vlan1000': { 'dhcp_servers': ['192.0.0.1'] - } + }, + 'Vlan200': {}, + }, + 'DHCPV4_RELAY': { + 'Vlan1000': { + 'dhcpv4_servers': ['192.0.0.1'] + }, + 'Vlan200': {}, }, 'DHCP_RELAY': { 'Vlan1000': { 'dhcpv6_servers': ['fc02:2000::1'] } + }, + 'DEVICE_METADATA': { + 'localhost': { + 'has_sonic_dhcpv4_relay' : "False" + } + }, + 'VRF': { + "default": { + 'VRF': 'default' + }, + "VrfRED": { + 'VRF': 'VrfRED' + }, + "VrfBLUE": { + 'VRF': 'VrfBLUE' + } + }, + 'PORTCHANNEL_INTERFACE': { + "PortChannel5": {}, + "PortChannel6": {}, + "PortChannel5|192.168.0.1/31": {}, + "PortChannel6|192.168.0.3/31": {} + }, + 'LOOPBACK_INTERFACE': { + "Loopback0": {}, + "Loopback2": {}, + "Loopback3": {}, + "Loopback0|10.1.0.1/32": {}, + "Loopback2|10.1.0.1/32": {}, + "Loopback3|10.1.0.2/32": {} + }, + 'INTERFACE': { + "Ethernet0": {}, + "Ethernet0|10.0.0.0/31": {}, + "Ethernet8": {}, + "Ethernet8|10.0.0.4/31": {}, + "Ethernet12": {}, + "Ethernet12|10.0.0.6/31": {}, } } @@ -34,6 +79,10 @@ def set_entry(table, key, data): def get_keys(table): return CONFIG[table].keys() + def get_table(table): + return CONFIG.get(table, {}) + + cfgdb.get_table = mock.Mock(side_effect=get_table) cfgdb.get_entry = mock.Mock(side_effect=get_entry) cfgdb.set_entry = mock.Mock(side_effect=set_entry) cfgdb.get_keys = mock.Mock(side_effect=get_keys) diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/mock_config.py b/dockers/docker-dhcp-relay/cli-plugin-tests/mock_config.py index 28370fd99f5..ed652b7aa7c 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/mock_config.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/mock_config.py @@ -42,6 +42,14 @@ "192.0.0.2" ] } + }, + "DHCPV4_RELAY": { + "Vlan1000": { + "dhcpv4_servers": [ + "192.0.0.1", + "192.0.0.2" + ] + } } } } @@ -170,5 +178,38 @@ } } } + ], + [ + "ipv4_dhcp", + { + "config_db": { + "DHCPV4_RELAY": { + "Vlan1000": { + "dhcpv4_servers": [ + "192.0.0.1", + "192.0.0.2" + ], + "source_interface": "Ethernet112", + "link_selection": "enable", + "server_vrf": "default", + "vrf_selection": "enable" + }, + "Vlan1001": { + "vlanid": "1001", + "dhcpv4_servers": [ + "192.0.0.3", + "192.0.0.4" + ], + "agent_relay_mode": "discard", + "max_hop_count": "5" + } + }, + 'DEVICE_METADATA': { + 'localhost': { + 'has_sonic_dhcpv4_relay' : "False" + } + } + } + } ] ] diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/test_clear_dhcp_relay.py b/dockers/docker-dhcp-relay/cli-plugin-tests/test_clear_dhcp_relay.py index a837258c6f5..a97d1511559 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/test_clear_dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_clear_dhcp_relay.py @@ -19,17 +19,17 @@ @pytest.fixture(scope="module") def patch_import_module(): # We need to mock import module because clear_dhcp_relay.py has below import - # dhcp6_relay = importlib.import_module('show.plugins.dhcp-relay') + # dhcprelay = importlib.import_module('show.plugins.dhcp-relay') # When install current container, sonic-application-extension would move below file to destination in switch # Src: dockers/docker-dhcp-relay/cli/show/plugins/show_dhcp_relay.py # Dst: python-package-patch/show/plugins/dhcp-relay.py # The dst path doesn't exist in UT env, hence we need to mock it - fake_dhcp6_relay = MagicMock() + fake_dhcprelay = MagicMock() with patch('importlib.import_module') as mock_import: def side_effect(name): if name == 'show.plugins.dhcp-relay': - return fake_dhcp6_relay + return fake_dhcprelay return original_import_module(name) # fallback mock_import.side_effect = side_effect @@ -54,7 +54,7 @@ def test_clear_dhcp_relay_ipv6_counter(interface, patch_import_module): gotten_interfaces = ["Ethernet0, Ethernet1"] mock_counter = MagicMock() - clear_dhcp_relay.dhcp6_relay.DHCPv6_Counter.return_value = mock_counter + clear_dhcp_relay.dhcprelay.DHCPv6_Counter.return_value = mock_counter mock_counter.get_interface.return_value = gotten_interfaces clear_dhcp_relay.clear_dhcp_relay_ipv6_counter(interface) if interface: @@ -319,3 +319,30 @@ def test_is_write_db_paused(patch_import_module): mock_db = MagicMock() clear_dhcp_relay.is_write_db_paused(mock_db, "table_name", "vlan_interface") mock_db.get.assert_called_once_with(ANY, "table_name|vlan_interface", "pause_write_to_db") + +@pytest.mark.parametrize("direction, pkt_type, interface", [ + (None, None, None), + ("TX", "DISCOVER", "Vlan1000"), + ("RX", "OFFER", None), +]) +def test_clear_dhcp_relay_ipv4_vlan_counter(patch_import_module, direction, pkt_type, interface): + clear_dhcp_relay = patch_import_module + mock_counter = MagicMock() + clear_dhcp_relay.dhcprelay.DHCPv4_Counter.return_value = mock_counter + clear_dhcp_relay.clear_dhcp_relay_ipv4_vlan_counter(direction, pkt_type, interface) + mock_counter.clear_table.assert_called_once_with(direction, pkt_type, interface) + +def test_dhcp4relay_clear_vlan_counters_command(patch_import_module): + clear_dhcp_relay = patch_import_module + runner = CliRunner() + # Patch the actual clear function to verify call + with patch.object(clear_dhcp_relay, "clear_dhcp_relay_ipv4_vlan_counter") as mock_clear: + result = runner.invoke(clear_dhcp_relay.dhcp4relay_clear.commands['dhcp4relay-vlan-counters'], ['-d', 'TX', 'Vlan1000']) + assert result.exit_code == 0 + +def test_register_adds_dhcp4relay_clear_vlan_counters(patch_import_module): + clear_dhcp_relay = patch_import_module + cli = MagicMock() + clear_dhcp_relay.register(cli) + # Check that dhcp4relay_clear_vlan_counters was registered as a command + cli.add_command.assert_any_call(clear_dhcp_relay.dhcp4relay_clear_vlan_counters) diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_dhcp_relay.py b/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_dhcp_relay.py index 295f38ab2a0..74d320e5e30 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_dhcp_relay.py @@ -11,6 +11,9 @@ Added DHCP relay address [{}] to Vlan1000 Restarting DHCP relay service... """ +config_dhcp_relay_update_output = """\ +Updated DHCPv4 Servers as {} to Vlan1000 +""" config_dhcp_relay_del_output = """\ Removed DHCP relay address [{}] from Vlan1000 Restarting DHCP relay service... @@ -20,6 +23,10 @@ "dhcp_servers": [ "192.0.0.1", "192.0.0.3"] }, + "ipv4_dhcp": { + "dhcpv4_servers": [ + "192.0.0.1", "192.0.0.3"] + }, "ipv6": { "dhcpv6_servers": [ "fc02:2000::1", "fc02:2000::3"] @@ -31,6 +38,11 @@ "192.0.0.1" ] }, + "ipv4_dhcp": { + "dhcpv4_servers": [ + "192.0.0.1" + ] + }, "ipv6": { "dhcpv6_servers": [ "fc02:2000::1" @@ -43,6 +55,11 @@ "192.0.0.1", "192.0.0.3", "192.0.0.4", "192.0.0.5" ] }, + "ipv4_dhcp": { + "dhcpv4_servers": [ + "192.0.0.1", "192.0.0.3", "192.0.0.4", "192.0.0.5" + ] + }, "ipv6": { "dhcpv6_servers": [ "fc02:2000::1", "fc02:2000::3", "fc02:2000::4", "fc02:2000::5" @@ -63,6 +80,18 @@ "invalid_ip": "192.0.0", "table": "VLAN" }, + "ipv4_dhcp": { + "command": "helper", + "ips": [ + "192.0.0.3", + "192.0.0.4", + "192.0.0.5" + ], + "exist_ip": "192.0.0.1", + "nonexist_ip": "192.0.0.2", + "invalid_ip": "192.0.0", + "table": "DHCPV4_RELAY" + }, "ipv6": { "command": "destination", "ips": [ @@ -78,8 +107,8 @@ } -@pytest.fixture(scope="module", params=["ipv4", "ipv6"]) -def ip_version(request): +@pytest.fixture(scope="module", params=["ipv4", "ipv4_dhcp", "ipv6"]) +def version(request): """ Parametrize Ip version @@ -140,13 +169,17 @@ def test_config_dhcp_relay_del_with_nonexist_vlanid_ipv6(self): assert "Error: Vlan1001 doesn't exist" in result.output assert mock_run_command.call_count == 0 - def test_config_add_del_dhcp_relay_with_invalid_ip(self, ip_version, op): + def test_config_add_del_dhcp_relay_with_invalid_ip(self, mock_cfgdb, version, op): runner = CliRunner() - invalid_ip = IP_VER_TEST_PARAM_MAP[ip_version]["invalid_ip"] - + db = Db() + db.cfgdb = mock_cfgdb + invalid_ip = IP_VER_TEST_PARAM_MAP[version]["invalid_ip"] + ip_version = "ipv6" if version == "ipv6" else "ipv4" + if version == "ipv4_dhcp": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] .commands[op], ["1000", invalid_ip]) print(result.exit_code) print(result.output) @@ -154,75 +187,122 @@ def test_config_add_del_dhcp_relay_with_invalid_ip(self, ip_version, op): assert "Error: {} is invalid IP address".format(invalid_ip) in result.output assert mock_run_command.call_count == 0 - def test_config_add_dhcp_with_exist_ip(self, mock_cfgdb, ip_version): + def test_config_add_dhcp_with_exist_ip(self, mock_cfgdb, version): runner = CliRunner() db = Db() db.cfgdb = mock_cfgdb - exist_ip = IP_VER_TEST_PARAM_MAP[ip_version]["exist_ip"] - + exist_ip = IP_VER_TEST_PARAM_MAP[version]["exist_ip"] + ip_version = "ipv6" if version == "ipv6" else "ipv4" + if version == "ipv4_dhcp": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) + table = db.cfgdb.get_entry("FEATURE", "dhcp_relay") + if('has_sonic_dhcpv4_relay' in table and table['has_sonic_dhcpv4_relay'] == 'True'): + print("has_sonic_dhcpv4_relay is set") with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] .commands["add"], ["1000", exist_ip], obj=db) print(result.exit_code) print(result.output) - assert result.exit_code == 0 - assert "{} is already a DHCP relay for Vlan1000".format(exist_ip) in result.output + assert result.exit_code != 0 + if version == "ipv4_dhcp": + assert "DHCPv4 relay entry for Vlan1000 already exists. Use 'update' instead." in result.output + else: + assert "{} is already a DHCP relay for Vlan1000".format(exist_ip) in result.output assert mock_run_command.call_count == 0 + db.cfgdb.set_entry.reset_mock() - def test_config_del_nonexist_dhcp_relay(self, mock_cfgdb, ip_version): + def test_config_del_nonexist_dhcp_relay(self, mock_cfgdb, version): runner = CliRunner() db = Db() db.cfgdb = mock_cfgdb - nonexist_ip = IP_VER_TEST_PARAM_MAP[ip_version]["nonexist_ip"] + nonexist_ip = IP_VER_TEST_PARAM_MAP[version]["nonexist_ip"] + ip_version = "ipv6" if version == "ipv6" else "ipv4" + if version == "ipv4_dhcp": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] .commands["del"], ["1000", nonexist_ip], obj=db) print(result.exit_code) print(result.output) assert result.exit_code != 0 - assert "Error: {} is not a DHCP relay for Vlan1000".format(nonexist_ip) in result.output + if version == "ipv4_dhcp": + assert "Error: {} is not a DHCPv4 relay for Vlan1000".format(nonexist_ip) in result.output + else: + assert "Error: {} is not a DHCP relay for Vlan1000".format(nonexist_ip) in result.output + assert mock_run_command.call_count == 0 + db.cfgdb.set_entry.reset_mock() - def test_config_add_del_dhcp_relay(self, mock_cfgdb, ip_version): + def test_config_add_del_dhcp_relay(self, mock_cfgdb, version): runner = CliRunner() db = Db() db.cfgdb = mock_cfgdb - test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] - config_db_table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + test_ip = IP_VER_TEST_PARAM_MAP[version]["ips"][0] + ip_version = "ipv6" if version == "ipv6" else "ipv4" + config_db_table = IP_VER_TEST_PARAM_MAP[version]["table"] + + if version == "ipv4_dhcp": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) + dhcp4_config_db_table = IP_VER_TEST_PARAM_MAP["ipv4_dhcp"]["table"] + op = "update" + else: + op = "add" with mock.patch("utilities_common.cli.run_command") as mock_run_command: # add new dhcp relay result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] - .commands["add"], ["1000", test_ip], obj=db) + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] + .commands[op], ["1000", test_ip], obj=db) print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_dhcp_relay_add_output.format(test_ip) - assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ - == expected_dhcp_relay_add_config_db_output[ip_version] - assert mock_run_command.call_count == 3 - db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", - expected_dhcp_relay_add_config_db_output[ip_version]) + if version != "ipv4_dhcp": + assert result.output == config_dhcp_relay_add_output.format(test_ip) + assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ + == expected_dhcp_relay_add_config_db_output[version] + assert mock_run_command.call_count == 3 + + if version == "ipv4_dhcp" : + assert result.output == config_dhcp_relay_update_output.format(test_ip) + expected_calls = [ + mock.call('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}), + mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_add_config_db_output["ipv4_dhcp"]) + ] + + assert db.cfgdb.set_entry.call_args_list == expected_calls + else: + db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", + expected_dhcp_relay_add_config_db_output[version]) db.cfgdb.set_entry.reset_mock() # del dhcp relay with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] .commands["del"], ["1000", test_ip], obj=db) print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_dhcp_relay_del_output.format(test_ip) - assert mock_run_command.call_count == 3 - assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ - == expected_dhcp_relay_del_config_db_output[ip_version] - db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", - expected_dhcp_relay_del_config_db_output[ip_version]) + if version != "ipv4_dhcp": + assert result.output == config_dhcp_relay_del_output.format(test_ip) + assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ + == expected_dhcp_relay_del_config_db_output[version] + assert mock_run_command.call_count == 3 + + if version == "ipv4_dhcp" : + assert "Removed DHCP relay address [{}] from Vlan1000".format(test_ip) in result.output + expected_calls = [ + mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_del_config_db_output["ipv4_dhcp"]) + ] + + assert db.cfgdb.set_entry.call_args_list == expected_calls + + else: + db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", + expected_dhcp_relay_del_config_db_output[version]) def test_config_add_del_dhcp_relay_with_enable_dhcp_server(self, mock_cfgdb): runner = CliRunner() @@ -254,59 +334,96 @@ def test_config_add_del_dhcp_relay_with_enable_dhcp_server(self, mock_cfgdb): assert result.exit_code == 0 assert "Cannot change ipv4 dhcp_relay configuration when dhcp_server feature is enabled" in result.output - def test_config_add_del_multiple_dhcp_relay(self, mock_cfgdb, ip_version): + def test_config_add_del_multiple_dhcp_relay(self, mock_cfgdb, version): runner = CliRunner() db = Db() db.cfgdb = mock_cfgdb - test_ips = IP_VER_TEST_PARAM_MAP[ip_version]["ips"] - config_db_table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + test_ips = IP_VER_TEST_PARAM_MAP[version]["ips"] + config_db_table = IP_VER_TEST_PARAM_MAP[version]["table"] + ip_version = "ipv6" if version == "ipv6" else "ipv4" + + if version == "ipv4_dhcp" : + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) + dhcp4_config_db_table = IP_VER_TEST_PARAM_MAP["ipv4_dhcp"]["table"] + op = "update" + else: + op = "add" with mock.patch("utilities_common.cli.run_command") as mock_run_command: # add new dhcp relay result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] - .commands["add"], ["1000"] + test_ips, obj=db) + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] + .commands[op], ["1000"] + test_ips, obj=db) print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_dhcp_relay_add_output.format(",".join(test_ips)) - assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ - == expected_dhcp_relay_add_multi_config_db_output[ip_version] - assert mock_run_command.call_count == 3 - db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", - expected_dhcp_relay_add_multi_config_db_output[ip_version]) + if version != "ipv4_dhcp": + assert result.output == config_dhcp_relay_add_output.format(",".join(test_ips)) + assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ + == expected_dhcp_relay_add_multi_config_db_output[version] + assert mock_run_command.call_count == 3 + + if version == "ipv4_dhcp" : + assert result.output == config_dhcp_relay_update_output.format(",".join(test_ips)) + expected_calls = [ + mock.call('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}), + mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_add_multi_config_db_output["ipv4_dhcp"]), + ] + + assert db.cfgdb.set_entry.call_args_list == expected_calls + else: + db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", + expected_dhcp_relay_add_multi_config_db_output[version]) db.cfgdb.set_entry.reset_mock() # del dhcp relay with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] .commands["del"], ["1000"] + test_ips, obj=db) print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_dhcp_relay_del_output.format(",".join(test_ips)) - assert mock_run_command.call_count == 3 - assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ - == expected_dhcp_relay_del_config_db_output[ip_version] - db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", - expected_dhcp_relay_del_config_db_output[ip_version]) + if version != "ipv4_dhcp": + assert result.output == config_dhcp_relay_del_output.format(",".join(test_ips)) + assert mock_run_command.call_count == 3 + assert db.cfgdb.get_entry(config_db_table, "Vlan1000") \ + == expected_dhcp_relay_del_config_db_output[version] + if ip_version == "ipv4_dhcp": + assert "Removed DHCP relay address [{}] from Vlan1000".format(",".join(test_ip)) in result.output + expected_calls = [ + mock.call(dhcp4_config_db_table, "Vlan1000", expected_dhcp_relay_del_config_db_output["ipv4_dhcp"]), + ] + + assert db.cfgdb.set_entry.call_args_list == expected_calls + else: + db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", + expected_dhcp_relay_del_config_db_output[version]) + + db.cfgdb.set_entry.reset_mock() - def test_config_add_del_duplicate_dhcp_relay(self, mock_cfgdb, ip_version, op): + def test_config_add_del_duplicate_dhcp_relay(self, mock_cfgdb, version, op): runner = CliRunner() db = Db() db.cfgdb = mock_cfgdb - test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] if op == "add" \ - else IP_VER_TEST_PARAM_MAP[ip_version]["exist_ip"] + test_ip = IP_VER_TEST_PARAM_MAP[version]["ips"][0] if op == "add" \ + else IP_VER_TEST_PARAM_MAP[version]["exist_ip"] + ip_version = "ipv6" if version == "ipv6" else "ipv4" + + if version == "ipv4_dhcp" : + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] - .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[IP_VER_TEST_PARAM_MAP[version]["command"]] .commands[op], ["1000", test_ip, test_ip], obj=db) print(result.exit_code) print(result.output) assert result.exit_code != 0 - assert "Error: Find duplicate DHCP relay ip {} in {} list".format(test_ip, op) in result.output + if version == "ipv4_dhcp" and op == "add": + assert "Error: DHCPv4 relay entry for Vlan1000 already exists. Use 'update' instead." in result.output + else: + assert "Error: Find duplicate DHCP relay ip {} in {} list".format(test_ip, op) in result.output assert mock_run_command.call_count == 0 def test_config_add_dhcp_relay_ipv6_with_non_entry(self, mock_cfgdb): @@ -330,3 +447,167 @@ def test_config_add_dhcp_relay_ipv6_with_non_entry(self, mock_cfgdb): assert result.exit_code == 0 assert db.cfgdb.get_entry(table, "Vlan1000") == {"dhcpv6_servers": [test_ip]} assert mock_run_command.call_count == 3 + + def test_add_dhcpv4_relay_compatibility_check(self, mock_cfgdb): + ip_version = "ipv4_dhcp" + test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + + #set feature flag + db.cfgdb.set_entry("DEVICE_METADATA", "localhost", {"has_sonic_dhcpv4_relay" : "True"}) + db.cfgdb.set_entry("DHCPV4_RELAY", "Vlan1000", None) + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + #Default 'ipv4 helper' command should work when has_sonic_dhcpv4_relay is enabled + result = runner.invoke(dhcp_relay.dhcp_relay.commands["ipv4"] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands["add"], [ + '1000', + test_ip + ], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + + # Configuring additional parameters also should work with new feature flag set + result = runner.invoke(dhcp_relay.dhcp_relay.commands["ipv4"] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands["update"], [ + '1000', + test_ip, + '--server-vrf', 'default', + '--source-interface', 'Loopback0', + '--link-selection', 'enable', + '--vrf-selection', 'enable', + '--server-id-override', 'enable', + '--agent-relay-mode', 'discard', + '--max-hop-count', '5' + ], obj=db) + + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + db.cfgdb.set_entry.reset_mock() + + def test_delete_dhcpv4_relay_compatibility_check(self, mock_cfgdb): + ip_version = "ipv4_dhcp" + test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["exist_ip"] + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + + #set feature flag + db.cfgdb.set_entry("DEVICE_METADATA", "localhost", {"has_sonic_dhcpv4_relay" : "True"}) + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Present delete cmd should work with new feature flag enabled + result = runner.invoke(dhcp_relay.dhcp_relay.commands["ipv4"] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands["del"], ['1000', test_ip], obj=db) + + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + + def test_add_dhcpv4_relay_compatibility_check_negative(self, mock_cfgdb): + op = "add" + ip_version = "ipv4_dhcp" + test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + + #unset feature flag + db.cfgdb.set_entry("DEVICE_METADATA", "localhost", {"has_sonic_dhcpv4_relay" : "False"}) + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + + # Parameters config should NOT work when new feature flag is disabled + result = runner.invoke(dhcp_relay.dhcp_relay.commands["ipv4"] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[op], [ + '1000', + test_ip, + '--server-vrf', 'default', + '--source-interface', 'Loopback0', + '--link-selection', 'enable', + '--vrf-selection', 'enable', + '--server-id-override', 'enable', + '--agent-relay-mode', 'discard', + '--max-hop-count', '5' + ], obj=db) + + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + assert "These parameters are applicable for new DHCPv4 Relay feature" in result.output + + def test_update_dhcpv4_relay(self, mock_cfgdb): + op = "update" + ip_version = "ipv4_dhcp" + test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + + #set feature flag + db.cfgdb.set_entry("DEVICE_METADATA", "localhost", {"has_sonic_dhcpv4_relay":"True"}) + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Updating parameters also should work with new feature flag set + result = runner.invoke(dhcp_relay.dhcp_relay.commands["ipv4"] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[op], [ + '1000', + test_ip, + '--server-vrf', 'default', + '--source-interface', 'Loopback0', + '--link-selection', 'enable', + '--vrf-selection', 'enable', + '--server-id-override', 'enable', + '--agent-relay-mode', 'discard', + '--max-hop-count', '5' + ], obj=db) + + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + + def test_update_dhcpv4_relay_negative(self, mock_cfgdb): + op = "update" + ip_version = "ipv4_dhcp" + test_ip = IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0] + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + table = IP_VER_TEST_PARAM_MAP[ip_version]["table"] + + #unset feature flag + db.cfgdb.set_entry("DEVICE_METADATA", "localhost", {"has_sonic_dhcpv4_relay":"False"}) + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Update cmd is available only for new feature, it should NOT work when flag is disabled + result = runner.invoke(dhcp_relay.dhcp_relay.commands["ipv4"] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[op], [ + '1000', + test_ip, + '--server-vrf', 'default', + '--source-interface', 'Loopback0', + '--link-selection', 'enable', + '--vrf-selection', 'enable', + '--server-id-override', 'enable', + '--agent-relay-mode', 'discard', + '--max-hop-count', '5' + ], obj=db) + + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + assert "This command is applicable for new DHCPv4 Relay feature" in result.output diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py b/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py index 46acda358b8..b94ab3609dd 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py @@ -42,6 +42,19 @@ Restarting DHCP relay service... """ +@pytest.fixture(scope="module", params=["isc", "new"]) +def mode(request): + """ + Parametrize dhcp mode + + Args: + request: pytest request object + + Returns: + dhcp mode needed for test case + """ + return request.param + class TestConfigVlanDhcpRelay(object): def test_plugin_registration(self): cli = mock.MagicMock() @@ -111,11 +124,14 @@ def test_config_vlan_add_dhcp_relay_with_exist_ip(self, mock_cfgdb): assert "192.0.0.1 is already a DHCP relay destination for Vlan1000" in result.output assert mock_run_command.call_count == 0 - def test_config_vlan_add_del_dhcp_relay_dest(self, mock_cfgdb): + def test_config_vlan_add_del_dhcp_relay_dest(self, mock_cfgdb, mode): runner = CliRunner() db = Db() db.cfgdb = mock_cfgdb - + if mode == "isc": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'False'}) + elif mode == "new": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) # add new relay dest with mock.patch("utilities_common.cli.run_command") as mock_run_command: result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], @@ -123,11 +139,28 @@ def test_config_vlan_add_del_dhcp_relay_dest(self, mock_cfgdb): print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_vlan_add_dhcp_relay_output - assert mock_run_command.call_count == 3 - db.cfgdb.set_entry.assert_called_once_with('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1', '192.0.0.100']}) + if mode == "isc": + assert result.output == config_vlan_add_dhcp_relay_output + assert mock_run_command.call_count == 3 + expected_calls = [ + mock.call('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'False'}), + mock.call('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1', '192.0.0.100']}) + ] + db.cfgdb.set_entry.assert_has_calls(expected_calls) + elif mode == "new": + assert "Added DHCP relay destination addresses ['192.0.0.100'] to Vlan1000" in result.output + expected_calls = [ + mock.call('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}), + #mock.call('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1', '192.0.0.100']}), + mock.call('DHCPV4_RELAY', 'Vlan1000', {'dhcpv4_servers': ['192.0.0.1', '192.0.0.100']}) + ] + db.cfgdb.set_entry.assert_has_calls(expected_calls) db.cfgdb.set_entry.reset_mock() + if mode == "isc": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'False'}) + elif mode == "new": + db.cfgdb.set_entry('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}) # del relay dest with mock.patch("utilities_common.cli.run_command") as mock_run_command: @@ -136,9 +169,22 @@ def test_config_vlan_add_del_dhcp_relay_dest(self, mock_cfgdb): print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_vlan_del_dhcp_relay_output - assert mock_run_command.call_count == 3 - db.cfgdb.set_entry.assert_called_once_with('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1']}) + if mode == "isc": + assert result.output == config_vlan_del_dhcp_relay_output + assert mock_run_command.call_count == 3 + expected_calls = [ + mock.call('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'False'}), + mock.call('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1']}) + ] + db.cfgdb.set_entry.assert_has_calls(expected_calls) + elif mode == "new": + assert "Removed DHCP relay destination addresses ('192.0.0.100',) from Vlan1000" in result.output + expected_calls = [ + mock.call('DEVICE_METADATA', 'localhost', {'has_sonic_dhcpv4_relay': 'True'}), + #mock.call(('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1']}), + mock.call('DHCPV4_RELAY', 'Vlan1000', {'dhcpv4_servers': ['192.0.0.1']}) + ] + db.cfgdb.set_entry.assert_has_calls(expected_calls) def test_config_vlan_add_del_dhcpv6_relay_dest(self, mock_cfgdb): runner = CliRunner() diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/test_dhcpv4_relay.py b/dockers/docker-dhcp-relay/cli-plugin-tests/test_dhcpv4_relay.py new file mode 100644 index 00000000000..3d0b4d043bf --- /dev/null +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_dhcpv4_relay.py @@ -0,0 +1,589 @@ +import os +import sys +import traceback +from unittest import mock + +from click.testing import CliRunner + +from utilities_common.db import Db +import pytest +sys.path.append('/usr/local/lib/python3.11/dist-packages/config/plugins') +import dhcp_relay + +config_dhcpv4_relay_add_output = """\ +Added DHCPv4 Servers as 3.3.3.3 to Vlan200 +""" + +config_dhcpv4_relay_update_output = """\ +Updated DHCPv4 Servers as 4.4.4.4 to Vlan200 +""" + +config_dhcpv4_relay_del_output = """\ +Removed DHCPv4 relay configuration for Vlan200 +""" + +config_dhcpv4_relay_add_multiple_ips_output = """\ +Added DHCPv4 Servers as {initial_servers} to Vlan200 +""" + +config_dhcpv4_relay_update_multiple_ips_output = """\ +Updated DHCPv4 Servers as {updated_servers} to Vlan200 +""" + +config_dhcpv4_relay_partial_del_output = """\ +Removed DHCPv4 relay configuration from Vlan200 for Servers [{partial_servers}] +""" + +config_dhcpv4_relay_add_source_interface_output = """\ +Added DHCPv4 Servers as 3.3.3.3, Source Interface as {interface} to Vlan200 +""" + +config_dhcpv4_relay_update_source_interface_output = """\ +Updated Source Interface as {updated_interface} to Vlan200 +""" + +config_dhcpv4_relay_add_server_vrf_output = """\ +Added DHCPv4 Servers as 3.3.3.3, Server VRF as VrfRED, link_selection as enable, vrf_selection as enable, server_id_override as enable to Vlan200 +""" + +config_dhcpv4_relay_update_server_vrf_output = """\ +Updated DHCPv4 Servers as 4.4.4.4, Server VRF as VrfBLUE to Vlan200 +""" + +config_dhcpv4_relay_add_server_vrf_default_output = """\ +Added DHCPv4 Servers as 3.3.3.3, Server VRF as default, link_selection as enable, vrf_selection as enable, server_id_override as enable to Vlan200 +""" + +config_dhcpv4_relay_update_server_vrf_default_output = """\ +Updated Server VRF as {server_vrf} to Vlan200 +""" + +config_dhcpv4_relay_en_selection_flag_output = """\ +Added DHCPv4 Servers as 3.3.3.3, {flag} as {value} to {vlan} +""" + +config_dhcpv4_relay_upd_selection_flag_output = """\ +Updated {flag} as {value} to {vlan} +""" + +config_dhcpv4_relay_add_agent_relay_mode_output = """\ +Added DHCPv4 Servers as 3.3.3.3, Agent Relay Mode as {mode} to Vlan200 +""" + +config_dhcpv4_relay_update_agent_relay_mode_output = """\ +Updated Agent Relay Mode as {mode} to Vlan200 +""" + +config_dhcpv4_relay_add_max_hop_count_output = """\ +Added DHCPv4 Servers as 3.3.3.3, Max Hop Count as 1 to Vlan200 +""" + +config_dhcpv4_relay_update_max_hop_count_output = """\ +Updated Max Hop Count as {count} to Vlan200 +""" + +config_dhcpv4_relay_add_all_option_output = """\ +Added DHCPv4 Servers as 3.3.3.3, Source Interface as Ethernet8, Server VRF as VrfRED, \ +link_selection as enable, vrf_selection as enable, server_id_override as enable, \ +Agent Relay Mode as discard, Max Hop Count as 8 to Vlan200 +""" + +class TestConfigDhcpv4Relay(object): + def test_plugin_registration(self): + cli = mock.MagicMock() + dhcp_relay.register(cli) + cli.commands['dhcpv4_relay'].add_command(dhcp_relay.dhcpv4_relay) + + def test_config_dhcpv4_relay_del_nonexistent_relay(self, mock_cfgdb): + """Deleting Non Existent Vlan from DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + mock_cfgdb.get_entry.return_value = {} + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan300"], obj=db) + assert result.exit_code != 0 + assert "Error: DHCPv4 relay configuration not found for Vlan Vlan300" in result.output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_update_nonexistent_vlan(self, mock_cfgdb): + """Updating DHCPv4 Relay config for a non-existent VLAN""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["update"], + ["--dhcpv4-servers", "1.1.1.1", "--source-interface", "Ethernet4", "Vlan786"], + obj=db + ) + assert result.exit_code != 0 + assert "Error: Vlan Vlan786 does not exist in the configDB" in result.output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_invalid_source_interface(self, mock_cfgdb): + """Validating error when source interface is not a valid Ethernet, PortChannel, or Loopback interface""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["add"], + [ + "--dhcpv4-servers", "3.3.3.3", + "--source-interface", "InvalidIntf123", + "Vlan200" + ], + obj=db + ) + assert result.exit_code != 0 + assert "Error: InvalidIntf123 is not a valid Ethernet, PortChannel, or Loopback interface." in result.output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_invalid_server_vrf(self, mock_cfgdb): + """Adding a Nonexistent VRF to DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["add"], + [ + "--dhcpv4-servers", "3.3.3.3", + "--link-selection", "enable", + "--vrf-selection", "enable", + "--server-id-override", "enable", + "--server-vrf", "Vrf99", + "Vlan200" + ], + obj=db + ) + assert result.exit_code != 0 + assert "Error: VRF Vrf99 does not exist in the VRF table." in result.output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_basic_relay(self, mock_cfgdb): + """Validating dhcpv4-servers in DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", "3.3.3.3", "Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_output + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + ["--dhcpv4-servers", "4.4.4.4", "Vlan200"], obj=db) + assert result.output == config_dhcpv4_relay_update_output + assert result.exit_code == 0 + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_multiple_servers_relay(self, mock_cfgdb): + """Validating multiple dhcpv4-servers in DHCPv4 Relay Config (Add, Update, Delete)""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + initial_servers = "1.1.1.1,2.2.2.2,3.3.3.3,4.4.4.4,5.5.5.5" + updated_servers = "6.6.6.6,7.7.7.7,8.8.8.8,9.9.9.9,10.10.10.10" + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Adding multiple dhcpv4-servers to dhcpv4 relay via 'initial_servers' + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", initial_servers, "Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_multiple_ips_output.format(initial_servers=initial_servers) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Update the dhcpv4-servers with another set of IPs via 'updated_servers' + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + ["--dhcpv4-servers", updated_servers, "Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_multiple_ips_output.format(updated_servers=updated_servers) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_multiple_servers_delete(self, mock_cfgdb): + """Validate partial deletion and complete cleanup of multiple DHCPv4 servers""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + initial_servers = "100.100.100.1,100.100.100.2,100.100.100.3,100.100.100.4,100.100.100.5" + delete_1 = "100.100.100.4,100.100.100.5" + delete_2 = "100.100.100.1,100.100.100.2,100.100.100.3" + + # Add multiple DHCPv4 servers + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", initial_servers, "Vlan200"], obj=db) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_multiple_ips_output.format(initial_servers=initial_servers) + assert mock_run_command.call_count == 0 + # Partially delete some of the servers + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + mock_cfgdb.get_entry.return_value = {"dhcpv4_servers": initial_servers} + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], + ["--dhcpv4-servers", delete_1, "Vlan200"], obj=db) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_partial_del_output.format(partial_servers=delete_1) + assert mock_run_command.call_count == 0 + # Delete the remaining servers, should clean up the full entry + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], + ["--dhcpv4-servers", delete_2, "Vlan200"], obj=db) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + + def test_config_dhcpv4_relay_source_interface(self, mock_cfgdb): + """Validating source interfaces in DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + interfaces = [ + ("Loopback2", "Loopback3"), + ("Ethernet8", "Ethernet12"), + ("PortChannel5", "PortChannel6") + ] + + for interface, updated_interface in interfaces: + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", "3.3.3.3", "--source-interface", interface, "Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_source_interface_output.format(interface=interface) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + ["--source-interface", updated_interface, "Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_source_interface_output.format(updated_interface=updated_interface) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + + def test_config_dhcpv4_relay_server_vrf(self, mock_cfgdb): + """Validating server vrf in DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["add"], + [ + "--dhcpv4-servers", "3.3.3.3", + "--link-selection", "enable", + "--vrf-selection", "enable", + "--server-id-override", "enable", + "--server-vrf", "VrfRED", + "Vlan200" + ], + obj=db + ) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_server_vrf_output + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + [ + "--dhcpv4-servers", "4.4.4.4", + "--server-vrf", "VrfBLUE", + "Vlan200" + ], + obj=db + ) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_server_vrf_output + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_default_server_vrf(self, mock_cfgdb): + """Validating default server VRF addition, update to VrfRED, update back to default, then delete""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Add with default VRF + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["add"], + [ + "--dhcpv4-servers", "3.3.3.3", + "--link-selection", "enable", + "--vrf-selection", "enable", + "--server-id-override", "enable", + "--server-vrf", "default", + "Vlan200" + ], + obj=db + ) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_server_vrf_default_output + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Update to VrfRED + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["update"], + [ + "--server-vrf", "VrfRED", + "Vlan200" + ], + obj=db + ) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_server_vrf_default_output.format(server_vrf="VrfRED") + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Update back to default + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["update"], + [ + "--server-vrf", "default", + "Vlan200" + ], + obj=db + ) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_server_vrf_default_output.format(server_vrf="default") + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + # Final delete + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["del"], + ["Vlan200"], + obj=db + ) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + + def test_config_dhcpv4_relay_server_vrf_link_selection_disabled(self, mock_cfgdb): + """Test error when --link-selection is disabled but --server-vrf is passed""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke( + dhcp_relay.dhcpv4_relay.commands["add"], + [ + "--dhcpv4-servers", "3.3.3.3", + "--link-selection", "disable", + "--vrf-selection", "enable", + "--server-id-override", "enable", + "--server-vrf", "VrfRED", + "Vlan200" + ], + obj=db + ) + assert result.exit_code != 0 + assert "Error: server-vrf requires link-selection, vrf-selection and server-id-override flags to be enabled." in result.output + assert mock_run_command.call_count == 0 + + db.cfgdb.ser_entry.reset_mock() + + def test_config_dhcpv4_relay_selection_flags(self, mock_cfgdb): + """Validating selection flags in DHCPv4 relay configs""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + flag_map = { + "link-selection": "link_selection", + "vrf-selection": "vrf_selection", + "server-id-override": "server_id_override" + } + for flag, config_key in flag_map.items(): + vlan, add_val, update_val = ("Vlan200", "enable", "disable") + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", "3.3.3.3", f"--{flag}", add_val, vlan], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_en_selection_flag_output.format(flag=config_key, value=add_val, vlan=vlan) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + [f"--{flag}", update_val, vlan], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_upd_selection_flag_output.format(flag=config_key, value=update_val, vlan=vlan) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], [vlan], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_agent_relay_mode(self, mock_cfgdb): + """Validating Agent Relay Modes in DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + agent_relay_modes = [ + "discard", + "append", + "replace" + ] + + for add_mode in agent_relay_modes: + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", "3.3.3.3", "--agent-relay-mode", add_mode, "Vlan200"], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_agent_relay_mode_output.format(mode=add_mode) + assert mock_run_command.call_count == 0 + + update_modes = [mode for mode in agent_relay_modes if mode != add_mode] + + for update_mode in update_modes: + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + ["--agent-relay-mode", update_mode, "Vlan200"], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_agent_relay_mode_output.format(mode=update_mode) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_max_hop_count(self, mock_cfgdb): + """Validating Max Hop Count in DHCPv4 Relay Config""" + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + #For add case, testing the min value + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", "3.3.3.3", "--max-hop-count", '1', "Vlan200"], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_max_hop_count_output + assert mock_run_command.call_count == 0 + + # For update case, testing default, mid and max values + max_hop_counts = ["4", "8", "16"] + for count in max_hop_counts: + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["update"], + ["--max-hop-count", count, "Vlan200"], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_update_max_hop_count_output.format(count=count) + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() + + def test_config_dhcpv4_relay_add_and_delete(self, mock_cfgdb): + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["add"], + ["--dhcpv4-servers", "3.3.3.3", + "--vrf-selection", "enable", + "--server-id-override", "enable", + "--source-interface", "Ethernet8", + "--link-selection", "enable", + "--agent-relay-mode", "discard", + "--max-hop-count", "8", + "--server-vrf", "VrfRED", + "Vlan200"], + obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_add_all_option_output + assert mock_run_command.call_count == 0 + + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.dhcpv4_relay.commands["del"], ["Vlan200"], obj=db) + assert result.exit_code == 0 + assert result.output == config_dhcpv4_relay_del_output + assert mock_run_command.call_count == 0 + + db.cfgdb.set_entry.reset_mock() diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp4relay_counters.py b/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp4relay_counters.py index 576e40fbbf4..e7e45524837 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp4relay_counters.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp4relay_counters.py @@ -12,6 +12,34 @@ "Unknown", "Discover", "Offer", "Request", "Decline", "Ack", "Nak", "Release", "Inform", "Bootp" ] SUPPORTED_DIR = ["TX", "RX"] +expected_counts_v4 = """\ +Packet type Abbr: Un - Unknown, Dis - Discover, Off - Offer, Req - Request, + Ack - Acknowledge, Nack - NegativeAcknowledge, Rel - Release, + Inf - Inform, Dec - Decline, Mal - Malformed, Drp - Dropped + ++-------------+------+-------+-------+-------+-------+--------+-------+-------+-------+-------+-------+ +| Vlan (TX) | Un | Dis | Off | Req | Ack | Nack | Rel | Inf | Dec | Mal | Drp | ++=============+======+=======+=======+=======+=======+========+=======+=======+=======+=======+=======+ ++-------------+------+-------+-------+-------+-------+--------+-------+-------+-------+-------+-------+ +""" +expected_counts_v4_rx = """\ +Packet type Abbr: Un - Unknown, Dis - Discover, Off - Offer, Req - Request, + Ack - Acknowledge, Nack - NegativeAcknowledge, Rel - Release, + Inf - Inform, Dec - Decline, Mal - Malformed, Drp - Dropped + ++-------------+------+-------+-------+-------+-------+--------+-------+-------+-------+-------+-------+ +| Vlan (RX) | Un | Dis | Off | Req | Ack | Nack | Rel | Inf | Dec | Mal | Drp | ++=============+======+=======+=======+=======+=======+========+=======+=======+=======+=======+=======+ ++-------------+------+-------+-------+-------+-------+--------+-------+-------+-------+-------+-------+ +""" + +expected_counts_v4_type = """\ ++-------------------+------+------+ +| Vlan (Discover) | TX | RX | ++===================+======+======+ ++-------------------+------+------+ +""" + def test_plugin_registration(): @@ -378,3 +406,41 @@ def test_dhcp_relay_ip4counters_incorrect_dir(args): runner = CliRunner() result = runner.invoke(show_dhcp_relay.dhcp_relay.commands["ipv4"].commands["counters"], args) assert result.exit_code != 0 + +class TestDhcpRelayCounters(object): + + def test_show_vlan_counts(self): + runner = CliRunner() + result = runner.invoke(show_dhcp_relay.dhcp4relay_counters.commands["vlan-counts"], ["Vlan1000"]) + print(result.output) + assert result.output == expected_counts_v4 + + def test_show_vlan_counts_dir(self): + runner = CliRunner() + result = runner.invoke(show_dhcp_relay.dhcp4relay_counters.commands["vlan-counts"], ["Vlan1000", "-d", "RX"]) + print(result.output) + assert result.output == expected_counts_v4_rx + + def test_show_vlan_counts_type(self): + runner = CliRunner() + result = runner.invoke(show_dhcp_relay.dhcp4relay_counters.commands["vlan-counts"], ["Vlan1000", "-t", "Discover"]) + print(result.output) + assert result.output == expected_counts_v4_type + + def test_show_ipv4_vlan_counters(self): + runner = CliRunner() + result = runner.invoke(show_dhcp_relay.dhcp_relay_ipv4.commands["vlan-counters"], ["Vlan1000"]) + print(result.output) + assert result.output == expected_counts_v4 + + def test_show_ipv4_vlan_counters_dir(self): + runner = CliRunner() + result = runner.invoke(show_dhcp_relay.dhcp_relay_ipv4.commands["vlan-counters"], ["Vlan1000", "-d", "RX"]) + print(result.output) + assert result.output == expected_counts_v4_rx + + def test_show_vlan_counts_type(self): + runner = CliRunner() + result = runner.invoke(show_dhcp_relay.dhcp_relay_ipv4.commands["vlan-counters"], ["Vlan1000", "-t", "Discover"]) + print(result.output) + assert result.output == expected_counts_v4_type diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp_relay.py b/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp_relay.py index de55af82d9d..7bef2f07604 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp_relay.py @@ -75,6 +75,18 @@ +-------------+----------------------+ """ +expected_dhcpv4_table_multi_with_header = """\ ++-------------+----------------------+--------------+--------------------+------------------+-----------------+----------------------+--------------------+-----------------+ +| Interface | DHCP Relay Address | Server Vrf | Source Interface | Link Selection | VRF Selection | Server ID Override | Agent Relay Mode | Max Hop Count | ++=============+======================+==============+====================+==================+=================+======================+====================+=================+ +| Vlan1000 | 192.0.0.1 | default | Ethernet112 | enable | enable | N/A | N/A | N/A | +| | 192.0.0.2 | | | | | | | | ++-------------+----------------------+--------------+--------------------+------------------+-----------------+----------------------+--------------------+-----------------+ +| Vlan1001 | 192.0.0.3 | N/A | N/A | N/A | N/A | N/A | discard | 5 | +| | 192.0.0.4 | | | | | | | | ++-------------+----------------------+--------------+--------------------+------------------+-----------------+----------------------+--------------------+-----------------+ +""" + DBCONFIG_PATH = '/var/run/redis/sonic-db/database_config.json' IP_VER_TEST_PARAM_MAP = { @@ -82,6 +94,11 @@ "entry": "dhcp_servers", "table": "VLAN" }, + "ipv4_dhcp": { + "entry": ["dhcpv4_servers", "server_vrf", "source_interface", "link_selection", + "vrf_selection", "server_id_override", "agent_relay_mode", "max_hop_count"], + "table": "DHCPV4_RELAY" + }, "ipv6": { "entry": "dhcpv6_servers", "table": "DHCP_RELAY" @@ -165,13 +182,29 @@ def test_show_multi_dhcp_relay(test_name, test_data, fs): config_db = MockConfigDb() ip_version = test_name table = config_db.get_table(IP_VER_TEST_PARAM_MAP[ip_version]["table"]) - result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"]) + + if ip_version == "ipv4_dhcp": + config_db.set_entry("DEVICE_METADATA", "localhost", {"has_sonic_dhcpv4_relay" : "True"}) + result = show.get_dhcpv4_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"]) + else: + result = show.get_dhcp_relay_data_with_header(table, IP_VER_TEST_PARAM_MAP[ip_version]["entry"]) if ip_version == "ipv4": expected_output = expected_ipv4_table_multi_with_header + elif ip_version == "ipv4_dhcp": + expected_output = expected_dhcpv4_table_multi_with_header else: expected_output = expected_ipv6_table_multi_with_header assert result == expected_output +def test_show_dhcp_relay_ipv4_counter_with_enabled_dhcp_server(): + with mock.patch.object(show, "is_dhcp_server_enabled", return_value=True), \ + mock.patch.object(swsscommon.ConfigDBConnector, "connect", return_value=None), \ + mock.patch.object(swsscommon.ConfigDBConnector, "get_table", return_value=None), \ + mock.patch.object(click, "echo", return_value=None) as mock_echo: + show.ipv4_counters("TX", "Discover", "Vlan1000") + expected_param = "Unsupport to check dhcp_relay ipv4 counter when dhcp_server feature is enabled" + mock_echo.assert_called_once_with(expected_param) + @pytest.mark.parametrize("enable_dhcp_server", [True, False]) def test_is_dhcp_server_enabled(enable_dhcp_server): diff --git a/dockers/docker-dhcp-relay/cli/clear/plugins/clear_dhcp_relay.py b/dockers/docker-dhcp-relay/cli/clear/plugins/clear_dhcp_relay.py index 2e753e6926f..e41385d6aca 100644 --- a/dockers/docker-dhcp-relay/cli/clear/plugins/clear_dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli/clear/plugins/clear_dhcp_relay.py @@ -8,7 +8,7 @@ import psutil import time import re -dhcp6_relay = importlib.import_module('show.plugins.dhcp-relay') +dhcprelay = importlib.import_module('show.plugins.dhcp-relay') import utilities_common.cli as clicommon @@ -25,7 +25,7 @@ def clear_dhcp_relay_ipv6_counter(interface): - counter = dhcp6_relay.DHCPv6_Counter() + counter = dhcprelay.DHCPv6_Counter() counter_intf = counter.get_interface() if interface: counter.clear_table(interface) @@ -33,6 +33,9 @@ def clear_dhcp_relay_ipv6_counter(interface): for intf in counter_intf: counter.clear_table(intf) +def clear_dhcp_relay_ipv4_vlan_counter(direction, pkt_type, interface): + counter = dhcprelay.DHCPv4_Counter() + counter.clear_table(direction, pkt_type, interface) def is_vlan_interface_valid(vlan_interface, db): # If not vlan interface specified, treat it as valid @@ -166,6 +169,18 @@ def clear_dhcp_relay_ipv6_counters(interface): """ Clear dhcp_relay ipv6 message counts """ clear_dhcp_relay_ipv6_counter(interface) +# sonic-clear dhcp6relay_counters +@click.group(cls=clicommon.AliasedGroup) +def dhcp4relay_clear(): + pass + +@dhcp4relay_clear.command('dhcp4relay-vlan-counters') +@click.option('-d', '--direction', required=False, type=click.Choice(['TX', 'RX']), help="Specify TX(egress) or RX(ingress)") +@click.option('-t', '--type', required=False, type=click.Choice(dhcprelay.dhcpv4_messages), help="Specify DHCP packet counter type") +@click.argument("vlan_interface", required=False) +def dhcp4relay_clear_vlan_counters(direction, type, vlan_interface): + """ Clear dhcp_relay ipv4 message counts """ + clear_dhcp_relay_ipv4_vlan_counter(direction, type, vlan_interface) @dhcp_relay.group(cls=clicommon.AliasedGroup, name="ipv4") def dhcp_relay_ipv4(): @@ -211,9 +226,11 @@ def clear_dhcp_relay_ipv4_counters(db, interface, dir, type): def register(cli): cli.add_command(dhcp6relay_clear_counters) + cli.add_command(dhcp4relay_clear_vlan_counters) cli.add_command(dhcp_relay) if __name__ == '__main__': dhcp6relay_clear_counters() + dhcp4relay_clear_vlan_counters() dhcp_relay() diff --git a/dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py b/dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py index b4cecf13b71..0c9cdd5f93a 100644 --- a/dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py @@ -9,8 +9,16 @@ VLAN_TABLE = "VLAN" DHCPV4_SERVERS = "dhcp_servers" IPV4 = 4 +DHCPV4_RELAY_TBL_SERVERS = "dhcpv4_servers" +DHCPV4_RELAY_TABLE = "DHCPV4_RELAY" +def check_sonic_dhcpv4_relay_flag(db): + table = db.cfgdb.get_entry("DEVICE_METADATA", "localhost") + if('has_sonic_dhcpv4_relay' in table and table['has_sonic_dhcpv4_relay'] == 'True'): + return True + return False + def validate_ips(ctx, ips, ip_version): for ip in ips: try: @@ -24,9 +32,15 @@ def validate_ips(ctx, ips, ip_version): def get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str, check_is_exist=True): if check_is_exist: - keys = db.cfgdb.get_keys(table_name) + # Check if vlan is created in VLAN_TABLE + keys = db.cfgdb.get_keys(VLAN_TABLE) if vlan_name not in keys: ctx.fail("{} doesn't exist".format(vlan_name)) + # Check if dhcp relay configs exist for a vlan + if VLAN_TABLE != table_name: + keys = db.cfgdb.get_keys(table_name) + if vlan_name not in keys: + return [],{} table = db.cfgdb.get_entry(table_name, vlan_name) dhcp_servers = table.get(dhcp_servers_str, []) @@ -34,10 +48,13 @@ def get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str, check_is_ return dhcp_servers, table -def restart_dhcp_relay_service(): +def restart_dhcp_relay_service(db, ip_version): """ Restart dhcp_relay service """ + if(ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db)): + # if 'has_sonic_dhcpv4_relay' flag is present in DEVICE_METADATA['localhost'] and is 'true' + return click.echo("Restarting DHCP relay service...") clicommon.run_command(['systemctl', 'stop', 'dhcp_relay'], display_cmd=False) clicommon.run_command(['systemctl', 'reset-failed', 'dhcp_relay'], display_cmd=False) @@ -48,33 +65,47 @@ def add_dhcp_relay(vid, dhcp_relay_ips, db, ip_version): table_name = DHCP_RELAY_TABLE if ip_version == 6 else VLAN_TABLE dhcp_servers_str = DHCPV6_SERVERS if ip_version == 6 else DHCPV4_SERVERS vlan_name = "Vlan{}".format(vid) + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + dhcp_table_name = DHCPV4_RELAY_TABLE + dhcpv4_servers_str = DHCPV4_RELAY_TBL_SERVERS ctx = click.get_current_context() # Verify ip addresses are valid validate_ips(ctx, dhcp_relay_ips, ip_version) # It's unnecessary for DHCPv6 Relay to verify entry exist check_config_exist = True if ip_version == 4 else False - dhcp_servers, table = get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str, check_config_exist) + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + dhcpv4_servers, v4_table = get_dhcp_servers(db, vlan_name, ctx, dhcp_table_name, dhcpv4_servers_str, check_config_exist) + else: + dhcp_servers, table = get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str, check_config_exist) added_ips = [] for dhcp_relay_ip in dhcp_relay_ips: # Verify ip addresses not duplicate in add list if dhcp_relay_ip in added_ips: - ctx.fail("Error: Find duplicate DHCP relay ip {} in add list".format(dhcp_relay_ip)) - # Verify ip addresses not exist in DB - if dhcp_relay_ip in dhcp_servers: - click.echo("{} is already a DHCP relay for {}".format(dhcp_relay_ip, vlan_name)) - return + ctx.fail("Find duplicate DHCP relay ip {} in add list".format(dhcp_relay_ip)) - dhcp_servers.append(dhcp_relay_ip) + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + if dhcp_relay_ip in dhcpv4_servers: + ctx.fail("{} is already a DHCPv4 relay for {}".format(dhcp_relay_ip, vlan_name)) + dhcpv4_servers.append(dhcp_relay_ip) + else: + if dhcp_relay_ip in dhcp_servers: + ctx.fail("{} is already a DHCP relay for {}".format(dhcp_relay_ip, vlan_name)) + dhcp_servers.append(dhcp_relay_ip) added_ips.append(dhcp_relay_ip) - table[dhcp_servers_str] = dhcp_servers + # for IPv4, we will add same entry to DHCPV4_RELAY table also + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + v4_table[dhcpv4_servers_str] = dhcpv4_servers + db.cfgdb.set_entry(dhcp_table_name, vlan_name, v4_table) + else: + table[dhcp_servers_str] = dhcp_servers + db.cfgdb.set_entry(table_name, vlan_name, table) - db.cfgdb.set_entry(table_name, vlan_name, table) click.echo("Added DHCP relay address [{}] to {}".format(",".join(dhcp_relay_ips), vlan_name)) try: - restart_dhcp_relay_service() + restart_dhcp_relay_service(db, ip_version) except SystemExit as e: ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) @@ -83,35 +114,56 @@ def del_dhcp_relay(vid, dhcp_relay_ips, db, ip_version): table_name = DHCP_RELAY_TABLE if ip_version == 6 else VLAN_TABLE dhcp_servers_str = DHCPV6_SERVERS if ip_version == 6 else DHCPV4_SERVERS vlan_name = "Vlan{}".format(vid) + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + dhcp_table_name = DHCPV4_RELAY_TABLE + dhcpv4_servers_str = DHCPV4_RELAY_TBL_SERVERS ctx = click.get_current_context() # Verify ip addresses are valid validate_ips(ctx, dhcp_relay_ips, ip_version) - dhcp_servers, table = get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str) + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + dhcpv4_servers, v4_table = get_dhcp_servers(db, vlan_name, ctx, dhcp_table_name, dhcpv4_servers_str) + else: + dhcp_servers, table = get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str) removed_ips = [] for dhcp_relay_ip in dhcp_relay_ips: # Verify ip addresses not duplicate in del list if dhcp_relay_ip in removed_ips: - ctx.fail("Error: Find duplicate DHCP relay ip {} in del list".format(dhcp_relay_ip)) - # Remove dhcp servers if they exist in the DB - if dhcp_relay_ip not in dhcp_servers: - ctx.fail("{} is not a DHCP relay for {}".format(dhcp_relay_ip, vlan_name)) - - dhcp_servers.remove(dhcp_relay_ip) - removed_ips.append(dhcp_relay_ip) + ctx.fail("Find duplicate DHCP relay ip {} in del list".format(dhcp_relay_ip)) - if len(dhcp_servers) == 0: - del table[dhcp_servers_str] + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + if dhcp_relay_ip in dhcpv4_servers: + dhcpv4_servers.remove(dhcp_relay_ip) + removed_ips.append(dhcp_relay_ip) + else: + ctx.fail("{} is not a DHCPv4 relay for {}".format(dhcp_relay_ip, vlan_name)) + else: + if dhcp_relay_ip in dhcp_servers: + dhcp_servers.remove(dhcp_relay_ip) + removed_ips.append(dhcp_relay_ip) + else: + ctx.fail("{} is not a DHCP relay for {}".format(dhcp_relay_ip, vlan_name)) + + # for IPv4, we will remove same entry from DHCPV4_RELAY table also + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + if len(dhcpv4_servers) == 0: + db.cfgdb.set_entry(dhcp_table_name, vlan_name, None) + else: + v4_table[dhcpv4_servers_str] = dhcpv4_servers + db.cfgdb.set_entry(dhcp_table_name, vlan_name, v4_table) else: - table[dhcp_servers_str] = dhcp_servers + if len(dhcp_servers) == 0: + del table[dhcp_servers_str] + else: + table[dhcp_servers_str] = dhcp_servers + db.cfgdb.set_entry(table_name, vlan_name, table) if ip_version == 6 and len(table.keys()) == 0: - table = None + db.cfgdb.set_entry(table_name, vlan_name, None) - db.cfgdb.set_entry(table_name, vlan_name, table) click.echo("Removed DHCP relay address [{}] from {}".format(",".join(dhcp_relay_ips), vlan_name)) try: - restart_dhcp_relay_service() + restart_dhcp_relay_service(db, ip_version) except SystemExit as e: ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) @@ -120,6 +172,301 @@ def is_dhcp_server_enabled(db): dhcp_server_feature_entry = db.cfgdb.get_entry("FEATURE", "dhcp_server") return "state" in dhcp_server_feature_entry and dhcp_server_feature_entry["state"] == "enabled" +@click.group(cls=clicommon.AbbreviationGroup, name="dhcpv4_relay") +def dhcpv4_relay(): + pass + +def validate_vrf_exists(db, vrf_name): + """Check if the given VRF exists in the ConfigDB""" + keys = db.cfgdb.get_keys("VRF") + if vrf_name in keys: + return True + return False + +def validate_vlan_exists(db, vlan_name): + """Check if the given Vlan exists in the ConfigDB""" + keys = db.cfgdb.get_keys("VLAN") + if vlan_name in keys: + return True + return False + +def validate_source_interface(vlan_name, source_interface, db): + config_db = db.cfgdb + ctx = click.get_current_context() + + interface_mapping = { + "Ethernet" : "INTERFACE", + "PortChannel" : "PORTCHANNEL_INTERFACE", + "Loopback" : "LOOPBACK_INTERFACE" + } + interface_table = None + for prefix, table in interface_mapping.items(): + if source_interface.startswith(prefix): + interface_table = table + break + + if not interface_table: + ctx.fail(f"{source_interface} is not a valid Ethernet, PortChannel, or Loopback interface.") + return False + + # Check if the interface exists in the corresponding table + interface_entries = config_db.get_table(interface_table) + if source_interface not in interface_entries: + ctx.fail(f"Interface {source_interface} not found in {interface_table} table. Please configure valid interface.") + return False + return True + +@dhcpv4_relay.command("update") +@click.option("--dhcpv4-servers", required=False, help="List of DHCPv4 relay servers to update") +@click.option("--source-interface", required=False, help="Source interface for DHCPv4 relay") +@click.option("--link-selection", required=False, type=click.Choice(["enable", "disable"]), help="Link selection flag for DHCPv4 Relay") +@click.option("--vrf-selection", required=False, type=click.Choice(["enable", "disable"]), help="VRF selection flag for DHCPv4 relay") +@click.option("--server-id-override", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable server ID override for DHCPv4 relay") +@click.option("--server-vrf", required=False, help="Server VRF name for DHCPv4 relay") +@click.option("--agent-relay-mode", required=False, type=click.Choice(["discard", "append", "replace"]), + help="Set agent relay mode for DHCPv4 relay") +@click.option("--max-hop-count", required=False, type=int, help="Maximum hop count for DHCPv4 relay") +@click.argument("vlan_name", metavar="", required=True) +@clicommon.pass_db +def update_dhcpv4_relay(db, vlan_name, dhcpv4_servers, source_interface, link_selection, + vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count): + """Update an existing DHCPv4 relay entry for a VLAN""" + config_db = db.cfgdb + ctx = click.get_current_context() + + # Check if VLAN exists + if not validate_vlan_exists(db, vlan_name): + ctx.fail(f"Error: Vlan {vlan_name} does not exist in the configDB".format(vlan_name)) + existing_entry = config_db.get_entry(DHCPV4_RELAY_TABLE, vlan_name) + if not existing_entry: + ctx.fail(f"Error: No existing DHCPv4 relay configuration found for {vlan_name}") + + updated_entry = existing_entry.copy() + updated_fields = [] + + # validating dhcpv4_servers: + if dhcpv4_servers: + new_servers = [ip.strip() for ip in dhcpv4_servers.split(",") if ip.strip()] + validate_ips(ctx, new_servers, ip_version=4) + + existing_servers = existing_entry.get("dhcpv4_servers", "") + if isinstance(existing_servers, str): + existing_servers = existing_servers.split(",") + combined_servers = existing_servers[:] + for ip in new_servers: + if ip not in combined_servers: + combined_servers.append(ip) + updated_entry["dhcpv4_servers"] = combined_servers + updated_fields.append(f"DHCPv4 Servers as {dhcpv4_servers}") + + if source_interface and existing_entry.get("source_interface") != source_interface: + if not validate_source_interface(vlan_name, source_interface, db): + ctx.fail(f"Invalid source interface {source_interface}") + updated_entry["source_interface"] = source_interface + updated_fields.append(f"Source Interface as {source_interface}") + + flags_dict = { + "link_selection": link_selection, + "vrf_selection": vrf_selection, + "server_id_override": server_id_override, + } + if existing_entry.get("server_vrf"): + for flag in ["link_selection", "vrf_selection", "server_id_override"]: + if flags_dict[flag] == 'disable': + ctx.fail(f"{flag} cannot be disabled when server-vrf is configured.") + + for flag in ["link_selection", "vrf_selection", "server_id_override"]: + value = flags_dict[flag] + if value and existing_entry.get(flag) != value: + updated_entry[flag] = value + updated_fields.append(f"{flag} as {value}") + + if server_vrf and existing_entry.get("server_vrf") != server_vrf: + if server_vrf != "default" and not validate_vrf_exists(db, server_vrf): + ctx.fail(f"VRF {server_vrf} does not exist in the VRF table.") + + for flag in ["link_selection", "vrf_selection", "server_id_override"]: + if updated_entry.get(flag, existing_entry.get(flag)) != "enable": + ctx.fail("server-vrf requires link-selection, vrf-selection and server-id-override flags to be enabled.") + + updated_entry["server_vrf"] = server_vrf + updated_fields.append(f"Server VRF as {server_vrf}") + + if agent_relay_mode and existing_entry.get("agent_relay_mode") != agent_relay_mode: + updated_entry["agent_relay_mode"] = agent_relay_mode + updated_fields.append(f"Agent Relay Mode as {agent_relay_mode}") + + if max_hop_count: + if not (1 <= max_hop_count <= 16): + ctx.fail("max-hop-count must be between 1 to 16") + updated_entry["max_hop_count"] = max_hop_count + updated_fields.append(f"Max Hop Count as {max_hop_count}") + + # Apply updated entry to the database + config_db.set_entry(DHCPV4_RELAY_TABLE, vlan_name, updated_entry) + + if updated_fields: + click.echo(f"Updated {', '.join(updated_fields)} to {vlan_name}") + else: + click.echo(f"No changes made to {vlan_name}") + + +@dhcpv4_relay.command("add") +@click.option("--dhcpv4-servers", required=True, help="List of DHCPv4 relay servers") +@click.option("--source-interface", required=False, help="Source interface for DHCPv4 relay") +@click.option("--link-selection", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable link selection for DHCPv4 relay") +@click.option("--vrf-selection", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable VRF selection for DHCPv4 relay") +@click.option("--server-id-override", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable server ID override for DHCPv4 relay") +@click.option("--server-vrf", required=False, help="Server VRF name for DHCPv4 relay") +@click.option("--agent-relay-mode", required=False, type=click.Choice(["discard", "append", "replace"]), + help="Set agent relay mode for DHCPv4 relay") +@click.option("--max-hop-count", required=False, type=int, help="Maximum hop count for DHCPv4 relay") +@click.argument("vlan_name", metavar="", required=True) +@clicommon.pass_db +def add_dhcpv4_relay(db, dhcpv4_servers, vlan_name, source_interface, link_selection, vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count): + """Add DHCPv4 relay configuration for a VLAN""" + config_db = db.cfgdb + ctx = click.get_current_context() + + # Check if VLAN exists + if not validate_vlan_exists(db, vlan_name): + ctx.fail(f"Error: Vlan {vlan_name} does not exist in the configDB".format(vlan_name)) + + vlan_entry = config_db.get_entry(DHCPV4_RELAY_TABLE, vlan_name) + if vlan_entry: + ctx.fail(f"DHCPv4 relay entry for {vlan_name} already exists. Use 'update' instead.") + + relay_entry = {} + added_fields = [] + + # Adding dhcpv4_servers - + new_server_list = list({ip.strip() for ip in dhcpv4_servers.split(",") if ip.strip()}) + validate_ips(ctx, new_server_list, ip_version=4) + relay_entry["dhcpv4_servers"] = new_server_list + added_fields.append(f"DHCPv4 Servers as {','.join(sorted(new_server_list))}") + + if source_interface: + if not validate_source_interface(vlan_name, source_interface, db): + ctx.fail(f"Invalid source interface {source_interface}") + relay_entry["source_interface"] = source_interface + added_fields.append(f"Source Interface as {source_interface}") + + if server_vrf: + if not all([link_selection == "enable", vrf_selection == "enable", server_id_override == "enable"]): + ctx.fail("server-vrf requires link-selection, vrf-selection and server-id-override flags to be enabled.") + + if server_vrf != "default" and not validate_vrf_exists(db, server_vrf): + ctx.fail(f"VRF {server_vrf} does not exist in the VRF table.") + relay_entry["server_vrf"] = server_vrf + added_fields.append(f"Server VRF as {server_vrf}") + + flags_dict = { + "link_selection": link_selection, + "vrf_selection": vrf_selection, + "server_id_override": server_id_override, + } + for flag in ["link_selection", "vrf_selection", "server_id_override"]: + value = flags_dict[flag] + if value: + relay_entry[flag] = value + added_fields.append(f"{flag} as {value}") + + if agent_relay_mode: + relay_entry["agent_relay_mode"] = agent_relay_mode + added_fields.append(f"Agent Relay Mode as {agent_relay_mode}") + + if max_hop_count: + if not (1 <= max_hop_count <= 16): + ctx.fail("max-hop-count must be between 1 to 16") + relay_entry["max_hop_count"] = max_hop_count + added_fields.append(f"Max Hop Count as {max_hop_count}") + + config_db.set_entry(DHCPV4_RELAY_TABLE, vlan_name, relay_entry) + click.echo(f"Added {', '.join(added_fields)} to {vlan_name}") + + +@dhcpv4_relay.command("del") +@click.option("--dhcpv4-servers", required=False, help="Delete list of DHCPv4 servers from DHCPv4 relay") +@click.option("--source-interface", required=False, is_flag=True, help="Delete source interface from DHCPv4 relay") +@click.option("--link-selection", required=False, is_flag=True, help="Delete link selection flag from DHCPv4 relay") +@click.option("--vrf-selection", required=False, is_flag=True, help="Delete VRF selection flag from DHCPv4 relay") +@click.option("--server-id-override", required=False, is_flag=True, help="Delete server ID override flag from DHCPv4 relay") +@click.option("--server-vrf", required=False, is_flag=True, help="Delete server VRF from DHCPv4 relay") +@click.option("--agent-relay-mode", required=False, is_flag=True, help="Delete agent relay mode from DHCPv4 relay") +@click.option("--max-hop-count", required=False, is_flag=True, help="Delete max hop count from DHCPv4 relay") +@click.argument("vlan_name", metavar="", required=True) +@clicommon.pass_db +def del_dhcpv4_relay(db, dhcpv4_servers, source_interface, link_selection, vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count, vlan_name): + """Delete DHCPv4 relay configuration from VLAN""" + ctx = click.get_current_context() + config_db = db.cfgdb + + relay_entry = config_db.get_entry(DHCPV4_RELAY_TABLE, vlan_name) + if not relay_entry: + ctx.fail(f"DHCPv4 relay configuration not found for Vlan {vlan_name}") + + # delete whole dhcpv4-relay configuration incase of vlan deletion + if len([v for k, v in ctx.params.items() if k != "vlan_name" and v]) == 0: + config_db.set_entry(DHCPV4_RELAY_TABLE, vlan_name, None) + click.echo(f"Removed DHCPv4 relay configuration for {vlan_name}") + return + + deleted_fields = [] + if dhcpv4_servers and "dhcpv4_servers" in relay_entry: + existing_servers = relay_entry.get("dhcpv4_servers") + servers_to_delete = [ip.strip() for ip in dhcpv4_servers.split(",") if ip.strip()] + validate_ips(ctx, servers_to_delete, ip_version=4) + + for ip in servers_to_delete: + if ip not in existing_servers: + ctx.fail(f"{ip} is not a configured DHCPv4 server on {vlan_name}") + + updated_servers = [ip for ip in existing_servers if ip not in servers_to_delete] + if not updated_servers: + config_db.set_entry(DHCPV4_RELAY_TABLE, vlan_name, None) + click.echo(f"Removed DHCPv4 relay configuration for {vlan_name}") + return + else: + relay_entry["dhcpv4_servers"] = updated_servers + deleted_fields.append(f"Servers [{dhcpv4_servers}]") + + if source_interface and "source_interface" in relay_entry: + del relay_entry["source_interface"] + deleted_fields.append("Source Interface") + + if server_vrf and "server_vrf" in relay_entry: + del relay_entry["server_vrf"] + deleted_fields.append("Server VRF") + + if "server_vrf" in relay_entry: + if link_selection or vrf_selection or server_id_override: + ctx.fail("Cannot delete link-selection, vrf-selection, or server-id-override when server-vrf is configured") + + if link_selection and "link_selection" in relay_entry: + del relay_entry["link_selection"] + deleted_fields.append("link-selection") + + if vrf_selection and "vrf_selection" in relay_entry: + del relay_entry["vrf_selection"] + deleted_fields.append("vrf-selection") + + if server_id_override and "server_id_override" in relay_entry: + del relay_entry["server_id_override"] + deleted_fields.append("server-id-override") + + if agent_relay_mode and "agent_relay_mode" in relay_entry: + del relay_entry["agent_relay_mode"] + deleted_fields.append("Agent Relay Mode") + + if max_hop_count and "max_hop_count" in relay_entry: + del relay_entry["max_hop_count"] + deleted_fields.append("Max Hop Count") + + config_db.set_entry(DHCPV4_RELAY_TABLE, vlan_name, relay_entry) + + if deleted_fields: + click.echo(f"Removed DHCPv4 relay configuration from {vlan_name} for {', '.join(deleted_fields)}") + @click.group(cls=clicommon.AbbreviationGroup, name="dhcp_relay") def dhcp_relay(): @@ -166,13 +513,52 @@ def dhcp_relay_ipv4_helper(): @dhcp_relay_ipv4_helper.command("add") @click.argument("vid", metavar="", required=True, type=int) @click.argument("dhcp_relay_helpers", nargs=-1, required=True) +@click.option("--source-interface", required=False, help="Source interface for DHCPv4 relay") +@click.option("--link-selection", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable link selection for DHCPv4 relay") +@click.option("--vrf-selection", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable VRF selection for DHCPv4 relay") +@click.option("--server-id-override", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable server ID override for DHCPv4 relay") +@click.option("--server-vrf", required=False, help="Server VRF name for DHCPv4 relay") +@click.option("--agent-relay-mode", required=False, type=click.Choice(["discard", "append", "replace"]), + help="Set agent relay mode for DHCPv4 relay") +@click.option("--max-hop-count", required=False, type=int, help="Maximum hop count for DHCPv4 relay") @clicommon.pass_db -def add_dhcp_relay_ipv4_helper(db, vid, dhcp_relay_helpers): +def add_dhcp_relay_ipv4_helper(db, vid, dhcp_relay_helpers, source_interface, link_selection, vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count): if is_dhcp_server_enabled(db): click.echo("Cannot change ipv4 dhcp_relay configuration when dhcp_server feature is enabled") return + + if check_sonic_dhcpv4_relay_flag(db): + add_dhcpv4_relay.callback(",".join(dhcp_relay_helpers), "Vlan"+str(vid), source_interface, link_selection, vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count) + return + else: + if source_interface or link_selection or vrf_selection or server_id_override or server_vrf or agent_relay_mode or max_hop_count: + click.echo(f"These parameters are applicable for new DHCPv4 Relay feature") + return + add_dhcp_relay(vid, dhcp_relay_helpers, db, IPV4) +@dhcp_relay_ipv4_helper.command("update") +@click.argument("vid", metavar="", required=True, type=int) +@click.argument("dhcp_relay_helpers", nargs=-1, required=True) +@click.option("--source-interface", required=False, help="Source interface for DHCPv4 relay") +@click.option("--link-selection", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable link selection for DHCPv4 relay") +@click.option("--vrf-selection", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable VRF selection for DHCPv4 relay") +@click.option("--server-id-override", required=False, type=click.Choice(["enable", "disable"]), help="Enable/Disable server ID override for DHCPv4 relay") +@click.option("--server-vrf", required=False, help="Server VRF name for DHCPv4 relay") +@click.option("--agent-relay-mode", required=False, type=click.Choice(["discard", "append", "replace"]), + help="Set agent relay mode for DHCPv4 relay") +@click.option("--max-hop-count", required=False, type=int, help="Maximum hop count for DHCPv4 relay") +@clicommon.pass_db +def update_dhcp_relay_ipv4_helper(db, vid, dhcp_relay_helpers, source_interface, link_selection, vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count): + if is_dhcp_server_enabled(db): + click.echo("Cannot change ipv4 dhcp_relay configuration when dhcp_server feature is enabled") + return + + if check_sonic_dhcpv4_relay_flag(db): + update_dhcpv4_relay.callback("Vlan"+str(vid), ",".join(dhcp_relay_helpers), source_interface, link_selection, vrf_selection, server_id_override, server_vrf, agent_relay_mode, max_hop_count) + else: + click.echo(f"This command is applicable for new DHCPv4 Relay feature") + @dhcp_relay_ipv4_helper.command("del") @click.argument("vid", metavar="", required=True, type=int) @@ -200,6 +586,7 @@ def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): ctx = click.get_current_context() added_servers = [] + relay_entry = {} # Verify vlan is valid vlan_name = 'Vlan{}'.format(vid) @@ -210,36 +597,47 @@ def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): # Verify all ip addresses are valid and not exist in DB dhcp_servers = vlan.get('dhcp_servers', []) dhcpv6_servers = vlan.get('dhcpv6_servers', []) + # Track if we need to update DHCPV4_RELAY table + relay_entry = db.cfgdb.get_entry('DHCPV4_RELAY', vlan_name) + dhcpv4_servers = relay_entry.get('dhcpv4_servers', []) if relay_entry else [] for ip_addr in dhcp_relay_destination_ips: try: ipaddress.ip_address(ip_addr) - if (ip_addr in dhcp_servers) or (ip_addr in dhcpv6_servers): + if (ip_addr in dhcp_servers) or (ip_addr in dhcpv6_servers) or (ip_addr in dhcpv4_servers): click.echo("{} is already a DHCP relay destination for {}".format(ip_addr, vlan_name)) continue if clicommon.ipaddress_type(ip_addr) == 4: if is_dhcp_server_enabled(db): click.echo("Cannot change dhcp_relay configuration when dhcp_server feature is enabled") return - dhcp_servers.append(ip_addr) + if not check_sonic_dhcpv4_relay_flag(db): + dhcp_servers.append(ip_addr) + else: + dhcpv4_servers.append(ip_addr) else: dhcpv6_servers.append(ip_addr) added_servers.append(ip_addr) except Exception: ctx.fail('{} is invalid IP address'.format(ip_addr)) - # Append new dhcp servers to config DB - if len(dhcp_servers): - vlan['dhcp_servers'] = dhcp_servers - if len(dhcpv6_servers): - vlan['dhcpv6_servers'] = dhcpv6_servers - - db.cfgdb.set_entry('VLAN', vlan_name, vlan) + ip_version = IPV4 if clicommon.ipaddress_type(ip_addr) == 4 else IPV6 + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db): + if len(dhcpv4_servers): + relay_entry['dhcpv4_servers'] = dhcpv4_servers + db.cfgdb.set_entry('DHCPV4_RELAY', vlan_name, relay_entry) + else: + # Append new dhcp servers to config DB + if len(dhcp_servers): + vlan['dhcp_servers'] = dhcp_servers + if len(dhcpv6_servers): + vlan['dhcpv6_servers'] = dhcpv6_servers + db.cfgdb.set_entry('VLAN', vlan_name, vlan) if len(added_servers): click.echo("Added DHCP relay destination addresses {} to {}".format(added_servers, vlan_name)) try: - restart_dhcp_relay_service() + restart_dhcp_relay_service(db, ip_version) except SystemExit as e: ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) @@ -263,43 +661,65 @@ def del_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): dhcp_servers = vlan.get('dhcp_servers', []) dhcpv6_servers = vlan.get('dhcpv6_servers', []) + # Track if we need to update DHCPV4_RELAY table + dhcpv4_relay_changed = False + relay_entry = db.cfgdb.get_entry('DHCPV4_RELAY', vlan_name) + dhcpv4_servers = relay_entry.get('dhcpv4_servers', []) if relay_entry else [] + for ip_addr in dhcp_relay_destination_ips: - if (ip_addr not in dhcp_servers) and (ip_addr not in dhcpv6_servers): + if (ip_addr not in dhcp_servers) and (ip_addr not in dhcpv6_servers) and (ip_addr not in dhcpv4_servers): ctx.fail("{} is not a DHCP relay destination for {}".format(ip_addr, vlan_name)) if clicommon.ipaddress_type(ip_addr) == 4: if is_dhcp_server_enabled(db): click.echo("Cannot change dhcp_relay configuration when dhcp_server feature is enabled") return - dhcp_servers.remove(ip_addr) + if not check_sonic_dhcpv4_relay_flag(db): + dhcp_servers.remove(ip_addr) + else: + if ip_addr in dhcpv4_servers: + dhcpv4_servers.remove(ip_addr) + dhcpv4_relay_changed = True else: dhcpv6_servers.remove(ip_addr) - # Update dhcp servers to config DB - if len(dhcp_servers): - vlan['dhcp_servers'] = dhcp_servers - else: - if 'dhcp_servers' in vlan.keys(): - del vlan['dhcp_servers'] + ip_version = IPV4 if clicommon.ipaddress_type(ip_addr) == 4 else IPV6 + # Update DHCPV4_RELAY table if needed + if ip_version == IPV4 and check_sonic_dhcpv4_relay_flag(db) : + if dhcpv4_relay_changed: + if len(dhcpv4_servers) == 0: + db.cfgdb.set_entry('DHCPV4_RELAY', vlan_name, None) + else: + relay_entry['dhcpv4_servers'] = dhcpv4_servers + db.cfgdb.set_entry('DHCPV4_RELAY', vlan_name, relay_entry) - if len(dhcpv6_servers): - vlan['dhcpv6_servers'] = dhcpv6_servers else: - if 'dhcpv6_servers' in vlan.keys(): - del vlan['dhcpv6_servers'] + # Update dhcp servers to config DB + if len(dhcp_servers): + vlan['dhcp_servers'] = dhcp_servers + else: + if 'dhcp_servers' in vlan.keys(): + del vlan['dhcp_servers'] - db.cfgdb.set_entry('VLAN', vlan_name, vlan) + if len(dhcpv6_servers): + vlan['dhcpv6_servers'] = dhcpv6_servers + else: + if 'dhcpv6_servers' in vlan.keys(): + del vlan['dhcpv6_servers'] + db.cfgdb.set_entry('VLAN', vlan_name, vlan) click.echo("Removed DHCP relay destination addresses {} from {}".format(dhcp_relay_destination_ips, vlan_name)) try: - restart_dhcp_relay_service() + restart_dhcp_relay_service(db, ip_version) except SystemExit as e: ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) def register(cli): cli.add_command(dhcp_relay) + cli.add_command(dhcpv4_relay) cli.commands['vlan'].add_command(vlan_dhcp_relay) if __name__ == '__main__': dhcp_relay() + dhcpv4_relay() vlan_dhcp_relay() diff --git a/dockers/docker-dhcp-relay/cli/show/plugins/show_dhcp_relay.py b/dockers/docker-dhcp-relay/cli/show/plugins/show_dhcp_relay.py index cdcad466ebf..83b6bcb720e 100644 --- a/dockers/docker-dhcp-relay/cli/show/plugins/show_dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli/show/plugins/show_dhcp_relay.py @@ -1,18 +1,28 @@ import click import json import re +import ast from natsort import natsorted from tabulate import tabulate import show.vlan as show_vlan import utilities_common.cli as clicommon +from typing import Dict, Optional from swsscommon.swsscommon import ConfigDBConnector from swsscommon.swsscommon import SonicV2Connector +# COUNTERS_DB Table +DHCPv4_COUNTER_TABLE = 'COUNTERS_DHCPV4' # STATE_DB Table DHCPv6_COUNTER_TABLE = 'DHCPv6_COUNTER_TABLE' +# DHCPv4 Counter Messages +dhcpv4_messages = [ + 'Unknown', 'Discover', 'Offer', 'Request', 'Acknowledge', 'NegativeAcknowledge', 'Release', + 'Inform', 'Decline', 'Malformed', 'Dropped' +] + # DHCPv6 Counter Messages messages = ["Unknown", "Solicit", "Advertise", "Request", "Confirm", "Renew", "Rebind", "Reply", "Release", "Decline", "Reconfigure", "Information-Request", "Relay-Forward", "Relay-Reply", "Malformed"] @@ -20,8 +30,11 @@ # DHCP_RELAY Config Table DHCP_RELAY = 'DHCP_RELAY' VLAN = "VLAN" +DHCPV4_RELAY_TABLE = 'DHCPV4_RELAY' DHCPV6_SERVERS = "dhcpv6_servers" DHCPV4_SERVERS = "dhcp_servers" +DHCPV4_TABLE_PARAMS = ["dhcpv4_servers", "server_vrf", "source_interface", "link_selection", + "vrf_selection", "server_id_override", "agent_relay_mode", "max_hop_count"] SUPPORTED_DHCPV4_TYPE = [ "Unknown", "Discover", "Offer", "Request", "Decline", "Ack", "Nak", "Release", "Inform", "Bootp" ] @@ -34,6 +47,16 @@ config_db = ConfigDBConnector() +def check_sonic_dhcpv4_relay_flag(): + if config_db is None: + return + + config_db.connect() + table = config_db.get_entry("DEVICE_METADATA", "localhost") + if('has_sonic_dhcpv4_relay' in table and table['has_sonic_dhcpv4_relay'] == 'True'): + return True + return False + def get_dhcp_helper_address(ctx, vlan): cfg, db = ctx vlan_dhcp_helper_data, _, _ = cfg @@ -50,6 +73,138 @@ def get_dhcp_helper_address(ctx, vlan): show_vlan.VlanBrief.register_column('DHCP Helper Address', get_dhcp_helper_address) +class DHCPv4_Counter(object): + def __init__(self): + self.db = SonicV2Connector(use_unix_socket_path=False) + self.db.connect(self.db.COUNTERS_DB) + self.table_name = DHCPv4_COUNTER_TABLE+ self.db.get_db_separator(self.db.COUNTERS_DB) + self.packet_abbr = ['Un', 'Dis', 'Off', 'Req', 'Ack', 'Nack', 'Rel', 'Inf', 'Dec', 'Mal', 'Drp'] + + def _fetch_db_data(self, vlan: str) -> Dict: + """Fetch DHCP counter data from Redis COUNTERS_DB.""" + dhcp_data = {} + + for key in self.db.keys(self.db.COUNTERS_DB): + if DHCPv4_COUNTER_TABLE in key: + table_data = self.db.get_all(self.db.COUNTERS_DB, key) + + intf_parts = key.split(self.db.get_db_separator(self.db.COUNTERS_DB)) + if len(intf_parts) > 1: + _intf = intf_parts[1] # Get VLAN name + + if _intf not in dhcp_data: + dhcp_data[_intf] = {} + + # Get TX and RX counters for this interface + if "TX" in key: + dhcp_data[_intf]['TX'] = table_data + if "RX" in key: + dhcp_data[_intf]['RX'] = table_data + + return {DHCPv4_COUNTER_TABLE: dhcp_data} + + def _get_interface_counters(self, vlan: str, direction: Optional[str] = None, pkt_type: Optional[str] = None) -> Dict: + """Extract counter data for interfaces from Redis data.""" + interface_counters = {} + redis_data = self._fetch_db_data(vlan) + + for key, value in redis_data[DHCPv4_COUNTER_TABLE].items(): + if vlan in key: + if direction: + counter_data = value[direction] + interface_counters[vlan] = counter_data + else: + rx_data = value['RX'] + tx_data = value['TX'] + interface_counters[vlan] = {'RX': rx_data, 'TX': tx_data} + + return interface_counters + + def show_direction_counters(self, vlan: str, direction: str): + """Generate table showing all packet types for a specific direction.""" + # Header with packet type abbreviations + abbr_header = [ + "Packet type Abbr: Un - Unknown, Dis - Discover, Off - Offer, Req - Request,", + " Ack - Acknowledge, Nack - NegativeAcknowledge, Rel - Release,", + " Inf - Inform, Dec - Decline, Mal - Malformed, Drp - Dropped", + "" + ] + + try: + interface_data = self._get_interface_counters(vlan, direction) + + # Prepare table headers and data + headers = [f"Vlan ({direction})"] + self.packet_abbr + table_data = [] + + for interface, counters in sorted(interface_data.items()): + row = [interface] + row.extend(str(counters.get(ptype, '0')) for ptype in dhcpv4_messages) + table_data.append(row) + + # Generate table using tabulate + table = tabulate(table_data, headers=headers, tablefmt='grid') + print("\n".join(abbr_header + [table])) + + except Exception as e: + print(f"Error fetching data from Redis: {str(e)}") + + def show_packet_type_counters(self, vlan: str, pkt_type: str, direction: Optional[str] = None): + """Generate table showing counters for a specific packet type.""" + try: + # Determine columns based on direction + columns = ['TX', 'RX'] if not direction else [direction] + + # Prepare table headers and data + headers = [f"Vlan ({pkt_type})"] + columns + table_data = [] + + interface_data = self._get_interface_counters(vlan) + for interface, counters in sorted(interface_data.items()): + row = [interface] + for col in columns: + count = counters[col][pkt_type] + row.append(count) + table_data.append(row) + + # Generate table using tabulate + print(tabulate(table_data, headers=headers, tablefmt='grid')) + + except Exception as e: + print(f"Error fetching data from Redis: {str(e)}") + + def clear_table(self, direction, pkt_type, vlan_intf): + """ Reset message counts to 0 """ + v4_cnts = {} + for msg in dhcpv4_messages: + v4_cnts[msg] = '0' + + for key in self.db.keys(self.db.COUNTERS_DB): + if DHCPv4_COUNTER_TABLE in key: + if vlan_intf and vlan_intf not in key: + continue + + self.db.hmset(self.db.COUNTERS_DB, key, (v4_cnts)) + + +def ipv4_counters(dir, pkt_type, vlan): + config_db.connect() + feature_tbl = config_db.get_table("FEATURE") + if is_dhcp_server_enabled(feature_tbl): + click.echo("Unsupport to check dhcp_relay ipv4 counter when dhcp_server feature is enabled") + return + counter = DHCPv4_Counter() + + if dir and pkt_type: + counter.show_packet_type_counters(vlan, pkt_type, dir) + elif pkt_type: + counter.show_packet_type_counters(vlan, pkt_type) + elif dir: + counter.show_direction_counters(vlan, dir) + else: + #when direction and message type both are not selected, for interface + #sending direction as "TX" by default. + counter.show_direction_counters(vlan, "TX") class DHCPv6_Counter(object): def __init__(self): @@ -84,7 +239,6 @@ def print_count(counter, intf): data.append(counter.get_dhcp6relay_msg_count(intf, i)) print(tabulate(data, headers=["Message Type", intf], tablefmt='simple', stralign='right') + "\n") - # # 'dhcp6relay_counters' group ### # @@ -122,6 +276,31 @@ def dhcp_relay_helper(): """Show DHCP_Relay helper information""" pass +def get_dhcpv4_relay_data_with_header(table_data, entry_names, dhcp_server_enabled=False): + vlan_relay = [] + vlans = table_data.keys() + for vlan in vlans: + vlan_data = table_data.get(vlan) + row = [vlan] + + if dhcp_server_enabled: + continue; + + for entry in entry_names: + entry_data = vlan_data.get(entry) + if entry_data is None or len(entry_data) == 0: + row.append("N/A") + else: + if isinstance(entry_data, list): + row.append("\n".join(entry_data)) + else: + row.append(entry_data) + vlan_relay.append(row) + + headers = ["Interface", "DHCP Relay Address", "Server Vrf", "Source Interface", "Link Selection", "VRF Selection", "Server ID Override", "Agent Relay Mode", "Max Hop Count"] + + return tabulate(vlan_relay, tablefmt='grid', stralign='right', headers=headers) + '\n' + def get_dhcp_relay_data_with_header(table_data, entry_name, dhcp_server_enabled=False): vlan_relay = {} @@ -162,12 +341,15 @@ def get_dhcp_relay(table_name, entry_name, with_header): return dhcp_server_enabled = False - if table_name == VLAN: + if table_name in {VLAN, DHCPV4_RELAY_TABLE}: feature_tbl = config_db.get_table("FEATURE") dhcp_server_enabled = is_dhcp_server_enabled(feature_tbl) if with_header: - output = get_dhcp_relay_data_with_header(table_data, entry_name, dhcp_server_enabled) + if table_name == DHCPV4_RELAY_TABLE: + output = get_dhcpv4_relay_data_with_header(table_data, entry_name, dhcp_server_enabled) + else: + output = get_dhcp_relay_data_with_header(table_data, entry_name, dhcp_server_enabled) print(output) else: vlans = config_db.get_keys(table_name) @@ -383,16 +565,37 @@ def dhcp_relay_ipv6(): def dhcp_relay_ipv4(): pass +@dhcp_relay_ipv4.command("vlan-counters") +@click.option('-d', '--direction', required=False, type=click.Choice(['TX', 'RX']), help="Specify TX(egress) or RX(ingress)") +@click.option('-t', '--type', required=False, type=click.Choice(dhcpv4_messages), help="Specify DHCP packet counter type") +@click.argument("vlan_interface", required=True) +def dhcp_relay_ip4_vlan_counters(direction, type, vlan_interface): + ipv4_counters(direction, type, vlan_interface) + @dhcp_relay_ipv4.command("helper") def dhcp_relay_ipv4_destination(): - get_dhcp_relay(VLAN, DHCPV4_SERVERS, with_header=True) + if check_sonic_dhcpv4_relay_flag(): + get_dhcp_relay(DHCPV4_RELAY_TABLE, DHCPV4_TABLE_PARAMS, with_header=True) + else: + get_dhcp_relay(VLAN, DHCPV4_SERVERS, with_header=True) @dhcp_relay_ipv6.command("destination") def dhcp_relay_ipv6_destination(): get_dhcp_relay(DHCP_RELAY, DHCPV6_SERVERS, with_header=True) +@click.group(cls=clicommon.AliasedGroup, name="dhcp4relay-counters") +def dhcp4relay_counters(): + """Show DHCPv4 counter""" + pass + +@dhcp4relay_counters.command("vlan-counts") +@click.option('-d', '--direction', required=False, type=click.Choice(['TX', 'RX']), help="Specify TX(egress) or RX(ingress)") +@click.option('-t', '--type', required=False, type=click.Choice(dhcpv4_messages), help="Specify DHCP packet counter type") +@click.argument("vlan_interface", required=True) +def counts(direction, type, vlan_interface): + ipv4_counters(direction, type, vlan_interface) @dhcp_relay_ipv6.command("counters") @click.option('-i', '--interface', required=False) @@ -416,5 +619,6 @@ def dhcp_relay_ip4counters(db, vlan_interface, dir, type): def register(cli): cli.add_command(dhcp6relay_counters) + cli.add_command(dhcp4relay_counters) cli.add_command(dhcp_relay_helper) cli.add_command(dhcp_relay)