From 919c10605daad213f71502d0b5925a187dca35f9 Mon Sep 17 00:00:00 2001 From: Salah Adel Date: Sun, 26 Apr 2026 17:04:36 +0000 Subject: [PATCH 1/5] Add loop guard CLI commands Signed-off-by: Salah Adel --- config/stp.py | 37 +++++++++++++++++++++++++++++++++++++ show/stp.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+) diff --git a/config/stp.py b/config/stp.py index 04adc507d4c..db83069882e 100644 --- a/config/stp.py +++ b/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} +# 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='', 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='', 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 # Specify configuring the port level priority for root bridge in seconds. # Default: 128, range 0-240 diff --git a/show/stp.py b/show/stp.py index a7e5443c4ce..c4cd33638cd 100644 --- a/show/stp.py +++ b/show/stp.py @@ -349,6 +349,40 @@ def show_stp_root_guard(ctx): click.echo("{:17}{:7}{}".format(ifname, 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, vlanid, state)) + + @spanning_tree.group('statistics', cls=clicommon.AliasedGroup, invoke_without_command=True) @click.pass_context def show_stp_statistics(ctx): From e84571b9c746b8539e79b4c44d95ca04fe6aaae6 Mon Sep 17 00:00:00 2001 From: Salah Adel Date: Sun, 26 Apr 2026 17:04:40 +0000 Subject: [PATCH 2/5] fix config mstp test Signed-off-by: Salah Adel --- tests/test_config_mstp.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_config_mstp.py b/tests/test_config_mstp.py index 961f113284f..5f30de7272c 100644 --- a/tests/test_config_mstp.py +++ b/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 } From a09579a580a7948efa39669394c0c33ea4bc482c Mon Sep 17 00:00:00 2001 From: Salah Adel Date: Sun, 26 Apr 2026 17:45:19 +0000 Subject: [PATCH 3/5] improve test coverage for loop guard Signed-off-by: Salah Adel --- tests/stp_test.py | 142 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) diff --git a/tests/stp_test.py b/tests/stp_test.py index ee736b125b1..bde96c9cd17 100644 --- a/tests/stp_test.py +++ b/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() From 5fec614b57cf470c2d396bc6cca9a41cde640e47 Mon Sep 17 00:00:00 2001 From: Salah Adel Date: Sun, 26 Apr 2026 18:21:09 +0000 Subject: [PATCH 4/5] minor test fix Signed-off-by: Salah Adel --- show/stp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/show/stp.py b/show/stp.py index c4cd33638cd..8323235f2f3 100644 --- a/show/stp.py +++ b/show/stp.py @@ -346,7 +346,7 @@ 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') From 5762ab41d83a61f60f187f99b9f4ef28ed3c6772 Mon Sep 17 00:00:00 2001 From: Salah Adel Date: Sun, 26 Apr 2026 19:12:04 +0000 Subject: [PATCH 5/5] minor test fix Signed-off-by: Salah Adel --- show/stp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/show/stp.py b/show/stp.py index 8323235f2f3..37b1d4e6719 100644 --- a/show/stp.py +++ b/show/stp.py @@ -380,7 +380,7 @@ def show_stp_loop_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.group('statistics', cls=clicommon.AliasedGroup, invoke_without_command=True)