[dhcp_relay] Adapt config/show CLI commands to support DHCPv6 relay (#8211)
#### Why I did it - Adapt config/show CLI commands to support DHCPv6 relay - Support multiple dhcp servers assignment in one command - Fix IP validation - Adapt UT and add new UT cases #### How I did it - Modify config/show dhcp relay files - Modify config/show UT files #### How to verify it This PR has a dependency on PR https://github.com/Azure/sonic-utilities/pull/1717 Build an image with the dependent PR and this PR Use config/show DHCPv6 relay commands.
This commit is contained in:
shlomibitton
Judy Joseph
parent
c7d4f5b8d8
commit
edd6f4086c
@ -13,12 +13,32 @@ sys.path.append('../cli/config/plugins/')
|
||||
import dhcp_relay
|
||||
|
||||
config_vlan_add_dhcp_relay_output="""\
|
||||
Added DHCP relay destination address 192.0.0.100 to Vlan1000
|
||||
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 address 192.0.0.100 from Vlan1000
|
||||
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...
|
||||
"""
|
||||
|
||||
@ -54,12 +74,14 @@ class TestConfigVlanDhcpRelay(object):
|
||||
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):
|
||||
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"])
|
||||
["1000", "192.0.0.1000"], obj=db)
|
||||
print(result.exit_code)
|
||||
print(result.output)
|
||||
# traceback.print_tb(result.exc_info[2])
|
||||
@ -67,6 +89,14 @@ class TestConfigVlanDhcpRelay(object):
|
||||
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()
|
||||
@ -110,6 +140,64 @@ class TestConfigVlanDhcpRelay(object):
|
||||
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()
|
||||
|
||||
@ -20,9 +20,9 @@ class TestVlanDhcpRelay(object):
|
||||
|
||||
def test_dhcp_relay_column_output(self):
|
||||
ctx = (
|
||||
({'Vlan100': {'dhcp_servers': ['192.0.0.1', '192.168.0.2']}}, {}, {}),
|
||||
({'Vlan100': {'dhcp_servers': ['192.0.0.1', '192.168.0.2'], 'dhcpv6_servers': ['fc02:2000::1', 'fc02:2000::2']}}, {}, {}),
|
||||
(),
|
||||
)
|
||||
assert show_dhcp_relay.get_dhcp_helper_address(ctx, 'Vlan100') == '192.0.0.1\n192.168.0.2'
|
||||
assert show_dhcp_relay.get_dhcp_helper_address(ctx, 'Vlan100') == '192.0.0.1\n192.168.0.2\nfc02:2000::1\nfc02:2000::2'
|
||||
|
||||
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import click
|
||||
import utilities_common.cli as clicommon
|
||||
import ipaddress
|
||||
|
||||
@click.group(cls=clicommon.AbbreviationGroup, name='dhcp_relay')
|
||||
def vlan_dhcp_relay():
|
||||
@ -7,66 +8,98 @@ def vlan_dhcp_relay():
|
||||
|
||||
@vlan_dhcp_relay.command('add')
|
||||
@click.argument('vid', metavar='<vid>', required=True, type=int)
|
||||
@click.argument('dhcp_relay_destination_ip', metavar='<dhcp_relay_destination_ip>', required=True)
|
||||
@click.argument('dhcp_relay_destination_ips', nargs=-1, required=True)
|
||||
@clicommon.pass_db
|
||||
def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ip):
|
||||
def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips):
|
||||
""" Add a destination IP address to the VLAN's DHCP relay """
|
||||
|
||||
ctx = click.get_current_context()
|
||||
added_servers = []
|
||||
|
||||
if not clicommon.is_ipaddress(dhcp_relay_destination_ip):
|
||||
ctx.fail('{} is invalid IP address'.format(dhcp_relay_destination_ip))
|
||||
|
||||
# Verify vlan is valid
|
||||
vlan_name = 'Vlan{}'.format(vid)
|
||||
vlan = db.cfgdb.get_entry('VLAN', vlan_name)
|
||||
if len(vlan) == 0:
|
||||
ctx.fail("{} doesn't exist".format(vlan_name))
|
||||
|
||||
dhcp_relay_dests = vlan.get('dhcp_servers', [])
|
||||
if dhcp_relay_destination_ip in dhcp_relay_dests:
|
||||
click.echo("{} is already a DHCP relay destination for {}".format(dhcp_relay_destination_ip, vlan_name))
|
||||
return
|
||||
# Verify all ip addresses are valid and not exist in DB
|
||||
dhcp_servers = vlan.get('dhcp_servers', [])
|
||||
dhcpv6_servers = vlan.get('dhcpv6_servers', [])
|
||||
|
||||
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):
|
||||
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:
|
||||
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
|
||||
|
||||
dhcp_relay_dests.append(dhcp_relay_destination_ip)
|
||||
vlan['dhcp_servers'] = dhcp_relay_dests
|
||||
db.cfgdb.set_entry('VLAN', vlan_name, vlan)
|
||||
click.echo("Added DHCP relay destination address {} to {}".format(dhcp_relay_destination_ip, 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)
|
||||
except SystemExit as e:
|
||||
ctx.fail("Restart service dhcp_relay failed with error {}".format(e))
|
||||
|
||||
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)
|
||||
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_ip', metavar='<dhcp_relay_destination_ip>', required=True)
|
||||
@click.argument('dhcp_relay_destination_ips', nargs=-1, required=True)
|
||||
@clicommon.pass_db
|
||||
def del_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ip):
|
||||
def del_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ips):
|
||||
""" Remove a destination IP address from the VLAN's DHCP relay """
|
||||
|
||||
ctx = click.get_current_context()
|
||||
|
||||
if not clicommon.is_ipaddress(dhcp_relay_destination_ip):
|
||||
ctx.fail('{} is invalid IP address'.format(dhcp_relay_destination_ip))
|
||||
|
||||
# Verify vlan is valid
|
||||
vlan_name = 'Vlan{}'.format(vid)
|
||||
vlan = db.cfgdb.get_entry('VLAN', vlan_name)
|
||||
if len(vlan) == 0:
|
||||
ctx.fail("{} doesn't exist".format(vlan_name))
|
||||
|
||||
dhcp_relay_dests = vlan.get('dhcp_servers', [])
|
||||
if not dhcp_relay_destination_ip in dhcp_relay_dests:
|
||||
ctx.fail("{} is not a DHCP relay destination for {}".format(dhcp_relay_destination_ip, vlan_name))
|
||||
# Remove dhcp servers if they exist in the DB
|
||||
dhcp_servers = vlan.get('dhcp_servers', [])
|
||||
dhcpv6_servers = vlan.get('dhcpv6_servers', [])
|
||||
|
||||
dhcp_relay_dests.remove(dhcp_relay_destination_ip)
|
||||
if len(dhcp_relay_dests) == 0:
|
||||
del vlan['dhcp_servers']
|
||||
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))
|
||||
if clicommon.ipaddress_type(ip_addr) == 4:
|
||||
dhcp_servers.remove(ip_addr)
|
||||
else:
|
||||
dhcpv6_servers.remove(ip_addr)
|
||||
|
||||
# Update dhcp servers to config DB
|
||||
if len(dhcp_servers):
|
||||
vlan['dhcp_servers'] = dhcp_servers
|
||||
else:
|
||||
vlan['dhcp_servers'] = dhcp_relay_dests
|
||||
if 'dhcp_servers' in vlan.keys():
|
||||
del vlan['dhcp_servers']
|
||||
|
||||
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 address {} from {}".format(dhcp_relay_destination_ip, vlan_name))
|
||||
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)
|
||||
|
||||
@ -9,8 +9,9 @@ def get_dhcp_helper_address(ctx, vlan):
|
||||
return ""
|
||||
|
||||
dhcp_helpers = vlan_config.get('dhcp_servers', [])
|
||||
dhcpv6_helpers = vlan_config.get('dhcpv6_servers', [])
|
||||
|
||||
return '\n'.join(natsorted(dhcp_helpers))
|
||||
return '\n'.join(natsorted(dhcp_helpers) + natsorted(dhcpv6_helpers))
|
||||
|
||||
|
||||
vlan.VlanBrief.register_column('DHCP Helper Address', get_dhcp_helper_address)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user