matthew/sonic-buildimage
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[dhcp-relay] Add support for dhcp_relay config cli (#13373) (#13584)

Browse Source
This commit is contained in:
mssonicbld 2023-02-02 06:30:53 +08:00 committed by GitHub
parent 900b809e12
commit 31297dcbdb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 606 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
dockers/docker-dhcp-relay/cli-plugin-tests/conftest.py
View File

@ -10,6 +10,11 @@ def mock_cfgdb():
'Vlan1000': {
'dhcp_servers': ['192.0.0.1']
}
},
'DHCP_RELAY': {
'Vlan1000': {
'dhcpv6_servers': ['fc02:2000::1']
}
}
}

349
dockers/docker-dhcp-relay/cli-plugin-tests/test_config_dhcp_relay.py
View File

@ -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"
}
}
@pytest.fixture(scope="module", params=["ipv4", "ipv6"])
def ip_version(request):
"""
Parametrize Ip version
config_vlan_del_dhcp_relay_output="""\
Removed DHCP relay destination addresses ('192.0.0.100',) from Vlan1000
Restarting DHCP relay service...
Args:
request: pytest request object
Returns:
Ip version needed for test case
"""
return request.param
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=["add", "del"])
def op(request):
"""
Parametrize operate tpye
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
Returns:
Operate tpye
"""
return request.param
class TestConfigVlanDhcpRelay(object):
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

229
dockers/docker-dhcp-relay/cli-plugin-tests/test_config_vlan_dhcp_relay.py Normal file
View File

@ -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

181
dockers/docker-dhcp-relay/cli/config/plugins/dhcp_relay.py
View File

@ -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="<vid>", 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="<vid>", 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="<vid>", 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="<vid>", 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='<vid>', required=True, type=int)
@click.argument('dhcp_relay_destination_ips', nargs=-1, required=True)
@ -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='<vid>', required=True, type=int)
@click.argument('dhcp_relay_destination_ips', nargs=-1, required=True)
@ -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()