diff --git a/config/muxcable.py b/config/muxcable.py index e5638c1f03..a9ace2a569 100644 --- a/config/muxcable.py +++ b/config/muxcable.py @@ -29,9 +29,19 @@ # Helper functions + def db_connect(db_name, namespace=EMPTY_NAMESPACE): return swsscommon.DBConnector(db_name, REDIS_TIMEOUT_MSECS, True, namespace) + +target_dict = { "NIC":"0", + "TORA":"1", + "TORB":"2", + "LOCAL":"3"} + +def parse_target(target): + return target_dict.get(target, None) + def get_value_for_key_in_dict(mdict, port, key, table_name): value = mdict.get(key, None) if value is None: @@ -39,6 +49,7 @@ def get_value_for_key_in_dict(mdict, port, key, table_name): sys.exit(CONFIG_FAIL) return value + def delete_all_keys_in_db_table(db_type, table_name): redis_db = {} @@ -55,13 +66,13 @@ def delete_all_keys_in_db_table(db_type, table_name): table[asic_id]._del(key) -def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_name, rsp_table_name, port, cmd_timeout_secs, arg=None): +def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_name, cmd_arg_table_name, rsp_table_name, port, cmd_timeout_secs, param_dict=None, arg=None): res_dict = {} state_db, appl_db = {}, {} firmware_rsp_tbl, firmware_rsp_tbl_keys = {}, {} firmware_rsp_sub_tbl = {} - firmware_cmd_tbl = {} + firmware_cmd_tbl, firmware_cmd_arg_tbl = {}, {} CMD_TIMEOUT_SECS = cmd_timeout_secs @@ -74,6 +85,8 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ state_db[asic_id] = db_connect("STATE_DB", namespace) appl_db[asic_id] = db_connect("APPL_DB", namespace) firmware_cmd_tbl[asic_id] = swsscommon.Table(appl_db[asic_id], cmd_table_name) + if cmd_arg_table_name is not None: + firmware_cmd_arg_tbl[asic_id] = swsscommon.Table(appl_db[asic_id], cmd_arg_table_name) firmware_rsp_sub_tbl[asic_id] = swsscommon.SubscriberStateTable(state_db[asic_id], rsp_table_name) firmware_rsp_tbl[asic_id] = swsscommon.Table(state_db[asic_id], rsp_table_name) firmware_rsp_tbl_keys[asic_id] = firmware_rsp_tbl[asic_id].getKeys() @@ -109,6 +122,11 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ else: cmd_arg = str(arg) + if param_dict is not None: + for key, value in param_dict.items(): + fvs = swsscommon.FieldValuePairs([(str(key), str(value))]) + firmware_cmd_arg_tbl[asic_index].set(port, fvs) + fvs = swsscommon.FieldValuePairs([(cmd_name, cmd_arg)]) firmware_cmd_tbl[asic_index].set(port, fvs) @@ -161,7 +179,6 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ # check if xcvrd got a probe command result = fvp_dict[rsp_name] - if result == exp_rsp: res_dict[1] = result res_dict[0] = 0 @@ -186,6 +203,7 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ return res_dict + def get_value_for_key_in_config_tbl(config_db, port, key, table): info_dict = {} info_dict = config_db.get_entry(table, port) @@ -201,6 +219,7 @@ def get_value_for_key_in_config_tbl(config_db, port, key, table): # 'muxcable' command ("config muxcable") # + @click.group(name='muxcable', cls=clicommon.AliasedGroup) def muxcable(): """SONiC command line - 'show muxcable' command""" @@ -340,35 +359,96 @@ def prbs(): @prbs.command() -@click.argument('port', required=True, default=None, type=click.INT) -@click.argument('target', required=True, default=None, type=click.INT) +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) @click.argument('mode_value', required=True, default=None, type=click.INT) -@click.argument('lane_map', required=True, default=None, type=click.INT) -def enable(port, target, mode_value, lane_map): - """Enable PRBS mode on a port""" - - 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) +@click.argument('lane_mask', required=True, default=None, type=click.INT) +@click.argument('prbs_direction', metavar=' PRBS_DIRECTION_BOTH = 0 PRBS_DIRECTION_GENERATOR = 1 PRBS_DIRECTION_CHECKER = 2', required=False, default="0", type=click.Choice(["0", "1", "2"])) +@clicommon.pass_db +def enable(db, port, target, mode_value, lane_mask, prbs_direction): + """Enable PRBS mode on a port args port target mode_value lane_mask prbs_direction + example sudo config mux prbs enable Ethernet48 0 3 3 0 + """ + + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + if port is not None: + click.confirm(('Muxcable at port {} will be changed to PRBS mode {} state; disable traffic Continue?'.format( + port, mode_value)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + param_dict["mode_value"] = mode_value + param_dict["lane_mask"] = lane_mask + param_dict["direction"] = prbs_direction + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_prbs", "status", "True", "XCVRD_CONFIG_PRBS_CMD", "XCVRD_CONFIG_PRBS_CMD_ARG", "XCVRD_CONFIG_PRBS_RSP", port, 30, param_dict, "enable") + + rc = res_dict[0] + + port = platform_sfputil_helper.get_interface_alias(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + if rc == 0: + click.echo("Success in PRBS mode port {} to {}".format(port, mode_value)) + else: + click.echo("ERR: Unable to set PRBS mode port {} to {}".format(port, mode_value)) + sys.exit(CONFIG_FAIL) @prbs.command() -@click.argument('port', required=True, default=None, type=click.INT) -@click.argument('target', required=True, default=None, type=click.INT) -def disable(port, target): - """Disable PRBS mode on a port""" - - 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 disable sucessful") - sys.exit(CONFIG_SUCCESSFUL) +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.argument('prbs_direction', metavar=' PRBS_DIRECTION_BOTH = 0 PRBS_DIRECTION_GENERATOR = 1 PRBS_DIRECTION_CHECKER = 2', required=False, default="0", type=click.Choice(["0", "1", "2"])) +@clicommon.pass_db +def disable(db, port, target, prbs_direction): + """Disable PRBS mode on a port + example sudo config mux prbs disable Ethernet48 0 + """ + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + if port is not None: + click.confirm(('Muxcable at port {} will be changed to disable PRBS mode {} target; disable traffic Continue?'.format( + port, target)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + param_dict["direction"] = prbs_direction + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_prbs", "status", "True", "XCVRD_CONFIG_PRBS_CMD", "XCVRD_CONFIG_PRBS_CMD_ARG", "XCVRD_CONFIG_PRBS_RSP", port, 30, param_dict, "disable") + + rc = res_dict[0] + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + port = platform_sfputil_helper.get_interface_alias(port, db) + if rc == 0: + click.echo("Success in disable PRBS mode port {} on target {}".format(port, target)) + else: + click.echo("ERR: Unable to disable PRBS mode port {} on target {}".format(port, target)) + sys.exit(CONFIG_FAIL) @muxcable.group(cls=clicommon.AbbreviationGroup) @@ -378,34 +458,89 @@ def loopback(): @loopback.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 enable(port, target, lane_map): - """Enable loopback mode on a port""" - - 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) +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.argument('lane_mask', required=True, default=None, type=click.INT) +@click.argument('mode_value', required=False, metavar=' 1 LOOPBACK_MODE_NEAR_END 2 LOOPBACK_MODE_FAR_END', default="1", type=click.Choice(["1", "2"])) +@clicommon.pass_db +def enable(db, port, target, lane_mask, mode_value): + """Enable loopback mode on a port args port target lane_map mode_value""" + + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_LOOP_RSP") + + if port is not None: + click.confirm(('Muxcable at port {} will be changed to LOOP mode {} state; disable traffic Continue?'.format( + port, mode_value)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + param_dict["mode_value"] = mode_value + param_dict["lane_mask"] = lane_mask + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_loop", "status", "True", "XCVRD_CONFIG_LOOP_CMD", "XCVRD_CONFIG_LOOP_CMD_ARG", "XCVRD_CONFIG_LOOP_RSP", port, 30, param_dict, "enable") + + rc = res_dict[0] + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_LOOP_RSP") + + port = platform_sfputil_helper.get_interface_alias(port, db) + if rc == 0: + click.echo("Success in LOOP mode port {} to {}".format(port, mode_value)) + else: + click.echo("ERR: Unable to set LOOP mode port {} to {}".format(port, mode_value)) + sys.exit(CONFIG_FAIL) @loopback.command() -@click.argument('port', required=True, default=None, type=click.INT) -@click.argument('target', required=True, default=None, type=click.INT) -def disable(port, target): +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@clicommon.pass_db +def disable(db, port, target): """Disable loopback mode on a port""" - 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) + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_LOOP_RSP") + + if port is not None: + click.confirm(('Muxcable at port {} will be changed to disable LOOP mode {} state; disable traffic Continue?'.format( + port, target)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_loop", "status", "True", "XCVRD_CONFIG_LOOP_CMD", "XCVRD_CONFIG_LOOP_CMD_ARG", "XCVRD_CONFIG_LOOP_RSP", port, 30, param_dict, "disable") + + rc = res_dict[0] + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_LOOP_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_LOOP_RSP") + + port = platform_sfputil_helper.get_interface_alias(port, db) + if rc == 0: + click.echo("Success in disable LOOP mode port {} to {}".format(port, target)) + else: + click.echo("ERR: Unable to set disable LOOP mode port {} to {}".format(port, target)) + sys.exit(CONFIG_FAIL) @muxcable.group(cls=clicommon.AbbreviationGroup) @@ -430,9 +565,10 @@ def state(db, state, port): click.confirm(('Muxcable at port {} will be changed to {} state. Continue?'.format(port, state)), abort=True) res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("config","result", "True", "XCVRD_CONFIG_HWMODE_DIR_CMD", "XCVRD_CONFIG_HWMODE_DIR_RSP", port, 1, state) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "config", "result", "True", "XCVRD_CONFIG_HWMODE_DIR_CMD", None, "XCVRD_CONFIG_HWMODE_DIR_RSP", port, 1, None, state) rc = res_dict[0] @@ -478,10 +614,11 @@ def state(db, state, port): continue res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = 'unknown' + res_dict[0] = CONFIG_FAIL + res_dict[1] = 'unknown' - res_dict = update_and_get_response_for_xcvr_cmd("config","result", "True", "XCVRD_CONFIG_HWMODE_DIR_CMD", "XCVRD_CONFIG_HWMODE_DIR_RSP", port, 1, state) + res_dict = update_and_get_response_for_xcvr_cmd( + "config", "result", "True", "XCVRD_CONFIG_HWMODE_DIR_CMD", None, "XCVRD_CONFIG_HWMODE_DIR_RSP", port, 1, None, state) rc = res_dict[0] @@ -508,7 +645,6 @@ def setswitchmode(db, state, port): port = platform_sfputil_helper.get_interface_name(port, db) - delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_HWMODE_SWMODE_CMD") delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_HWMODE_SWMODE_RSP") @@ -516,10 +652,10 @@ def setswitchmode(db, state, port): click.confirm(('Muxcable at port {} will be changed to {} switching mode. Continue?'.format(port, state)), abort=True) res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("config", "result", "True", "XCVRD_CONFIG_HWMODE_SWMODE_CMD", "XCVRD_CONFIG_HWMODE_SWMODE_RSP", port, 1, state) - + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "config", "result", "True", "XCVRD_CONFIG_HWMODE_SWMODE_CMD", None, "XCVRD_CONFIG_HWMODE_SWMODE_RSP", port, 1, None, state) rc = res_dict[0] @@ -565,9 +701,10 @@ def setswitchmode(db, state, port): continue res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("config", "result", "True", "XCVRD_CONFIG_HWMODE_SWMODE_CMD", "XCVRD_CONFIG_HWMODE_SWMODE_RSP", port, 1, state) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "config", "result", "True", "XCVRD_CONFIG_HWMODE_SWMODE_CMD", None, "XCVRD_CONFIG_HWMODE_SWMODE_RSP", port, 1, None, state) rc = res_dict[0] @@ -606,9 +743,10 @@ def download(db, fwfile, port): if port is not None and port != "all": res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("download_firmware", "status", "0", "XCVRD_DOWN_FW_CMD", "XCVRD_DOWN_FW_RSP", port, 1000, fwfile) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "download_firmware", "status", "0", "XCVRD_DOWN_FW_CMD", None, "XCVRD_DOWN_FW_RSP", port, 1000, None, fwfile) rc = res_dict[0] @@ -655,9 +793,10 @@ def download(db, fwfile, port): res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("download_firmware", "status", "0", "XCVRD_DOWN_FW_CMD", "XCVRD_DOWN_FW_RSP", port, 1000, fwfile) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "download_firmware", "status", "0", "XCVRD_DOWN_FW_CMD", "XCVRD_DOWN_FW_RSP", port, 1000, fwfile) rc = res_dict[0] @@ -678,21 +817,28 @@ def download(db, fwfile, port): @firmware.command() @click.argument('port', metavar='', required=True, default=None) @click.argument('fwfile', metavar='', required=False, default=None) +@click.option('--nonhitless', 'nonhitless', required=False, is_flag=True, type=click.BOOL) @clicommon.pass_db -def activate(db, port, fwfile): +def activate(db, port, fwfile, nonhitless): """Config muxcable firmware activate""" port = platform_sfputil_helper.get_interface_name(port, db) delete_all_keys_in_db_table("STATE_DB", "XCVRD_ACTI_FW_RSP") delete_all_keys_in_db_table("APPL_DB", "XCVRD_ACTI_FW_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_ACTI_FW_CMD_ARG") if port is not None and port != "all": res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("activate_firmware", "status", "0", "XCVRD_ACTI_FW_CMD", "XCVRD_ACTI_FW_RSP", port, 60, fwfile) + param_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + if nonhitless: + param_dict["hitless"] = "1" + + res_dict = update_and_get_response_for_xcvr_cmd( + "activate_firmware", "status", "0", "XCVRD_ACTI_FW_CMD", "XCVRD_ACTI_FW_CMD_ARG", "XCVRD_ACTI_FW_RSP", port, 60, param_dict, fwfile) rc = res_dict[0] @@ -738,12 +884,14 @@ def activate(db, port, fwfile): res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("activate_firmware", "status", "0", "XCVRD_ACTI_FW_CMD", "XCVRD_ACTI_FW_RSP", port, 60, fwfile) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "activate_firmware", "status", "0", "XCVRD_ACTI_FW_CMD", None, "XCVRD_ACTI_FW_RSP", port, 60, None, fwfile) delete_all_keys_in_db_table("STATE_DB", "XCVRD_ACTI_FW_RSP") delete_all_keys_in_db_table("APPL_DB", "XCVRD_ACTI_FW_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_ACTI_FW_CMD_ARG") rc = res_dict[0] @@ -773,9 +921,10 @@ def rollback(db, port, fwfile): if port is not None and port != "all": res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("rollback_firmware", "status", "0", "XCVRD_ROLL_FW_CMD", "XCVRD_ROLL_FW_RSP", port, 60, fwfile) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "rollback_firmware", "status", "0", "XCVRD_ROLL_FW_CMD", None, "XCVRD_ROLL_FW_RSP", port, 60, None, fwfile) delete_all_keys_in_db_table("STATE_DB", "XCVRD_ROLL_FW_RSP") delete_all_keys_in_db_table("APPL_DB", "XCVRD_ROLL_FW_CMD") @@ -821,9 +970,10 @@ def rollback(db, port, fwfile): res_dict = {} - res_dict [0] = CONFIG_FAIL - res_dict [1] = "unknown" - res_dict = update_and_get_response_for_xcvr_cmd("rollback_firmware", "status", "0", "XCVRD_ROLL_FW_CMD", "XCVRD_ROLL_FW_RSP", port, 60, fwfile) + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict = update_and_get_response_for_xcvr_cmd( + "rollback_firmware", "status", "0", "XCVRD_ROLL_FW_CMD", None, "XCVRD_ROLL_FW_RSP", port, 60, None, fwfile) delete_all_keys_in_db_table("STATE_DB", "XCVRD_ROLL_FW_RSP") delete_all_keys_in_db_table("APPL_DB", "XCVRD_ROLL_FW_CMD") @@ -839,3 +989,128 @@ def rollback(db, port, fwfile): rc_exit = CONFIG_FAIL sys.exit(rc_exit) + + +@muxcable.command() +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@clicommon.pass_db +def reset(db, port, target): + """reset a target on the cable NIC TORA TORB Local """ + + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + if port is not None: + click.confirm(('Muxcable at port {} will be reset; CAUTION: disable traffic Continue?'.format(port)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_prbs", "status", "True", "XCVRD_CONFIG_PRBS_CMD", "XCVRD_CONFIG_PRBS_CMD_ARG", "XCVRD_CONFIG_PRBS_RSP", port, 30, param_dict, "reset") + + rc = res_dict[0] + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + port = platform_sfputil_helper.get_interface_alias(port, db) + if rc == 0: + click.echo("Success in reset port {}".format(port)) + else: + click.echo("ERR: Unable to reset port {}".format(port)) + sys.exit(CONFIG_FAIL) + + +@muxcable.command() +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.argument('mode', required=True, metavar=' 0 disable 1 enable', default=True, type=click.Choice(["0", "1"])) +@clicommon.pass_db +def set_anlt(db, port, target, mode): + """Enable anlt mode on a port args port NIC TORA TORB LOCAL enable/disable 1/0""" + + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + if port is not None: + click.confirm( + ('Muxcable at port {} will be changed to enable/disable anlt mode {} state; disable traffic Continue?'.format(port, mode)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + param_dict["mode"] = mode + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_prbs", "status", "True", "XCVRD_CONFIG_PRBS_CMD", "XCVRD_CONFIG_PRBS_CMD_ARG", "XCVRD_CONFIG_PRBS_RSP", port, 30, param_dict, "anlt") + + rc = res_dict[0] + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + port = platform_sfputil_helper.get_interface_alias(port, db) + if rc == 0: + click.echo("Success in anlt enable/disable port {} to {}".format(port, mode)) + else: + click.echo("ERR: Unable to set anlt enable/disable port {} to {}".format(port, mode)) + sys.exit(CONFIG_FAIL) + +@muxcable.command() +@click.argument('port', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.argument('mode', required=True, metavar=' FEC_MODE_NONE 0 FEC_MODE_RS 1 FEC_MODE_FC 2', default=True, type=click.Choice(["0", "1", "2"])) +@clicommon.pass_db +def set_fec(db, port, target, mode): + """Enable fec mode on a port args port NIC TORA TORB LOCAL FEC_MODE_NONE 0 FEC_MODE_RS 1 FEC_MODE_FC 2 """ + + port = platform_sfputil_helper.get_interface_name(port, db) + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + if port is not None: + click.confirm( + ('Muxcable at port {} will be changed to enable/disable fec mode {} state; disable traffic Continue?'.format(port, mode)), abort=True) + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + param_dict["mode"] = mode + + res_dict = update_and_get_response_for_xcvr_cmd( + "config_prbs", "status", "True", "XCVRD_CONFIG_PRBS_CMD", "XCVRD_CONFIG_PRBS_CMD_ARG", "XCVRD_CONFIG_PRBS_RSP", port, 30, param_dict, "fec") + + rc = res_dict[0] + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_CONFIG_PRBS_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_CONFIG_PRBS_RSP") + + port = platform_sfputil_helper.get_interface_alias(port, db) + if rc == 0: + click.echo("Success in fec enable/disable port {} to {}".format(port, mode)) + else: + click.echo("ERR: Unable to set fec enable/disable port {} to {}".format(port, mode)) + sys.exit(CONFIG_FAIL) diff --git a/doc/Command-Reference.md b/doc/Command-Reference.md index 06e9dd75dd..c575d629c0 100644 --- a/doc/Command-Reference.md +++ b/doc/Command-Reference.md @@ -5344,10 +5344,10 @@ While displaying the muxcable configuration, users can configure the following f **show muxcable ber-info** This command displays the ber(Bit error rate) of the port user provides on the target user provides. The target provided as an integer corresponds to actual target as. -0 -> local -1 -> tor 1 -2 -> tor 2 -3 -> nic +NIC +TORA +TORB +LOCAL - Usage: ``` @@ -5360,19 +5360,20 @@ This command displays the ber(Bit error rate) of the port user provides on the t - Example: ``` - admin@sonic:~$ show muxcable ber-info 1 1 + admin@sonic:~$ show muxcable ber-info Ethernet48 NIC Lane1 Lane2 ------- ------- 0 0 ``` -**show muxcable ber-info** +**show muxcable eye-info** This command displays the eye info in mv(milli volts) of the port user provides on the target user provides. The target provided as an integer corresponds to actual target as. -0 -> local -1 -> tor 1 -2 -> tor 2 -3 -> nic +NIC +TORA +TORB +LOCAL + - Usage: ``` @@ -5384,12 +5385,119 @@ This command displays the eye info in mv(milli volts) of the port user provides - Example: ``` - admin@sonic:~$ show muxcable ber-info 1 1 + admin@sonic:~$ show muxcable eye-info Ethernet48 NIC Lane1 Lane2 ------- ------- 632 622 ``` +**show muxcable alivecablestatus** + +This command displays whether the cable is alive if connected to a muxcable port, which basically means if the NIC side is configured and powered on + + +- Usage: + ``` + Usage: Usage: show mux alivecablestatus [OPTIONS] + ``` + +- PORT required - Port number should be a valid port + +- Example: + ``` + admin@sonic:~$ show mux alivecablestatus Ethernet48 + PORT ATTR VALUE + ---------- ------ ------- + Ethernet48 cable False + + ``` + + **show muxcable pcsstatistics** + +This command displays pcs layer statistics on the target which the user enters. The attributes will be predefined by the vendor. Target can be any of +NIC +TORA +TORB +LOCAL + +- Usage: + ``` + : show mux pcsstatistics [OPTIONS] NIC TORA TORB LOCAL + + ``` + +- PORT required - Port number should be a valid port +- TARGET required - the actual target to get the ber info of. + +- Example: + ``` + admin@sonic:~$ show mux pcsstatistics Ethernet48 NIC + PORT ATTR VALUE + ---------- -------------------- ------- + Ethernet48 Rx Frames OK 0 + Ethernet48 Rx Chk SEQ Errs 0 + Ethernet48 Rx Alignment Errs 0 + Ethernet48 Rx In Errs 0 + Ethernet48 Rx FrameTooLong Errs 0 + Ethernet48 Rx Octets OK 0 + Ethernet48 Tx Frames OK 0 + Ethernet48 Tx Out Errs 0 + Ethernet48 Tx Octets OK 0 + ``` +**show muxcable fecstatistics** + +This command displays fec layer statistics on the target which the user enters. The attributes will be predefined by the vendor. Target can be any of +NIC +TORA +TORB +LOCAL + + +- Usage: + ``` + Usage: show mux fecstatistics [OPTIONS] NIC TORA TORB LOCAL + ``` + +- PORT required - Port number should be a valid port +- TARGET required - the actual target to get the eye info of. +- Example: + ``` + show mux fec Ethernet48 NIC + PORT ATTR VALUE + ---------- ------------------------- ------- + Ethernet48 Total recevied CW 0 + Ethernet48 Total correct CW 0 + Ethernet48 Total corrected CW 0 + Ethernet48 Total uncorrectable CW 0 + Ethernet48 Corrected CW ( 1 sym err) 0 + ``` + +**show muxcable get-fec-anlt-speed** + +This command displays the fec, anlt and speed related configurations for the cable + + +- Usage: + ``` + Usage: Usage: show mux get-fec-anlt-speed [OPTIONS] + ``` + +- PORT required - Port number should be a valid port + +- Example: + ``` + admin@sonic:~$ show mux get-fec-anlt-speed Ethernet48 + PORT ATTR VALUE + ---------- ---------- ------- + Ethernet48 fec_nic 0 + Ethernet48 fec_tor_a 0 + Ethernet48 fec_tor_b 0 + Ethernet48 speed 50000 + Ethernet48 anlt_nic False + Ethernet48 anlt_tor_a False + Ethernet48 anlt_tor_b False + ``` + ### Muxcable Config commands @@ -5454,10 +5562,10 @@ While configuring the muxcable, users needs to configure the following fields fo - PORT required - Port number should be a valid port - TARGET required - the actual target to run the prbs on - 0 -> local side, - 1 -> TOR 1 - 2 -> TOR 2 - 3 -> NIC + NIC + TORA + TORB + LOCAL - MODE_VALUE required - the mode/type for configuring the PRBS mode. 0x00 = PRBS 9, 0x01 = PRBS 15, 0x02 = PRBS 23, 0x03 = PRBS 31 - LANE_MAP required - an integer representing the lane_map to be run PRBS on @@ -5465,9 +5573,9 @@ While configuring the muxcable, users needs to configure the following fields fo for example 3 -> 0b'0011 , means running on lane0 and lane1 - Example: ``` - admin@sonic:~$ sudo config muxcable prbs enable 1 1 3 3 + admin@sonic:~$ sudo config muxcable prbs enable Ethernet48 NIC 3 3 PRBS config sucessful - admin@sonic:~$ sudo config muxcable prbs disable 1 0 + admin@sonic:~$ sudo config muxcable prbs disable Ethernet48 NIC PRBS disable sucessful ``` @@ -5481,26 +5589,103 @@ This command is used for setting the configuration and enable/disable of loopbac config muxcable loopback disable [OPTIONS] PORT TARGET ``` -While configuring the muxcable, users needs to configure the following fields for the operation +While configuring the muxcable, users needs to provide the following fields for the operation - PORT required - Port number should be a valid port - TARGET required - the actual target to run the loopback on - 0 -> local side, - 1 -> TOR 1 - 2 -> TOR 2 - 3 -> NIC + NIC + TORA + TORB + LOCAL - LANE_MAP required - an integer representing the lane_map to be run loopback on 0bit for lane 0, 1bit for lane1 and so on. for example 3 -> 0b'0011 , means running on lane0 and lane1 - Example: ``` - admin@sonic:~$ sudo config muxcable loopback enable 1 1 3 + admin@sonic:~$ sudo config muxcable loopback enable Ethernet48 NIC 3 loopback config sucessful - admin@sonic:~$ sudo config muxcable loopback disable 1 0 + admin@sonic:~$ sudo config muxcable loopback disable Ethernet48 NIC loopback disable sucessfull ``` + +**config muxcable set-anlt enable/disable** + +This command is used for setting the configuration and enable/disable of anlt on the port and target on the cable. + +- Usage: + ``` + config mux set-anlt [OPTIONS] PORT NIC TORA TORB LOCAL 0 disable 1 enable + ``` + +While configuring the muxcable for fec, users needs to provide the following fields for the operation + +- PORT required - Port number should be a valid port +- TARGET required - the actual target to run the loopback on + NIC + TORA + TORB + LOCAL + +- Example: + ``` + admin@sonic:~$ sudo config mux set-fec Ethernet48 NIC 0 + + Success in fec mode port Ethernet48 to 0 + ``` + +**config muxcable set-fec mode** + +This command is used for setting the configuration and enable/disable of fec mode on the port and target on the cable. The fec could be one of FEC_MODE_NONE FEC_MODE_RS FEC_MODE_FC + +- Usage: + ``` + config mux set-fec [OPTIONS] PORT NIC TORA TORB LOCAL FEC_MODE_NONE 0 FEC_MODE_RS 1 FEC_MODE_FC 2 + ``` + +While configuring the muxcable for anlt, users needs to configure the following fields for the operation + +- PORT required - Port number should be a valid port +- TARGET required - the actual target to run the loopback on + NIC + TORA + TORB + LOCAL + +- Example: + ``` + admin@sonic:~$ sudo config mux set-anlt Ethernet48 NIC 0 + Success in anlt enable/disable port Ethernet48 to 0 + ``` + + +**config muxcable reset** + +This command is used for hard reset for the configuration +of the MCU on the side which the user wants to. + +- Usage: + ``` + config mux reset [OPTIONS] PORT NIC TORA TORB LOCAL + ``` + +While configuring the muxcable for reset, users needs to provide the following fields for the operation + +- PORT required - Port number should be a valid port +- TARGET required - the actual target to run the loopback on + NIC + TORA + TORB + LOCAL + +- Example: + ``` + admin@sonic:~$ sudo config mux reset Ethernet48 NIC Muxcable at port Ethernet48 will be reset; CAUTION: disable traffic Continue? [y/N]: y + Success in reset port Ethernet48 + ``` + + Go Back To [Beginning of the document](#) or [Beginning of this section](#muxcable) ## Mirroring diff --git a/show/muxcable.py b/show/muxcable.py index 6d4b1bab76..d47793041e 100644 --- a/show/muxcable.py +++ b/show/muxcable.py @@ -1,5 +1,4 @@ import json -import os import sys import time @@ -53,6 +52,48 @@ def delete_all_keys_in_db_table(db_type, table_name): for key in table_keys[asic_id]: table[asic_id]._del(key) +target_dict = { "NIC":"0", + "TORA":"1", + "TORB":"2", + "LOCAL":"3"} + +def parse_target(target): + return target_dict.get(target, None) + +def check_port_in_mux_cable_table(port): + + per_npu_configdb = {} + mux_tbl_cfg_db = {} + port_mux_tbl_keys = {} + + # 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) + # TO-DO replace the macros with correct swsscommon names + per_npu_configdb[asic_id] = ConfigDBConnector(use_unix_socket_path=False, namespace=namespace) + per_npu_configdb[asic_id].connect() + mux_tbl_cfg_db[asic_id] = per_npu_configdb[asic_id].get_table("MUX_CABLE") + port_mux_tbl_keys[asic_id] = mux_tbl_cfg_db[asic_id].keys() + + asic_index = None + if platform_sfputil is not None: + asic_index = platform_sfputil_helper.get_asic_id_for_logical_port(port) + + if asic_index is None: + # TODO this import is only for unit test purposes, and should be removed once sonic_platform_base + # is fully mocked + import sonic_platform_base.sonic_sfp.sfputilhelper + asic_index = sonic_platform_base.sonic_sfp.sfputilhelper.SfpUtilHelper().get_asic_id_for_logical_port(port) + if asic_index is None: + click.echo("Got invalid asic index for port {}, cant retreive mux status".format(port)) + return False + + if port in port_mux_tbl_keys[asic_index]: + return True + return False + def get_response_for_version(port, mux_info_dict): state_db = {} @@ -99,14 +140,89 @@ def get_response_for_version(port, mux_info_dict): return mux_info_dict +def get_event_logs(port, res_dict, mux_info_dict): + state_db = {} + xcvrd_show_fw_res_tbl = {} + + namespaces = multi_asic.get_front_end_namespaces() + for namespace in namespaces: + asic_id = multi_asic.get_asic_index_from_namespace(namespace) + state_db[asic_id] = db_connect("STATE_DB", namespace) + xcvrd_show_fw_res_tbl[asic_id] = swsscommon.Table(state_db[asic_id], "XCVRD_EVENT_LOG_RES") + + logical_port_list = platform_sfputil_helper.get_logical_list() + if port not in logical_port_list: + click.echo("ERR: This is not a valid port, valid ports ({})".format(", ".join(logical_port_list))) + rc = EXIT_FAIL + res_dict[1] = rc + return mux_info_dict + + asic_index = None + if platform_sfputil is not None: + asic_index = platform_sfputil_helper.get_asic_id_for_logical_port(port) + if asic_index is None: + # TODO this import is only for unit test purposes, and should be removed once sonic_platform_base + # is fully mocked + import sonic_platform_base.sonic_sfp.sfputilhelper + asic_index = sonic_platform_base.sonic_sfp.sfputilhelper.SfpUtilHelper().get_asic_id_for_logical_port(port) + if asic_index is None: + click.echo("Got invalid asic index for port {}, cant retreive mux status".format(port)) + rc = CONFIG_FAIL + res_dict[1] = rc + return mux_info_dict + + (status, fvp) = xcvrd_show_fw_res_tbl[asic_index].get(port) + res_dir = dict(fvp) + + for key, value in res_dir.items(): + mux_info_dict[key] = value; + + return mux_info_dict + +def get_result(port, res_dict, cmd ,result, table_name): + state_db = {} + xcvrd_show_fw_res_tbl = {} + + namespaces = multi_asic.get_front_end_namespaces() + for namespace in namespaces: + asic_id = multi_asic.get_asic_index_from_namespace(namespace) + state_db[asic_id] = db_connect("STATE_DB", namespace) + xcvrd_show_fw_res_tbl[asic_id] = swsscommon.Table(state_db[asic_id], table_name) + + logical_port_list = platform_sfputil_helper.get_logical_list() + if port not in logical_port_list: + click.echo("ERR: This is not a valid port, valid ports ({})".format(", ".join(logical_port_list))) + rc = EXIT_FAIL + res_dict[1] = rc + return result + + asic_index = None + if platform_sfputil is not None: + asic_index = platform_sfputil_helper.get_asic_id_for_logical_port(port) + if asic_index is None: + # TODO this import is only for unit test purposes, and should be removed once sonic_platform_base + # is fully mocked + import sonic_platform_base.sonic_sfp.sfputilhelper + asic_index = sonic_platform_base.sonic_sfp.sfputilhelper.SfpUtilHelper().get_asic_id_for_logical_port(port) + if asic_index is None: + click.echo("Got invalid asic index for port {}, cant retreive mux status".format(port)) + rc = CONFIG_FAIL + res_dict[1] = rc + return result + + (status, fvp) = xcvrd_show_fw_res_tbl[asic_index].get(port) + res_dir = dict(fvp) -def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_name, rsp_table_name, port, cmd_timeout_secs, arg=None): + return res_dir + +def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_name, cmd_arg_table_name, rsp_table_name ,port, cmd_timeout_secs, param_dict= None, arg=None): res_dict = {} state_db, appl_db = {}, {} firmware_rsp_tbl, firmware_rsp_tbl_keys = {}, {} firmware_rsp_sub_tbl = {} firmware_cmd_tbl = {} + firmware_cmd_arg_tbl = {} CMD_TIMEOUT_SECS = cmd_timeout_secs @@ -121,6 +237,8 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ firmware_cmd_tbl[asic_id] = swsscommon.Table(appl_db[asic_id], cmd_table_name) firmware_rsp_sub_tbl[asic_id] = swsscommon.SubscriberStateTable(state_db[asic_id], rsp_table_name) firmware_rsp_tbl[asic_id] = swsscommon.Table(state_db[asic_id], rsp_table_name) + if cmd_arg_table_name is not None: + firmware_cmd_arg_tbl[asic_id] = swsscommon.Table(appl_db[asic_id], cmd_arg_table_name) firmware_rsp_tbl_keys[asic_id] = firmware_rsp_tbl[asic_id].getKeys() for key in firmware_rsp_tbl_keys[asic_id]: firmware_rsp_tbl[asic_id]._del(key) @@ -154,6 +272,11 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ else: cmd_arg = str(arg) + if param_dict is not None: + for key, value in param_dict.items(): + fvs = swsscommon.FieldValuePairs([(str(key), str(value))]) + firmware_cmd_arg_tbl[asic_index].set(port, fvs) + fvs = swsscommon.FieldValuePairs([(cmd_name, cmd_arg)]) firmware_cmd_tbl[asic_index].set(port, fvs) @@ -573,47 +696,301 @@ def config(db, port, json_output): @muxcable.command() -@click.argument('port', required=True, default=None, type=click.INT) -@click.argument('target', required=True, default=None, type=click.INT) -def berinfo(port, target): +@click.argument('port', metavar='', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def berinfo(db, port, target, json_output): """Show muxcable BER (bit error rate) information""" - if os.geteuid() != 0: - click.echo("Root privileges are required for this operation") - sys.exit(EXIT_FAIL) - 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', 'Lane3', 'Lane4'] - lane_data = [] - lane_data.append(res) - click.echo(tabulate(lane_data, headers=headers)) - sys.exit(EXIT_SUCCESS) + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + if port is not None: + + res_dict = {} + result = {} + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_ber", "status", "True", "XCVRD_GET_BER_CMD", "XCVRD_GET_BER_CMD_ARG", "XCVRD_GET_BER_RSP", port, 10, param_dict, "ber") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_BER_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + + else: + click.echo("Did not get a valid Port for ber value".format(port)) + sys.exit(CONFIG_FAIL) @muxcable.command() -@click.argument('port', required=True, default=None, type=click.INT) -@click.argument('target', required=True, default=None, type=click.INT) -def eyeinfo(port, target): +@click.argument('port', metavar='', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def eyeinfo(db, port, target, json_output): """Show muxcable eye information in mv""" - if os.geteuid() != 0: - click.echo("Root privileges are required for this operation") - sys.exit(EXIT_FAIL) - 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', 'Lane3', 'Lane4'] - lane_data = [] - lane_data.append(res) - click.echo(tabulate(lane_data, headers=headers)) - sys.exit(EXIT_SUCCESS) + port = platform_sfputil_helper.get_interface_alias(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + if port is not None: + + res_dict = {} + result = {} + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_ber", "status", "True", "XCVRD_GET_BER_CMD", "XCVRD_GET_BER_CMD_ARG", "XCVRD_GET_BER_RSP", port, 10, param_dict, "eye") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_BER_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + + else: + click.echo("Did not get a valid Port for ber value".format(port)) + sys.exit(CONFIG_FAIL) +@muxcable.command() +@click.argument('port', metavar='', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def fecstatistics(db, port, target, json_output): + """Show muxcable fec layer statistics information, target NIC TORA TORB""" + + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + if port is not None: + + res_dict = {} + result = {} + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_ber", "status", "True", "XCVRD_GET_BER_CMD", "XCVRD_GET_BER_CMD_ARG", "XCVRD_GET_BER_RSP", port, 10, param_dict, "fec_stats") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_BER_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + + else: + click.echo("Did not get a valid Port for ber value".format(port)) + sys.exit(CONFIG_FAIL) + + +@muxcable.command() +@click.argument('port', metavar='', required=True, default=None) +@click.argument('target', metavar=' NIC TORA TORB LOCAL', required=True, default=None, type=click.Choice(["NIC", "TORA", "TORB", "LOCAL"])) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def pcsstatistics(db, port, target, json_output): + """Show muxcable pcs layer statistics information""" + + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + if port is not None: + + res_dict = {} + result = {} + param_dict = {} + target = parse_target(target) + param_dict["target"] = target + + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_ber", "status", "True", "XCVRD_GET_BER_CMD", "XCVRD_GET_BER_CMD_ARG", "XCVRD_GET_BER_RSP", port, 10, param_dict, "pcs_stats") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_BER_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + + else: + click.echo("Did not get a valid Port for pcs statistics".format(port)) + sys.exit(CONFIG_FAIL) + +@muxcable.command() +@click.argument('port', metavar='', required=True, default=None) +@click.argument('option', required=False, default=None) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def debugdumpregisters(db, port, option, json_output): + """Show muxcable debug deump registers information, preagreed by vendors""" + + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + if port is not None: + + res_dict = {} + result = {} + param_dict = {} + param_dict["option"] = option + + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_ber", "status", "True", "XCVRD_GET_BER_CMD", "XCVRD_GET_BER_CMD_ARG", "XCVRD_GET_BER_RSP", port, 10, param_dict, "debug_dump") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_BER_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD_ARG") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + else: + click.echo("Did not get a valid Port for debug dump registers".format(port)) + sys.exit(CONFIG_FAIL) + +@muxcable.command() +@click.argument('port', metavar='', required=True, default=None) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def alivecablestatus(db, port, json_output): + """Show muxcable alive information """ + + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + if port is not None: + + res_dict = {} + result = {} + + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_ber", "status", "True", "XCVRD_GET_BER_CMD", None, "XCVRD_GET_BER_RSP", port, 10, None, "cable_alive") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_BER_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_BER_CMD") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_BER_RES") + + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + else: + click.echo("Did not get a valid Port for cable alive status".format(port)) + sys.exit(CONFIG_FAIL) + @muxcable.command() @click.argument('port', required=True, default=None) @clicommon.pass_db @@ -667,11 +1044,15 @@ def muxdirection(db, port): if port is not None: + if check_port_in_mux_cable_table(port) == False: + click.echo("Not Y-cable port") + return CONFIG_FAIL + res_dict = {} res_dict[0] = CONFIG_FAIL res_dict[1] = "unknown" res_dict = update_and_get_response_for_xcvr_cmd( - "state", "state", "True", "XCVRD_SHOW_HWMODE_DIR_CMD", "XCVRD_SHOW_HWMODE_DIR_RSP", port, 1, "probe") + "state", "state", "True", "XCVRD_SHOW_HWMODE_DIR_CMD", None, "XCVRD_SHOW_HWMODE_DIR_RSP", port, 1, None, "probe") body = [] temp_list = [] @@ -706,6 +1087,9 @@ def muxdirection(db, port): if len(physical_port_list) != 1: continue + if not check_port_in_mux_cable_table(port): + continue + physical_port = physical_port_list[0] logical_port_list_for_physical_port = platform_sfputil_helper.get_physical_to_logical() @@ -724,7 +1108,7 @@ def muxdirection(db, port): res_dict[0] = CONFIG_FAIL res_dict[1] = "unknown" res_dict = update_and_get_response_for_xcvr_cmd( - "state", "state", "True", "XCVRD_SHOW_HWMODE_DIR_CMD", "XCVRD_SHOW_HWMODE_DIR_RSP", port, 1, "probe") + "state", "state", "True", "XCVRD_SHOW_HWMODE_DIR_CMD", None, "XCVRD_SHOW_HWMODE_DIR_RSP", port, 1, None, "probe") port = platform_sfputil_helper.get_interface_alias(port, db) temp_list.append(port) temp_list.append(res_dict[1]) @@ -755,11 +1139,15 @@ def switchmode(db, port): if port is not None: + if check_port_in_mux_cable_table(port) == False: + click.echo("Not Y-cable port") + return CONFIG_FAIL + res_dict = {} res_dict[0] = CONFIG_FAIL res_dict[1] = "unknown" res_dict = update_and_get_response_for_xcvr_cmd( - "state", "state", "True", "XCVRD_SHOW_HWMODE_SWMODE_CMD", "XCVRD_SHOW_HWMODE_SWMODE_RSP", port, 1, "probe") + "state", "state", "True", "XCVRD_SHOW_HWMODE_SWMODE_CMD", None, "XCVRD_SHOW_HWMODE_SWMODE_RSP", port, 1, None, "probe") body = [] temp_list = [] @@ -794,6 +1182,9 @@ def switchmode(db, port): if len(physical_port_list) != 1: continue + if not check_port_in_mux_cable_table(port): + continue + physical_port = physical_port_list[0] logical_port_list_for_physical_port = platform_sfputil_helper.get_physical_to_logical() @@ -812,7 +1203,7 @@ def switchmode(db, port): res_dict[0] = CONFIG_FAIL res_dict[1] = "unknown" res_dict = update_and_get_response_for_xcvr_cmd( - "state", "state", "True", "XCVRD_SHOW_HWMODE_SWMODE_CMD", "XCVRD_SHOW_HWMODE_SWMODE_RSP", port, 1, "probe") + "state", "state", "True", "XCVRD_SHOW_HWMODE_SWMODE_CMD", None, "XCVRD_SHOW_HWMODE_SWMODE_RSP", port, 1, None, "probe") port = platform_sfputil_helper.get_interface_alias(port, db) temp_list.append(port) temp_list.append(res_dict[1]) @@ -997,7 +1388,7 @@ def version(db, port, active): mux_info_dict["version_self_next"] = "N/A" res_dict = update_and_get_response_for_xcvr_cmd( - "firmware_version", "status", "True", "XCVRD_SHOW_FW_CMD", "XCVRD_SHOW_FW_RSP", port, 20, "probe") + "firmware_version", "status", "True", "XCVRD_SHOW_FW_CMD", None, "XCVRD_SHOW_FW_RSP", port, 20, None, "probe") if res_dict[1] == "True": mux_info_dict = get_response_for_version(port, mux_info_dict) @@ -1083,3 +1474,89 @@ def metrics(db, port, json_output): headers = ['PORT', 'EVENT', 'TIME'] click.echo(tabulate(print_data, headers=headers)) + +@muxcable.command() +@click.argument('port', metavar='', required=True, default=None) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def event_log(db, port, json_output): + """Show muxcable event log """ + + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_EVENT_LOG_CMD") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_EVENT_LOG_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_EVENT_LOG_RES") + + if port is not None: + + res_dict = {} + result = {} + mux_info_dict = {} + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "show_event", "status", "True", "XCVRD_EVENT_LOG_CMD", None, "XCVRD_EVENT_LOG_RSP", port, 1000, None, "probe") + + if res_dict[1] == "True": + result = get_event_logs(port, res_dict, mux_info_dict) + + + delete_all_keys_in_db_table("STATE_DB", "XCVRD_EVENT_LOG_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_EVENT_LOG_RES") + delete_all_keys_in_db_table("APPL_DB", "XCVRD_EVENT_LOG_CMD") + port = platform_sfputil_helper.get_interface_alias(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + else: + click.echo("Did not get a valid Port for event log".format(port)) + sys.exit(CONFIG_FAIL) + +@muxcable.command() +@click.argument('port', metavar='', required=True, default=None) +@click.option('--json', 'json_output', required=False, is_flag=True, type=click.BOOL, help="display the output in json format") +@clicommon.pass_db +def get_fec_anlt_speed(db, port, json_output): + """Show muxcable configurations for fec anlt speed """ + + port = platform_sfputil_helper.get_interface_name(port, db) + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_FEC_CMD") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_FEC_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_FEC_RES") + + if port is not None: + + res_dict = {} + result = {} + + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + + res_dict = update_and_get_response_for_xcvr_cmd( + "get_fec", "status", "True", "XCVRD_GET_FEC_CMD", None, "XCVRD_GET_FEC_RSP", port, 10, None, "probe") + + if res_dict[1] == "True": + result = get_result(port, res_dict, "fec" , result, "XCVRD_GET_FEC_RES") + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_GET_FEC_CMD") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_FEC_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_GET_FEC_RES") + port = platform_sfputil_helper.get_interface_name(port, db) + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['PORT', 'ATTR', 'VALUE'] + res = [[port]+[key] + [val] for key, val in result.items()] + click.echo(tabulate(res, headers=headers)) + else: + click.echo("Did not get a valid Port for fec value speed anlt".format(port)) + sys.exit(CONFIG_FAIL) + diff --git a/tests/muxcable_test.py b/tests/muxcable_test.py index 498b4761ef..e9fcc065d4 100644 --- a/tests/muxcable_test.py +++ b/tests/muxcable_test.py @@ -778,69 +778,703 @@ def test_config_muxcable_tabular_port_with_incorrect_port(self): assert result.exit_code == 1 - @mock.patch('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.get_eye_info', mock.MagicMock(return_value=[0, 0])) + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @mock.patch('show.muxcable.get_result', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) def test_show_muxcable_eye_info(self): runner = CliRunner() db = Db() result = runner.invoke(show.cli.commands["muxcable"].commands["eyeinfo"], - ["0", "0"], obj=db) + ["Ethernet0", "NIC"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_result', mock.MagicMock(return_value={0: 0, + 1: "active"})) + def test_show_mux_debugdeumpregisters(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["debugdumpregisters"], + ["Ethernet0"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_result', mock.MagicMock(return_value={0: 0, + 1: "active"})) + def test_show_mux_pcsstatistics(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["pcsstatistics"], + ["Ethernet0", "NIC"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_result', mock.MagicMock(return_value={0: 0, + 1: "active"})) + def test_show_mux_fecstatistics(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["fecstatistics"], + ["Ethernet0", "NIC"], obj=db) assert result.exit_code == 0 - @mock.patch('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.get_ber_info', mock.MagicMock(return_value=[0, 0])) + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_event_logs', mock.MagicMock(return_value={0: 0, + 2: "log"})) + def test_show_mux_event_log(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["event-log"], + ["Ethernet0"], obj=db) + + assert result.exit_code == 0 + + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_result', mock.MagicMock(return_value={0: 0, + 1: "active"})) + def test_show_mux_get_fec_anlt_speed(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["get-fec-anlt-speed"], + ["Ethernet0"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) def test_show_muxcable_ber_info(self): runner = CliRunner() db = Db() result = runner.invoke(show.cli.commands["muxcable"].commands["berinfo"], - ["0", "0"], obj=db) + ["Ethernet0", "NIC"], obj=db) assert result.exit_code == 0 - @mock.patch('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.enable_prbs_mode', mock.MagicMock(return_value=1)) + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) def test_config_muxcable_enable_prbs(self): runner = CliRunner() db = Db() result = runner.invoke(config.config.commands["muxcable"].commands["prbs"].commands["enable"], - ["0", "0", "0", "0"], obj=db) + ["Ethernet0", "NIC", "0", "0"], obj=db) assert result.exit_code == 0 - @mock.patch('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.enable_loopback_mode', mock.MagicMock(return_value=1)) + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) def test_config_muxcable_enable_loopback(self): runner = CliRunner() db = Db() result = runner.invoke(config.config.commands["muxcable"].commands["loopback"].commands["enable"], - ["0", "0", "0"], obj=db) + ["Ethernet0", "NIC", "0"], obj=db) assert result.exit_code == 0 - @mock.patch('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.disable_prbs_mode', mock.MagicMock(return_value=1)) + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) def test_config_muxcable_disble_prbs(self): runner = CliRunner() db = Db() result = runner.invoke(config.config.commands["muxcable"].commands["prbs"].commands["disable"], - ["0", "0"], obj=db) + ["Ethernet0", "NIC"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_disable_loopback(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["loopback"].commands["disable"], + ["Ethernet0", "NIC"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('sonic_y_cable.y_cable.get_part_number', mock.MagicMock(return_value=("CACL1X321P2PA1M"))) + @mock.patch('sonic_y_cable.y_cable.get_vendor', mock.MagicMock(return_value=("Credo "))) + @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value=1)) + @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0])) + def test_show_muxcable_cableinfo(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["cableinfo"], + ["Ethernet0"], obj=db) + + assert result.exit_code == 0 + assert result.output == expected_muxcable_cableinfo_output + + @mock.patch('sonic_y_cable.y_cable.get_part_number', mock.MagicMock(return_value=(False))) + @mock.patch('sonic_y_cable.y_cable.get_vendor', mock.MagicMock(return_value=(False))) + @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value=1)) + @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0])) + def test_show_muxcable_cableinfo_incorrect_port(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["cableinfo"], + ["Ethernet0"], obj=db) + assert result.exit_code == 1 + + @mock.patch('sonic_y_cable.y_cable.get_part_number', mock.MagicMock(return_value=(False))) + @mock.patch('sonic_y_cable.y_cable.get_vendor', mock.MagicMock(return_value=(False))) + @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value=1)) + @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=0)) + def test_show_muxcable_cableinfo_incorrect_port_return_value(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["cableinfo"], + ["Ethernet0"], obj=db) + assert result.exit_code == 1 + + @mock.patch('sonic_y_cable.y_cable.get_part_number', mock.MagicMock(return_value=(False))) + @mock.patch('sonic_y_cable.y_cable.get_vendor', mock.MagicMock(return_value=(False))) + @mock.patch('show.muxcable.platform_sfputil', mock.MagicMock(return_value=1)) + @mock.patch('utilities_common.platform_sfputil_helper.logical_port_name_to_physical_port_list', mock.MagicMock(return_value=[0, 1])) + def test_show_muxcable_cableinfo_incorrect_logical_port_return_value(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["cableinfo"], + ["Ethernet0"], obj=db) + assert result.exit_code == 1 + + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('re.match', mock.MagicMock(return_value=(True))) + def test_show_muxcable_hwmode_muxdirection_port_active(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["hwmode"].commands["muxdirection"], + ["Ethernet12"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_hwmode_muxdirection_active_expected_output + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('re.match', mock.MagicMock(return_value=(True))) + def test_show_muxcable_hwmode_muxdirection_active_expected_output_alias(self): + runner = CliRunner() + db = Db() + os.environ['SONIC_CLI_IFACE_MODE'] = "alias" + result = runner.invoke(show.cli.commands["muxcable"].commands["hwmode"].commands["muxdirection"], + ["etp4"], obj=db) + os.environ['SONIC_CLI_IFACE_MODE'] = "default" + assert result.exit_code == 0 + assert result.output == show_muxcable_hwmode_muxdirection_active_expected_output_alias + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "standby"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('re.match', mock.MagicMock(return_value=(True))) + def test_show_muxcable_hwmode_muxdirection_active(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["hwmode"].commands["muxdirection"], obj=db) + assert result.exit_code == 0 + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "standby"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(2))) + @mock.patch('re.match', mock.MagicMock(return_value=(True))) + def test_show_muxcable_hwmode_muxdirection_port_standby(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["hwmode"].commands["muxdirection"], + ["Ethernet12"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_hwmode_muxdirection_standby_expected_output + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "standby"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(2))) + @mock.patch('re.match', mock.MagicMock(return_value=(True))) + def test_show_muxcable_hwmode_muxdirection_port_standby_alias(self): + runner = CliRunner() + db = Db() + + os.environ['SONIC_CLI_IFACE_MODE'] = "alias" + result = runner.invoke(show.cli.commands["muxcable"].commands["hwmode"].commands["muxdirection"], + ["etp4"], obj=db) + os.environ['SONIC_CLI_IFACE_MODE'] = "default" assert result.exit_code == 0 + assert result.output == show_muxcable_hwmode_muxdirection_standby_expected_output_alias - @mock.patch('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.disable_loopback_mode', mock.MagicMock(return_value=1)) + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "sucess"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(2))) + @mock.patch('re.match', mock.MagicMock(return_value=(True))) + def test_show_muxcable_hwmode_muxdirection_standby(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["hwmode"].commands["muxdirection"], obj=db) + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "sucess"})) + @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.swsscommon.SubscriberStateTable', mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torA', mock.MagicMock(return_value=(True))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torB', mock.MagicMock(return_value=(True))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_hwmode_state_port_active(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["hwmode"].commands["state"], + ["active", "Ethernet12"], obj=db) + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "sucess"})) + @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.swsscommon.SubscriberStateTable', mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torA', mock.MagicMock(return_value=(True))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torB', mock.MagicMock(return_value=(True))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_hwmode_state_active(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["hwmode"].commands["state"], + ["active", "all"], obj=db) + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "standby"})) + @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.swsscommon.SubscriberStateTable', mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torA', mock.MagicMock(return_value=(True))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torB', mock.MagicMock(return_value=(True))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_hwmode_state_port_standby(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["hwmode"].commands["state"], + ["standby", "Ethernet12"], obj=db) + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "standby"})) + @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.swsscommon.SubscriberStateTable', mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.check_mux_direction', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torA', mock.MagicMock(return_value=(True))) + @mock.patch('sonic_y_cable.y_cable.toggle_mux_to_torB', mock.MagicMock(return_value=(True))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_hwmode_state_standby(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["hwmode"].commands["state"], + ["standby", "all"], obj=db) + assert result.exit_code == 0 + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_response_for_version', mock.MagicMock(return_value={"version_self_active": "0.6MS", + "version_self_inactive": "0.6MS", + "version_self_next": "0.6MS", + "version_peer_active": "0.6MS", + "version_peer_inactive": "0.6MS", + "version_peer_next": "0.6MS", + "version_nic_active": "0.6MS", + "version_nic_inactive": "0.6MS", + "version_nic_next": "0.6MS"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + @mock.patch('sonic_y_cable.y_cable.get_firmware_version', mock.MagicMock(return_value={"version_active": "0.6MS", + "version_inactive": "0.6MS", + "version_next": "0.6MS"})) + def test_show_muxcable_firmware_version(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["firmware"].commands["version"], [ + "Ethernet0"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_firmware_version_expected_output + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "sucess"})) + @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.swsscommon.SubscriberStateTable', mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + @mock.patch('sonic_y_cable.y_cable.download_fimware', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.FIRMWARE_DOWNLOAD_SUCCESS', mock.MagicMock(return_value=(1))) + def test_config_muxcable_download_firmware(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["firmware"].commands["download"], [ + "fwfile", "Ethernet0"], obj=db) + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "sucess"})) + @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.swsscommon.SubscriberStateTable', mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + @mock.patch('sonic_y_cable.y_cable.activate_firmware', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.FIRMWARE_ACTIVATE_SUCCESS', mock.MagicMock(return_value=(1))) + def test_config_muxcable_activate_firmware(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["firmware"].commands["activate"], [ + "Ethernet0"], obj=db) + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "sucess"})) + @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.swsscommon.SubscriberStateTable", mock.MagicMock(return_value=0)) + @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('config.muxcable.platform_sfputil', mock.MagicMock(return_value={0: ["Ethernet12", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + @mock.patch('sonic_y_cable.y_cable.rollback_firmware', mock.MagicMock(return_value=(1))) + @mock.patch('sonic_y_cable.y_cable.FIRMWARE_ROLLBACK_SUCCESS', mock.MagicMock(return_value=(1))) + def test_config_muxcable_rollback_firmware(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["firmware"].commands["rollback"], [ + "Ethernet0"], obj=db) + assert result.exit_code == 0 + + @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_metrics_port(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["metrics"], + ["Ethernet0"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_metrics_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_metrics_port_alias(self): + runner = CliRunner() + db = Db() + + os.environ['SONIC_CLI_IFACE_MODE'] = "alias" + result = runner.invoke(show.cli.commands["muxcable"].commands["metrics"], + ["etp1"], obj=db) + + os.environ['SONIC_CLI_IFACE_MODE'] = "default" + assert result.exit_code == 0 + assert result.output == show_muxcable_metrics_expected_output_alias + + @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_metrics_port_json(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["metrics"], + ["Ethernet0", "--json"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_metrics_expected_output_json + + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "True"})) + @mock.patch('show.muxcable.get_response_for_version', mock.MagicMock(return_value={"version_self_active": "0.6MS", + "version_self_inactive": "0.6MS", + "version_self_next": "0.6MS", + "version_peer_active": "0.6MS", + "version_peer_inactive": "0.6MS", + "version_peer_next": "0.6MS", + "version_nic_active": "0.6MS", + "version_nic_inactive": "0.6MS", + "version_nic_next": "0.6MS"})) + @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.get_physical_to_logical', 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])) + @mock.patch('sonic_y_cable.y_cable.check_read_side', mock.MagicMock(return_value=(1))) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + @mock.patch('sonic_y_cable.y_cable.get_firmware_version', mock.MagicMock(return_value={"version_active": "0.6MS", + "version_inactive": "0.6MS", + "version_next": "0.6MS"})) + def test_show_muxcable_firmware_active_version(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["firmware"].commands["version"], [ + "Ethernet0", "--active"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_firmware_version_active_expected_output + + @classmethod + def teardown_class(cls): + os.environ['UTILITIES_UNIT_TESTING'] = "0" + print("TEARDOWN") + @mock.patch('show.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('show.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_show_muxcable_ber_info(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["berinfo"], + ["Ethernet0", "NIC"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_enable_prbs(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["prbs"].commands["enable"], + ["Ethernet0", "NIC", "0", "0"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_enable_loopback(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["loopback"].commands["enable"], + ["Ethernet0", "NIC", "0"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) + def test_config_muxcable_disble_prbs(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(config.config.commands["muxcable"].commands["prbs"].commands["disable"], + ["Ethernet0", "NIC"], obj=db) + + assert result.exit_code == 0 + + @mock.patch('config.muxcable.delete_all_keys_in_db_table', mock.MagicMock(return_value=0)) + @mock.patch('config.muxcable.update_and_get_response_for_xcvr_cmd', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @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('utilities_common.platform_sfputil_helper.get_physical_to_logical', 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])) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) def test_config_muxcable_disable_loopback(self): runner = CliRunner() db = Db() result = runner.invoke(config.config.commands["muxcable"].commands["loopback"].commands["disable"], - ["0", "0"], obj=db) + ["Ethernet0", "NIC"], obj=db) assert result.exit_code == 0