diff --git a/config/muxcable.py b/config/muxcable.py index 7a94ab4f75..f6b5a6a39d 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(): """Show muxcable information""" @@ -416,35 +435,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) @@ -454,34 +534,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) @@ -506,9 +641,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] @@ -554,10 +690,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] @@ -584,7 +721,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") @@ -592,10 +728,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] @@ -641,9 +777,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] @@ -682,9 +819,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] @@ -731,9 +869,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] @@ -754,21 +893,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] @@ -814,12 +960,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] @@ -849,9 +997,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") @@ -897,9 +1046,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") @@ -915,3 +1065,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/show/muxcable.py b/show/muxcable.py index 34f4224fc6..190fc37d15 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 retrieve mux cable table entries".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") -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): + 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) + + 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) @@ -223,6 +346,7 @@ def update_and_get_response_for_xcvr_cmd(cmd_name, rsp_name, exp_rsp, cmd_table_ return res_dict + # 'muxcable' command ("show muxcable") # @@ -277,12 +401,23 @@ def get_switch_name(config_db): def create_json_dump_per_port_status(db, port_status_dict, muxcable_info_dict, muxcable_health_dict, muxcable_metrics_dict, asic_index, port): + res_dict = {} status_value = get_value_for_key_in_dict(muxcable_info_dict[asic_index], port, "state", "MUX_CABLE_TABLE") port_name = platform_sfputil_helper.get_interface_alias(port, db) port_status_dict["MUX_CABLE"][port_name] = {} port_status_dict["MUX_CABLE"][port_name]["STATUS"] = status_value health_value = get_value_for_key_in_dict(muxcable_health_dict[asic_index], port, "state", "MUX_LINKMGR_TABLE") port_status_dict["MUX_CABLE"][port_name]["HEALTH"] = health_value + res_dict = get_hwmode_mux_direction_port(db, port) + if res_dict[2] == "False": + hwstatus = "absent" + elif res_dict[1] == "not Y-Cable port": + hwstatus = "not Y-Cable port" + elif res_dict[1] == status_value: + hwstatus = "consistent" + else: + hwstatus = "inconsistent" + port_status_dict["MUX_CABLE"][port_name]["HWSTATUS"] = hwstatus last_switch_end_time = "" if "linkmgrd_switch_standby_end" in muxcable_metrics_dict[asic_index]: @@ -294,7 +429,9 @@ def create_json_dump_per_port_status(db, port_status_dict, muxcable_info_dict, m def create_table_dump_per_port_status(db, print_data, muxcable_info_dict, muxcable_health_dict, muxcable_metrics_dict, asic_index, port): print_port_data = [] + res_dict = {} + res_dict = get_hwmode_mux_direction_port(db, port) status_value = get_value_for_key_in_dict(muxcable_info_dict[asic_index], port, "state", "MUX_CABLE_TABLE") #status_value = get_value_for_key_in_tbl(y_cable_asic_table, port, "status") health_value = get_value_for_key_in_dict(muxcable_health_dict[asic_index], port, "state", "MUX_LINKMGR_TABLE") @@ -309,6 +446,15 @@ def create_table_dump_per_port_status(db, print_data, muxcable_info_dict, muxcab print_port_data.append(port_name) print_port_data.append(status_value) print_port_data.append(health_value) + if res_dict[2] == "False": + hwstatus = "absent" + elif res_dict[1] == "not Y-Cable port": + hwstatus = "not Y-Cable port" + elif res_dict[1] == status_value: + hwstatus = "consistent" + else: + hwstatus = "inconsistent" + print_port_data.append(hwstatus) print_port_data.append(last_switch_end_time) print_data.append(print_port_data) @@ -415,7 +561,7 @@ def status(db, port, json_output): create_table_dump_per_port_status(db, print_data, muxcable_info_dict, muxcable_health_dict, muxcable_metrics_dict, asic_index, port) - headers = ['PORT', 'STATUS', 'HEALTH', 'LAST_SWITCHOVER_TIME'] + headers = ['PORT', 'STATUS', 'HEALTH', 'HWSTATUS', 'LAST_SWITCHOVER_TIME'] click.echo(tabulate(print_data, headers=headers)) sys.exit(STATUS_SUCCESSFUL) @@ -465,7 +611,7 @@ def status(db, port, json_output): create_table_dump_per_port_status(db, print_data, muxcable_info_dict, muxcable_health_dict, muxcable_metrics_dict, asic_id, port) - headers = ['PORT', 'STATUS', 'HEALTH', 'LAST_SWITCHOVER_TIME'] + headers = ['PORT', 'STATUS', 'HEALTH', 'HWSTATUS','LAST_SWITCHOVER_TIME'] click.echo(tabulate(print_data, headers=headers)) sys.exit(STATUS_SUCCESSFUL) @@ -604,47 +750,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, 100, 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 @@ -679,6 +1079,31 @@ def cableinfo(db, port): click.echo(tabulate(body, headers=headers)) + +def get_hwmode_mux_direction_port(db, port): + + + delete_all_keys_in_db_table("APPL_DB", "XCVRD_SHOW_HWMODE_DIR_CMD") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RES") + + res_dict = {} + res_dict[0] = CONFIG_FAIL + res_dict[1] = "unknown" + res_dict[2] = "unknown" + result = {} + if port is not None: + + res_dict = update_and_get_response_for_xcvr_cmd( + "state", "state", "True", "XCVRD_SHOW_HWMODE_DIR_CMD", "XCVRD_SHOW_HWMODE_DIR_RES", "XCVRD_SHOW_HWMODE_DIR_RSP", port, 1, None, "probe") + + result = get_result(port, res_dict, "muxdirection" , result, "XCVRD_SHOW_HWMODE_DIR_RES") + + res_dict[2] = result.get("presence","unknown") + + return res_dict + + @muxcable.group(cls=clicommon.AbbreviationGroup) def hwmode(): """Shows the muxcable hardware information directly""" @@ -695,21 +1120,23 @@ def muxdirection(db, port): delete_all_keys_in_db_table("APPL_DB", "XCVRD_SHOW_HWMODE_DIR_CMD") delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RES") if port is not None: - 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") + if check_port_in_mux_cable_table(port) == False: + click.echo("Not Y-cable port") + return CONFIG_FAIL + + res_dict = get_hwmode_mux_direction_port(db, port) body = [] temp_list = [] - headers = ['Port', 'Direction'] + headers = ['Port', 'Direction', 'Presence'] port = platform_sfputil_helper.get_interface_alias(port, db) temp_list.append(port) temp_list.append(res_dict[1]) + temp_list.append(res_dict[2]) body.append(temp_list) rc = res_dict[0] @@ -717,6 +1144,7 @@ def muxdirection(db, port): delete_all_keys_in_db_table("APPL_DB", "XCVRD_SHOW_HWMODE_DIR_CMD") delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RES") return rc @@ -737,6 +1165,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() @@ -751,25 +1182,25 @@ def muxdirection(db, port): continue temp_list = [] - 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") + + res_dict = get_hwmode_mux_direction_port(db, port) + port = platform_sfputil_helper.get_interface_alias(port, db) temp_list.append(port) temp_list.append(res_dict[1]) + temp_list.append(res_dict[2]) body.append(temp_list) rc = res_dict[0] if rc != 0: rc_exit = False - headers = ['Port', 'Direction'] + headers = ['Port', 'Direction', 'Presence'] click.echo(tabulate(body, headers=headers)) delete_all_keys_in_db_table("APPL_DB", "XCVRD_SHOW_HWMODE_DIR_CMD") delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RSP") + delete_all_keys_in_db_table("STATE_DB", "XCVRD_SHOW_HWMODE_DIR_RES") if rc_exit == False: sys.exit(EXIT_FAIL) @@ -786,11 +1217,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 = [] @@ -825,6 +1260,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() @@ -843,7 +1281,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]) @@ -1028,7 +1466,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) @@ -1115,6 +1553,91 @@ def metrics(db, port, json_output): 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) + @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") diff --git a/tests/muxcable_test.py b/tests/muxcable_test.py index c0dfb1819d..30798e3915 100644 --- a/tests/muxcable_test.py +++ b/tests/muxcable_test.py @@ -25,58 +25,65 @@ tabular_data_status_output_expected = """\ -PORT STATUS HEALTH LAST_SWITCHOVER_TIME ----------- -------- --------- --------------------------- -Ethernet0 active healthy 2021-May-13 10:01:15.696728 -Ethernet4 standby healthy -Ethernet8 standby unhealthy -Ethernet12 unknown unhealthy -Ethernet16 standby healthy -Ethernet32 active healthy +PORT STATUS HEALTH HWSTATUS LAST_SWITCHOVER_TIME +---------- -------- --------- ------------ --------------------------- +Ethernet0 active healthy inconsistent 2021-May-13 10:01:15.696728 +Ethernet4 standby healthy consistent +Ethernet8 standby unhealthy consistent +Ethernet12 unknown unhealthy inconsistent +Ethernet16 standby healthy consistent +Ethernet32 active healthy inconsistent """ tabular_data_status_output_expected_alias = """\ -PORT STATUS HEALTH LAST_SWITCHOVER_TIME ------- -------- --------- --------------------------- -etp1 active healthy 2021-May-13 10:01:15.696728 -etp2 standby healthy -etp3 standby unhealthy -etp4 unknown unhealthy -etp5 standby healthy -etp9 active healthy +PORT STATUS HEALTH HWSTATUS LAST_SWITCHOVER_TIME +------ -------- --------- ------------ --------------------------- +etp1 active healthy inconsistent 2021-May-13 10:01:15.696728 +etp2 standby healthy consistent +etp3 standby unhealthy consistent +etp4 unknown unhealthy inconsistent +etp5 standby healthy consistent +etp9 active healthy inconsistent """ + json_data_status_output_expected = """\ { "MUX_CABLE": { "Ethernet0": { "STATUS": "active", "HEALTH": "healthy", + "HWSTATUS": "inconsistent", "LAST_SWITCHOVER_TIME": "2021-May-13 10:01:15.696728" }, "Ethernet4": { "STATUS": "standby", "HEALTH": "healthy", + "HWSTATUS": "consistent", "LAST_SWITCHOVER_TIME": "" }, "Ethernet8": { "STATUS": "standby", "HEALTH": "unhealthy", + "HWSTATUS": "consistent", "LAST_SWITCHOVER_TIME": "" }, "Ethernet12": { "STATUS": "unknown", "HEALTH": "unhealthy", + "HWSTATUS": "inconsistent", "LAST_SWITCHOVER_TIME": "" }, "Ethernet16": { "STATUS": "standby", "HEALTH": "healthy", + "HWSTATUS": "consistent", "LAST_SWITCHOVER_TIME": "" }, "Ethernet32": { "STATUS": "active", "HEALTH": "healthy", + "HWSTATUS": "inconsistent", "LAST_SWITCHOVER_TIME": "" } } @@ -89,31 +96,37 @@ "etp1": { "STATUS": "active", "HEALTH": "healthy", + "HWSTATUS": "inconsistent", "LAST_SWITCHOVER_TIME": "2021-May-13 10:01:15.696728" }, "etp2": { "STATUS": "standby", "HEALTH": "healthy", + "HWSTATUS": "consistent", "LAST_SWITCHOVER_TIME": "" }, "etp3": { "STATUS": "standby", "HEALTH": "unhealthy", + "HWSTATUS": "consistent", "LAST_SWITCHOVER_TIME": "" }, "etp4": { "STATUS": "unknown", "HEALTH": "unhealthy", + "HWSTATUS": "inconsistent", "LAST_SWITCHOVER_TIME": "" }, "etp5": { "STATUS": "standby", "HEALTH": "healthy", + "HWSTATUS": "consistent", "LAST_SWITCHOVER_TIME": "" }, "etp9": { "STATUS": "active", "HEALTH": "healthy", + "HWSTATUS": "inconsistent", "LAST_SWITCHOVER_TIME": "" } } @@ -354,27 +367,27 @@ """ show_muxcable_hwmode_muxdirection_active_expected_output = """\ -Port Direction ----------- ----------- -Ethernet12 active +Port Direction Presence +---------- ----------- ---------- +Ethernet12 active True """ show_muxcable_hwmode_muxdirection_active_expected_output_alias = """\ -Port Direction ------- ----------- -etp4 active +Port Direction Presence +------ ----------- ---------- +etp4 active True """ show_muxcable_hwmode_muxdirection_standby_expected_output = """\ -Port Direction ----------- ----------- -Ethernet12 standby +Port Direction Presence +---------- ----------- ---------- +Ethernet12 standby True """ show_muxcable_hwmode_muxdirection_standby_expected_output_alias = """\ -Port Direction ------- ----------- -etp4 standby +Port Direction Presence +------ ----------- ---------- +etp4 standby True """ show_muxcable_firmware_version_expected_output = """\ @@ -453,6 +466,9 @@ def setup_class(cls): #show.muxcable.platform_sfputil.logical = mock.Mock(return_value=["Ethernet0", "Ethernet4"]) print("SETUP") + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status(self): runner = CliRunner() db = Db() @@ -461,6 +477,9 @@ def test_muxcable_status(self): assert result.exit_code == 0 assert result.output == tabular_data_status_output_expected + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_alias(self): runner = CliRunner() db = Db() @@ -471,6 +490,9 @@ def test_muxcable_status_alias(self): assert result.exit_code == 0 assert result.output == tabular_data_status_output_expected_alias + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json(self): runner = CliRunner() db = Db() @@ -480,6 +502,9 @@ def test_muxcable_status_json(self): assert result.exit_code == 0 assert result.output == json_data_status_output_expected + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json_alias(self): runner = CliRunner() db = Db() @@ -540,6 +565,9 @@ def test_muxcable_config_json_with_incorrect_port(self): assert result.exit_code == 1 + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json_with_correct_port(self): runner = CliRunner() db = Db() @@ -549,6 +577,9 @@ def test_muxcable_status_json_with_correct_port(self): assert result.exit_code == 0 + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json_with_correct_port_alias(self): runner = CliRunner() db = Db() @@ -561,6 +592,9 @@ def test_muxcable_status_json_with_correct_port_alias(self): assert result.exit_code == 0 + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json_port_incorrect_index(self): runner = CliRunner() db = Db() @@ -576,6 +610,9 @@ def test_muxcable_status_with_patch(self): result = runner.invoke(show.cli.commands["muxcable"], obj=db) + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json_with_incorrect_port(self): runner = CliRunner() db = Db() @@ -621,6 +658,9 @@ def test_muxcable_config_json_with_incorrect_port_patch(self): assert result.exit_code == 1 + @mock.patch('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value=({ 0 :"active", + 1 :"standby", + 2 : "True"}))) def test_muxcable_status_json_port_eth0(self): runner = CliRunner() db = Db() @@ -820,6 +860,12 @@ def test_config_muxcable_packetloss_reset_Ethernet0(self): 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('show.muxcable.get_result', mock.MagicMock(return_value={0: 0, + 1: "active"})) + @mock.patch('click.confirm', mock.MagicMock(return_value=("y"))) @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])) def test_show_muxcable_eye_info(self): @@ -827,62 +873,728 @@ def test_show_muxcable_eye_info(self): 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('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_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('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('os.geteuid', mock.MagicMock(return_value=0)) - @mock.patch('sonic_y_cable.y_cable.disable_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_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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "active", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "active", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) + @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('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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "sucess", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "sucess", + 2: "True"})) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "sucess", + 2: "True"})) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @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 @@ -940,6 +1652,10 @@ def test_show_muxcable_cableinfo_incorrect_logical_port_return_value(self): @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_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "active", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) @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"]})) @@ -960,6 +1676,10 @@ def test_show_muxcable_hwmode_muxdirection_port_active(self): @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_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "active", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) @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"]})) @@ -982,6 +1702,10 @@ def test_show_muxcable_hwmode_muxdirection_active_expected_output_alias(self): @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) @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"]})) @@ -1000,6 +1724,10 @@ def test_show_muxcable_hwmode_muxdirection_active(self): @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) @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"]})) @@ -1020,6 +1748,10 @@ def test_show_muxcable_hwmode_muxdirection_port_standby(self): @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "standby", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) @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"]})) @@ -1042,6 +1774,10 @@ def test_show_muxcable_hwmode_muxdirection_port_standby_alias(self): @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('show.muxcable.get_hwmode_mux_direction_port', mock.MagicMock(return_value={0: 0, + 1: "sucess", + 2: "True"})) + @mock.patch('show.muxcable.check_port_in_mux_cable_table', mock.MagicMock(return_value=True)) @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"]}))