From 39c1f878b32cac920408467deceea0eab660e8be Mon Sep 17 00:00:00 2001 From: Yaqiang Zhu <1512099831@qq.com> Date: Mon, 30 Jan 2023 17:48:01 -0800 Subject: [PATCH] [dhcp-relay] Add support for dhcp_relay config cli (#13373) Why I did it Currently the config cli of dhcpv4 is may cause confusion and config of dhcpv6 is missing. How I did it Add dhcp_relay config cli and test cases. config dhcp_relay ipv4 helper (add | del) config dhcp_relay ipv6 destination (add | del) Updated docs for it in sonic-utilities: https://github.com/sonic-net/sonic-utilities/pull/2598/files How to verify it Build docker-dhcp-relay.gz with and without INCLUDE_DHCP_RELAY, and check target/docker-dhcp-relay.gz.log --- .../cli-plugin-tests/conftest.py | 5 + .../test_config_dhcp_relay.py | 357 ++++++++++-------- .../test_config_vlan_dhcp_relay.py | 229 +++++++++++ .../cli/config/plugins/dhcp_relay.py | 189 +++++++++- 4 files changed, 606 insertions(+), 174 deletions(-) create mode 100644 dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py diff --git a/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py b/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py index 4eb79edd0b..37aec0b8b2 100644 --- a/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py @@ -10,6 +10,11 @@ def mock_cfgdb(): 'Vlan1000': { 'dhcp_servers': ['192.0.0.1'] } + }, + 'DHCP_RELAY': { + 'Vlan1000': { + 'dhcpv6_servers': ['fc02:2000::1'] + } } } 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 46acda358b..2c9a5c19a9 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 @@ -1,229 +1,264 @@ -import os import sys -import traceback +from utilities_common.db import Db from unittest import mock - from click.testing import CliRunner -from utilities_common.db import Db - import pytest - -sys.path.append('../cli/config/plugins/') +sys.path.append("../cli/config/plugins/") import dhcp_relay -config_vlan_add_dhcp_relay_output="""\ -Added DHCP relay destination addresses ['192.0.0.100'] to Vlan1000 +config_dhcp_relay_add_output = """\ +Added DHCP relay address [{}] to Vlan1000 Restarting DHCP relay service... """ - -config_vlan_add_dhcpv6_relay_output="""\ -Added DHCP relay destination addresses ['fc02:2000::1'] to Vlan1000 +config_dhcp_relay_del_output = """\ +Removed DHCP relay address [{}] from Vlan1000 Restarting DHCP relay service... """ +expected_dhcp_relay_add_config_db_output = { + "ipv4": { + "dhcp_servers": [ + "192.0.0.1", "192.0.0.3"] + }, + "ipv6": { + "dhcpv6_servers": [ + "fc02:2000::1", "fc02:2000::3"] + } +} +expected_dhcp_relay_del_config_db_output = { + "ipv4": { + "dhcp_servers": [ + "192.0.0.1" + ] + }, + "ipv6": { + "dhcpv6_servers": [ + "fc02:2000::1" + ] + } +} +expected_dhcp_relay_add_multi_config_db_output = { + "ipv4": { + "dhcp_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" + ] + } +} -config_vlan_add_multiple_dhcpv6_relay_output="""\ -Added DHCP relay destination addresses ['fc02:2000::1', 'fc02:2000::2', 'fc02:2000::3'] to Vlan1000 -Restarting DHCP relay service... -""" +IP_VER_TEST_PARAM_MAP = { + "ipv4": { + "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": "VLAN" + }, + "ipv6": { + "command": "destination", + "ips": [ + "fc02:2000::3", + "fc02:2000::4", + "fc02:2000::5" + ], + "exist_ip": "fc02:2000::1", + "nonexist_ip": "fc02:2000::2", + "invalid_ip": "fc02:2000:", + "table": "DHCP_RELAY" + } +} -config_vlan_del_dhcp_relay_output="""\ -Removed DHCP relay destination addresses ('192.0.0.100',) from Vlan1000 -Restarting DHCP relay service... -""" -config_vlan_del_dhcpv6_relay_output="""\ -Removed DHCP relay destination addresses ('fc02:2000::1',) from Vlan1000 -Restarting DHCP relay service... -""" +@pytest.fixture(scope="module", params=["ipv4", "ipv6"]) +def ip_version(request): + """ + Parametrize Ip version -config_vlan_del_multiple_dhcpv6_relay_output="""\ -Removed DHCP relay destination addresses ('fc02:2000::1', 'fc02:2000::2', 'fc02:2000::3') from Vlan1000 -Restarting DHCP relay service... -""" + Args: + request: pytest request object -class TestConfigVlanDhcpRelay(object): + Returns: + Ip version needed for test case + """ + return request.param + + +@pytest.fixture(scope="module", params=["add", "del"]) +def op(request): + """ + Parametrize operate tpye + + Args: + request: pytest request object + + Returns: + Operate tpye + """ + return request.param + + +class TestConfigDhcpRelay(object): def test_plugin_registration(self): cli = mock.MagicMock() dhcp_relay.register(cli) - cli.commands['vlan'].add_command.assert_called_once_with(dhcp_relay.vlan_dhcp_relay) - def test_config_vlan_add_dhcp_relay_with_nonexist_vlanid(self): + def test_config_dhcp_relay_add_del_with_nonexist_vlanid(self, ip_version, op): runner = CliRunner() - with mock.patch('utilities_common.cli.run_command') as mock_run_command: - result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], - ["1001", "192.0.0.100"]) + 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[op], ["1001", IP_VER_TEST_PARAM_MAP[ip_version]["ips"][0]]) print(result.exit_code) - print(result.output) - # traceback.print_tb(result.exc_info[2]) + print(result.stdout) assert result.exit_code != 0 assert "Error: Vlan1001 doesn't exist" in result.output assert mock_run_command.call_count == 0 - def test_config_vlan_add_dhcp_relay_with_invalid_vlanid(self): + def test_config_add_del_dhcp_relay_with_invalid_ip(self, ip_version, op): runner = CliRunner() + invalid_ip = IP_VER_TEST_PARAM_MAP[ip_version]["invalid_ip"] - with mock.patch('utilities_common.cli.run_command') as mock_run_command: - result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], - ["4096", "192.0.0.100"]) - print(result.exit_code) - print(result.output) - # traceback.print_tb(result.exc_info[2]) - assert result.exit_code != 0 - assert "Error: Vlan4096 doesn't exist" in result.output - assert mock_run_command.call_count == 0 - - def test_config_vlan_add_dhcp_relay_with_invalid_ip(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.vlan_dhcp_relay.commands["add"], - ["1000", "192.0.0.1000"], obj=db) - print(result.exit_code) - print(result.output) - # traceback.print_tb(result.exc_info[2]) - assert result.exit_code != 0 - assert "Error: 192.0.0.1000 is invalid IP address" in result.output - assert mock_run_command.call_count == 0 - - result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], - ["1000", "192.0.0."], obj=db) - print(result.exit_code) - print(result.output) - assert result.exit_code != 0 - assert "Error: 192.0.0. is invalid IP address" in result.output - assert mock_run_command.call_count == 0 - - def test_config_vlan_add_dhcp_relay_with_exist_ip(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.vlan_dhcp_relay.commands["add"], - ["1000", "192.0.0.1"], obj=db) - print(result.exit_code) - print(result.output) - assert result.exit_code == 0 - 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): - runner = CliRunner() - db = Db() - db.cfgdb = mock_cfgdb - - # 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"], - ["1000", "192.0.0.100"], obj=db) + result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[op], ["1000", invalid_ip]) + print(result.exit_code) + print(result.output) + assert result.exit_code != 0 + 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): + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + exist_ip = IP_VER_TEST_PARAM_MAP[ip_version]["exist_ip"] + + 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["add"], ["1000", exist_ip], obj=db) print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_vlan_add_dhcp_relay_output + assert "{} is already a DHCP relay for Vlan1000".format(exist_ip) in result.output + assert mock_run_command.call_count == 0 + + def test_config_del_nonexist_dhcp_relay(self, mock_cfgdb, ip_version): + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + nonexist_ip = IP_VER_TEST_PARAM_MAP[ip_version]["nonexist_ip"] + + 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["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 + assert mock_run_command.call_count == 0 + + def test_config_add_del_dhcp_relay(self, mock_cfgdb, ip_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"] + + 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) + 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('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1', '192.0.0.100']}) + db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", + expected_dhcp_relay_add_config_db_output[ip_version]) db.cfgdb.set_entry.reset_mock() - - # del relay dest + # del dhcp relay with mock.patch("utilities_common.cli.run_command") as mock_run_command: - result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["del"], - ["1000", "192.0.0.100"], obj=db) + result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] + .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 - assert result.output == config_vlan_del_dhcp_relay_output + assert result.output == config_dhcp_relay_del_output.format(test_ip) assert mock_run_command.call_count == 3 - db.cfgdb.set_entry.assert_called_once_with('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1']}) + 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]) - def test_config_vlan_add_del_dhcpv6_relay_dest(self, mock_cfgdb): + def test_config_add_del_multiple_dhcp_relay(self, mock_cfgdb, ip_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"] - # 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"], - ["1000", "fc02:2000::1"], obj=db) + # 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) print(result.exit_code) print(result.output) assert result.exit_code == 0 - assert result.output == config_vlan_add_dhcpv6_relay_output + 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('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1'], 'dhcpv6_servers': ['fc02:2000::1']}) + db.cfgdb.set_entry.assert_called_once_with(config_db_table, "Vlan1000", + expected_dhcp_relay_add_multi_config_db_output[ip_version]) db.cfgdb.set_entry.reset_mock() - - # del relay dest + # del dhcp relay with mock.patch("utilities_common.cli.run_command") as mock_run_command: - result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["del"], - ["1000", "fc02:2000::1"], obj=db) + result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] + .commands[IP_VER_TEST_PARAM_MAP[ip_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_vlan_del_dhcpv6_relay_output + assert result.output == config_dhcp_relay_del_output.format(",".join(test_ips)) assert mock_run_command.call_count == 3 - db.cfgdb.set_entry.assert_called_once_with('VLAN', 'Vlan1000', {'dhcp_servers': ['192.0.0.1']}) + 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]) - def test_config_vlan_add_del_multiple_dhcpv6_relay_dest(self, mock_cfgdb): + def test_config_add_del_duplicate_dhcp_relay(self, mock_cfgdb, ip_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"] - # 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"], - ["1000", "fc02:2000::1", "fc02:2000::2", "fc02:2000::3"], obj=db) + result = runner.invoke(dhcp_relay.dhcp_relay.commands[ip_version] + .commands[IP_VER_TEST_PARAM_MAP[ip_version]["command"]] + .commands[op], ["1000", test_ip, test_ip], obj=db) print(result.exit_code) print(result.output) - assert result.exit_code == 0 - assert result.output == config_vlan_add_multiple_dhcpv6_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'], 'dhcpv6_servers': ['fc02:2000::1', 'fc02:2000::2', 'fc02:2000::3']}) - - db.cfgdb.set_entry.reset_mock() - - # del relay dest - with mock.patch("utilities_common.cli.run_command") as mock_run_command: - result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["del"], - ["1000", "fc02:2000::1", "fc02:2000::2", "fc02:2000::3"], obj=db) - print(result.exit_code) - print(result.output) - assert result.exit_code == 0 - assert result.output == config_vlan_del_multiple_dhcpv6_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']}) - - def test_config_vlan_remove_nonexist_dhcp_relay_dest(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.vlan_dhcp_relay.commands["del"], - ["1000", "192.0.0.100"], obj=db) - print(result.exit_code) - print(result.output) - # traceback.print_tb(result.exc_info[2]) assert result.exit_code != 0 - assert "Error: 192.0.0.100 is not a DHCP relay destination for Vlan1000" in result.output - assert mock_run_command.call_count == 0 - - def test_config_vlan_remove_dhcp_relay_dest_with_nonexist_vlanid(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.vlan_dhcp_relay.commands["del"], - ["1001", "192.0.0.1"], obj=Db) - print(result.exit_code) - print(result.output) - # traceback.print_tb(result.exc_info[2]) - assert result.exit_code != 0 - assert "Error: Vlan1001 doesn't exist" in result.output + assert "Error: Find duplicate DHCP relay ip {} in {} list".format(test_ip, op) in result.output assert mock_run_command.call_count == 0 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 new file mode 100644 index 0000000000..46acda358b --- /dev/null +++ b/dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py @@ -0,0 +1,229 @@ +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('../cli/config/plugins/') +import dhcp_relay + +config_vlan_add_dhcp_relay_output="""\ +Added DHCP relay destination addresses ['192.0.0.100'] to Vlan1000 +Restarting DHCP relay service... +""" + +config_vlan_add_dhcpv6_relay_output="""\ +Added DHCP relay destination addresses ['fc02:2000::1'] to Vlan1000 +Restarting DHCP relay service... +""" + +config_vlan_add_multiple_dhcpv6_relay_output="""\ +Added DHCP relay destination addresses ['fc02:2000::1', 'fc02:2000::2', 'fc02:2000::3'] to Vlan1000 +Restarting DHCP relay service... +""" + +config_vlan_del_dhcp_relay_output="""\ +Removed DHCP relay destination addresses ('192.0.0.100',) from Vlan1000 +Restarting DHCP relay service... +""" + +config_vlan_del_dhcpv6_relay_output="""\ +Removed DHCP relay destination addresses ('fc02:2000::1',) from Vlan1000 +Restarting DHCP relay service... +""" + +config_vlan_del_multiple_dhcpv6_relay_output="""\ +Removed DHCP relay destination addresses ('fc02:2000::1', 'fc02:2000::2', 'fc02:2000::3') from Vlan1000 +Restarting DHCP relay service... +""" + +class TestConfigVlanDhcpRelay(object): + def test_plugin_registration(self): + cli = mock.MagicMock() + dhcp_relay.register(cli) + cli.commands['vlan'].add_command.assert_called_once_with(dhcp_relay.vlan_dhcp_relay) + + def test_config_vlan_add_dhcp_relay_with_nonexist_vlanid(self): + runner = CliRunner() + + with mock.patch('utilities_common.cli.run_command') as mock_run_command: + result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], + ["1001", "192.0.0.100"]) + print(result.exit_code) + print(result.output) + # traceback.print_tb(result.exc_info[2]) + assert result.exit_code != 0 + assert "Error: Vlan1001 doesn't exist" in result.output + assert mock_run_command.call_count == 0 + + def test_config_vlan_add_dhcp_relay_with_invalid_vlanid(self): + runner = CliRunner() + + with mock.patch('utilities_common.cli.run_command') as mock_run_command: + result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], + ["4096", "192.0.0.100"]) + print(result.exit_code) + print(result.output) + # traceback.print_tb(result.exc_info[2]) + assert result.exit_code != 0 + assert "Error: Vlan4096 doesn't exist" in result.output + assert mock_run_command.call_count == 0 + + def test_config_vlan_add_dhcp_relay_with_invalid_ip(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.vlan_dhcp_relay.commands["add"], + ["1000", "192.0.0.1000"], obj=db) + print(result.exit_code) + print(result.output) + # traceback.print_tb(result.exc_info[2]) + assert result.exit_code != 0 + assert "Error: 192.0.0.1000 is invalid IP address" in result.output + assert mock_run_command.call_count == 0 + + result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["add"], + ["1000", "192.0.0."], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code != 0 + assert "Error: 192.0.0. is invalid IP address" in result.output + assert mock_run_command.call_count == 0 + + def test_config_vlan_add_dhcp_relay_with_exist_ip(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.vlan_dhcp_relay.commands["add"], + ["1000", "192.0.0.1"], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + 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): + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + # 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"], + ["1000", "192.0.0.100"], obj=db) + 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']}) + + db.cfgdb.set_entry.reset_mock() + + # del relay dest + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["del"], + ["1000", "192.0.0.100"], obj=db) + 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']}) + + def test_config_vlan_add_del_dhcpv6_relay_dest(self, mock_cfgdb): + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + # 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"], + ["1000", "fc02:2000::1"], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_vlan_add_dhcpv6_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'], 'dhcpv6_servers': ['fc02:2000::1']}) + + db.cfgdb.set_entry.reset_mock() + + # del relay dest + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["del"], + ["1000", "fc02:2000::1"], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_vlan_del_dhcpv6_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']}) + + def test_config_vlan_add_del_multiple_dhcpv6_relay_dest(self, mock_cfgdb): + runner = CliRunner() + db = Db() + db.cfgdb = mock_cfgdb + + # 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"], + ["1000", "fc02:2000::1", "fc02:2000::2", "fc02:2000::3"], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_vlan_add_multiple_dhcpv6_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'], 'dhcpv6_servers': ['fc02:2000::1', 'fc02:2000::2', 'fc02:2000::3']}) + + db.cfgdb.set_entry.reset_mock() + + # del relay dest + with mock.patch("utilities_common.cli.run_command") as mock_run_command: + result = runner.invoke(dhcp_relay.vlan_dhcp_relay.commands["del"], + ["1000", "fc02:2000::1", "fc02:2000::2", "fc02:2000::3"], obj=db) + print(result.exit_code) + print(result.output) + assert result.exit_code == 0 + assert result.output == config_vlan_del_multiple_dhcpv6_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']}) + + def test_config_vlan_remove_nonexist_dhcp_relay_dest(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.vlan_dhcp_relay.commands["del"], + ["1000", "192.0.0.100"], obj=db) + print(result.exit_code) + print(result.output) + # traceback.print_tb(result.exc_info[2]) + assert result.exit_code != 0 + assert "Error: 192.0.0.100 is not a DHCP relay destination for Vlan1000" in result.output + assert mock_run_command.call_count == 0 + + def test_config_vlan_remove_dhcp_relay_dest_with_nonexist_vlanid(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.vlan_dhcp_relay.commands["del"], + ["1001", "192.0.0.1"], obj=Db) + print(result.exit_code) + print(result.output) + # traceback.print_tb(result.exc_info[2]) + assert result.exit_code != 0 + assert "Error: Vlan1001 doesn't exist" in result.output + assert mock_run_command.call_count == 0 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 33a798fc87..57848161f6 100644 --- a/dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py +++ b/dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py @@ -1,11 +1,177 @@ import click -import utilities_common.cli as clicommon import ipaddress +import utilities_common.cli as clicommon +DHCP_RELAY_TABLE = "DHCP_RELAY" +DHCPV6_SERVERS = "dhcpv6_servers" +IPV6 = 6 + +VLAN_TABLE = "VLAN" +DHCPV4_SERVERS = "dhcp_servers" +IPV4 = 4 + + +def validate_ips(ctx, ips, ip_version): + for ip in ips: + try: + ip_address = ipaddress.ip_address(ip) + except Exception: + ctx.fail("{} is invalid IP address".format(ip)) + + if ip_address.version != ip_version: + ctx.fail("{} is not IPv{} address".format(ip, ip_version)) + + +def get_dhcp_servers(db, vlan_name, ctx, table_name, dhcp_servers_str): + table = db.cfgdb.get_entry(table_name, vlan_name) + if len(table.keys()) == 0: + ctx.fail("{} doesn't exist".format(vlan_name)) + + dhcp_servers = table.get(dhcp_servers_str, []) + + return dhcp_servers, table + + +def restart_dhcp_relay_service(): + """ + Restart dhcp_relay service + """ + 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) + clicommon.run_command("systemctl start dhcp_relay", display_cmd=False) + + +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) + 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) + 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 + + dhcp_servers.append(dhcp_relay_ip) + added_ips.append(dhcp_relay_ip) + + table[dhcp_servers_str] = dhcp_servers + + 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() + except SystemExit as e: + ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) + + +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) + 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) + 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) + + 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) + click.echo("Removed DHCP relay address [{}] from {}".format(",".join(dhcp_relay_ips), vlan_name)) + try: + restart_dhcp_relay_service() + except SystemExit as e: + ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) + + +@click.group(cls=clicommon.AbbreviationGroup, name="dhcp_relay") +def dhcp_relay(): + """config DHCP_Relay information""" + pass + + +@dhcp_relay.group(cls=clicommon.AbbreviationGroup, name="ipv6") +def dhcp_relay_ipv6(): + pass + + +@dhcp_relay_ipv6.group(cls=clicommon.AbbreviationGroup, name="destination") +def dhcp_relay_ipv6_destination(): + pass + + +@dhcp_relay_ipv6_destination.command("add") +@click.argument("vid", metavar="", required=True, type=int) +@click.argument("dhcp_relay_destinations", nargs=-1, required=True) +@clicommon.pass_db +def add_dhcp_relay_ipv6_destination(db, vid, dhcp_relay_destinations): + add_dhcp_relay(vid, dhcp_relay_destinations, db, IPV6) + + +@dhcp_relay_ipv6_destination.command("del") +@click.argument("vid", metavar="", required=True, type=int) +@click.argument("dhcp_relay_destinations", nargs=-1, required=True) +@clicommon.pass_db +def del_dhcp_relay_ipv6_destination(db, vid, dhcp_relay_destinations): + del_dhcp_relay(vid, dhcp_relay_destinations, db, IPV6) + + +@dhcp_relay.group(cls=clicommon.AbbreviationGroup, name="ipv4") +def dhcp_relay_ipv4(): + pass + + +@dhcp_relay_ipv4.group(cls=clicommon.AbbreviationGroup, name="helper") +def dhcp_relay_ipv4_helper(): + pass + + +@dhcp_relay_ipv4_helper.command("add") +@click.argument("vid", metavar="", required=True, type=int) +@click.argument("dhcp_relay_helpers", nargs=-1, required=True) +@clicommon.pass_db +def add_dhcp_relay_ipv4_helper(db, vid, dhcp_relay_helpers): + add_dhcp_relay(vid, dhcp_relay_helpers, db, IPV4) + + +@dhcp_relay_ipv4_helper.command("del") +@click.argument("vid", metavar="", required=True, type=int) +@click.argument("dhcp_relay_helpers", nargs=-1, required=True) +@clicommon.pass_db +def del_dhcp_relay_ipv4_helper(db, vid, dhcp_relay_helpers): + del_dhcp_relay(vid, dhcp_relay_helpers, db, IPV4) + + +# subcommand of vlan @click.group(cls=clicommon.AbbreviationGroup, name='dhcp_relay') def vlan_dhcp_relay(): pass + @vlan_dhcp_relay.command('add') @click.argument('vid', metavar='', required=True, type=int) @click.argument('dhcp_relay_destination_ips', nargs=-1, required=True) @@ -30,8 +196,8 @@ def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): try: ipaddress.ip_address(ip_addr) if (ip_addr in dhcp_servers) or (ip_addr in dhcpv6_servers): - click.echo("{} is already a DHCP relay destination for {}".format(ip_addr, vlan_name)) - continue + click.echo("{} is already a DHCP relay destination for {}".format(ip_addr, vlan_name)) + continue if clicommon.ipaddress_type(ip_addr) == 4: dhcp_servers.append(ip_addr) else: @@ -51,13 +217,11 @@ def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): if len(added_servers): click.echo("Added DHCP relay destination addresses {} to {}".format(added_servers, vlan_name)) try: - 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) - clicommon.run_command("systemctl start dhcp_relay", display_cmd=False) + restart_dhcp_relay_service() except SystemExit as e: ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) + @vlan_dhcp_relay.command('del') @click.argument('vid', metavar='', required=True, type=int) @click.argument('dhcp_relay_destination_ips', nargs=-1, required=True) @@ -79,9 +243,9 @@ def del_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): for ip_addr in dhcp_relay_destination_ips: if (ip_addr not in dhcp_servers) and (ip_addr not in dhcpv6_servers): - ctx.fail("{} is not a DHCP relay destination for {}".format(ip_addr, vlan_name)) + ctx.fail("{} is not a DHCP relay destination for {}".format(ip_addr, vlan_name)) if clicommon.ipaddress_type(ip_addr) == 4: - dhcp_servers.remove(ip_addr) + dhcp_servers.remove(ip_addr) else: dhcpv6_servers.remove(ip_addr) @@ -101,17 +265,16 @@ def del_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips): db.cfgdb.set_entry('VLAN', vlan_name, vlan) click.echo("Removed DHCP relay destination addresses {} from {}".format(dhcp_relay_destination_ips, vlan_name)) try: - 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) - clicommon.run_command("systemctl start dhcp_relay", display_cmd=False) + restart_dhcp_relay_service() except SystemExit as e: ctx.fail("Restart service dhcp_relay failed with error {}".format(e)) def register(cli): + cli.add_command(dhcp_relay) cli.commands['vlan'].add_command(vlan_dhcp_relay) if __name__ == '__main__': + dhcp_relay() vlan_dhcp_relay()