emqx4.3/apps/emqx_bridge/test/emqx_bridge_v2_api_SUITE.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emqx_bridge_v2_api_SUITE).

-compile(nowarn_export_all).
-compile(export_all).

-import(emqx_mgmt_api_test_util, [uri/1]).

-include_lib("eunit/include/eunit.hrl").
-include_lib("common_test/include/ct.hrl").
-include_lib("snabbkaffe/include/test_macros.hrl").

-define(ACTIONS_ROOT, "actions").
-define(SOURCES_ROOT, "sources").

-define(ACTION_CONNECTOR_NAME, <<"my_connector">>).
-define(SOURCE_CONNECTOR_NAME, <<"my_connector">>).

-define(RESOURCE(NAME, TYPE), #{
    <<"enable">> => true,
    %<<"ssl">> => #{<<"enable">> => false},
    <<"type">> => TYPE,
    <<"name">> => NAME
}).

-define(ACTION_CONNECTOR_TYPE_STR, "kafka_producer").
-define(ACTION_CONNECTOR_TYPE, <<?ACTION_CONNECTOR_TYPE_STR>>).
-define(KAFKA_BOOTSTRAP_HOST, <<"127.0.0.1:9092">>).
-define(KAFKA_CONNECTOR(Name, BootstrapHosts), ?RESOURCE(Name, ?ACTION_CONNECTOR_TYPE)#{
    <<"authentication">> => <<"none">>,
    <<"bootstrap_hosts">> => BootstrapHosts,
    <<"connect_timeout">> => <<"5s">>,
    <<"metadata_request_timeout">> => <<"5s">>,
    <<"min_metadata_refresh_interval">> => <<"3s">>,
    <<"socket_opts">> =>
        #{
            <<"nodelay">> => true,
            <<"recbuf">> => <<"1024KB">>,
            <<"sndbuf">> => <<"1024KB">>,
            <<"tcp_keepalive">> => <<"none">>
        }
}).

-define(ACTIONS_CONNECTOR(Name), ?KAFKA_CONNECTOR(Name, ?KAFKA_BOOTSTRAP_HOST)).
-define(ACTIONS_CONNECTOR, ?ACTIONS_CONNECTOR(?ACTION_CONNECTOR_NAME)).

-define(MQTT_LOCAL_TOPIC, <<"mqtt/local/topic">>).
-define(BRIDGE_NAME, (atom_to_binary(?FUNCTION_NAME))).
-define(ACTION_TYPE_STR, "kafka_producer").
-define(ACTION_TYPE, <<?ACTION_TYPE_STR>>).
-define(KAFKA_BRIDGE(Name, Connector), ?RESOURCE(Name, ?ACTION_TYPE)#{
    <<"connector">> => Connector,
    <<"kafka">> => #{
        <<"buffer">> => #{
            <<"memory_overload_protection">> => true,
            <<"mode">> => <<"hybrid">>,
            <<"per_partition_limit">> => <<"2GB">>,
            <<"segment_bytes">> => <<"100MB">>
        },
        <<"compression">> => <<"no_compression">>,
        <<"kafka_ext_headers">> => [
            #{
                <<"kafka_ext_header_key">> => <<"clientid">>,
                <<"kafka_ext_header_value">> => <<"${clientid}">>
            },
            #{
                <<"kafka_ext_header_key">> => <<"topic">>,
                <<"kafka_ext_header_value">> => <<"${topic}">>
            }
        ],
        <<"kafka_header_value_encode_mode">> => <<"none">>,
        <<"kafka_headers">> => <<"${pub_props}">>,
        <<"max_batch_bytes">> => <<"896KB">>,
        <<"max_inflight">> => 10,
        <<"message">> => #{
            <<"key">> => <<"${.clientid}">>,
            <<"timestamp">> => <<"${.timestamp}">>,
            <<"value">> => <<"${.}">>
        },
        <<"partition_count_refresh_interval">> => <<"60s">>,
        <<"partition_strategy">> => <<"random">>,
        <<"required_acks">> => <<"all_isr">>,
        <<"topic">> => <<"kafka-topic">>
    },
    <<"local_topic">> => ?MQTT_LOCAL_TOPIC,
    <<"resource_opts">> => #{
        <<"health_check_interval">> => <<"32s">>
    }
}).
-define(KAFKA_BRIDGE(Name), ?KAFKA_BRIDGE(Name, ?ACTION_CONNECTOR_NAME)).

-define(KAFKA_BRIDGE_UPDATE(Name, Connector),
    maps:without([<<"name">>, <<"type">>], ?KAFKA_BRIDGE(Name, Connector))
).

-define(SOURCE_TYPE_STR, "mqtt").
-define(SOURCE_TYPE, <<?SOURCE_TYPE_STR>>).

-define(APPSPECS, [
    emqx_conf,
    emqx,
    emqx_auth,
    emqx_management,
    emqx_connector,
    {emqx_bridge, "actions {}"},
    {emqx_rule_engine, "rule_engine { rules {} }"}
]).

-define(APPSPEC_DASHBOARD,
    {emqx_dashboard, "dashboard.listeners.http { enable = true, bind = 18083 }"}
).

%%------------------------------------------------------------------------------
%% CT boilerplate
%%------------------------------------------------------------------------------

-if(?EMQX_RELEASE_EDITION == ee).
%% For now we got only kafka implementing `bridge_v2` and that is enterprise only.
all() ->
    All0 = emqx_common_test_helpers:all(?MODULE),
    All = All0 -- matrix_cases(),
    Groups = lists:map(fun({G, _, _}) -> {group, G} end, groups()),
    Groups ++ All.
-else.
all() ->
    [].
-endif.

matrix_cases() ->
    emqx_common_test_helpers:all(?MODULE).

groups() ->
    emqx_common_test_helpers:matrix_to_groups(?MODULE, matrix_cases()).

suite() ->
    [{timetrap, {seconds, 60}}].

init_per_suite(Config) ->
    Config.

end_per_suite(_Config) ->
    ok.

init_per_group(cluster = Name, Config) ->
    Nodes = [NodePrimary | _] = mk_cluster(Name, Config),
    init_api([{group, Name}, {cluster_nodes, Nodes}, {node, NodePrimary} | Config]);
init_per_group(cluster_later_join = Name, Config) ->
    Nodes = [NodePrimary | _] = mk_cluster(Name, Config, #{join_to => undefined}),
    init_api([{group, Name}, {cluster_nodes, Nodes}, {node, NodePrimary} | Config]);
init_per_group(single = Group, Config) ->
    WorkDir = filename:join(?config(priv_dir, Config), Group),
    Apps = emqx_cth_suite:start(?APPSPECS ++ [?APPSPEC_DASHBOARD], #{work_dir => WorkDir}),
    init_api([{group, single}, {group_apps, Apps}, {node, node()} | Config]);
init_per_group(actions, Config) ->
    [{bridge_kind, action} | Config];
init_per_group(sources, Config) ->
    [{bridge_kind, source} | Config];
init_per_group(_Group, Config) ->
    Config.

init_api(Config) ->
    Node = ?config(node, Config),
    {ok, ApiKey} = erpc:call(Node, emqx_common_test_http, create_default_app, []),
    [{api_key, ApiKey} | Config].

mk_cluster(Name, Config) ->
    mk_cluster(Name, Config, #{}).

mk_cluster(Name, Config, Opts) ->
    Node1Apps = ?APPSPECS ++ [?APPSPEC_DASHBOARD],
    Node2Apps = ?APPSPECS,
    emqx_cth_cluster:start(
        [
            {emqx_bridge_v2_api_SUITE_1, Opts#{role => core, apps => Node1Apps}},
            {emqx_bridge_v2_api_SUITE_2, Opts#{role => core, apps => Node2Apps}}
        ],
        #{work_dir => emqx_cth_suite:work_dir(Name, Config)}
    ).

end_per_group(Group, Config) when
    Group =:= cluster;
    Group =:= cluster_later_join
->
    ok = emqx_cth_cluster:stop(?config(cluster_nodes, Config));
end_per_group(single, Config) ->
    emqx_cth_suite:stop(?config(group_apps, Config)),
    ok;
end_per_group(_Group, _Config) ->
    ok.

init_per_testcase(t_action_types, Config) ->
    case ?config(cluster_nodes, Config) of
        undefined ->
            init_mocks();
        Nodes ->
            [erpc:call(Node, ?MODULE, init_mocks, []) || Node <- Nodes]
    end,
    Config;
init_per_testcase(_TestCase, Config) ->
    case ?config(cluster_nodes, Config) of
        undefined ->
            init_mocks();
        Nodes ->
            [erpc:call(Node, ?MODULE, init_mocks, []) || Node <- Nodes]
    end,
    case ?config(bridge_kind, Config) of
        action ->
            {ok, 201, _} = request(post, uri(["connectors"]), ?ACTIONS_CONNECTOR, Config);
        source ->
            {ok, 201, _} = request(
                post,
                uri(["connectors"]),
                source_connector_create_config(#{}),
                Config
            )
    end,
    Config.

end_per_testcase(_TestCase, Config) ->
    Node = ?config(node, Config),
    ok = erpc:call(Node, fun clear_resources/0),
    case ?config(cluster_nodes, Config) of
        undefined ->
            meck:unload();
        ClusterNodes ->
            [erpc:call(ClusterNode, meck, unload, []) || ClusterNode <- ClusterNodes]
    end,
    ok = emqx_common_test_helpers:call_janitor(),
    ok.

%%------------------------------------------------------------------------------
%% Helper fns
%%------------------------------------------------------------------------------

-define(CONNECTOR_IMPL, emqx_bridge_v2_dummy_connector).
init_mocks() ->
    meck:new(emqx_connector_resource, [passthrough, no_link]),
    meck:expect(emqx_connector_resource, connector_to_resource_type, 1, ?CONNECTOR_IMPL),
    meck:new(?CONNECTOR_IMPL, [non_strict, no_link]),
    meck:expect(?CONNECTOR_IMPL, callback_mode, 0, async_if_possible),
    meck:expect(
        ?CONNECTOR_IMPL,
        on_start,
        fun
            (<<"connector:", ?ACTION_CONNECTOR_TYPE_STR, ":bad_", _/binary>>, _C) ->
                {ok, bad_connector_state};
            (_I, _C) ->
                {ok, connector_state}
        end
    ),
    meck:expect(?CONNECTOR_IMPL, on_stop, 2, ok),
    meck:expect(
        ?CONNECTOR_IMPL,
        on_get_status,
        fun
            (_, bad_connector_state) -> connecting;
            (_, _) -> connected
        end
    ),
    meck:expect(?CONNECTOR_IMPL, on_add_channel, 4, {ok, connector_state}),
    meck:expect(?CONNECTOR_IMPL, on_remove_channel, 3, {ok, connector_state}),
    meck:expect(?CONNECTOR_IMPL, on_get_channel_status, 3, connected),
    ok = meck:expect(?CONNECTOR_IMPL, on_get_channels, fun(ResId) ->
        emqx_bridge_v2:get_channels_for_connector(ResId)
    end),
    meck:expect(?CONNECTOR_IMPL, on_query_async, fun(_ResId, _Req, ReplyFunAndArgs, _ConnState) ->
        emqx_resource:apply_reply_fun(ReplyFunAndArgs, ok),
        {ok, self()}
    end),
    ok.

clear_resources() ->
    emqx_bridge_v2_testlib:delete_all_bridges_and_connectors().

expect_on_all_nodes(Mod, Function, Fun, Config) ->
    case ?config(cluster_nodes, Config) of
        undefined ->
            ok = meck:expect(Mod, Function, Fun);
        Nodes ->
            [erpc:call(Node, meck, expect, [Mod, Function, Fun]) || Node <- Nodes]
    end,
    ok.

connector_operation(Config, ConnectorType, ConnectorName, OperationName) ->
    case ?config(group, Config) of
        cluster ->
            case ?config(cluster_nodes, Config) of
                undefined ->
                    Node = ?config(node, Config),
                    ok = rpc:call(
                        Node,
                        emqx_connector_resource,
                        OperationName,
                        [ConnectorType, ConnectorName],
                        500
                    );
                Nodes ->
                    erpc:multicall(
                        Nodes,
                        emqx_connector_resource,
                        OperationName,
                        [ConnectorType, ConnectorName],
                        500
                    )
            end;
        _ ->
            ok = emqx_connector_resource:OperationName(ConnectorType, ConnectorName)
    end.

listen_on_random_port() ->
    SockOpts = [binary, {active, false}, {packet, raw}, {reuseaddr, true}, {backlog, 1000}],
    case gen_tcp:listen(0, SockOpts) of
        {ok, Sock} ->
            {ok, Port} = inet:port(Sock),
            {Port, Sock};
        {error, Reason} when Reason /= eaddrinuse ->
            {error, Reason}
    end.

request(Method, URL, Config) ->
    request(Method, URL, [], Config).

request(Method, {operation, Type, Op, BridgeID}, Body, Config) ->
    URL = operation_path(Type, Op, BridgeID, Config),
    request(Method, URL, Body, Config);
request(Method, URL, Body, Config) ->
    AuthHeader = emqx_common_test_http:auth_header(?config(api_key, Config)),
    Opts = #{compatible_mode => true, httpc_req_opts => [{body_format, binary}]},
    emqx_mgmt_api_test_util:request_api(Method, URL, [], AuthHeader, Body, Opts).

request(Method, URL, Body, Decoder, Config) ->
    case request(Method, URL, Body, Config) of
        {ok, Code, Response} ->
            case Decoder(Response) of
                {error, _} = Error -> Error;
                Decoded -> {ok, Code, Decoded}
            end;
        Otherwise ->
            Otherwise
    end.

request_json(Method, URLLike, Config) ->
    request(Method, URLLike, [], fun json/1, Config).

request_json(Method, URLLike, Body, Config) ->
    request(Method, URLLike, Body, fun json/1, Config).

operation_path(node, Oper, BridgeID, Config) ->
    [_SingleOrCluster, Kind | _] = group_path(Config),
    #{api_root_key := APIRootKey} = get_common_values(Kind, <<"unused">>),
    uri(["nodes", ?config(node, Config), APIRootKey, BridgeID, Oper]);
operation_path(cluster, Oper, BridgeID, Config) ->
    [_SingleOrCluster, Kind | _] = group_path(Config),
    #{api_root_key := APIRootKey} = get_common_values(Kind, <<"unused">>),
    uri([APIRootKey, BridgeID, Oper]).

enable_path(Enable, BridgeID) ->
    uri([?ACTIONS_ROOT, BridgeID, "enable", Enable]).

publish_message(Topic, Body, Config) ->
    Node = ?config(node, Config),
    publish_message(Topic, Body, Node, Config).

publish_message(Topic, Body, Node, _Config) ->
    erpc:call(Node, emqx, publish, [emqx_message:make(Topic, Body)]).

update_config(Path, Value, Config) ->
    Node = ?config(node, Config),
    erpc:call(Node, emqx, update_config, [Path, Value]).

get_raw_config(Path, Config) ->
    Node = ?config(node, Config),
    erpc:call(Node, emqx, get_raw_config, [Path]).

add_user_auth(Chain, AuthenticatorID, User, Config) ->
    Node = ?config(node, Config),
    erpc:call(Node, emqx_authentication, add_user, [Chain, AuthenticatorID, User]).

delete_user_auth(Chain, AuthenticatorID, User, Config) ->
    Node = ?config(node, Config),
    erpc:call(Node, emqx_authentication, delete_user, [Chain, AuthenticatorID, User]).

str(S) when is_list(S) -> S;
str(S) when is_binary(S) -> binary_to_list(S).

json(B) when is_binary(B) ->
    case emqx_utils_json:safe_decode(B, [return_maps]) of
        {ok, Term} ->
            Term;
        {error, Reason} = Error ->
            ct:pal("Failed to decode json: ~p~n~p", [Reason, B]),
            Error
    end.

group_path(Config) ->
    case emqx_common_test_helpers:group_path(Config) of
        [] ->
            undefined;
        Path ->
            Path
    end.

source_connector_config_base() ->
    #{
        <<"enable">> => true,
        <<"description">> => <<"my connector">>,
        <<"pool_size">> => 3,
        <<"proto_ver">> => <<"v5">>,
        <<"server">> => <<"127.0.0.1:1883">>,
        <<"resource_opts">> => #{
            <<"health_check_interval">> => <<"15s">>,
            <<"start_after_created">> => true,
            <<"start_timeout">> => <<"5s">>
        }
    }.

source_connector_create_config(Overrides0) ->
    Overrides = emqx_utils_maps:binary_key_map(Overrides0),
    Conf0 = maps:merge(
        source_connector_config_base(),
        #{
            <<"enable">> => true,
            <<"type">> => ?SOURCE_TYPE,
            <<"name">> => ?SOURCE_CONNECTOR_NAME
        }
    ),
    maps:merge(
        Conf0,
        Overrides
    ).

source_config_base() ->
    #{
        <<"enable">> => true,
        <<"connector">> => ?SOURCE_CONNECTOR_NAME,
        <<"parameters">> =>
            #{
                <<"topic">> => <<"remote/topic">>,
                <<"qos">> => 2
            },
        <<"resource_opts">> => #{
            <<"health_check_interval">> => <<"15s">>,
            <<"resume_interval">> => <<"15s">>
        }
    }.

source_create_config(Overrides0) ->
    Overrides = emqx_utils_maps:binary_key_map(Overrides0),
    Conf0 = maps:merge(
        source_config_base(),
        #{
            <<"enable">> => true,
            <<"type">> => ?SOURCE_TYPE
        }
    ),
    maps:merge(
        Conf0,
        Overrides
    ).

source_update_config(Overrides0) ->
    Overrides = emqx_utils_maps:binary_key_map(Overrides0),
    maps:merge(
        source_config_base(),
        Overrides
    ).

get_common_values(Kind, FnName) ->
    case Kind of
        actions ->
            #{
                api_root_key => ?ACTIONS_ROOT,
                type => ?ACTION_TYPE,
                default_connector_name => ?ACTION_CONNECTOR_NAME,
                create_config_fn =>
                    fun(Overrides) ->
                        Name = maps:get(name, Overrides, FnName),
                        ConnectorName = maps:get(connector, Overrides, ?ACTION_CONNECTOR_NAME),
                        ?KAFKA_BRIDGE(Name, ConnectorName)
                    end,
                update_config_fn =>
                    fun(Overrides) ->
                        Name = maps:get(name, Overrides, FnName),
                        ConnectorName = maps:get(connector, Overrides, ?ACTION_CONNECTOR_NAME),
                        ?KAFKA_BRIDGE_UPDATE(Name, ConnectorName)
                    end,
                create_connector_config_fn =>
                    fun(Overrides) ->
                        ConnectorName = maps:get(name, Overrides, ?ACTION_CONNECTOR_NAME),
                        ?ACTIONS_CONNECTOR(ConnectorName)
                    end
            };
        sources ->
            #{
                api_root_key => ?SOURCES_ROOT,
                type => ?SOURCE_TYPE,
                default_connector_name => ?SOURCE_CONNECTOR_NAME,
                create_config_fn => fun(Overrides0) ->
                    Overrides =
                        case Overrides0 of
                            #{name := _} -> Overrides0;
                            _ -> Overrides0#{name => FnName}
                        end,
                    source_create_config(Overrides)
                end,
                update_config_fn => fun source_update_config/1,
                create_connector_config_fn => fun source_connector_create_config/1
            }
    end.

maybe_get_other_node(Config) ->
    %% In the single node test group, this simply returns the lone node.  Otherwise, it'll
    %% return a node that's not the primary one that receives API calls.
    PrimaryNode = ?config(node, Config),
    case proplists:get_value(cluster_nodes, Config, []) -- [PrimaryNode] of
        [] ->
            PrimaryNode;
        [OtherNode | _] ->
            OtherNode
    end.

%%------------------------------------------------------------------------------
%% Testcases
%%------------------------------------------------------------------------------

%% We have to pretend testing a kafka bridge since at this point that's the
%% only one that's implemented.

t_bridges_lifecycle(matrix) ->
    [
        [single, actions],
        [single, sources],
        [cluster, actions],
        [cluster, sources]
    ];
t_bridges_lifecycle(Config) ->
    [_SingleOrCluster, Kind | _] = group_path(Config),
    FnName = atom_to_binary(?FUNCTION_NAME),
    #{
        api_root_key := APIRootKey,
        type := Type,
        default_connector_name := DefaultConnectorName,
        create_config_fn := CreateConfigFn,
        update_config_fn := UpdateConfigFn,
        create_connector_config_fn := CreateConnectorConfigFn
    } = get_common_values(Kind, FnName),
    %% assert we there's no bridges at first
    {ok, 200, []} = request_json(get, uri([APIRootKey]), Config),

    {ok, 404, _} = request(get, uri([APIRootKey, "foo"]), Config),
    {ok, 404, _} = request(get, uri([APIRootKey, "kafka_producer:foo"]), Config),

    %% need a var for patterns below
    BridgeName = FnName,
    CreateRes = request_json(
        post,
        uri([APIRootKey]),
        CreateConfigFn(#{}),
        Config
    ),
    ?assertMatch(
        {ok, 201, #{
            <<"type">> := Type,
            <<"name">> := BridgeName,
            <<"enable">> := true,
            <<"status">> := <<"connected">>,
            <<"node_status">> := [_ | _],
            <<"connector">> := DefaultConnectorName,
            <<"parameters">> := #{},
            <<"resource_opts">> := _
        }},
        CreateRes,
        #{name => BridgeName, type => Type, connector => DefaultConnectorName}
    ),
    case Kind of
        actions ->
            ?assertMatch({ok, 201, #{<<"local_topic">> := _}}, CreateRes);
        sources ->
            ok
    end,

    %% list all bridges, assert bridge is in it
    ?assertMatch(
        {ok, 200, [
            #{
                <<"type">> := Type,
                <<"name">> := BridgeName,
                <<"enable">> := true,
                <<"status">> := _,
                <<"node_status">> := [_ | _]
            }
        ]},
        request_json(get, uri([APIRootKey]), Config)
    ),

    %% list all bridges, assert bridge is in it
    ?assertMatch(
        {ok, 200, [
            #{
                <<"type">> := Type,
                <<"name">> := BridgeName,
                <<"enable">> := true,
                <<"status">> := _,
                <<"node_status">> := [_ | _]
            }
        ]},
        request_json(get, uri([APIRootKey]), Config)
    ),

    %% get the bridge by id
    BridgeID = emqx_bridge_resource:bridge_id(Type, ?BRIDGE_NAME),
    ?assertMatch(
        {ok, 200, #{
            <<"type">> := Type,
            <<"name">> := BridgeName,
            <<"enable">> := true,
            <<"status">> := _,
            <<"node_status">> := [_ | _]
        }},
        request_json(get, uri([APIRootKey, BridgeID]), Config)
    ),

    ?assertMatch(
        {ok, 400, #{
            <<"code">> := <<"BAD_REQUEST">>,
            <<"message">> := _
        }},
        request_json(post, uri([APIRootKey, BridgeID, "brababbel"]), Config)
    ),

    %% update bridge config
    {ok, 201, _} = request(
        post,
        uri(["connectors"]),
        CreateConnectorConfigFn(#{name => <<"foobla">>}),
        Config
    ),
    ?assertMatch(
        {ok, 200, #{
            <<"type">> := Type,
            <<"name">> := BridgeName,
            <<"connector">> := <<"foobla">>,
            <<"enable">> := true,
            <<"status">> := _,
            <<"node_status">> := [_ | _]
        }},
        request_json(
            put,
            uri([APIRootKey, BridgeID]),
            UpdateConfigFn(#{connector => <<"foobla">>}),
            Config
        )
    ),

    %% update bridge with unknown connector name
    {ok, 400, #{
        <<"code">> := <<"BAD_REQUEST">>,
        <<"message">> := Message1
    }} =
        request_json(
            put,
            uri([APIRootKey, BridgeID]),
            UpdateConfigFn(#{connector => <<"does_not_exist">>}),
            Config
        ),
    ?assertMatch(
        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
        emqx_utils_json:decode(Message1)
    ),

    %% update bridge with connector of wrong type
    {ok, 201, _} =
        request(
            post,
            uri(["connectors"]),
            (?ACTIONS_CONNECTOR(<<"foobla2">>))#{
                <<"type">> => <<"azure_event_hub_producer">>,
                <<"authentication">> => #{
                    <<"username">> => <<"emqxuser">>,
                    <<"password">> => <<"topSecret">>,
                    <<"mechanism">> => <<"plain">>
                },
                <<"ssl">> => #{
                    <<"enable">> => true,
                    <<"server_name_indication">> => <<"auto">>,
                    <<"verify">> => <<"verify_none">>,
                    <<"versions">> => [<<"tlsv1.3">>, <<"tlsv1.2">>]
                }
            },
            Config
        ),
    {ok, 400, #{
        <<"code">> := <<"BAD_REQUEST">>,
        <<"message">> := Message2
    }} =
        request_json(
            put,
            uri([APIRootKey, BridgeID]),
            UpdateConfigFn(#{connector => <<"foobla2">>}),
            Config
        ),
    ?assertMatch(
        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
        emqx_utils_json:decode(Message2)
    ),

    %% delete the bridge
    {ok, 204, <<>>} = request(delete, uri([APIRootKey, BridgeID]), Config),
    {ok, 200, []} = request_json(get, uri([APIRootKey]), Config),

    %% try create with unknown connector name
    {ok, 400, #{
        <<"code">> := <<"BAD_REQUEST">>,
        <<"message">> := Message3
    }} =
        request_json(
            post,
            uri([APIRootKey]),
            CreateConfigFn(#{connector => <<"does_not_exist">>}),
            Config
        ),
    ?assertMatch(
        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
        emqx_utils_json:decode(Message3)
    ),

    %% try create bridge with connector of wrong type
    {ok, 400, #{
        <<"code">> := <<"BAD_REQUEST">>,
        <<"message">> := Message4
    }} =
        request_json(
            post,
            uri([APIRootKey]),
            CreateConfigFn(#{connector => <<"foobla2">>}),
            Config
        ),
    ?assertMatch(
        #{<<"reason">> := <<"connector_not_found_or_wrong_type">>},
        emqx_utils_json:decode(Message4)
    ),

    %% make sure nothing has been created above
    {ok, 200, []} = request_json(get, uri([APIRootKey]), Config),

    %% update a deleted bridge returns an error
    ?assertMatch(
        {ok, 404, #{
            <<"code">> := <<"NOT_FOUND">>,
            <<"message">> := _
        }},
        request_json(
            put,
            uri([APIRootKey, BridgeID]),
            UpdateConfigFn(#{}),
            Config
        )
    ),

    %% deleting a non-existing bridge should result in an error
    ?assertMatch(
        {ok, 404, #{
            <<"code">> := <<"NOT_FOUND">>,
            <<"message">> := _
        }},
        request_json(delete, uri([APIRootKey, BridgeID]), Config)
    ),

    %% try delete unknown bridge id
    ?assertMatch(
        {ok, 404, #{
            <<"code">> := <<"NOT_FOUND">>,
            <<"message">> := <<"Invalid bridge ID", _/binary>>
        }},
        request_json(delete, uri([APIRootKey, "foo"]), Config)
    ),

    %% Try create bridge with bad characters as name
    {ok, 400, _} = request(
        post, uri([APIRootKey]), CreateConfigFn(#{name => <<"隋达"/utf8>>}), Config
    ),
    {ok, 400, _} = request(post, uri([APIRootKey]), CreateConfigFn(#{name => <<"a.b">>}), Config),
    ok.

t_broken_bridge_config(matrix) ->
    [
        [single, actions]
    ];
t_broken_bridge_config(Config) ->
    emqx_cth_suite:stop_apps([emqx_bridge]),
    BridgeName = ?BRIDGE_NAME,
    StartOps =
        #{
            config =>
                "actions {\n"
                "  "
                ?ACTION_TYPE_STR
                " {\n"
                "    " ++ binary_to_list(BridgeName) ++
                " {\n"
                "      connector = does_not_exist\n"
                "      enable = true\n"
                "      kafka {\n"
                "        topic = test-topic-one-partition\n"
                "      }\n"
                "      local_topic = \"mqtt/local/topic\"\n"
                "      resource_opts {health_check_interval = 32s}\n"
                "    }\n"
                "  }\n"
                "}\n"
                "\n",
            schema_mod => emqx_bridge_v2_schema
        },
    emqx_cth_suite:start_app(emqx_bridge, StartOps),

    ?assertMatch(
        {ok, 200, [
            #{
                <<"name">> := BridgeName,
                <<"type">> := ?ACTION_TYPE,
                <<"connector">> := <<"does_not_exist">>,
                <<"status">> := <<"disconnected">>,
                <<"error">> := <<"Not installed">>
            }
        ]},
        request_json(get, uri([?ACTIONS_ROOT]), Config)
    ),

    BridgeID = emqx_bridge_resource:bridge_id(?ACTION_TYPE, ?BRIDGE_NAME),
    ?assertEqual(
        {ok, 204, <<>>},
        request(delete, uri([?ACTIONS_ROOT, BridgeID]), Config)
    ),

    ?assertEqual(
        {ok, 200, []},
        request_json(get, uri([?ACTIONS_ROOT]), Config)
    ),

    ok.

t_fix_broken_bridge_config(matrix) ->
    [
        [single, actions]
    ];
t_fix_broken_bridge_config(Config) ->
    emqx_cth_suite:stop_apps([emqx_bridge]),
    BridgeName = ?BRIDGE_NAME,
    StartOps =
        #{
            config =>
                "actions {\n"
                "  "
                ?ACTION_TYPE_STR
                " {\n"
                "    " ++ binary_to_list(BridgeName) ++
                " {\n"
                "      connector = does_not_exist\n"
                "      enable = true\n"
                "      kafka {\n"
                "        topic = test-topic-one-partition\n"
                "      }\n"
                "      local_topic = \"mqtt/local/topic\"\n"
                "      resource_opts {health_check_interval = 32s}\n"
                "    }\n"
                "  }\n"
                "}\n"
                "\n",
            schema_mod => emqx_bridge_v2_schema
        },
    emqx_cth_suite:start_app(emqx_bridge, StartOps),

    ?assertMatch(
        {ok, 200, [
            #{
                <<"name">> := BridgeName,
                <<"type">> := ?ACTION_TYPE,
                <<"connector">> := <<"does_not_exist">>,
                <<"status">> := <<"disconnected">>,
                <<"error">> := <<"Not installed">>
            }
        ]},
        request_json(get, uri([?ACTIONS_ROOT]), Config)
    ),

    BridgeID = emqx_bridge_resource:bridge_id(?ACTION_TYPE, ?BRIDGE_NAME),
    request_json(
        put,
        uri([?ACTIONS_ROOT, BridgeID]),
        ?KAFKA_BRIDGE_UPDATE(?BRIDGE_NAME, ?ACTION_CONNECTOR_NAME),
        Config
    ),

    ?assertMatch(
        {ok, 200, #{
            <<"connector">> := ?ACTION_CONNECTOR_NAME,
            <<"status">> := <<"connected">>
        }},
        request_json(get, uri([?ACTIONS_ROOT, BridgeID]), Config)
    ),

    ok.

t_start_bridge_unknown_node(matrix) ->
    [
        [single, actions],
        [cluster, actions]
    ];
t_start_bridge_unknown_node(Config) ->
    {ok, 404, _} =
        request(
            post,
            uri(["nodes", "thisbetterbenotanatomyet", ?ACTIONS_ROOT, "kafka_producer:foo", start]),
            Config
        ),
    {ok, 404, _} =
        request(
            post,
            uri(["nodes", "undefined", ?ACTIONS_ROOT, "kafka_producer:foo", start]),
            Config
        ).

t_start_bridge_node(matrix) ->
    [
        [single, actions],
        [single, sources],
        [cluster, actions],
        [cluster, sources]
    ];
t_start_bridge_node(Config) ->
    do_start_bridge(node, Config).

t_start_bridge_cluster(matrix) ->
    [
        [single, actions],
        [single, sources],
        [cluster, actions],
        [cluster, sources]
    ];
t_start_bridge_cluster(Config) ->
    do_start_bridge(cluster, Config).

do_start_bridge(TestType, Config) ->
    [_SingleOrCluster, Kind | _] = group_path(Config),
    Name = atom_to_binary(TestType),
    #{
        api_root_key := APIRootKey,
        type := Type,
        default_connector_name := DefaultConnectorName,
        create_config_fn := CreateConfigFn
    } = get_common_values(Kind, Name),
    %% assert we there's no bridges at first
    {ok, 200, []} = request_json(get, uri([APIRootKey]), Config),

    ?assertMatch(
        {ok, 201, #{
            <<"type">> := Type,
            <<"name">> := Name,
            <<"enable">> := true,
            <<"status">> := <<"connected">>,
            <<"node_status">> := [_ | _]
        }},
        request_json(
            post,
            uri([APIRootKey]),
            CreateConfigFn(#{name => Name}),
            Config
        )
    ),

    BridgeID = emqx_bridge_resource:bridge_id(Type, Name),

    %% start again
    {ok, 204, <<>>} = request(post, {operation, TestType, start, BridgeID}, Config),
    ?assertMatch(
        {ok, 200, #{<<"status">> := <<"connected">>}},
        request_json(get, uri([APIRootKey, BridgeID]), Config)
    ),
    %% start a started bridge
    {ok, 204, <<>>} = request(post, {operation, TestType, start, BridgeID}, Config),
    ?assertMatch(
        {ok, 200, #{<<"status">> := <<"connected">>}},
        request_json(get, uri([APIRootKey, BridgeID]), Config)
    ),

    {ok, 400, _} = request(post, {operation, TestType, invalidop, BridgeID}, Config),

    %% Make start bridge fail
    expect_on_all_nodes(
        ?CONNECTOR_IMPL,
        on_add_channel,
        fun(_, _, _ResId, _Channel) -> {error, <<"my_error">>} end,
        Config
    ),

    connector_operation(Config, Type, DefaultConnectorName, stop),
    connector_operation(Config, Type, DefaultConnectorName, start),

    {ok, 400, _} = request(post, {operation, TestType, start, BridgeID}, Config),

    %% Make start bridge succeed

    expect_on_all_nodes(
        ?CONNECTOR_IMPL,
        on_add_channel,
        fun(_, _, _ResId, _Channel) -> {ok, connector_state} end,
        Config
    ),

    %% try to start again
    {ok, 204, <<>>} = request(post, {operation, TestType, start, BridgeID}, Config),

    %% delete the bridge
    {ok, 204, <<>>} = request(delete, uri([APIRootKey, BridgeID]), Config),
    {ok, 200, []} = request_json(get, uri([APIRootKey]), Config),

    %% Fail parse-id check
    {ok, 404, _} = request(post, {operation, TestType, start, <<"wreckbook_fugazi">>}, Config),
    %% Looks ok but doesn't exist
    {ok, 404, _} = request(post, {operation, TestType, start, <<"webhook:cptn_hook">>}, Config),
    ok.

%% t_start_stop_inconsistent_bridge_node(Config) ->
%%     start_stop_inconsistent_bridge(node, Config).

%% t_start_stop_inconsistent_bridge_cluster(Config) ->
%%     start_stop_inconsistent_bridge(cluster, Config).

%% start_stop_inconsistent_bridge(Type, Config) ->
%%     Node = ?config(node, Config),

%%     erpc:call(Node, fun() ->
%%         meck:new(emqx_bridge_resource, [passthrough, no_link]),
%%         meck:expect(
%%             emqx_bridge_resource,
%%             stop,
%%             fun
%%                 (_, <<"bridge_not_found">>) -> {error, not_found};
%%                 (BridgeType, Name) -> meck:passthrough([BridgeType, Name])
%%             end
%%         )
%%     end),

%%     emqx_common_test_helpers:on_exit(fun() ->
%%         erpc:call(Node, fun() ->
%%             meck:unload([emqx_bridge_resource])
%%         end)
%%     end),

%%     {ok, 201, _Bridge} = request(
%%         post,
%%         uri([?ROOT]),
%%         ?KAFKA_BRIDGE(<<"bridge_not_found">>),
%%         Config
%%     ),
%%     {ok, 503, _} = request(
%%         post, {operation, Type, stop, <<"kafka:bridge_not_found">>}, Config
%%     ).

%% [TODO] This is a mess, need to clarify what the actual behavior needs to be
%% like.
%% t_enable_disable_bridges(Config) ->
%%     %% assert we there's no bridges at first
%%     {ok, 200, []} = request_json(get, uri([?ROOT]), Config),

%%     Name = ?BRIDGE_NAME,
%%     ?assertMatch(
%%         {ok, 201, #{
%%             <<"type">> := ?BRIDGE_TYPE,
%%             <<"name">> := Name,
%%             <<"enable">> := true,
%%             <<"status">> := <<"connected">>,
%%             <<"node_status">> := [_ | _]
%%         }},
%%         request_json(
%%             post,
%%             uri([?ROOT]),
%%             ?KAFKA_BRIDGE(Name),
%%             Config
%%         )
%%     ),
%%     BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
%%     %% disable it
%%     meck:expect(?CONNECTOR_IMPL, on_get_channel_status, 3, connecting),
%%     {ok, 204, <<>>} = request(put, enable_path(false, BridgeID), Config),
%%     ?assertMatch(
%%         {ok, 200, #{<<"status">> := <<"stopped">>}},
%%         request_json(get, uri([?ROOT, BridgeID]), Config)
%%     ),
%%     %% enable again
%%     meck:expect(?CONNECTOR_IMPL, on_get_channel_status, 3, connected),
%%     {ok, 204, <<>>} = request(put, enable_path(true, BridgeID), Config),
%%     ?assertMatch(
%%         {ok, 200, #{<<"status">> := <<"connected">>}},
%%         request_json(get, uri([?ROOT, BridgeID]), Config)
%%     ),
%%     %% enable an already started bridge
%%     {ok, 204, <<>>} = request(put, enable_path(true, BridgeID), Config),
%%     ?assertMatch(
%%         {ok, 200, #{<<"status">> := <<"connected">>}},
%%         request_json(get, uri([?ROOT, BridgeID]), Config)
%%     ),
%%     %% disable it again
%%     {ok, 204, <<>>} = request(put, enable_path(false, BridgeID), Config),

%%     %% bad param
%%     {ok, 404, _} = request(put, enable_path(foo, BridgeID), Config),
%%     {ok, 404, _} = request(put, enable_path(true, "foo"), Config),
%%     {ok, 404, _} = request(put, enable_path(true, "webhook:foo"), Config),

%%     {ok, 400, Res} = request(post, {operation, node, start, BridgeID}, <<>>, fun json/1, Config),
%%     ?assertEqual(
%%         #{
%%             <<"code">> => <<"BAD_REQUEST">>,
%%             <<"message">> => <<"Forbidden operation, bridge not enabled">>
%%         },
%%         Res
%%     ),
%%     {ok, 400, Res} = request(
%%         post, {operation, cluster, start, BridgeID}, <<>>, fun json/1, Config
%%     ),

%%     %% enable a stopped bridge
%%     {ok, 204, <<>>} = request(put, enable_path(true, BridgeID), Config),
%%     ?assertMatch(
%%         {ok, 200, #{<<"status">> := <<"connected">>}},
%%         request_json(get, uri([?ROOT, BridgeID]), Config)
%%     ),
%%     %% delete the bridge
%%     {ok, 204, <<>>} = request(delete, uri([?ROOT, BridgeID]), Config),
%%     {ok, 200, []} = request_json(get, uri([?ROOT]), Config).

t_bridges_probe(matrix) ->
    [
        [single, actions]
    ];
t_bridges_probe(Config) ->
    {ok, 204, <<>>} = request(
        post,
        uri(["actions_probe"]),
        ?KAFKA_BRIDGE(?BRIDGE_NAME),
        Config
    ),

    %% second time with same name is ok since no real bridge created
    {ok, 204, <<>>} = request(
        post,
        uri(["actions_probe"]),
        ?KAFKA_BRIDGE(?BRIDGE_NAME),
        Config
    ),

    meck:expect(?CONNECTOR_IMPL, on_start, 2, {error, on_start_error}),

    ?assertMatch(
        {ok, 400, #{
            <<"code">> := <<"TEST_FAILED">>,
            <<"message">> := _
        }},
        request_json(
            post,
            uri(["actions_probe"]),
            ?KAFKA_BRIDGE(<<"broken_bridge">>, <<"brokenhost:1234">>),
            Config
        )
    ),

    meck:expect(?CONNECTOR_IMPL, on_start, 2, {ok, bridge_state}),

    ?assertMatch(
        {ok, 400, #{<<"code">> := <<"BAD_REQUEST">>}},
        request_json(
            post,
            uri(["actions_probe"]),
            ?RESOURCE(<<"broken_bridge">>, <<"unknown_type">>),
            Config
        )
    ),
    ok.

t_cascade_delete_actions(matrix) ->
    [
        [single, actions],
        [cluster, actions]
    ];
t_cascade_delete_actions(Config) ->
    %% assert we there's no bridges at first
    {ok, 200, []} = request_json(get, uri([?ACTIONS_ROOT]), Config),
    %% then we add a a bridge, using POST
    %% POST /actions/ will create a bridge
    BridgeID = emqx_bridge_resource:bridge_id(?ACTION_TYPE, ?BRIDGE_NAME),
    {ok, 201, _} = request(
        post,
        uri([?ACTIONS_ROOT]),
        ?KAFKA_BRIDGE(?BRIDGE_NAME),
        Config
    ),
    {ok, 201, #{<<"id">> := RuleId}} = request_json(
        post,
        uri(["rules"]),
        #{
            <<"name">> => <<"t_http_crud_apis">>,
            <<"enable">> => true,
            <<"actions">> => [BridgeID],
            <<"sql">> => <<"SELECT * from \"t\"">>
        },
        Config
    ),
    %% delete the bridge will also delete the actions from the rules
    {ok, 204, _} = request(
        delete,
        uri([?ACTIONS_ROOT, BridgeID]) ++ "?also_delete_dep_actions=true",
        Config
    ),
    {ok, 200, []} = request_json(get, uri([?ACTIONS_ROOT]), Config),
    ?assertMatch(
        {ok, 200, #{<<"actions">> := []}},
        request_json(get, uri(["rules", RuleId]), Config)
    ),
    {ok, 204, <<>>} = request(delete, uri(["rules", RuleId]), Config),

    {ok, 201, _} = request(
        post,
        uri([?ACTIONS_ROOT]),
        ?KAFKA_BRIDGE(?BRIDGE_NAME),
        Config
    ),
    {ok, 201, _} = request(
        post,
        uri(["rules"]),
        #{
            <<"name">> => <<"t_http_crud_apis">>,
            <<"enable">> => true,
            <<"actions">> => [BridgeID],
            <<"sql">> => <<"SELECT * from \"t\"">>
        },
        Config
    ),
    {ok, 400, Body} = request(
        delete,
        uri([?ACTIONS_ROOT, BridgeID]),
        Config
    ),
    ?assertMatch(#{<<"rules">> := [_ | _]}, emqx_utils_json:decode(Body, [return_maps])),
    {ok, 200, [_]} = request_json(get, uri([?ACTIONS_ROOT]), Config),
    %% Cleanup
    {ok, 204, _} = request(
        delete,
        uri([?ACTIONS_ROOT, BridgeID]) ++ "?also_delete_dep_actions=true",
        Config
    ),
    {ok, 200, []} = request_json(get, uri([?ACTIONS_ROOT]), Config).

t_action_types(matrix) ->
    [
        [single, actions],
        [cluster, actions]
    ];
t_action_types(Config) ->
    Res = request_json(get, uri(["action_types"]), Config),
    ?assertMatch({ok, 200, _}, Res),
    {ok, 200, Types} = Res,
    ?assert(is_list(Types), #{types => Types}),
    ?assert(lists:all(fun is_binary/1, Types), #{types => Types}),
    ok.

t_bad_name(matrix) ->
    [
        [single, actions],
        [single, sources],
        [cluster, actions],
        [cluster, sources]
    ];
t_bad_name(Config) ->
    [_SingleOrCluster, Kind | _] = group_path(Config),
    Name = <<"_bad_name">>,
    #{
        api_root_key := APIRootKey,
        create_config_fn := CreateConfigFn
    } = get_common_values(Kind, Name),
    Res = request_json(
        post,
        uri([APIRootKey]),
        CreateConfigFn(#{}),
        Config
    ),
    ?assertMatch({ok, 400, #{<<"message">> := _}}, Res),
    {ok, 400, #{<<"message">> := Msg0}} = Res,
    Msg = emqx_utils_json:decode(Msg0, [return_maps]),
    ?assertMatch(
        #{
            <<"kind">> := <<"validation_error">>,
            <<"reason">> := <<"Invalid name format.", _/binary>>
        },
        Msg
    ),
    ok.

t_metrics(matrix) ->
    [
        [single, actions],
        [cluster, actions]
    ];
t_metrics(Config) ->
    {ok, 200, []} = request_json(get, uri([?ACTIONS_ROOT]), Config),

    ActionName = ?BRIDGE_NAME,
    ?assertMatch(
        {ok, 201, _},
        request_json(
            post,
            uri([?ACTIONS_ROOT]),
            ?KAFKA_BRIDGE(?BRIDGE_NAME),
            Config
        )
    ),

    ActionID = emqx_bridge_resource:bridge_id(?ACTION_TYPE, ActionName),

    ?assertMatch(
        {ok, 200, #{
            <<"metrics">> := #{<<"matched">> := 0},
            <<"node_metrics">> := [#{<<"metrics">> := #{<<"matched">> := 0}} | _]
        }},
        request_json(get, uri([?ACTIONS_ROOT, ActionID, "metrics"]), Config)
    ),

    {ok, 200, Bridge} = request_json(get, uri([?ACTIONS_ROOT, ActionID]), Config),
    ?assertNot(maps:is_key(<<"metrics">>, Bridge)),
    ?assertNot(maps:is_key(<<"node_metrics">>, Bridge)),

    Body = <<"my msg">>,
    _ = publish_message(?MQTT_LOCAL_TOPIC, Body, Config),

    %% check for non-empty bridge metrics
    ?retry(
        _Sleep0 = 200,
        _Retries0 = 20,
        ?assertMatch(
            {ok, 200, #{
                <<"metrics">> := #{<<"matched">> := 1},
                <<"node_metrics">> := [#{<<"metrics">> := #{<<"matched">> := 1}} | _]
            }},
            request_json(get, uri([?ACTIONS_ROOT, ActionID, "metrics"]), Config)
        )
    ),

    %% check for absence of metrics when listing all bridges
    {ok, 200, Bridges} = request_json(get, uri([?ACTIONS_ROOT]), Config),
    ?assertNotMatch(
        [
            #{
                <<"metrics">> := #{},
                <<"node_metrics">> := [_ | _]
            }
        ],
        Bridges
    ),
    ok.

t_reset_metrics(matrix) ->
    [
        [single, actions],
        [cluster, actions]
    ];
t_reset_metrics(Config) ->
    %% assert there's no bridges at first
    {ok, 200, []} = request_json(get, uri([?ACTIONS_ROOT]), Config),

    ActionName = ?BRIDGE_NAME,
    ?assertMatch(
        {ok, 201, _},
        request_json(
            post,
            uri([?ACTIONS_ROOT]),
            ?KAFKA_BRIDGE(?BRIDGE_NAME),
            Config
        )
    ),
    ActionID = emqx_bridge_resource:bridge_id(?ACTION_TYPE, ActionName),

    Body = <<"my msg">>,
    OtherNode = maybe_get_other_node(Config),
    _ = publish_message(?MQTT_LOCAL_TOPIC, Body, OtherNode, Config),
    ?retry(
        _Sleep0 = 200,
        _Retries0 = 20,
        ?assertMatch(
            {ok, 200, #{
                <<"metrics">> := #{<<"matched">> := 1},
                <<"node_metrics">> := [#{<<"metrics">> := #{}} | _]
            }},
            request_json(get, uri([?ACTIONS_ROOT, ActionID, "metrics"]), Config)
        )
    ),

    {ok, 204, <<>>} = request(put, uri([?ACTIONS_ROOT, ActionID, "metrics", "reset"]), Config),

    Res = ?retry(
        _Sleep0 = 200,
        _Retries0 = 20,
        begin
            Res0 = request_json(get, uri([?ACTIONS_ROOT, ActionID, "metrics"]), Config),
            ?assertMatch(
                {ok, 200, #{
                    <<"metrics">> := #{<<"matched">> := 0},
                    <<"node_metrics">> := [#{<<"metrics">> := #{}} | _]
                }},
                Res0
            ),
            Res0
        end
    ),
    {ok, 200, #{<<"node_metrics">> := NodeMetrics}} = Res,
    ?assert(
        lists:all(
            fun(#{<<"metrics">> := #{<<"matched">> := Matched}}) ->
                Matched == 0
            end,
            NodeMetrics
        ),
        #{node_metrics => NodeMetrics}
    ),

    ok.

t_cluster_later_join_metrics(matrix) ->
    [
        [cluster_later_join, actions]
    ];
t_cluster_later_join_metrics(Config) ->
    [PrimaryNode, OtherNode | _] = ?config(cluster_nodes, Config),
    Name = ?BRIDGE_NAME,
    ActionParams = ?KAFKA_BRIDGE(Name),
    ActionID = emqx_bridge_resource:bridge_id(?ACTION_TYPE, Name),
    ?check_trace(
        begin
            %% Create a bridge on only one of the nodes.
            ?assertMatch(
                {ok, 201, _}, request_json(post, uri([?ACTIONS_ROOT]), ActionParams, Config)
            ),
            %% Pre-condition.
            ?assertMatch(
                {ok, 200, #{
                    <<"metrics">> := #{<<"success">> := _},
                    <<"node_metrics">> := [#{<<"metrics">> := #{}} | _]
                }},
                request_json(get, uri([?ACTIONS_ROOT, ActionID, "metrics"]), Config)
            ),
            %% Now join the other node join with the api node.
            ok = erpc:call(OtherNode, ekka, join, [PrimaryNode]),
            %% Check metrics; shouldn't crash even if the bridge is not
            %% ready on the node that just joined the cluster.
            ?assertMatch(
                {ok, 200, #{
                    <<"metrics">> := #{<<"success">> := _},
                    <<"node_metrics">> := [#{<<"metrics">> := #{}} | _]
                }},
                request_json(get, uri([?ACTIONS_ROOT, ActionID, "metrics"]), Config)
            ),
            ok
        end,
        []
    ),
    ok.

t_raw_config_response_defaults(matrix) ->
    [
        [single, actions],
        [single, sources],
        [cluster, actions],
        [cluster, sources]
    ];
t_raw_config_response_defaults(Config) ->
    [_SingleOrCluster, Kind | _] = group_path(Config),
    Name = atom_to_binary(?FUNCTION_NAME),
    #{
        api_root_key := APIRootKey,
        create_config_fn := CreateConfigFn
    } = get_common_values(Kind, Name),
    Params = maps:remove(<<"enable">>, CreateConfigFn(#{})),
    ?assertMatch(
        {ok, 201, #{<<"enable">> := true}},
        request_json(
            post,
            uri([APIRootKey]),
            Params,
            Config
        )
    ),
    ok.

t_older_version_nodes_in_cluster(matrix) ->
    [
        [cluster, actions],
        [cluster, sources]
    ];
t_older_version_nodes_in_cluster(Config) ->
    [_, Kind | _] = group_path(Config),
    PrimaryNode = ?config(node, Config),
    OtherNode = maybe_get_other_node(Config),
    ?assertNotEqual(OtherNode, PrimaryNode),
    Name = atom_to_binary(?FUNCTION_NAME),
    ?check_trace(
        begin
            #{api_root_key := APIRootKey} = get_common_values(Kind, Name),
            erpc:call(PrimaryNode, fun() ->
                meck:new(emqx_bpapi, [no_history, passthrough, no_link]),
                meck:expect(emqx_bpapi, supported_version, fun(N, Api) ->
                    case N =:= OtherNode of
                        true -> 1;
                        false -> meck:passthrough([N, Api])
                    end
                end)
            end),
            erpc:call(OtherNode, fun() ->
                meck:new(emqx_bridge_v2, [no_history, passthrough, no_link]),
                meck:expect(emqx_bridge_v2, list, fun(_ConfRootKey) ->
                    error(should_not_be_called)
                end)
            end),
            ?assertMatch(
                {ok, 200, _},
                request_json(
                    get,
                    uri([APIRootKey]),
                    Config
                )
            ),
            ok
        end,
        []
    ),

    ok.