From 0b35ebf22a3699b05df298ea5da1eb7323262629 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Fri, 22 May 2026 16:08:06 +0530 Subject: [PATCH 01/11] Add link event damping support Signed-off-by: Sivakumar Thirukkanna Thevar --- config/main.py | 91 ++++++++++++++++++++++++++++++++ doc/Command-Reference.md | 75 ++++++++++++++++++++++++++ tests/config_int_damping_test.py | 81 ++++++++++++++++++++++++++++ 3 files changed, 247 insertions(+) create mode 100644 tests/config_int_damping_test.py diff --git a/config/main.py b/config/main.py index 6ab10f634ff..3605ed9bb1b 100644 --- a/config/main.py +++ b/config/main.py @@ -5422,6 +5422,97 @@ def interface_type(ctx, interface_name, interface_type_value, verbose): command += ["-vv"] clicommon.run_command(command, display_cmd=verbose) +# +# 'damping' subgroup ('config interface damping ...') +# + +@interface.group(cls=clicommon.AbbreviationGroup) +@click.pass_context +def damping(ctx): + """Set interface damping configurations""" + pass + +# +# 'algo' subcommand ('config interface damping algo ...') +# + +@damping.command() +@click.pass_context +@click.argument('interface_name', metavar='', required=True) +@click.argument('algo_type', metavar='', required=True, type=click.Choice(["aied", "disabled"])) +def algo(ctx, interface_name, algo_type): + """Set link event damping algorithm""" + # Get the config_db connector + 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_dict = config_db.get_table('PORT') + if interface_name not in port_dict: + ctx.fail("Invalid port {}".format(interface_name)) + + log.log_info("Executing: interface link_event_damping_algorithm {} {}".format(interface_name, algo_type)) + + config_db.mod_entry("PORT", interface_name, {"link_event_damping_algorithm": algo_type}) + +# +# 'aied-param' subcommand ('config interface damping aied-param ...') +# + +@damping.command() +@click.pass_context +@click.argument('interface_name', metavar='', required=True) +@click.option('--max-suppress-time', required=False, type=int, help="Set max suppress time in ms") +@click.option('--decay-half-life', required=False, type=int, help="Set decay half life in ms") +@click.option('--suppress-threshold', required=False, type=int, help="Set suppress threshold") +@click.option('--reuse-threshold', required=False, type=int, help="Set reuse threshold") +@click.option('--flap-penalty', required=False, type=int, help="Set flap penalty") +def aied_param(ctx, interface_name, max_suppress_time, decay_half_life, suppress_threshold, reuse_threshold, flap_penalty): + """Set AIED link event damping configuration""" + # Get the config_db connector + 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_dict = config_db.get_table('PORT') + if interface_name not in port_dict: + ctx.fail("Invalid port {}".format(interface_name)) + + config_set = {} + + if max_suppress_time is not None: + if max_suppress_time < 0: + ctx.fail("Invalid max_suppress_time value {}. It should be >= 0".format(max_suppress_time)) + config_set['max_suppress_time'] = max_suppress_time + if decay_half_life is not None: + if decay_half_life < 0: + ctx.fail("Invalid decay_half_life value {}. It should be >= 0".format(decay_half_life)) + config_set['decay_half_life'] = decay_half_life + if suppress_threshold is not None: + if suppress_threshold < 0: + ctx.fail("Invalid suppress_threshold value {}. It should be >= 0".format(suppress_threshold)) + config_set['suppress_threshold'] = suppress_threshold + if reuse_threshold is not None: + if reuse_threshold < 0: + ctx.fail("Invalid reuse_threshold value {}. It should be >= 0".format(reuse_threshold)) + config_set['reuse_threshold'] = reuse_threshold + if flap_penalty is not None: + if flap_penalty < 0: + ctx.fail("Invalid flap_penalty value {}. It should be >= 0".format(flap_penalty)) + config_set['flap_penalty'] = flap_penalty + + if len(config_set) == 0: + ctx.fail("Expected at least one valid AIED config parameter") + + log.log_info("Executing: interface aied_config {}".format(interface_name)) + config_db.mod_entry("PORT", interface_name, config_set) + # # 'advertised-interface-types' subcommand # diff --git a/doc/Command-Reference.md b/doc/Command-Reference.md index 022963be128..24cfd015288 100644 --- a/doc/Command-Reference.md +++ b/doc/Command-Reference.md @@ -6931,6 +6931,7 @@ This sub-section explains the following list of configuration on the interfaces. 11) mpls - To add or remove MPLS operation for the interface 12) loopback-action - to set action for packet that ingress and gets routed on the same IP interface 13) link-training - to set interface link-training mode +14) damping - to set link event damping configuration on an interface From 201904 release onwards, the “config interface” command syntax is changed and the format is as follows: @@ -7540,6 +7541,80 @@ This command is used for setting link-training mode of a interface. admin@sonic:~$ sudo config interface link-training Ethernet0 off ``` +**config interface damping <...> (Versions >= 202311)** + +This sub-section contains the config commands that are supported for configuring link event damping on an interface. + - Link event damping algorithm. + - Link event damping configuration. + +***config interface damping algo (Versions >= 202311)*** + + This command is used to configure link event damping algorithm on an interface. + +- Usage + ``` + config interface damping algo --help + Usage: config interface damping algo [OPTIONS] + + Set link event damping algorithm + + Options: + -h, -?, --help Show this message and exit. + ``` + Currently link event damping supports `aied` (Additive Increase Exponential Decrease) algorithm, so expected algo value is either to set `aied` or disable the algorithm using `disabled` value. + +- Example + + To set `aied` algorithm: + ``` + config interface damping algo Ethernet20 aied + ``` + + To disable the link event damping algorithm: + ``` + config interface damping algo Ethernet20 disabled + ``` + +***config interface damping aied-param (Versions >= 202311*** + +This command is used to configure link event damping AIED parameters on an interface. + +- Usage + ``` + config interface damping aied-param --help + Usage: config interface damping aied-param [OPTIONS] + + Set AIED link event damping configuration + + Options: + --max-suppress-time INTEGER Set max suppress time in ms + --decay-half-life INTEGER Set decay half life in ms + --suppress-threshold INTEGER Set suppress threshold + --reuse-threshold INTEGER Set reuse threshold + --flap-penalty INTEGER Set flap penalty + -h, -?, --help Show this message and exit. + + ``` + + One or more AIED link event damping config params can be configured at a time. + +- Examples + + Set all the config parameters: + ``` + config interface damping aied-param Ethernet20 --suppress-threshold 1200 --decay-half-life 15000 --max-suppress-time 30000 --flap-penalty 1000 --reuse-threshold 1000 + ``` + + Set only flap penalty: + ``` + config interface damping aied-param Ethernet20 --flap-penalty 500 + ``` + + Set suppress threshold and reuse threshold: + ``` + config interface damping aied-param Ethernet20 --suppress-threshold 1500 --reuse-threshold 1100 + ``` + Go Back To [Beginning of the document](#) or [Beginning of this section](#interfaces) ## Interface Naming Mode diff --git a/tests/config_int_damping_test.py b/tests/config_int_damping_test.py new file mode 100644 index 00000000000..06825db6ca6 --- /dev/null +++ b/tests/config_int_damping_test.py @@ -0,0 +1,81 @@ +import click +import config.main as config +import operator +import os +import pytest +import sys + +from click.testing import CliRunner +from utilities_common.db import Db + +test_path = os.path.dirname(os.path.abspath(__file__)) +modules_path = os.path.dirname(test_path) +scripts_path = os.path.join(modules_path, "scripts") +sys.path.insert(0, modules_path) + + +@pytest.fixture(scope='module') +def ctx(scope='module'): + db = Db() + obj = {'config_db':db.cfgdb, 'namespace': ''} + yield obj + + +class TestDampingConfig(object): + @classmethod + def setup_class(cls): + print("SETUP") + os.environ["UTILITIES_UNIT_TESTING"] = "1" + + def test_damping_algorithm(self, ctx): + self.basic_check("algo", ["Ethernet0", "aied"], ctx) + self.basic_check("algo", ["Ethernet0", "disabled"], ctx) + + def test_invalid_damping_algorithm(self, ctx): + self.basic_check("algo", ["Ethernet0", "invalid"], ctx, operator.ne) + result = self.basic_check("algo", ["Invalid", "aied"], ctx, op=operator.ne) + assert "Error: Invalid port" in result.output + + def test_invalid_aied_config(self, ctx): + result = self.basic_check("aied-param", ["Invalid"], ctx, op=operator.ne) + assert "Error: Invalid port" in result.output + result = self.basic_check("aied-param", ["Ethernet0"], ctx, op=operator.ne) + assert "Error: Expected at least one valid AIED config parameter" in result.output + result = self.basic_check("aied-param", ["Ethernet0", "--suppress-threshold", "10", "--max-suppress-time", "10", "--decay-half-life", "-1"], ctx, op=operator.ne) + assert "Error: Invalid decay_half_life value -1. It should be >= 0" in result.output + + def test_max_suppress_time_config(self, ctx): + result = self.basic_check("aied-param", ["Ethernet0", "--max-suppress-time", "-1"], ctx, op=operator.ne) + assert "Error: Invalid max_suppress_time value" in result.output + self.basic_check("aied-param", ["Ethernet0", "--max-suppress-time", "50"], ctx) + + def test_decay_half_life_config(self, ctx): + result = self.basic_check("aied-param", ["Ethernet0", "--decay-half-life", "-1"], ctx, op=operator.ne) + assert "Error: Invalid decay_half_life value" in result.output + self.basic_check("aied-param", ["Ethernet0", "--decay-half-life", "50"], ctx) + + def test_suppress_threshold_config(self, ctx): + result = self.basic_check("aied-param", ["Ethernet0", "--suppress-threshold", "-1"], ctx, op=operator.ne) + assert "Error: Invalid suppress_threshold value" in result.output + self.basic_check("aied-param", ["Ethernet0", "--suppress-threshold", "50"], ctx) + + def test_reuse_threshold_config(self, ctx): + result = self.basic_check("aied-param", ["Ethernet0", "--reuse-threshold", "-1"], ctx, op=operator.ne) + assert "Error: Invalid reuse_threshold value" in result.output + self.basic_check("aied-param", ["Ethernet0", "--reuse-threshold", "50"], ctx) + + def test_flap_penalty_config(self, ctx): + result = self.basic_check("aied-param", ["Ethernet0", "--flap-penalty", "-1"], ctx, op=operator.ne) + assert "Error: Invalid flap_penalty value" in result.output + self.basic_check("aied-param", ["Ethernet0", "--flap-penalty", "50"], ctx) + + def test_all_config(self, ctx): + self.basic_check("aied-param", ["Ethernet0", "--decay-half-life", "1001", "--suppress-threshold", "190", "--max-suppress-time", "500", "--flap-penalty", "1000", "--reuse-threshold", "170"], ctx) + + def basic_check(self, command_name, para_list, ctx, op=operator.eq, expect_result=0): + runner = CliRunner() + result = runner.invoke(config.config.commands["interface"].commands["damping"].commands[command_name], para_list, obj = ctx) + print(result.exit_code, result.output) + assert op(result.exit_code, expect_result) + return result + From 47c3529d6dadf6d547728d25b5ea15f971a220a1 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 27 May 2026 10:31:04 +0530 Subject: [PATCH 02/11] Addressing code review comments Signed-off-by: Sivakumar Thirukkanna Thevar --- clear/main.py | 55 +++++ show/interfaces/__init__.py | 139 +++++++++++++ tests/link_event_damping_test.py | 336 +++++++++++++++++++++++++++++++ 3 files changed, 530 insertions(+) create mode 100644 tests/link_event_damping_test.py diff --git a/clear/main.py b/clear/main.py index b4df9e48f5f..d70b7c3bbf4 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: @@ -762,6 +763,60 @@ def asic_sdk_health_event(db, namespace): for key in keys: 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 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)) + sys.exit(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/show/interfaces/__init__.py b/show/interfaces/__init__.py index cb00755a75a..8ee6b72759a 100644 --- a/show/interfaces/__init__.py +++ b/show/interfaces/__init__.py @@ -1453,6 +1453,145 @@ def display_phy_taps_attribute(attr_display_name, attr_json): click.echo(tabulate(body, header, tablefmt='simple', numalign="left")) 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) diff --git a/tests/link_event_damping_test.py b/tests/link_event_damping_test.py new file mode 100644 index 00000000000..8f6abbb6f9c --- /dev/null +++ b/tests/link_event_damping_test.py @@ -0,0 +1,336 @@ +import os +from click.testing import CliRunner +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 eaf92bb34ad27e4d87f82c4907ad50988f108224 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Thu, 28 May 2026 11:52:37 +0530 Subject: [PATCH 03/11] Fix build errors Signed-off-by: Sivakumar Thirukkanna Thevar --- clear/main.py | 1 + config/main.py | 49 ++++++++++++++++++++++++++++---- show/interfaces/__init__.py | 1 + tests/config_int_damping_test.py | 27 ++++++++++++++---- 4 files changed, 67 insertions(+), 11 deletions(-) diff --git a/clear/main.py b/clear/main.py index d70b7c3bbf4..986c564028c 100755 --- a/clear/main.py +++ b/clear/main.py @@ -763,6 +763,7 @@ def asic_sdk_health_event(db, namespace): for key in keys: state_db.delete(state_db.STATE_DB, key); + # # 'interfaces' group ("sonic-clear interfaces ...") # diff --git a/config/main.py b/config/main.py index 3605ed9bb1b..41c1941edca 100644 --- a/config/main.py +++ b/config/main.py @@ -5422,6 +5422,7 @@ def interface_type(ctx, interface_name, interface_type_value, verbose): command += ["-vv"] clicommon.run_command(command, display_cmd=verbose) + # # 'damping' subgroup ('config interface damping ...') # @@ -5432,6 +5433,7 @@ def damping(ctx): """Set interface damping configurations""" pass + # # 'algo' subcommand ('config interface damping algo ...') # @@ -5458,6 +5460,7 @@ def algo(ctx, interface_name, algo_type): config_db.mod_entry("PORT", interface_name, {"link_event_damping_algorithm": algo_type}) + # # 'aied-param' subcommand ('config interface damping aied-param ...') # @@ -5465,12 +5468,45 @@ def algo(ctx, interface_name, algo_type): @damping.command() @click.pass_context @click.argument('interface_name', metavar='', required=True) -@click.option('--max-suppress-time', required=False, type=int, help="Set max suppress time in ms") -@click.option('--decay-half-life', required=False, type=int, help="Set decay half life in ms") -@click.option('--suppress-threshold', required=False, type=int, help="Set suppress threshold") -@click.option('--reuse-threshold', required=False, type=int, help="Set reuse threshold") -@click.option('--flap-penalty', required=False, type=int, help="Set flap penalty") -def aied_param(ctx, interface_name, max_suppress_time, decay_half_life, suppress_threshold, reuse_threshold, flap_penalty): +@click.option( + '--max-suppress-time', + required=False, + type=int, + help="Set max suppress time in ms" + ) +@click.option( + '--decay-half-life', + required=False, + type=int, + help="Set decay half life in ms" + ) +@click.option( + '--suppress-threshold', + required=False, + type=int, + help="Set suppress threshold" + ) +@click.option( + '--reuse-threshold', + required=False, + type=int, + help="Set reuse threshold" + ) +@click.option( + '--flap-penalty', + required=False, + type=int, + help="Set flap penalty" + ) +def aied_param( + ctx, + interface_name, + max_suppress_time, + decay_half_life, + suppress_threshold, + reuse_threshold, + flap_penalty + ): """Set AIED link event damping configuration""" # Get the config_db connector config_db = ctx.obj['config_db'] @@ -5513,6 +5549,7 @@ def aied_param(ctx, interface_name, max_suppress_time, decay_half_life, suppress log.log_info("Executing: interface aied_config {}".format(interface_name)) config_db.mod_entry("PORT", interface_name, config_set) + # # 'advertised-interface-types' subcommand # diff --git a/show/interfaces/__init__.py b/show/interfaces/__init__.py index 8ee6b72759a..e8411871df3 100644 --- a/show/interfaces/__init__.py +++ b/show/interfaces/__init__.py @@ -1453,6 +1453,7 @@ def display_phy_taps_attribute(attr_display_name, attr_json): click.echo(tabulate(body, header, tablefmt='simple', numalign="left")) click.echo("") + # # 'dampening' subcommand ("show interfaces dampening") # diff --git a/tests/config_int_damping_test.py b/tests/config_int_damping_test.py index 06825db6ca6..f516700722f 100644 --- a/tests/config_int_damping_test.py +++ b/tests/config_int_damping_test.py @@ -1,4 +1,3 @@ -import click import config.main as config import operator import os @@ -17,7 +16,7 @@ @pytest.fixture(scope='module') def ctx(scope='module'): db = Db() - obj = {'config_db':db.cfgdb, 'namespace': ''} + obj = {'config_db': db.cfgdb, 'namespace': ''} yield obj @@ -41,7 +40,16 @@ def test_invalid_aied_config(self, ctx): assert "Error: Invalid port" in result.output result = self.basic_check("aied-param", ["Ethernet0"], ctx, op=operator.ne) assert "Error: Expected at least one valid AIED config parameter" in result.output - result = self.basic_check("aied-param", ["Ethernet0", "--suppress-threshold", "10", "--max-suppress-time", "10", "--decay-half-life", "-1"], ctx, op=operator.ne) + result = self.basic_check( + "aied-param", + [ + "Ethernet0", + "--suppress-threshold", "10", + "--max-suppress-time", "10", + "--decay-half-life", "-1" + ], + ctx, + op=operator.ne) assert "Error: Invalid decay_half_life value -1. It should be >= 0" in result.output def test_max_suppress_time_config(self, ctx): @@ -70,7 +78,17 @@ def test_flap_penalty_config(self, ctx): self.basic_check("aied-param", ["Ethernet0", "--flap-penalty", "50"], ctx) def test_all_config(self, ctx): - self.basic_check("aied-param", ["Ethernet0", "--decay-half-life", "1001", "--suppress-threshold", "190", "--max-suppress-time", "500", "--flap-penalty", "1000", "--reuse-threshold", "170"], ctx) + self.basic_check( + "aied-param", + [ + "Ethernet0", + "--decay-half-life", "1001", + "--suppress-threshold", "190", + "--max-suppress-time", "500", + "--flap-penalty", "1000", + "--reuse-threshold", "170" + ], + ctx) def basic_check(self, command_name, para_list, ctx, op=operator.eq, expect_result=0): runner = CliRunner() @@ -78,4 +96,3 @@ def basic_check(self, command_name, para_list, ctx, op=operator.eq, expect_resul print(result.exit_code, result.output) assert op(result.exit_code, expect_result) return result - From 25a114e94145a4ed1841e8aee686b6315511e4f2 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 3 Jun 2026 14:46:53 +0530 Subject: [PATCH 04/11] Fixing flake8 errors Signed-off-by: Sivakumar Thirukkanna Thevar --- tests/config_int_damping_test.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/config_int_damping_test.py b/tests/config_int_damping_test.py index f516700722f..d44dc0f4641 100644 --- a/tests/config_int_damping_test.py +++ b/tests/config_int_damping_test.py @@ -41,7 +41,7 @@ def test_invalid_aied_config(self, ctx): result = self.basic_check("aied-param", ["Ethernet0"], ctx, op=operator.ne) assert "Error: Expected at least one valid AIED config parameter" in result.output result = self.basic_check( - "aied-param", + "aied-param", [ "Ethernet0", "--suppress-threshold", "10", @@ -92,7 +92,13 @@ def test_all_config(self, ctx): def basic_check(self, command_name, para_list, ctx, op=operator.eq, expect_result=0): runner = CliRunner() - result = runner.invoke(config.config.commands["interface"].commands["damping"].commands[command_name], para_list, obj = ctx) + result = runner.invoke( + config.config.commands["interface"] + .commands["damping"] + .commands[command_name], + para_list, + obj=ctx + ) print(result.exit_code, result.output) assert op(result.exit_code, expect_result) return result From 2ae5a3da661d63318591d83a6bbde882b85a7041 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 3 Jun 2026 15:37:30 +0530 Subject: [PATCH 05/11] Fixing indentation issue Signed-off-by: Sivakumar Thirukkanna Thevar --- config/main.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/config/main.py b/config/main.py index 41c1941edca..a3e9805de6c 100644 --- a/config/main.py +++ b/config/main.py @@ -5473,31 +5473,31 @@ def algo(ctx, interface_name, algo_type): required=False, type=int, help="Set max suppress time in ms" - ) +) @click.option( '--decay-half-life', required=False, type=int, help="Set decay half life in ms" - ) +) @click.option( '--suppress-threshold', required=False, type=int, help="Set suppress threshold" - ) +) @click.option( '--reuse-threshold', required=False, type=int, help="Set reuse threshold" - ) +) @click.option( '--flap-penalty', required=False, type=int, help="Set flap penalty" - ) +) def aied_param( ctx, interface_name, @@ -5506,7 +5506,7 @@ def aied_param( suppress_threshold, reuse_threshold, flap_penalty - ): +): """Set AIED link event damping configuration""" # Get the config_db connector config_db = ctx.obj['config_db'] From 177b024e50f0e52a695d4ffce1b9ada197e56a4c Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 3 Jun 2026 16:42:18 +0530 Subject: [PATCH 06/11] Modified the link_event_damping_test.py to suite with damping CLIs Signed-off-by: Sivakumar Thirukkanna Thevar --- tests/link_event_damping_test.py | 335 +++++++++---------------------- 1 file changed, 97 insertions(+), 238 deletions(-) diff --git a/tests/link_event_damping_test.py b/tests/link_event_damping_test.py index 8f6abbb6f9c..f301e7dc8f2 100644 --- a/tests/link_event_damping_test.py +++ b/tests/link_event_damping_test.py @@ -1,14 +1,11 @@ import os from click.testing import CliRunner -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): +class TestConfigInterfaceDamping(object): @classmethod def setup_class(cls): os.environ['UTILITIES_UNIT_TESTING'] = "1" @@ -17,320 +14,182 @@ def setup_class(cls): def teardown_class(cls): os.environ['UTILITIES_UNIT_TESTING'] = "0" - def test_enable_dampening_defaults(self): - """Test enabling dampening with default parameters""" + def test_set_algo_aied(self): + """Test setting damping algorithm to AIED""" runner = CliRunner() db = Db() obj = {'config_db': db.cfgdb} result = runner.invoke( - config.config.commands["interface"].commands["dampening"].commands["enable"], - ["Ethernet0"], obj=obj) + config.config.commands["interface"].commands["damping"].commands["algo"], + ["Ethernet0", "aied"], + 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""" + def test_set_algo_disabled(self): + """Test disabling damping""" 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) + config.config.commands["interface"].commands["damping"].commands["algo"], + ["Ethernet0", "disabled"], + 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""" + def test_set_algo_invalid_interface(self): + """Test invalid interface for algo""" runner = CliRunner() db = Db() obj = {'config_db': db.cfgdb} result = runner.invoke( - config.config.commands["interface"].commands["dampening"].commands["enable"], - ["Ethernet0", "--monitor"], obj=obj) + config.config.commands["interface"].commands["damping"].commands["algo"], + ["EthernetINVALID", "aied"], + obj=obj + ) print(result.exit_code, result.output) - assert result.exit_code == 0 - assert "monitor" in result.output.lower() + assert result.exit_code != 0 - def test_enable_dampening_invalid_interface(self): - """Test enabling dampening on non-existent interface""" + def test_set_aied_params_valid(self): + """Test setting valid AIED parameters""" runner = CliRunner() db = Db() obj = {'config_db': db.cfgdb} result = runner.invoke( - config.config.commands["interface"].commands["dampening"].commands["enable"], - ["EthernetINVALID"], obj=obj) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + [ + "Ethernet0", + "--suppress-threshold", "1400", + "--decay-half-life", "20000", + "--max-suppress-time", "40000", + "--flap-penalty", "1000", + "--reuse-threshold", "1100" + ], + obj=obj + ) print(result.exit_code, result.output) - assert result.exit_code != 0 - assert "does not exist" in result.output + assert result.exit_code == 0 - def test_enable_dampening_reuse_ge_suppress(self): - """Test that reuse >= suppress is rejected""" + def test_set_aied_params_single_param(self): + """Test setting only one parameter""" 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) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + ["Ethernet0", "--decay-half-life", "20000"], + obj=obj + ) print(result.exit_code, result.output) - assert result.exit_code != 0 - assert "Reuse threshold" in result.output + assert result.exit_code == 0 - def test_enable_dampening_halflife_gt_maxsuppress(self): - """Test that half-life > max-suppress-time is rejected""" + def test_aied_params_no_args(self): + """Test failure if no params provided""" 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) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + ["Ethernet0"], + obj=obj + ) print(result.exit_code, result.output) assert result.exit_code != 0 - assert "Half-life" in result.output + assert "Expected at least one valid AIED config parameter" in result.output - def test_enable_dampening_invalid_halflife(self): - """Test that out-of-range half-life is rejected by click.IntRange""" + def test_aied_negative_value(self): + """Test invalid negative values""" 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) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + ["Ethernet0", "--flap-penalty", "-1"], + obj=obj + ) print(result.exit_code, result.output) assert result.exit_code != 0 - assert "Invalid value" in result.output + assert "Invalid flap_penalty" 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""" +def test_aied_reuse_ge_suppress(self): + """Test that reuse_threshold >= suppress_threshold is rejected""" runner = CliRunner() db = Db() obj = {'config_db': db.cfgdb} result = runner.invoke( - config.config.commands["interface"].commands["dampening"].commands["disable"], - ["Ethernet0"], obj=obj) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + [ + "Ethernet0", + "--reuse-threshold", "3000", + "--suppress-threshold", "2000" + ], + obj=obj + ) print(result.exit_code, result.output) - assert result.exit_code == 0 - assert "disabled" in result.output + assert result.exit_code != 0 + assert "reuse_threshold" in result.output.lower() or "Reuse threshold" in result.output + - def test_disable_dampening_invalid_interface(self): - """Test disabling dampening on non-existent interface""" + def test_aied_halflife_gt_maxsuppress(self): + """Test that decay_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["disable"], - ["EthernetINVALID"], obj=obj) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + [ + "Ethernet0", + "--decay-half-life", "30000", + "--max-suppress-time", "20000" + ], + obj=obj + ) print(result.exit_code, result.output) assert result.exit_code != 0 - assert "does not exist" in result.output + assert "decay_half_life" in result.output.lower() or "Half-life" 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""" + def test_aied_invalid_halflife(self): + """Test invalid (negative) decay_half_life""" 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 + obj = {'config_db': db.cfgdb} - 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) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + [ + "Ethernet0", + "--decay-half-life", "-1" + ], + obj=obj + ) print(result.exit_code, result.output) assert result.exit_code != 0 + assert "Invalid decay_half_life" in result.output or "Invalid value" in result.output - def test_show_dampening_configured(self): - """Test show when dampening is configured on an interface""" + def test_aied_invalid_interface(self): + """Test invalid interface for aied-param""" runner = CliRunner() db = Db() + obj = {'config_db': db.cfgdb} - # 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"]) + config.config.commands["interface"].commands["damping"].commands["aied-param"], + ["EthernetINVALID", "--decay-half-life", "1000"], + obj=obj + ) print(result.exit_code, result.output) assert result.exit_code != 0 - assert "does not exist" in result.output From 1c0466700925ce8f7d794d91cddd5333ae03964f Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 3 Jun 2026 17:21:34 +0530 Subject: [PATCH 07/11] Fixing indentation issue in link_event_damping_test.py Signed-off-by: Sivakumar Thirukkanna Thevar --- tests/link_event_damping_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/link_event_damping_test.py b/tests/link_event_damping_test.py index f301e7dc8f2..0eb661f7389 100644 --- a/tests/link_event_damping_test.py +++ b/tests/link_event_damping_test.py @@ -122,7 +122,7 @@ def test_aied_negative_value(self): assert "Invalid flap_penalty" in result.output -def test_aied_reuse_ge_suppress(self): + def test_aied_reuse_ge_suppress(self): """Test that reuse_threshold >= suppress_threshold is rejected""" runner = CliRunner() db = Db() From 413ad2213e3b7d42eada06639d91d4fae2acb23f Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 3 Jun 2026 18:28:50 +0530 Subject: [PATCH 08/11] Adding checks for damping config params Signed-off-by: Sivakumar Thirukkanna Thevar --- config/main.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/config/main.py b/config/main.py index a3e9805de6c..4449b9f7872 100644 --- a/config/main.py +++ b/config/main.py @@ -5543,6 +5543,20 @@ def aied_param( ctx.fail("Invalid flap_penalty value {}. It should be >= 0".format(flap_penalty)) config_set['flap_penalty'] = flap_penalty + if reuse_threshold is not None and suppress_threshold is not None: + if reuse_threshold >= suppress_threshold: + ctx.fail( + "Invalid configuration: reuse_threshold ({}) must be less than suppress_threshold ({})" + .format(reuse_threshold, suppress_threshold) + ) + + if decay_half_life is not None and max_suppress_time is not None: + if decay_half_life > max_suppress_time: + ctx.fail( + "Invalid configuration: decay_half_life ({}) must be <= max_suppress_time ({})" + .format(decay_half_life, max_suppress_time) + ) + if len(config_set) == 0: ctx.fail("Expected at least one valid AIED config parameter") From 993034a6a25d19184912386de13f4f80cfec9501 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Wed, 3 Jun 2026 19:40:43 +0530 Subject: [PATCH 09/11] Fixed invalid config params and extra blank lines Signed-off-by: Sivakumar Thirukkanna Thevar --- tests/config_int_damping_test.py | 8 ++++---- tests/link_event_damping_test.py | 3 --- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/tests/config_int_damping_test.py b/tests/config_int_damping_test.py index d44dc0f4641..9b126c01fa2 100644 --- a/tests/config_int_damping_test.py +++ b/tests/config_int_damping_test.py @@ -82,11 +82,11 @@ def test_all_config(self, ctx): "aied-param", [ "Ethernet0", - "--decay-half-life", "1001", - "--suppress-threshold", "190", - "--max-suppress-time", "500", + "--decay-half-life", "15000", + "--suppress-threshold", "1600", + "--max-suppress-time", "30000", "--flap-penalty", "1000", - "--reuse-threshold", "170" + "--reuse-threshold", "1200" ], ctx) diff --git a/tests/link_event_damping_test.py b/tests/link_event_damping_test.py index 0eb661f7389..ee6122680a7 100644 --- a/tests/link_event_damping_test.py +++ b/tests/link_event_damping_test.py @@ -121,7 +121,6 @@ def test_aied_negative_value(self): assert result.exit_code != 0 assert "Invalid flap_penalty" in result.output - def test_aied_reuse_ge_suppress(self): """Test that reuse_threshold >= suppress_threshold is rejected""" runner = CliRunner() @@ -141,7 +140,6 @@ def test_aied_reuse_ge_suppress(self): assert result.exit_code != 0 assert "reuse_threshold" in result.output.lower() or "Reuse threshold" in result.output - def test_aied_halflife_gt_maxsuppress(self): """Test that decay_half_life > max_suppress_time is rejected""" runner = CliRunner() @@ -161,7 +159,6 @@ def test_aied_halflife_gt_maxsuppress(self): assert result.exit_code != 0 assert "decay_half_life" in result.output.lower() or "Half-life" in result.output - def test_aied_invalid_halflife(self): """Test invalid (negative) decay_half_life""" runner = CliRunner() From 74dc6a4cdca6a2bd7ffb2f7e661096e1555e041a Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Thu, 4 Jun 2026 12:14:02 +0530 Subject: [PATCH 10/11] Removing show & clear CLI related code Signed-off-by: Sivakumar Thirukkanna Thevar --- clear/main.py | 55 -------------- show/interfaces/__init__.py | 140 ------------------------------------ 2 files changed, 195 deletions(-) diff --git a/clear/main.py b/clear/main.py index 986c564028c..9fb9fe2a0c1 100755 --- a/clear/main.py +++ b/clear/main.py @@ -764,60 +764,5 @@ 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 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)) - sys.exit(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/show/interfaces/__init__.py b/show/interfaces/__init__.py index e8411871df3..cb00755a75a 100644 --- a/show/interfaces/__init__.py +++ b/show/interfaces/__init__.py @@ -1454,146 +1454,6 @@ 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 30e05da77935c9e337c04b7d14d4274deb1a27a9 Mon Sep 17 00:00:00 2001 From: Sivakumar Thirukkanna Thevar Date: Thu, 4 Jun 2026 12:17:06 +0530 Subject: [PATCH 11/11] Removing the missed line in clear.py Signed-off-by: Sivakumar Thirukkanna Thevar --- clear/main.py | 1 - 1 file changed, 1 deletion(-) diff --git a/clear/main.py b/clear/main.py index 9fb9fe2a0c1..b4df9e48f5f 100755 --- a/clear/main.py +++ b/clear/main.py @@ -11,7 +11,6 @@ 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: