Adapt config/show CLI commands to support DHCPv6 relay feature

Support multiple dhcp servers assignment in one command
Fix IP validation
Adapt and add new UT cases

Signed-off-by: Shlomi Bitton <[email protected]>

Author: Shlomi Bitton

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

@@ -13,12 +13,32 @@
 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,19 +74,29 @@ def test_config_vlan_add_dhcp_relay_with_invalid_vlanid(self):
             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])
             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()
@@ -110,6 +140,64 @@ def test_config_vlan_add_del_dhcp_relay_dest(self, mock_cfgdb):
             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()

dockers/docker-dhcp-relay/cli-plugin-tests/test_show_dhcp_relay.py

@@ -20,9 +20,9 @@ def test_plugin_registration(self):
 
     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'
 
 

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

@@ -1,72 +1,105 @@
 import click
 import utilities_common.cli as clicommon
+import ipaddress
 
 @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_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:
+            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', [])
+
+    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:
+        if 'dhcp_servers' in vlan.keys():
+            del vlan['dhcp_servers']
 
-    dhcp_relay_dests.remove(dhcp_relay_destination_ip)
-    if len(dhcp_relay_dests) == 0:
-        del vlan['dhcp_servers']
+    if len(dhcpv6_servers):
+        vlan['dhcpv6_servers'] = dhcpv6_servers
     else:
-        vlan['dhcp_servers'] = dhcp_relay_dests
+        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)

dockers/docker-dhcp-relay/cli/show/plugins/show_dhcp_relay.py

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