[config][show] cli support for retrieving ber, eye-info and configuring prbs, loopback on Y-cable

config/muxcable.py

@@ -5,7 +5,6 @@
 import click
 import utilities_common.cli as clicommon
 from sonic_py_common import multi_asic
-from sonic_y_cable import y_cable
 from swsssdk import ConfigDBConnector
 from swsscommon import swsscommon
 from tabulate import tabulate
@@ -196,25 +195,53 @@ def mode(state, port, json_output):
 @click.argument('target', required=True, default=None, type=click.INT)
 @click.argument('mode_value', required=True, default=None, type=click.INT)
 @click.argument('lane_map', required=True, default=None, type=click.INT)
-def prbs(port, target, mode_value, lane_map):
+def enable_prbs(port, target, mode_value, lane_map):
 
-    res = y_cable.enable_prbs_mode(port, target, mode_value, lane_map)
+    import sonic_y_cable.y_cable
+    res = sonic_y_cable.y_cable.enable_prbs_mode(port, target, mode_value, lane_map)
     if res != True:
         click.echo("PRBS config unsuccesful")
         sys.exit(CONFIG_FAIL)
     click.echo("PRBS config sucessful")
     sys.exit(CONFIG_SUCCESSFUL)
 
+@muxcable.command()
+@click.argument('port', required=True, default=None, type=click.INT)
+@click.argument('target', required=True, default=None, type=click.INT)
+def disable_prbs(port, target):
+
+    import sonic_y_cable.y_cable
+    res = sonic_y_cable.y_cable.disable_prbs_mode(port, target)
+    if res != True:
+        click.echo("PRBS disable unsuccesful")
+        sys.exit(CONFIG_FAIL)
+    click.echo("PRBS disbale sucessful")
+    sys.exit(CONFIG_SUCCESSFUL)
+
 
 @muxcable.command()
 @click.argument('port', required=True, default=None, type=click.INT)
 @click.argument('target', required=True, default=None, type=click.INT)
 @click.argument('lane_map', required=True, default=None, type=click.INT)
-def loopback(port, target, lane_map):
+def enable_loopback(port, target, lane_map):
 
-    res = y_cable.enable_loopback_mode(port, target, lane_map)
+    import sonic_y_cable.y_cable
+    res = sonic_y_cable.y_cable.enable_loopback_mode(port, target, lane_map)
     if res != True:
         click.echo("loopback config unsuccesful")
         sys.exit(CONFIG_FAIL)
     click.echo("loopback config sucessful")
     sys.exit(CONFIG_SUCCESSFUL)
+
+@muxcable.command()
+@click.argument('port', required=True, default=None, type=click.INT)
+@click.argument('target', required=True, default=None, type=click.INT)
+def disable_loopback(port, target):
+
+    import sonic_y_cable.y_cable
+    res = sonic_y_cable.y_cable.disable_loopback_mode(port, target)
+    if res != True:
+        click.echo("loopback disable unsuccesful")
+        sys.exit(CONFIG_FAIL)
+    click.echo("loopback disable sucessful")
+    sys.exit(CONFIG_SUCCESSFUL)

show/muxcable.py

@@ -5,7 +5,6 @@
 import click
 import utilities_common.cli as clicommon
 from sonic_py_common import multi_asic
-from sonic_y_cable import y_cable
 from swsscommon import swsscommon
 from swsssdk import ConfigDBConnector
 from tabulate import tabulate
@@ -350,14 +349,16 @@ def berinfo(port, target):
     if os.geteuid() != 0:
         click.echo("Root privileges are required for this operation")
         sys.exit(EXIT_FAIL)
-    res = y_cable.get_ber_info(port, target)
+    import sonic_y_cable.y_cable
+    res = sonic_y_cable.y_cable.get_ber_info(port, target)
     if res == False or res == -1:
         click.echo("Unable to fetch ber info")
         sys.exit(EXIT_FAIL)
     headers = ['Lane1', 'Lane2']
     lane_data = []
     lane_data.append(res)
     click.echo(tabulate(lane_data, headers=headers))
+    sys.exit(EXIT_SUCCESS)
 
 
 @muxcable.command()
@@ -368,11 +369,13 @@ def eyeinfo(port, target):
     if os.geteuid() != 0:
         click.echo("Root privileges are required for this operation")
         sys.exit(EXIT_FAIL)
-    res = y_cable.get_eye_info(port, target)
+    import sonic_y_cable.y_cable
+    res = sonic_y_cable.y_cable.get_eye_info(port, target)
     if res == False or res == -1:
         click.echo("Unable to fetch eye info")
         sys.exit(EXIT_FAIL)
     headers = ['Lane1', 'Lane2']
     lane_data = []
     lane_data.append(res)
     click.echo(tabulate(lane_data, headers=headers))
+    sys.exit(EXIT_SUCCESS)

tests/muxcable_test.py

@@ -10,6 +10,11 @@
 sys.modules['sonic_platform_base'] = mock.Mock()
 sys.modules['sonic_platform_base.sonic_sfp'] = mock.Mock()
 sys.modules['sonic_platform_base.sonic_sfp.sfputilhelper'] = mock.Mock()
+sys.modules['sonic_y_cable'] = mock.Mock()
+sys.modules['y_cable'] = mock.Mock()
+sys.modules['sonic_y_cable.y_cable'] = mock.Mock()
+#sys.modules['os'] = mock.Mock()
+#sys.modules['os.geteuid'] = mock.Mock()
 #sys.modules['platform_sfputil'] = mock.Mock()
 import config.main as config
 import show.main as show
@@ -416,6 +421,84 @@ def test_config_muxcable_tabular_port_with_incorrect_port(self):
 
         assert(result.exit_code == 1)
 
+    def test_show_muxcable_eye_info(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('os.geteuid') as patched_util_outer:
+            os.geteuid.return_value = 0
+            with mock.patch('sonic_y_cable.y_cable') as patched_util:
+                patched_util.get_eye_info.return_value = [0, 0]
+                result = runner.invoke(show.cli.commands["muxcable"].commands["eyeinfo"], [
+                                       "0", "0"], obj=db)
+
+        assert(result.exit_code == 0)
+
+    def test_show_muxcable_ber_info(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('os.geteuid') as patched_util_outer:
+            os.geteuid.return_value = 0
+            with mock.patch('sonic_y_cable.y_cable') as patched_util:
+                patched_util.get_ber_info.return_value = [0, 0]
+                result = runner.invoke(show.cli.commands["muxcable"].commands["berinfo"], [
+                                       "0", "0"], obj=db)
+
+        assert(result.exit_code == 0)
+
+    def test_config_muxcable_enable_prbs(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('os.geteuid') as patched_util_outer:
+            os.geteuid.return_value = 0
+            with mock.patch('sonic_y_cable.y_cable') as patched_util:
+                patched_util.enable_prbs_mode.return_value = 1
+                result = runner.invoke(config.config.commands["muxcable"].commands["enable-prbs"], [
+                                       "0", "0", "0", "0"], obj=db)
+
+        assert(result.exit_code == 100)
+
+    def test_config_muxcable_enable_loopback(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('os.geteuid') as patched_util_outer:
+            os.geteuid.return_value = 0
+            with mock.patch('sonic_y_cable.y_cable') as patched_util:
+                patched_util.enable_loopback_mode.return_value = 1
+                result = runner.invoke(config.config.commands["muxcable"].commands["enable-loopback"], [
+                                       "0", "0", "0"], obj=db)
+
+        assert(result.exit_code == 100)
+
+    def test_config_muxcable_disble_prbs(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('os.geteuid') as patched_util_outer:
+            os.geteuid.return_value = 0
+            with mock.patch('sonic_y_cable.y_cable') as patched_util:
+                patched_util.disable_prbs_mode.return_value = 1
+                result = runner.invoke(config.config.commands["muxcable"].commands["disable-prbs"], [
+                                       "0", "0"], obj=db)
+
+        assert(result.exit_code == 100)
+
+    def test_config_muxcable_disable_loopback(self):
+        runner = CliRunner()
+        db = Db()
+
+        with mock.patch('os.geteuid') as patched_util_outer:
+            os.geteuid.return_value = 0
+            with mock.patch('sonic_y_cable.y_cable') as patched_util:
+                patched_util.disable_loopback_mode.return_value = 1
+                result = runner.invoke(config.config.commands["muxcable"].commands["disable-loopback"], [
+                                       "0", "0"], obj=db)
+
+        assert(result.exit_code == 100)
+
     @classmethod
     def teardown_class(cls):
         os.environ['UTILITIES_UNIT_TESTING'] = "0"