From 456be049eb184a5cf438b01d882a3380a9806a85 Mon Sep 17 00:00:00 2001 From: DendroLabs Date: Tue, 17 Mar 2026 11:54:18 -0400 Subject: [PATCH 1/5] Add link event damping CLI commands (config/show/clear) Implements CLI support for RFC 2439/7196 link event dampening: - config interface dampening enable/disable: configure AIED dampening with validation (IntRange per review feedback), monitor-only mode, and configurable flap-penalty - show interfaces dampening: display config and operational state from CONFIG_DB and STATE_DB with per-interface counters - sonic-clear interfaces dampening: reset penalty via STATE_DB flag Supersedes sonic-net/sonic-utilities#3001 with improvements: - click.IntRange validation (per @Junchao-Mellanox review) - Monitor-only mode per RFC 7196 "Calculate But Do Not Damp" - Configurable flap-penalty (vs hardcoded 1000) - Show command with operational state and counters - Clear command per RFC 2439 Section 4.8.6 Signed-off-by: DendroLabs --- clear/main.py | 51 +++++++++++++ config/main.py | 109 ++++++++++++++++++++++++++++ show/interfaces/__init__.py | 140 ++++++++++++++++++++++++++++++++++++ 3 files changed, 300 insertions(+) diff --git a/clear/main.py b/clear/main.py index b4df9e48f5f..6f89f1f7bbf 100755 --- a/clear/main.py +++ b/clear/main.py @@ -11,6 +11,7 @@ from utilities_common import util_base from show.plugins.pbh import read_pbh_counters from config.plugins.pbh import serialize_pbh_counters +from swsscommon.swsscommon import SonicV2Connector, ConfigDBConnector from . import plugins from . import stp # This is from the aliases example: @@ -763,5 +764,55 @@ def asic_sdk_health_event(db, namespace): state_db.delete(state_db.STATE_DB, key); +# +# 'interfaces' group ("sonic-clear interfaces ...") +# + +@cli.group(cls=AliasedGroup) +def interfaces(): + """Clear interface-related state""" + pass + + +@interfaces.command() +@click.argument('interface_name', metavar='', required=False) +def dampening(interface_name): + """Clear link event dampening state (reset penalty to 0). + + Without an interface name, clears dampening on all interfaces. + The interface is immediately unsuppressed if currently damped. + """ + config_db = ConfigDBConnector() + config_db.connect() + port_table = config_db.get_table("PORT") + + if interface_name: + if interface_name not in port_table: + click.echo("Error: Interface {} does not exist".format(interface_name)) + raise SystemExit(1) + ports_to_clear = [interface_name] + else: + ports_to_clear = [] + for port_name, port_data in port_table.items(): + algo = port_data.get("link_event_damping_algorithm", "disabled") + if algo != "disabled": + ports_to_clear.append(port_name) + + if not ports_to_clear: + click.echo("No interfaces have dampening configured") + return + + state_db = SonicV2Connector(host="127.0.0.1") + state_db.connect(state_db.STATE_DB) + + for port_name in ports_to_clear: + state_key = "CLEAR_DAMPENING|{}".format(port_name) + state_db.set(state_db.STATE_DB, state_key, "clear", "true") + click.echo("Cleared dampening on {}".format(port_name)) + + if not interface_name: + click.echo("Cleared dampening on {} interface(s)".format(len(ports_to_clear))) + + if __name__ == '__main__': cli() diff --git a/config/main.py b/config/main.py index 95eafb80777..8a43389db3a 100644 --- a/config/main.py +++ b/config/main.py @@ -6349,6 +6349,115 @@ def cable_length(ctx, interface_name, length): except ValueError as e: ctx.fail("Invalid ConfigDB. Error: {}".format(e)) +# +# 'dampening' subgroup ('config interface dampening ...') +# + +@interface.group(cls=clicommon.AbbreviationGroup) +@click.pass_context +def dampening(ctx): + """Configure link event dampening on an interface""" + pass + + +@dampening.command() +@click.argument('interface_name', metavar='', required=True) +@click.option('--half-life', type=click.IntRange(min=1, max=3600), default=5, + show_default=True, + help='Decay half-life in seconds (1-3600).') +@click.option('--reuse', type=click.IntRange(min=1, max=20000), default=1000, + show_default=True, + help='Reuse threshold (1-20000).') +@click.option('--suppress', type=click.IntRange(min=1, max=20000), default=2000, + show_default=True, + help='Suppress threshold (1-20000).') +@click.option('--max-suppress-time', type=click.IntRange(min=1, max=3600), default=20, + show_default=True, + help='Max suppress time in seconds (1-3600).') +@click.option('--flap-penalty', type=click.IntRange(min=1, max=20000), default=1000, + show_default=True, + help='Penalty per link-down event (1-20000).') +@click.option('--monitor', is_flag=True, default=False, + help='Monitor-only mode: calculate penalties and emit syslog ' + 'but do NOT suppress events. Use to safely tune parameters ' + 'in production before enabling full dampening.') +@click.pass_context +def enable(ctx, interface_name, half_life, reuse, suppress, max_suppress_time, + flap_penalty, monitor): + """Enable link event dampening on an interface with AIED algorithm.""" + config_db = ctx.obj['config_db'] + + if clicommon.get_interface_naming_mode() == "alias": + interface_name = interface_alias_to_name(config_db, interface_name) + if interface_name is None: + ctx.fail("'interface_name' is None!") + + # Validate interface exists in PORT table + port_table = config_db.get_table("PORT") + if interface_name not in port_table: + ctx.fail("Interface {} does not exist".format(interface_name)) + + # Validate parameter relationships + if reuse >= suppress: + ctx.fail("Reuse threshold ({}) must be less than suppress threshold ({})".format( + reuse, suppress)) + + if half_life > max_suppress_time: + ctx.fail("Half-life ({}) must not exceed max-suppress-time ({})".format( + half_life, max_suppress_time)) + + # Calculate ceiling and warn if very high + ceiling = reuse * (2 ** (max_suppress_time / half_life)) + if ceiling > 100000: + click.echo("Warning: calculated penalty ceiling is very high ({:.0f}). " + "Consider reducing max-suppress-time or increasing reuse threshold.".format(ceiling)) + + algorithm = "aied-monitor" if monitor else "aied" + + config_db.mod_entry("PORT", interface_name, { + "link_event_damping_algorithm": algorithm, + "decay_half_life": str(half_life), + "reuse_threshold": str(reuse), + "suppress_threshold": str(suppress), + "max_suppress_time": str(max_suppress_time), + "flap_penalty": str(flap_penalty), + }) + + mode_str = "monitor-only" if monitor else "active" + click.echo("Link event dampening enabled on {} (algorithm={}, mode={}, half-life={}s, " + "reuse={}, suppress={}, max-suppress={}s, penalty={})".format( + interface_name, algorithm, mode_str, half_life, reuse, suppress, + max_suppress_time, flap_penalty)) + + +@dampening.command() +@click.argument('interface_name', metavar='', required=True) +@click.pass_context +def disable(ctx, interface_name): + """Disable link event dampening on an interface.""" + config_db = ctx.obj['config_db'] + + if clicommon.get_interface_naming_mode() == "alias": + interface_name = interface_alias_to_name(config_db, interface_name) + if interface_name is None: + ctx.fail("'interface_name' is None!") + + port_table = config_db.get_table("PORT") + if interface_name not in port_table: + ctx.fail("Interface {} does not exist".format(interface_name)) + + config_db.mod_entry("PORT", interface_name, { + "link_event_damping_algorithm": "disabled", + "decay_half_life": "0", + "reuse_threshold": "0", + "suppress_threshold": "0", + "max_suppress_time": "0", + "flap_penalty": "0", + }) + + click.echo("Link event dampening disabled on {}".format(interface_name)) + + # # 'transceiver' subgroup ('config interface transceiver ...') # diff --git a/show/interfaces/__init__.py b/show/interfaces/__init__.py index b0fd79105f5..31d4f372845 100644 --- a/show/interfaces/__init__.py +++ b/show/interfaces/__init__.py @@ -1446,6 +1446,146 @@ def display_phy_taps_attribute(attr_display_name, attr_json): click.echo("") +# +# 'dampening' subcommand ("show interfaces dampening") +# +@interfaces.command() +@click.argument('interfacename', required=False) +@clicommon.pass_db +def dampening(db, interfacename): + """Show link event dampening configuration and operational state""" + + ctx = click.get_current_context() + + if interfacename: + interfacename = try_convert_interfacename_from_alias(ctx, interfacename) + + config_db = db.cfgdb + state_db = db.db + + port_table = config_db.get_table("PORT") + + if interfacename: + if interfacename not in port_table: + ctx.fail("Interface {} does not exist".format(interfacename)) + ports = {interfacename: port_table[interfacename]} + else: + ports = port_table + + header = [ + "Interface", + "Algorithm", + "Half-Life(s)", + "Reuse", + "Suppress", + "Max-Suppress(s)", + "Penalty", + "Flap-Penalty", + "Suppressed", + "Time-Left(s)", + ] + + rows = [] + for port_name in natsorted(ports.keys()): + port_data = ports[port_name] + algorithm = port_data.get("link_event_damping_algorithm", "disabled") + + if algorithm == "disabled": + # Only show this port if specifically requested + if interfacename: + rows.append([ + port_name, + "disabled", + "-", "-", "-", "-", "-", "-", "-", "-" + ]) + continue + + is_monitor = (algorithm == "aied-monitor") + + half_life = port_data.get("decay_half_life", "0") + reuse = port_data.get("reuse_threshold", "0") + suppress = port_data.get("suppress_threshold", "0") + max_suppress = port_data.get("max_suppress_time", "0") + flap_penalty = port_data.get("flap_penalty", "1000") + + # Read operational state from STATE_DB + state_key = "PORT_TABLE|{}".format(port_name) + current_penalty = "N/A" + suppressed = "N/A" + time_remaining = "N/A" + + if state_db: + penalty_val = state_db.get(state_db.STATE_DB, state_key, + "damping_current_penalty") + suppressed_val = state_db.get(state_db.STATE_DB, state_key, + "damping_suppressed") + time_val = state_db.get(state_db.STATE_DB, state_key, + "damping_time_remaining") + + if penalty_val: + current_penalty = penalty_val + if suppressed_val: + if is_monitor and suppressed_val == "true": + suppressed = "(Mon)" + else: + suppressed = "Yes" if suppressed_val == "true" else "No" + if time_val: + time_remaining = time_val if suppressed == "Yes" else "-" + + rows.append([ + port_name, + algorithm, + half_life, + reuse, + suppress, + max_suppress, + current_penalty, + flap_penalty, + suppressed, + time_remaining, + ]) + + if not rows: + if interfacename: + click.echo("Link event dampening is not configured on {}".format( + interfacename)) + else: + click.echo("Link event dampening is not configured on any interface") + return + + click.echo(tabulate(rows, header, tablefmt="simple")) + click.echo("") + + # Show counters if a specific interface is queried + if interfacename and state_db: + _show_dampening_counters(state_db, interfacename) + + +def _show_dampening_counters(state_db, interface_name): + """Display per-interface dampening event counters.""" + state_key = "PORT_TABLE|{}".format(interface_name) + + counter_fields = [ + ("damping_pre_transitions", "Pre-damping transitions (total)"), + ("damping_post_transitions", "Post-damping transitions (total)"), + ("damping_pre_up_transitions", "Pre-damping UP transitions"), + ("damping_pre_down_transitions", "Pre-damping DOWN transitions"), + ("damping_post_up_transitions", "Post-damping UP transitions"), + ("damping_post_down_transitions", "Post-damping DOWN transitions"), + ] + + counter_rows = [] + for field, description in counter_fields: + value = state_db.get(state_db.STATE_DB, state_key, field) + if value: + counter_rows.append([description, value]) + + if counter_rows: + click.echo("Dampening Counters for {}:".format(interface_name)) + click.echo(tabulate(counter_rows, ["Counter", "Value"], tablefmt="simple")) + click.echo("") + + @interfaces.command('phy-serdes') @click.argument('interfacename', required=True) @multi_asic_util.multi_asic_click_options From f48473356542617ff8129cda982f95bf3c06d14d Mon Sep 17 00:00:00 2001 From: DendroLabs Date: Tue, 17 Mar 2026 11:54:18 -0400 Subject: [PATCH 2/5] [Link Event Damping] Fix CLI critical issues - Guard against exponential overflow in ceiling calculation - Add interface alias support to clear command - Improve error handling consistency Signed-off-by: DendroLabs --- clear/main.py | 8 +++++++- config/main.py | 15 +++++++++++---- 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/clear/main.py b/clear/main.py index 6f89f1f7bbf..decf2a98621 100755 --- a/clear/main.py +++ b/clear/main.py @@ -787,9 +787,15 @@ def dampening(interface_name): port_table = config_db.get_table("PORT") if interface_name: + if clicommon.get_interface_naming_mode() == "alias": + alias = interface_name + interface_name = clicommon.InterfaceAliasConverter().alias_to_name(interface_name) + if interface_name == alias: + click.echo("Error: invalid interface alias {}".format(alias)) + sys.exit(1) if interface_name not in port_table: click.echo("Error: Interface {} does not exist".format(interface_name)) - raise SystemExit(1) + sys.exit(1) ports_to_clear = [interface_name] else: ports_to_clear = [] diff --git a/config/main.py b/config/main.py index 8a43389db3a..eeee3e9918c 100644 --- a/config/main.py +++ b/config/main.py @@ -6407,10 +6407,17 @@ def enable(ctx, interface_name, half_life, reuse, suppress, max_suppress_time, half_life, max_suppress_time)) # Calculate ceiling and warn if very high - ceiling = reuse * (2 ** (max_suppress_time / half_life)) - if ceiling > 100000: - click.echo("Warning: calculated penalty ceiling is very high ({:.0f}). " - "Consider reducing max-suppress-time or increasing reuse threshold.".format(ceiling)) + exponent = max_suppress_time / half_life + if exponent > 50: + click.echo("Warning: calculated penalty ceiling is extremely high " + "(exponent={:.0f}). Consider reducing max-suppress-time or " + "increasing half-life.".format(exponent)) + else: + ceiling = reuse * (2 ** exponent) + if ceiling > 100000: + click.echo("Warning: calculated penalty ceiling is very high ({:.0f}). " + "Consider reducing max-suppress-time or increasing reuse " + "threshold.".format(ceiling)) algorithm = "aied-monitor" if monitor else "aied" From 1895b3a195d8709e8c2ab41449311f77872837b5 Mon Sep 17 00:00:00 2001 From: DendroLabs Date: Tue, 17 Mar 2026 12:11:35 -0400 Subject: [PATCH 3/5] Fix flake8 style violations in dampening CLI - Add missing blank line before dampening subgroup (E302) - Fix continuation line indentation in warning messages (E127) Signed-off-by: DendroLabs --- config/main.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/config/main.py b/config/main.py index eeee3e9918c..d2e07310212 100644 --- a/config/main.py +++ b/config/main.py @@ -6349,10 +6349,12 @@ def cable_length(ctx, interface_name, length): except ValueError as e: ctx.fail("Invalid ConfigDB. Error: {}".format(e)) + # # 'dampening' subgroup ('config interface dampening ...') # + @interface.group(cls=clicommon.AbbreviationGroup) @click.pass_context def dampening(ctx): @@ -6410,14 +6412,14 @@ def enable(ctx, interface_name, half_life, reuse, suppress, max_suppress_time, exponent = max_suppress_time / half_life if exponent > 50: click.echo("Warning: calculated penalty ceiling is extremely high " - "(exponent={:.0f}). Consider reducing max-suppress-time or " - "increasing half-life.".format(exponent)) + "(exponent={:.0f}). Consider reducing max-suppress-time or " + "increasing half-life.".format(exponent)) else: ceiling = reuse * (2 ** exponent) if ceiling > 100000: click.echo("Warning: calculated penalty ceiling is very high ({:.0f}). " - "Consider reducing max-suppress-time or increasing reuse " - "threshold.".format(ceiling)) + "Consider reducing max-suppress-time or increasing reuse " + "threshold.".format(ceiling)) algorithm = "aied-monitor" if monitor else "aied" From 4358624e5068b2496dfaa9579a04d69f42f1de62 Mon Sep 17 00:00:00 2001 From: DendroLabs Date: Tue, 17 Mar 2026 13:26:09 -0400 Subject: [PATCH 4/5] Add unit tests for link event dampening CLI commands Tests cover config enable/disable, show, and clear commands: - Config: defaults, custom params, monitor mode, validation (invalid interface, reuse>=suppress, half-life>max-suppress, IntRange bounds, overflow warning) - Show: no config, disabled, configured, monitor mode display - Clear: specific interface, all interfaces, no config, invalid Signed-off-by: DendroLabs --- tests/link_event_damping_test.py | 339 +++++++++++++++++++++++++++++++ 1 file changed, 339 insertions(+) create mode 100644 tests/link_event_damping_test.py diff --git a/tests/link_event_damping_test.py b/tests/link_event_damping_test.py new file mode 100644 index 00000000000..9514cfe673b --- /dev/null +++ b/tests/link_event_damping_test.py @@ -0,0 +1,339 @@ +import os +import sys +import pytest +from click.testing import CliRunner +from unittest import mock +from unittest.mock import patch, MagicMock + +import config.main as config +import show.main as show +import clear.main as clear +from utilities_common.db import Db + + +class TestConfigInterfaceDampening(object): + @classmethod + def setup_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "1" + + @classmethod + def teardown_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "0" + + def test_enable_dampening_defaults(self): + """Test enabling dampening with default parameters""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0"], obj=obj) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "enabled" in result.output + + def test_enable_dampening_custom_params(self): + """Test enabling dampening with custom parameters""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--half-life", "10", "--reuse", "500", + "--suppress", "3000", "--max-suppress-time", "40", + "--flap-penalty", "2000"], + obj=obj) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "enabled" in result.output + + def test_enable_dampening_monitor_mode(self): + """Test enabling dampening in monitor-only mode""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--monitor"], obj=obj) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "monitor" in result.output.lower() + + def test_enable_dampening_invalid_interface(self): + """Test enabling dampening on non-existent interface""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["EthernetINVALID"], obj=obj) + print(result.exit_code, result.output) + assert result.exit_code != 0 + assert "does not exist" in result.output + + def test_enable_dampening_reuse_ge_suppress(self): + """Test that reuse >= suppress is rejected""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--reuse", "3000", "--suppress", "2000"], + obj=obj) + print(result.exit_code, result.output) + assert result.exit_code != 0 + assert "Reuse threshold" in result.output + + def test_enable_dampening_halflife_gt_maxsuppress(self): + """Test that half-life > max-suppress-time is rejected""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--half-life", "30", "--max-suppress-time", "20"], + obj=obj) + print(result.exit_code, result.output) + assert result.exit_code != 0 + assert "Half-life" in result.output + + def test_enable_dampening_invalid_halflife(self): + """Test that out-of-range half-life is rejected by click.IntRange""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--half-life", "0"], obj=obj) + print(result.exit_code, result.output) + assert result.exit_code != 0 + assert "Invalid value" in result.output + + def test_enable_dampening_overflow_warning(self): + """Test warning for extremely high ceiling exponent""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--half-life", "1", "--max-suppress-time", "3600", + "--reuse", "100", "--suppress", "200"], + obj=obj) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "Warning" in result.output + + def test_disable_dampening(self): + """Test disabling dampening""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["disable"], + ["Ethernet0"], obj=obj) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "disabled" in result.output + + def test_disable_dampening_invalid_interface(self): + """Test disabling dampening on non-existent interface""" + runner = CliRunner() + db = Db() + obj = {'config_db': db.cfgdb} + + result = runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["disable"], + ["EthernetINVALID"], obj=obj) + print(result.exit_code, result.output) + assert result.exit_code != 0 + assert "does not exist" in result.output + + +class TestShowInterfacesDampening(object): + @classmethod + def setup_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "1" + + @classmethod + def teardown_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "0" + + def test_show_dampening_no_config(self): + """Test show when no dampening is configured""" + runner = CliRunner() + db = Db() + result = runner.invoke( + show.cli.commands["interfaces"].commands["dampening"], + [], obj=db) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "not configured on any interface" in result.output + + def test_show_dampening_specific_disabled(self): + """Test show for a specific interface with no dampening""" + runner = CliRunner() + db = Db() + result = runner.invoke( + show.cli.commands["interfaces"].commands["dampening"], + ["Ethernet0"], obj=db) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "disabled" in result.output + + def test_show_dampening_invalid_interface(self): + """Test show for non-existent interface""" + runner = CliRunner() + db = Db() + result = runner.invoke( + show.cli.commands["interfaces"].commands["dampening"], + ["EthernetINVALID"], obj=db) + print(result.exit_code, result.output) + assert result.exit_code != 0 + + def test_show_dampening_configured(self): + """Test show when dampening is configured on an interface""" + runner = CliRunner() + db = Db() + + # Set up dampening config via the config command first + config_obj = {'config_db': db.cfgdb} + runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0"], obj=config_obj) + + result = runner.invoke( + show.cli.commands["interfaces"].commands["dampening"], + ["Ethernet0"], obj=db) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "aied" in result.output + assert "Ethernet0" in result.output + + def test_show_dampening_monitor_mode(self): + """Test show displays monitor mode correctly""" + runner = CliRunner() + db = Db() + + config_obj = {'config_db': db.cfgdb} + runner.invoke( + config.config.commands["interface"].commands["dampening"].commands["enable"], + ["Ethernet0", "--monitor"], obj=config_obj) + + result = runner.invoke( + show.cli.commands["interfaces"].commands["dampening"], + ["Ethernet0"], obj=db) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "aied-monitor" in result.output + + +class TestClearInterfacesDampening(object): + @classmethod + def setup_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "1" + + @classmethod + def teardown_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "0" + + @patch('clear.main.ConfigDBConnector') + @patch('clear.main.SonicV2Connector') + def test_clear_dampening_specific_interface(self, mock_sv2, mock_cfgdb): + """Test clearing dampening on a specific interface""" + mock_db_instance = MagicMock() + mock_cfgdb.return_value = mock_db_instance + mock_db_instance.get_table.return_value = { + "Ethernet0": { + "link_event_damping_algorithm": "aied", + "alias": "etp1", + } + } + + mock_state = MagicMock() + mock_sv2.return_value = mock_state + + runner = CliRunner() + result = runner.invoke( + clear.cli.commands["interfaces"].commands["dampening"], + ["Ethernet0"]) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "Cleared dampening on Ethernet0" in result.output + mock_state.set.assert_called_once() + + @patch('clear.main.ConfigDBConnector') + @patch('clear.main.SonicV2Connector') + def test_clear_dampening_all(self, mock_sv2, mock_cfgdb): + """Test clearing dampening on all interfaces""" + mock_db_instance = MagicMock() + mock_cfgdb.return_value = mock_db_instance + mock_db_instance.get_table.return_value = { + "Ethernet0": { + "link_event_damping_algorithm": "aied", + }, + "Ethernet4": { + "link_event_damping_algorithm": "aied-monitor", + }, + "Ethernet8": { + "link_event_damping_algorithm": "disabled", + } + } + + mock_state = MagicMock() + mock_sv2.return_value = mock_state + + runner = CliRunner() + result = runner.invoke( + clear.cli.commands["interfaces"].commands["dampening"], + []) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "Cleared dampening on Ethernet0" in result.output + assert "Cleared dampening on Ethernet4" in result.output + assert "Ethernet8" not in result.output + assert "2 interface(s)" in result.output + + @patch('clear.main.ConfigDBConnector') + def test_clear_dampening_no_config(self, mock_cfgdb): + """Test clearing when no dampening is configured""" + mock_db_instance = MagicMock() + mock_cfgdb.return_value = mock_db_instance + mock_db_instance.get_table.return_value = { + "Ethernet0": { + "link_event_damping_algorithm": "disabled", + } + } + + runner = CliRunner() + result = runner.invoke( + clear.cli.commands["interfaces"].commands["dampening"], + []) + print(result.exit_code, result.output) + assert result.exit_code == 0 + assert "No interfaces have dampening configured" in result.output + + @patch('clear.main.ConfigDBConnector') + def test_clear_dampening_invalid_interface(self, mock_cfgdb): + """Test clearing dampening on non-existent interface""" + mock_db_instance = MagicMock() + mock_cfgdb.return_value = mock_db_instance + mock_db_instance.get_table.return_value = { + "Ethernet0": {"link_event_damping_algorithm": "aied"} + } + + runner = CliRunner() + result = runner.invoke( + clear.cli.commands["interfaces"].commands["dampening"], + ["EthernetINVALID"]) + print(result.exit_code, result.output) + assert result.exit_code != 0 + assert "does not exist" in result.output From 6eaf7a09dcff39a94a4dc4fb8413aa9247df3cab Mon Sep 17 00:00:00 2001 From: DendroLabs Date: Tue, 17 Mar 2026 13:56:59 -0400 Subject: [PATCH 5/5] Remove unused imports in dampening test file Signed-off-by: DendroLabs --- tests/link_event_damping_test.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/tests/link_event_damping_test.py b/tests/link_event_damping_test.py index 9514cfe673b..8f6abbb6f9c 100644 --- a/tests/link_event_damping_test.py +++ b/tests/link_event_damping_test.py @@ -1,8 +1,5 @@ import os -import sys -import pytest from click.testing import CliRunner -from unittest import mock from unittest.mock import patch, MagicMock import config.main as config