[config][muxcable] add support to enable/disable ycable telemetry (sonic-net#2297)

This PR provides a capability to sonic-utilities CLI to enable/disable telemetry for ycabled.
Basically there is a periodic loop for ycabled which posts telemetry data for that configured interval of time(currently 60 sec). This PR diables this data posting, and does not call platform API calls for ycable.
This PR is required for the initiative of getting some failover/switchover not get interfered because of sonic-telemetry API calls.
The CLI for enabling/disabling telemetry is
config muxcable telemetry enable/disable

What I did
How I did it
How to verify it
Previous command output (if the output of a command-line utility has changed)
New command output (if the output of a command-line utility has changed)
Dependent on sonic-net/sonic-platform-daemons#279 and submodule update

Signed-off-by: vaibhav-dahiya <[email protected]>

Author: vdahiya12

config/muxcable.py

@@ -1190,3 +1190,46 @@ def set_fec(db, port, target, mode):
         else:
             click.echo("ERR: Unable to set fec enable/disable port {} to {}".format(port, mode))
             sys.exit(CONFIG_FAIL)
+
+def update_configdb_ycable_telemetry_data(config_db, key, val):
+    log_verbosity = get_value_for_key_in_config_tbl(config_db, key, "log_verbosity", "XCVRD_LOG")
+
+    config_db.set_entry("XCVRD_LOG", key, {"log_verbosity": log_verbosity,
+                                                "disable_telemetry": val})
+    return 0
+
+@muxcable.command()
+@click.argument('state', metavar='<enable/disable telemetry>', required=True, type=click.Choice(["enable", "disable"]))
+@clicommon.pass_db
+def telemetry(db, state):
+    """Enable/Disable Telemetry for ycabled """
+
+    per_npu_configdb = {}
+    xcvrd_log_cfg_db_tbl = {}
+
+    if state == 'enable':
+        val = 'False'
+    elif state == 'disable':
+        val = 'True'
+
+
+    # Getting all front asic namespace and correspding config and state DB connector
+
+    namespaces = multi_asic.get_front_end_namespaces()
+    for namespace in namespaces:
+        asic_id = multi_asic.get_asic_index_from_namespace(namespace)
+        # replace these with correct macros
+        per_npu_configdb[asic_id] = ConfigDBConnector(use_unix_socket_path=True, namespace=namespace)
+        per_npu_configdb[asic_id].connect()
+
+        xcvrd_log_cfg_db_tbl[asic_id] = per_npu_configdb[asic_id].get_table("XCVRD_LOG")
+
+    asic_index = multi_asic.get_asic_index_from_namespace(EMPTY_NAMESPACE)
+    rc = update_configdb_ycable_telemetry_data(per_npu_configdb[asic_index], "Y_CABLE", val)
+
+
+    if rc == 0:
+        click.echo("Success in ycabled telemetry state to {}".format(state))
+    else:
+        click.echo("ERR: Unable to set ycabled telemetry state to {}".format(state))
+        sys.exit(CONFIG_FAIL)

tests/muxcable_test.py

@@ -2095,6 +2095,94 @@ def test_show_muxcable_packetloss_port_json(self):
         assert result.exit_code == 0
         assert result.output == show_muxcable_packetloss_expected_output_json
 
+    @mock.patch('utilities_common.platform_sfputil_helper.get_logical_list', mock.MagicMock(return_value=["Ethernet0", "Ethernet12"]))
+    @mock.patch('utilities_common.platform_sfputil_helper.get_asic_id_for_logical_port', mock.MagicMock(return_value=0))
+    @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]}))
+    @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0]))
+    def test_show_muxcable_tunnel_route(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(show.cli.commands["muxcable"].commands["tunnel-route"], obj=db)
+
+        assert result.exit_code == 0
+        assert result.output == show_muxcable_tunnel_route_expected_output
+    
+    @mock.patch('utilities_common.platform_sfputil_helper.get_logical_list', mock.MagicMock(return_value=["Ethernet0", "Ethernet12"]))
+    @mock.patch('utilities_common.platform_sfputil_helper.get_asic_id_for_logical_port', mock.MagicMock(return_value=0))
+    @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]}))
+    @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0]))
+    def test_show_muxcable_tunnel_route_json(self):
+        runner = CliRunner()
+        db = Db()
+        
+        result = runner.invoke(show.cli.commands["muxcable"].commands["tunnel-route"],
+                               ["--json"], obj=db)
+
+        assert result.exit_code == 0
+        assert result.output == show_muxcable_tunnel_route_expected_output_json
+
+    @mock.patch('utilities_common.platform_sfputil_helper.get_logical_list', mock.MagicMock(return_value=["Ethernet0", "Ethernet12"]))
+    @mock.patch('utilities_common.platform_sfputil_helper.get_asic_id_for_logical_port', mock.MagicMock(return_value=0))
+    @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]}))
+    @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0]))
+    def test_show_muxcable_tunnel_route_port(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(show.cli.commands["muxcable"].commands["tunnel-route"],
+                               ["Ethernet0"], obj=db)
+
+        assert result.exit_code == 0
+        assert result.output == show_muxcable_tunnel_route_expected_port_output
+
+    @mock.patch('utilities_common.platform_sfputil_helper.get_logical_list', mock.MagicMock(return_value=["Ethernet0", "Ethernet12"]))
+    @mock.patch('utilities_common.platform_sfputil_helper.get_asic_id_for_logical_port', mock.MagicMock(return_value=0))
+    @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]}))
+    @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0]))
+    def test_show_muxcable_tunnel_route_json_port(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(show.cli.commands["muxcable"].commands["tunnel-route"],
+                               ["Ethernet0", "--json"], obj=db)
+        assert result.exit_code == 0
+        assert result.output == show_muxcable_tunnel_route_expected_output_port_json
+
+    @mock.patch('config.muxcable.swsscommon.DBConnector', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.swsscommon.Table', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.swsscommon.Select', mock.MagicMock(return_value=0))
+    def test_config_muxcable_telemetry_enable_without_patch(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(config.config.commands["muxcable"].commands["telemetry"], [
+                               "enable"], obj=db)
+        assert result.exit_code == 1
+
+    @mock.patch('config.muxcable.swsscommon.DBConnector', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.swsscommon.Table', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.swsscommon.Select', mock.MagicMock(return_value=0))
+    def test_config_muxcable_telemetry_disable_without_patch(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(config.config.commands["muxcable"].commands["telemetry"], [
+                               "disable"], obj=db)
+        assert result.exit_code == 1
+
+    @mock.patch('config.muxcable.swsscommon.DBConnector', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.swsscommon.Table', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.swsscommon.Select', mock.MagicMock(return_value=0))
+    @mock.patch('config.muxcable.update_configdb_ycable_telemetry_data', mock.MagicMock(return_value=0))
+    def test_config_muxcable_telemetry_enable(self):
+        runner = CliRunner()
+        db = Db()
+
+        result = runner.invoke(config.config.commands["muxcable"].commands["telemetry"], [
+                               "enable"], obj=db)
+        assert result.exit_code == 0
+
     @classmethod
     def teardown_class(cls):
         os.environ['UTILITIES_UNIT_TESTING'] = "0"