diff --git a/show/muxcable.py b/show/muxcable.py index d9f0a94f15..b640d32135 100644 --- a/show/muxcable.py +++ b/show/muxcable.py @@ -36,6 +36,21 @@ VENDOR_MODEL_REGEX = re.compile(r"CAC\w{3}321P2P\w{2}MS") +def get_asic_index_for_port(port): + 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: + port_name = platform_sfputil_helper.get_interface_alias(port, db) + click.echo("Got invalid asic index for port {}, cant retreive mux status".format(port_name)) + return 0 + return asic_index + def db_connect(db_name, namespace=EMPTY_NAMESPACE): return swsscommon.DBConnector(db_name, REDIS_TIMEOUT_MSECS, True, namespace) @@ -1239,6 +1254,67 @@ def get_hwmode_mux_direction_port(db, port): return res_dict +def create_active_active_mux_direction_json_result(result, port, db): + + port = platform_sfputil_helper.get_interface_alias(port, db) + result["HWMODE"][port] = {} + res_dict = get_grpc_cached_version_mux_direction_per_port(db, port) + result["HWMODE"][port]["Direction"] = res_dict["self_mux_direction"] + result["HWMODE"][port]["Presence"] = res_dict["presence"] + result["HWMODE"][port]["PeerDirection"] = res_dict["peer_mux_direction"] + result["HWMODE"][port]["ConnectivityState"] = res_dict["grpc_connection_status"] + + rc = res_dict["rc"] + + return rc + +def create_active_standby_mux_direction_json_result(result, port, db): + + res_dict = get_hwmode_mux_direction_port(db, port) + port = platform_sfputil_helper.get_interface_alias(port, db) + result["HWMODE"][port] = {} + result["HWMODE"][port]["Direction"] = res_dict[1] + result["HWMODE"][port]["Presence"] = res_dict[2] + + rc = res_dict[0] + + return rc + +def create_active_active_mux_direction_result(body, port, db): + + res_dict = get_grpc_cached_version_mux_direction_per_port(db, port) + temp_list = [] + port = platform_sfputil_helper.get_interface_alias(port, db) + temp_list.append(port) + temp_list.append(res_dict["self_mux_direction"]) + temp_list.append(res_dict["presence"]) + temp_list.append(res_dict["peer_mux_direction"]) + temp_list.append(res_dict["grpc_connection_status"]) + body.append(temp_list) + + rc = res_dict["rc"] + + return rc + +def create_active_standby_mux_direction_result(body, port, db): + + res_dict = get_hwmode_mux_direction_port(db, port) + + temp_list = [] + 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] + + 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 + @muxcable.group(cls=clicommon.AbbreviationGroup) def hwmode(): """Shows the muxcable hardware information directly""" @@ -1247,8 +1323,9 @@ def hwmode(): @hwmode.command() @click.argument('port', metavar='', 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 muxdirection(db, port): +def muxdirection(db, port, json_output): """Shows the current direction of the muxcable {active/standy}""" port = platform_sfputil_helper.get_interface_name(port, db) @@ -1256,30 +1333,42 @@ 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") + per_npu_configdb = {} - if port is not None: + namespaces = multi_asic.get_front_end_namespaces() + for namespace in namespaces: + asic_id = multi_asic.get_asic_index_from_namespace(namespace) + per_npu_configdb[asic_id] = ConfigDBConnector(use_unix_socket_path=False, namespace=namespace) + per_npu_configdb[asic_id].connect() + + if port is not None: + + asic_index = get_asic_index_for_port(port) + cable_type = get_optional_value_for_key_in_config_tbl(per_npu_configdb[asic_index], port, "cable_type", "MUX_CABLE") 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', '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] - click.echo(tabulate(body, headers=headers)) + if json_output: + result = {} + result ["HWMODE"] = {} + if cable_type == "active-active": + rc = create_active_active_mux_direction_json_result(result, port, db) + else: + rc = False + rc = create_active_standby_mux_direction_json_result(result, port, db) + click.echo("{}".format(json.dumps(result, indent=4))) - 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") + else: + body = [] + if cable_type == "active-active": + headers = ['Port', 'Direction', 'Presence', 'PeerDirection', 'ConnectivityState'] + rc = create_active_active_mux_direction_result(body, port, db) + else: + rc = create_active_standby_mux_direction_result(body, port, db) + headers = ['Port', 'Direction', 'Presence'] + click.echo(tabulate(body, headers=headers)) return rc @@ -1289,8 +1378,12 @@ def muxdirection(db, port): rc_exit = True body = [] + active_active = False + if json_output: + result = {} + result ["HWMODE"] = {} - for port in logical_port_list: + for port in natsorted(logical_port_list): if platform_sfputil is not None: physical_port_list = platform_sfputil_helper.logical_port_name_to_physical_port_list(port) @@ -1316,26 +1409,37 @@ def muxdirection(db, port): if port != logical_port_list_per_port[0]: continue - temp_list = [] + + asic_index = get_asic_index_for_port(port) + cable_type = get_optional_value_for_key_in_config_tbl(per_npu_configdb[asic_index], port, "cable_type", "MUX_CABLE") + if json_output: + if cable_type == "active-active": + rc = create_active_active_mux_direction_json_result(result, port, db) + active_active = True + else: + rc = create_active_standby_mux_direction_json_result(result, port, db) - res_dict = get_hwmode_mux_direction_port(db, port) + else: + if cable_type == 'active-active': + rc = create_active_active_mux_direction_result(body, port, db) + active_active = True + else: + rc = create_active_standby_mux_direction_result(body, port, db) + if rc != 0: + rc_exit = False - 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', 'Presence'] - click.echo(tabulate(body, headers=headers)) + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + if active_active: + + headers = ['Port', 'Direction', 'Presence', 'PeerDirection', 'ConnectivityState'] + else: + 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) @@ -2003,3 +2107,152 @@ def tunnel_route(db, port, json_output): click.echo(tabulate(print_data, headers=headers)) sys.exit(STATUS_SUCCESSFUL) + + +def get_grpc_cached_version_mux_direction_per_port(db, port): + + + state_db = {} + mux_info_dict = {} + mux_info_full_dict = {} + trans_info_full_dict = {} + mux_info_dict["rc"] = False + + # 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) + state_db[asic_id] = swsscommon.SonicV2Connector(use_unix_socket_path=False, namespace=namespace) + state_db[asic_id].connect(state_db[asic_id].STATE_DB) + + 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 mux_info_dict + + + mux_info_full_dict[asic_index] = state_db[asic_index].get_all( + state_db[asic_index].STATE_DB, 'MUX_CABLE_INFO|{}'.format(port)) + trans_info_full_dict[asic_index] = state_db[asic_index].get_all( + state_db[asic_index].STATE_DB, 'TRANSCEIVER_STATUS|{}'.format(port)) + + res_dir = {} + res_dir = mux_info_full_dict[asic_index] + mux_info_dict["self_mux_direction"] = res_dir.get("self_mux_direction", None) + mux_info_dict["peer_mux_direction"] = res_dir.get("peer_mux_direction", None) + mux_info_dict["grpc_connection_status"] = res_dir.get("grpc_connection_status", None) + + trans_dir = {} + trans_dir = trans_info_full_dict[asic_index] + + status = trans_dir.get("status", "0") + presence = "True" if status == "1" else "False" + + mux_info_dict["presence"] = presence + + mux_info_dict["rc"] = True + + return mux_info_dict + + +@muxcable.group(cls=clicommon.AbbreviationGroup) +def grpc(): + """Shows the muxcable hardware information directly""" + pass + + +@grpc.command() +@click.argument('port', metavar='', 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 muxdirection(db, port, json_output): + """Shows the current direction of the FPGA facing port on Tx Side {active/standy}""" + + port = platform_sfputil_helper.get_interface_name(port, db) + + + if port is not None: + + if check_port_in_mux_cable_table(port) == False: + click.echo("Not Y-cable port") + return CONFIG_FAIL + + if json_output: + result = {} + result ["HWMODE"] = {} + rc = create_active_active_mux_direction_json_result(result, port, db) + click.echo("{}".format(json.dumps(result, indent=4))) + + else: + body = [] + + headers = ['Port', 'Direction', 'Presence', 'PeerDirection', 'ConnectivityState'] + rc = create_active_active_mux_direction_result(body, port, db) + click.echo(tabulate(body, headers=headers)) + + return rc + + else: + + + logical_port_list = platform_sfputil_helper.get_logical_list() + + rc_exit = True + body = [] + if json_output: + result = {} + result ["HWMODE"] = {} + + for port in natsorted(logical_port_list): + + if platform_sfputil is not None: + physical_port_list = platform_sfputil_helper.logical_port_name_to_physical_port_list(port) + + if not isinstance(physical_port_list, list): + continue + 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() + + logical_port_list_per_port = logical_port_list_for_physical_port.get(physical_port, None) + + """ This check is required for checking whether or not this logical port is the one which is + actually mapped to physical port and by convention it is always the first port. + TODO: this should be removed with more logic to check which logical port maps to actual physical port + being used""" + + if port != logical_port_list_per_port[0]: + continue + + if json_output: + rc = create_active_active_mux_direction_json_result(result, port, db) + else: + rc = create_active_active_mux_direction_result(body, port, db) + + if rc != True: + rc_exit = False + + if json_output: + click.echo("{}".format(json.dumps(result, indent=4))) + else: + headers = ['Port', 'Direction', 'Presence', 'PeerDirection', 'ConnectivityState'] + + click.echo(tabulate(body, headers=headers)) + + if rc_exit == False: + sys.exit(EXIT_FAIL) + + diff --git a/tests/mock_tables/state_db.json b/tests/mock_tables/state_db.json index 8462248a92..97633b3ef8 100644 --- a/tests/mock_tables/state_db.json +++ b/tests/mock_tables/state_db.json @@ -549,6 +549,24 @@ "version_self_next": "0.2MS", "version_self_active": "0.2MS", "version_self_inactive": "0.2MS", + "peer_mux_direction": "active", + "self_mux_direction": "active", + "grpc_connection_status": "READY", + "Value": "AABB" + }, + "MUX_CABLE_INFO|Ethernet4": { + "version_peer_next": "0.2MS", + "version_peer_active": "0.2MS", + "version_peer_inactive": "0.2MS", + "version_nic_next": "0.2MS", + "version_nic_active": "0.2MS", + "version_nic_inactive": "0.2MS", + "version_self_next": "0.2MS", + "version_self_active": "0.2MS", + "version_self_inactive": "0.2MS", + "peer_mux_direction": "active", + "self_mux_direction": "standby", + "grpc_connection_status": "READY", "Value": "AABB" }, "MUX_CABLE_INFO|Ethernet12": { diff --git a/tests/muxcable_test.py b/tests/muxcable_test.py index b8eb3dce62..7e4b4b250c 100644 --- a/tests/muxcable_test.py +++ b/tests/muxcable_test.py @@ -378,6 +378,50 @@ } """ +show_muxcable_grpc_muxdirection_active_expected_output = """\ +Port Direction Presence PeerDirection ConnectivityState +---------- ----------- ---------- --------------- ------------------- +Ethernet12 active True active READY +""" + +show_muxcable_grpc_muxdirection_standby_expected_output = """\ +Port Direction Presence PeerDirection ConnectivityState +--------- ----------- ---------- --------------- ------------------- +Ethernet4 standby True active READY +""" + +show_muxcable_grpc_muxdirection_active_expected_all_output = """\ +Port Direction Presence PeerDirection ConnectivityState +--------- ----------- ---------- --------------- ------------------- +Ethernet0 active False active READY +""" + +show_muxcable_grpc_muxdirection_active_expected_all_output_json = """\ +{ + "HWMODE": { + "Ethernet0": { + "Direction": "active", + "Presence": "False", + "PeerDirection": "active", + "ConnectivityState": "READY" + } + } +} +""" + +show_muxcable_grpc_muxdirection_standby_expected_output_json = """\ +{ + "HWMODE": { + "Ethernet4": { + "Direction": "standby", + "Presence": "True", + "PeerDirection": "active", + "ConnectivityState": "READY" + } + } +} +""" + expected_muxcable_cableinfo_output = """\ Vendor Model -------- --------------- @@ -395,6 +439,17 @@ Ethernet12 active True """ +show_muxcable_hwmode_muxdirection_active_expected_output_json = """\ +{ + "HWMODE": { + "Ethernet12": { + "Direction": "active", + "Presence": "True" + } + } +} +""" + show_muxcable_hwmode_muxdirection_active_expected_output_alias = """\ Port Direction Presence ------ ----------- ---------- @@ -2367,6 +2422,138 @@ def test_config_muxcable_telemetry_enable(self): "enable"], 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_grpc_cached_version_mux_direction_per_port', mock.MagicMock(return_value={"self_mux_direction": "active", + "peer_mux_direction": "active", + "presence": "True", + "rc": 0, + "grpc_connection_status": "READY"})) + @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_grpc_muxdirection_port_standby_with_patch(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["grpc"].commands["muxdirection"], + ["Ethernet12"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_grpc_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: "standby"})) + @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", "Ethernet4"])) + @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: ["Ethernet4", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', mock.MagicMock(return_value={0: ["Ethernet4", "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_grpc_muxdirection_port_standby(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["grpc"].commands["muxdirection"], + ["Ethernet4"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_grpc_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.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", "Ethernet4"])) + @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: ["Ethernet4", "Ethernet0"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', mock.MagicMock(return_value={0: ["Ethernet4", "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_grpc_muxdirection_port_standby_json(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["grpc"].commands["muxdirection"], + ["Ethernet4", "--json"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_grpc_muxdirection_standby_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: "standby"})) + @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", "Ethernet4"])) + @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: ["Ethernet0", "Ethernet4"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', mock.MagicMock(return_value={0: ["Ethernet0", "Ethernet4"]})) + @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_grpc_muxdirection_port_all(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["grpc"].commands["muxdirection"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_grpc_muxdirection_active_expected_all_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.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", "Ethernet4"])) + @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: ["Ethernet0", "Ethernet4"]})) + @mock.patch('utilities_common.platform_sfputil_helper.get_physical_to_logical', mock.MagicMock(return_value={0: ["Ethernet0", "Ethernet4"]})) + @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_grpc_muxdirection_port_all_json(self): + runner = CliRunner() + db = Db() + + result = runner.invoke(show.cli.commands["muxcable"].commands["grpc"].commands["muxdirection"], ["--json"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_grpc_muxdirection_active_expected_all_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: "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", "--json"], obj=db) + assert result.exit_code == 0 + assert result.output == show_muxcable_hwmode_muxdirection_active_expected_output_json + @classmethod def teardown_class(cls): os.environ['UTILITIES_UNIT_TESTING'] = "0"