[config/show]: split vlan into separate file (sonic-net#1038)

- add vlan.py in config and show
- add extensive unit tests for vlan

Signed-off-by: Guohan Lu <[email protected]>

config/aaa.py

@@ -2,19 +2,9 @@
 # -*- coding: utf-8 -*-
 
 import click
-import netaddr
-from swsssdk import ConfigDBConnector
-
-
-def is_ipaddress(val):
-    if not val:
-        return False
-    try:
-        netaddr.IPAddress(str(val))
-    except ValueError:
-        return False
-    return True
 
+from swsssdk import ConfigDBConnector
+import utilities_common.cli as clicommon
 
 def add_table_kv(table, entry, key, val):
     config_db = ConfigDBConnector()
@@ -31,7 +21,6 @@ def del_table_key(table, entry, key):
             del data[key]
         config_db.set_entry(table, entry, data)
 
-
 @click.group()
 def aaa():
     """AAA command line"""
@@ -164,7 +153,7 @@ def passkey(ctx, secret):
 @click.option('-m', '--use-mgmt-vrf', help="Management vrf, default is no vrf", is_flag=True)
 def add(address, timeout, key, auth_type, port, pri, use_mgmt_vrf):
     """Specify a TACACS+ server"""
-    if not is_ipaddress(address):
+    if not clicommon.is_ipaddress(address):
         click.echo('Invalid ip address')
         return
 
@@ -196,7 +185,7 @@ def add(address, timeout, key, auth_type, port, pri, use_mgmt_vrf):
 @click.argument('address', metavar='<ip_address>')
 def delete(address):
     """Delete a TACACS+ server"""
-    if not is_ipaddress(address):
+    if not clicommon.is_ipaddress(address):
         click.echo('Invalid ip address')
         return
 

config/mlnx.py

@@ -7,12 +7,11 @@
 
 try:
     import os
-    import subprocess
-    import sys
     import time
 
     import click
     from sonic_py_common import logger
+    import utilities_common.cli as clicommon
 except ImportError as e:
     raise ImportError("%s - required module not found" % str(e))
 
@@ -41,24 +40,6 @@
 # Global logger instance
 log = logger.Logger(SNIFFER_SYSLOG_IDENTIFIER)
 
-
-# run command
-def run_command(command, display_cmd=False, ignore_error=False):
-    """Run bash command and print output to stdout
-    """
-    if display_cmd == True:
-        click.echo(click.style("Running command: ", fg='cyan') + click.style(command, fg='green'))
-
-    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE)
-    (out, err) = proc.communicate()
-
-    if len(out) > 0:
-        click.echo(out)
-
-    if proc.returncode != 0 and not ignore_error:
-        sys.exit(proc.returncode)
-
-
 # generate sniffer target file name include a time stamp.
 def sniffer_filename_generate(path, filename_prefix, filename_ext):
     time_stamp = time.strftime("%Y%m%d%H%M%S")
@@ -99,12 +80,12 @@ def env_variable_delete(delete_line):
 
 def conf_file_copy(src, dest):
     command = 'docker cp ' + src + ' ' + dest
-    run_command(command)
+    clicommon.run_command(command)
 
 
 def conf_file_receive():
     command = "docker exec {} bash -c 'touch {}'".format(CONTAINER_NAME, SNIFFER_CONF_FILE)
-    run_command(command)
+    clicommon.run_command(command)
     conf_file_copy(SNIFFER_CONF_FILE_IN_CONTAINER, TMP_SNIFFER_CONF_FILE)
 
 
@@ -134,15 +115,15 @@ def sniffer_env_variable_set(enable, env_variable_name, env_variable_string=""):
         config_file_send()
 
     command = 'rm -rf {}'.format(TMP_SNIFFER_CONF_FILE)
-    run_command(command)
+    clicommon.run_command(command)
 
     return ignore
 
 
 # restart the swss service with command 'service swss restart'
 def restart_swss():
     try:
-        run_command(COMMAND_RESTART_SWSS)
+        clicommon.run_command(COMMAND_RESTART_SWSS)
     except OSError as e:
         log.log_error("Not able to restart swss service, %s" % str(e), True)
         return 1

config/utils.py

@@ -0,0 +1,6 @@
+from sonic_py_common import logger
+
+SYSLOG_IDENTIFIER = "config"
+
+# Global logger instance
+log = logger.Logger(SYSLOG_IDENTIFIER)

config/vlan.py

@@ -0,0 +1,203 @@
+import click
+
+import utilities_common.cli as clicommon
+from .utils import log
+
+#
+# 'vlan' group ('config vlan ...')
+#
+@click.group(cls=clicommon.AbbreviationGroup, name='vlan')
+def vlan():
+    """VLAN-related configuration tasks"""
+    pass
+
+@vlan.command('add')
+@click.argument('vid', metavar='<vid>', required=True, type=int)
+@clicommon.pass_db
+def add_vlan(db, vid):
+    """Add VLAN"""
+
+    ctx = click.get_current_context()
+
+    if not clicommon.is_vlanid_in_range(vid):
+        ctx.fail("Invalid VLAN ID {} (1-4094)".format(vid))
+
+    vlan = 'Vlan{}'.format(vid)
+    if clicommon.check_if_vlanid_exist(db.cfgdb, vlan):
+        ctx.fail("{} already exists".format(vlan))
+
+    db.cfgdb.set_entry('VLAN', vlan, {'vlanid': vid})
+
+@vlan.command('del')
+@click.argument('vid', metavar='<vid>', required=True, type=int)
+@clicommon.pass_db
+def del_vlan(db, vid):
+    """Delete VLAN"""
+
+    log.log_info("'vlan del {}' executing...".format(vid))
+
+    ctx = click.get_current_context()
+
+    if not clicommon.is_vlanid_in_range(vid):
+        ctx.fail("Invalid VLAN ID {} (1-4094)".format(vid))
+
+    vlan = 'Vlan{}'.format(vid)
+    if clicommon.check_if_vlanid_exist(db.cfgdb, vlan) == False:
+        ctx.fail("{} does not exist".format(vlan))
+
+    keys = [ (k, v) for k, v in db.cfgdb.get_table('VLAN_MEMBER') if k == 'Vlan{}'.format(vid) ]
+    for k in keys:
+        db.cfgdb.set_entry('VLAN_MEMBER', k, None)
+    db.cfgdb.set_entry('VLAN', 'Vlan{}'.format(vid), None)
+
+#
+# 'member' group ('config vlan member ...')
+#
+@vlan.group(cls=clicommon.AbbreviationGroup, name='member')
+def vlan_member():
+    pass
+
+@vlan_member.command('add')
+@click.argument('vid', metavar='<vid>', required=True, type=int)
+@click.argument('port', metavar='port', required=True)
+@click.option('-u', '--untagged', is_flag=True)
+@clicommon.pass_db
+def add_vlan_member(db, vid, port, untagged):
+    """Add VLAN member"""
+
+    ctx = click.get_current_context()
+
+    log.log_info("'vlan member add {} {}' executing...".format(vid, port))
+
+    if not clicommon.is_vlanid_in_range(vid):
+        ctx.fail("Invalid VLAN ID {} (1-4094)".format(vid))
+
+    vlan = 'Vlan{}'.format(vid)
+    if clicommon.check_if_vlanid_exist(db.cfgdb, vlan) == False:
+        ctx.fail("{} does not exist".format(vlan))
+
+    if clicommon.get_interface_naming_mode() == "alias":
+        alias = port
+        iface_alias_converter = clicommon.InterfaceAliasConverter(db)
+        port = iface_alias_converter.alias_to_name(alias)
+        if port is None:
+            ctx.fail("cannot find port name for alias {}".format(alias))
+
+    if clicommon.is_port_mirror_dst_port(db.cfgdb, port):
+        ctx.fail("{} is configured as mirror destination port".format(port))
+
+    if clicommon.is_port_vlan_member(db.cfgdb, port, vlan):
+        ctx.fail("{} is already a member of {}".format(port, vlan))
+
+    if clicommon.is_valid_port(db.cfgdb, port):
+        is_port = True
+    elif clicommon.is_valid_portchannel(db.cfgdb, port):
+        is_port = False
+    else:
+        ctx.fail("{} does not exist".format(port))
+
+    if (is_port and clicommon.is_port_router_interface(db.cfgdb, port)) or \
+       (not is_port and clicommon.is_pc_router_interface(db.cfgdb, port)):
+        ctx.fail("{} is a router interface!".format(port))
+
+    db.cfgdb.set_entry('VLAN_MEMBER', (vlan, port), {'tagging_mode': "untagged" if untagged else "tagged" })
+
+@vlan_member.command('del')
+@click.argument('vid', metavar='<vid>', required=True, type=int)
+@click.argument('port', metavar='<port>', required=True)
+@clicommon.pass_db
+def del_vlan_member(db, vid, port):
+    """Delete VLAN member"""
+
+    ctx = click.get_current_context()
+
+    log.log_info("'vlan member del {} {}' executing...".format(vid, port))
+
+    if not clicommon.is_vlanid_in_range(vid):
+        ctx.fail("Invalid VLAN ID {} (1-4094)".format(vid))
+
+    vlan = 'Vlan{}'.format(vid)
+    if clicommon.check_if_vlanid_exist(db.cfgdb, vlan) == False:
+        ctx.fail("{} does not exist".format(vlan))
+
+    if clicommon.get_interface_naming_mode() == "alias":
+        alias = port
+        iface_alias_converter = clicommon.InterfaceAliasConverter(db)
+        port = iface_alias_converter.alias_to_name(alias)
+        if port is None:
+            ctx.fail("cannot find port name for alias {}".format(alias))
+
+    if not clicommon.is_port_vlan_member(db.cfgdb, port, vlan):
+        ctx.fail("{} is not a member of {}".format(port, vlan))
+
+    db.cfgdb.set_entry('VLAN_MEMBER', (vlan, port), None)
+
+@vlan.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)
+@clicommon.pass_db
+def add_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ip):
+    """ Add a destination IP address to 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))
+
+    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
+
+    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 restart 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)
+@clicommon.pass_db
+def del_vlan_dhcp_relay_destination(db, vid, dhcp_relay_destination_ip):
+    """ 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))
+
+    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))
+
+    dhcp_relay_dests.remove(dhcp_relay_destination_ip)
+    if len(dhcp_relay_dests) == 0:
+        del vlan['dhcp_servers']
+    else:
+        vlan['dhcp_servers'] = dhcp_relay_dests
+    db.cfgdb.set_entry('VLAN', vlan_name, vlan)
+    click.echo("Removed DHCP relay destination address {} from {}".format(dhcp_relay_destination_ip, vlan_name))
+    try:
+        click.echo("Restarting DHCP relay service...")
+        clicommon.run_command("systemctl restart dhcp_relay", display_cmd=False)
+    except SystemExit as e:
+        ctx.fail("Restart service dhcp_relay failed with error {}".format(e))

show/bgp_frr_v4.py

@@ -1,5 +1,8 @@
 import click
-from show.main import AliasedGroup, ip, run_command, get_bgp_summary_extended
+
+import utilities_common.cli as clicommon
+
+from show.main import ip, run_command, get_bgp_summary_extended
 
 
 ###############################################################################
@@ -9,7 +12,7 @@
 ###############################################################################
 
 
-@ip.group(cls=AliasedGroup)
+@ip.group(cls=clicommon.AliasedGroup)
 def bgp():
     """Show IPv4 BGP (Border Gateway Protocol) information"""
     pass

show/bgp_frr_v6.py

@@ -1,5 +1,7 @@
 import click
-from show.main import AliasedGroup, ipv6, run_command, get_bgp_summary_extended
+
+import utilities_common.cli as clicommon
+from show.main import ipv6, run_command, get_bgp_summary_extended
 
 
 ###############################################################################
@@ -9,7 +11,7 @@
 ###############################################################################
 
 
-@ipv6.group(cls=AliasedGroup)
+@ipv6.group(cls=clicommon.AliasedGroup)
 def bgp():
     """Show IPv6 BGP (Border Gateway Protocol) information"""
     pass