Add loop guard CLI commands

config/stp.py

@@ -292,6 +292,7 @@ def vlan_enable_stp(db, vlan_name):
 def interface_enable_stp(db, interface_name):
     fvs = {'enabled': 'true',
            'root_guard': 'false',
+           'loop_guard': 'false',
            'bpdu_guard': 'false',
            'bpdu_guard_do_disable': 'false',
            'portfast': 'false',
@@ -361,6 +362,7 @@ def is_portchannel_member_port(db, interface_name):
 def enable_stp_for_interfaces(db):
     fvs = {'enabled': 'true',
            'root_guard': 'false',
+           'loop_guard': 'false',
            'bpdu_guard': 'false',
            'bpdu_guard_do_disable': 'false',
            'portfast': 'false',
@@ -446,6 +448,7 @@ def enable_mst_for_interfaces(db):
         'bpdu_guard': 'false',
         'bpdu_guard_do': 'false',
         'root_guard': 'false',
+        'loop_guard': 'false',
         'path_cost': MST_DEFAULT_PORT_PATH_COST,
         'priority': MST_DEFAULT_PORT_PRIORITY
         }
@@ -1350,6 +1353,7 @@ def stp_interface_enable(_db, interface_name):
     fvs = {
         'enabled': 'true',
         'root_guard': 'false',
+        'loop_guard': 'false',
         'bpdu_guard': 'false',
         'bpdu_guard_do_disable': 'false'
     }
@@ -1533,6 +1537,39 @@ def stp_interface_root_guard_disable(_db, interface_name):
     db.mod_entry('STP_PORT', interface_name, fvs)
 
 
+# config spanning_tree interface loop_guard {enable|disable} <ifname>
+# This command allow enabling or disabling of loop_guard on an interface.
+@spanning_tree_interface.group('loop_guard')
+@clicommon.pass_db
+def spanning_tree_interface_loop_guard(_db):
+    """Configure STP loop guard for interface"""
+    pass
+
+
+@spanning_tree_interface_loop_guard.command('enable')
+@click.argument('interface_name', metavar='<interface_name>', required=True)
+@clicommon.pass_db
+def stp_interface_loop_guard_enable(_db, interface_name):
+    """Enable STP loop guard for interface"""
+    ctx = click.get_current_context()
+    db = _db.cfgdb
+    check_if_stp_enabled_for_interface(ctx, db, interface_name)
+    check_if_interface_is_valid(ctx, db, interface_name)
+    db.mod_entry('STP_PORT', interface_name, {'loop_guard': 'true'})
+
+
+@spanning_tree_interface_loop_guard.command('disable')
+@click.argument('interface_name', metavar='<interface_name>', required=True)
+@clicommon.pass_db
+def stp_interface_loop_guard_disable(_db, interface_name):
+    """Disable STP loop guard for interface"""
+    ctx = click.get_current_context()
+    db = _db.cfgdb
+    check_if_stp_enabled_for_interface(ctx, db, interface_name)
+    check_if_interface_is_valid(ctx, db, interface_name)
+    db.mod_entry('STP_PORT', interface_name, {'loop_guard': 'false'})
+
+
 # config spanning_tree interface priority <ifname> <port_priority-value>
 # Specify configuring the port level priority for root bridge in seconds.
 # Default: 128, range 0-240

show/stp.py

@@ -346,7 +346,41 @@ def show_stp_root_guard(ctx):
                         if vlanid:
                             click.echo("{:17}{:7}{}".format(ifname, vlanid.group(1), state))
                         else:
-                            click.echo("{:17}{:7}{}".format(ifname, vlanid, state))
+                            click.echo("{:17}{:7}{}".format(ifname, str(vlanid), state))
+
+
+@spanning_tree.command('loop_guard')
+@click.pass_context
+def show_stp_loop_guard(ctx):
+    """Show spanning_tree loop_guard"""
+
+    print_header = 1
+    ifname_all = g_stp_cfg_db.get_keys("STP_PORT")
+    for ifname in ifname_all:
+        entry = g_stp_cfg_db.get_entry("STP_PORT", ifname)
+        if entry.get('loop_guard') == 'true' and entry.get('enabled') == 'true':
+            if print_header:
+                click.echo("{:17}{:7}{}".format("Port", "VLAN", "Current State"))
+                click.echo("-------------------------------------------")
+                print_header = 0
+
+            state = ''
+            vlanid = ''
+            keys = g_stp_appl_db.keys(g_stp_appl_db.APPL_DB, "*STP_VLAN_PORT_TABLE:*:{}".format(ifname))
+            if keys:
+                for key in keys:
+                    entry = g_stp_appl_db.get_all(g_stp_appl_db.APPL_DB, key)
+                    if entry and 'loop_guard_active' in entry:
+                        if entry['loop_guard_active'] == '0':
+                            state = 'Consistent state'
+                        else:
+                            state = 'Loop-inconsistent state'
+
+                        vlanid = re.search(':Vlan(.*):', key)
+                        if vlanid:
+                            click.echo("{:17}{:7}{}".format(ifname, vlanid.group(1), state))
+                        else:
+                            click.echo("{:17}{:7}{}".format(ifname, str(vlanid), state))
 
 
 @spanning_tree.group('statistics', cls=clicommon.AliasedGroup, invoke_without_command=True)

tests/stp_test.py

@@ -9,6 +9,7 @@
 
 import config.main as config
 import show.main as show
+import show.stp as show_stp
 from utilities_common.db import Db
 
 show_spanning_tree = """\
@@ -77,6 +78,20 @@
 """
 
 
+show_spanning_tree_loop_guard = """\
+Port             VLAN   Current State
+-------------------------------------------
+Ethernet0        100    Consistent state
+Ethernet0        None   Loop-inconsistent state
+"""
+
+
+show_spanning_tree_loop_guard_empty = """\
+Port             VLAN   Current State
+-------------------------------------------
+"""
+
+
 class TestStp(object):
     @classmethod
     def setup_class(cls):
@@ -155,6 +170,64 @@ def test_show_spanning_tree_root_guard(self):
             assert result.exit_code == 0
             assert result.output == show_spanning_tree_root_guard
 
+    def test_show_spanning_tree_loop_guard(self):
+        cli_runner = CliRunner()
+        db = Db()
+
+        mock_cfg_db = MagicMock()
+        mock_appl_db = MagicMock()
+
+        mock_cfg_db.get_keys.return_value = ["Ethernet0"]
+        mock_cfg_db.get_entry.side_effect = lambda table, key: (
+            {"mode": "pvst"} if table == "STP" and key == "GLOBAL" else
+            {"loop_guard": "true", "enabled": "true"} if table == "STP_PORT" and key == "Ethernet0" else
+            {}
+        )
+
+        mock_appl_db.APPL_DB = "APPL_DB"
+        mock_appl_db.keys.return_value = [
+            "STP_VLAN_PORT_TABLE:Vlan100:Ethernet0",
+            "STP_VLAN_PORT_TABLE:Eth:Ethernet0"
+        ]
+
+        def get_all_side_effect(_db, key):
+            if "Vlan100" in key:
+                return {"loop_guard_active": "0"}
+            return {"loop_guard_active": "1"}
+
+        mock_appl_db.get_all.side_effect = get_all_side_effect
+
+        show_stp.g_stp_cfg_db = mock_cfg_db
+        show_stp.g_stp_appl_db = mock_appl_db
+
+        result = cli_runner.invoke(show.cli.commands["spanning-tree"].commands["loop_guard"], [], obj=db)
+
+        assert result.exit_code == 0
+        assert result.output == show_spanning_tree_loop_guard
+
+    def test_show_spanning_tree_loop_guard_no_vlan_entries(self):
+        cli_runner = CliRunner()
+        db = Db()
+
+        mock_cfg_db = MagicMock()
+        mock_appl_db = MagicMock()
+
+        mock_cfg_db.get_keys.return_value = ["Ethernet0"]
+        mock_cfg_db.get_entry.side_effect = lambda table, key: (
+            {"loop_guard": "true", "enabled": "true"} if table == "STP_PORT" and key == "Ethernet0" else {}
+        )
+
+        mock_appl_db.APPL_DB = "APPL_DB"
+        mock_appl_db.keys.return_value = []
+
+        show_stp.g_stp_cfg_db = mock_cfg_db
+        show_stp.g_stp_appl_db = mock_appl_db
+
+        result = cli_runner.invoke(show.cli.commands["spanning-tree"].commands["loop_guard"], [], obj=db)
+
+        assert result.exit_code == 0
+        assert result.output == show_spanning_tree_loop_guard_empty
+
     def test_disable_enable_global_pvst(self):
         cli_runner = CliRunner()
         db = Db()
@@ -2190,6 +2263,75 @@ def test_root_guard_disable_invalid_interface(self, mock_check_valid):
         assert "Invalid interface" in result.output
 
 
+class TestStpInterfaceLoopGuard:
+    def setup_method(self):
+        self.runner = CliRunner()
+        self.cfgdb = MagicMock()
+        self.db = Db()
+        self.db.cfgdb = self.cfgdb
+
+    @patch('config.stp.check_if_interface_is_valid')
+    @patch('config.stp.check_if_stp_enabled_for_interface')
+    def test_loop_guard_enable(self, mock_check_enabled, mock_check_valid):
+        result = self.runner.invoke(
+            config.config.commands["spanning-tree"]
+            .commands["interface"]
+            .commands["loop_guard"],
+            ["enable", "Ethernet0"],
+            obj=self.db
+        )
+
+        assert result.exit_code == 0
+        self.cfgdb.mod_entry.assert_called_with("STP_PORT", "Ethernet0", {"loop_guard": "true"})
+        mock_check_enabled.assert_called_once()
+        mock_check_valid.assert_called_once()
+
+    @patch('config.stp.check_if_interface_is_valid')
+    @patch('config.stp.check_if_stp_enabled_for_interface')
+    def test_loop_guard_disable(self, mock_check_enabled, mock_check_valid):
+        result = self.runner.invoke(
+            config.config.commands["spanning-tree"]
+            .commands["interface"]
+            .commands["loop_guard"],
+            ["disable", "Ethernet0"],
+            obj=self.db
+        )
+
+        assert result.exit_code == 0
+        self.cfgdb.mod_entry.assert_called_with("STP_PORT", "Ethernet0", {"loop_guard": "false"})
+        mock_check_enabled.assert_called_once()
+        mock_check_valid.assert_called_once()
+
+    @patch('config.stp.check_if_interface_is_valid', side_effect=click.ClickException("Invalid interface"))
+    @patch('config.stp.check_if_stp_enabled_for_interface')
+    def test_loop_guard_enable_invalid_interface(self, mock_check_enabled, mock_check_valid):
+        result = self.runner.invoke(
+            config.config.commands["spanning-tree"]
+            .commands["interface"]
+            .commands["loop_guard"],
+            ["enable", "Ethernet99"],
+            obj=self.db
+        )
+
+        assert result.exit_code != 0
+        assert "Invalid interface" in result.output
+
+    @patch('config.stp.check_if_interface_is_valid')
+    @patch('config.stp.check_if_stp_enabled_for_interface', side_effect=click.ClickException("STP not enabled"))
+    def test_loop_guard_enable_stp_not_enabled(self, mock_check_enabled, mock_check_valid):
+        result = self.runner.invoke(
+            config.config.commands["spanning-tree"]
+            .commands["interface"]
+            .commands["loop_guard"],
+            ["enable", "Ethernet0"],
+            obj=self.db
+        )
+
+        assert result.exit_code != 0
+        assert "STP not enabled" in result.output
+        mock_check_valid.assert_not_called()
+
+
 class TestStpInterfaceRootGuardEnable:
     def setup_method(self):
         self.runner = CliRunner()

tests/test_config_mstp.py

@@ -347,6 +347,7 @@ def test_enable_mst_for_interfaces():
         'bpdu_guard': 'false',
         'bpdu_guard_do': 'false',
         'root_guard': 'false',
+        'loop_guard': 'false',
         'path_cost': MST_DEFAULT_PORT_PATH_COST,
         'priority': MST_DEFAULT_PORT_PRIORITY
     }