997 lines
33 KiB
Erlang
997 lines
33 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2020-2023 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_api_SUITE).
|
|
|
|
-compile(nowarn_export_all).
|
|
-compile(export_all).
|
|
|
|
-import(emqx_mgmt_api_test_util, [request/3, uri/1]).
|
|
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
-include_lib("common_test/include/ct.hrl").
|
|
-define(CONF_DEFAULT, <<"bridges: {}">>).
|
|
-define(BRIDGE_TYPE, <<"webhook">>).
|
|
-define(BRIDGE_NAME, (atom_to_binary(?FUNCTION_NAME))).
|
|
-define(URL(PORT, PATH),
|
|
list_to_binary(
|
|
io_lib:format(
|
|
"http://localhost:~s/~s",
|
|
[integer_to_list(PORT), PATH]
|
|
)
|
|
)
|
|
).
|
|
-define(BRIDGE(NAME, TYPE), #{
|
|
<<"ssl">> => #{<<"enable">> => false},
|
|
<<"type">> => TYPE,
|
|
<<"name">> => NAME
|
|
}).
|
|
|
|
-define(BRIDGE_TYPE_MQTT, <<"mqtt">>).
|
|
-define(MQTT_BRIDGE(SERVER, NAME), ?BRIDGE(NAME, ?BRIDGE_TYPE_MQTT)#{
|
|
<<"server">> => SERVER,
|
|
<<"username">> => <<"user1">>,
|
|
<<"password">> => <<"">>,
|
|
<<"proto_ver">> => <<"v5">>
|
|
}).
|
|
-define(MQTT_BRIDGE(SERVER), ?MQTT_BRIDGE(SERVER, <<"mqtt_egress_test_bridge">>)).
|
|
|
|
-define(HTTP_BRIDGE(URL, TYPE, NAME), ?BRIDGE(NAME, TYPE)#{
|
|
<<"url">> => URL,
|
|
<<"local_topic">> => <<"emqx_webhook/#">>,
|
|
<<"method">> => <<"post">>,
|
|
<<"body">> => <<"${payload}">>,
|
|
<<"headers">> => #{
|
|
<<"content-type">> => <<"application/json">>
|
|
}
|
|
}).
|
|
|
|
all() ->
|
|
emqx_common_test_helpers:all(?MODULE).
|
|
|
|
groups() ->
|
|
[].
|
|
|
|
suite() ->
|
|
[{timetrap, {seconds, 60}}].
|
|
|
|
init_per_suite(Config) ->
|
|
_ = application:load(emqx_conf),
|
|
%% some testcases (may from other app) already get emqx_connector started
|
|
_ = application:stop(emqx_resource),
|
|
_ = application:stop(emqx_connector),
|
|
ok = emqx_mgmt_api_test_util:init_suite(
|
|
[emqx_rule_engine, emqx_bridge, emqx_authn]
|
|
),
|
|
ok = emqx_common_test_helpers:load_config(
|
|
emqx_rule_engine_schema,
|
|
<<"rule_engine {rules {}}">>
|
|
),
|
|
ok = emqx_common_test_helpers:load_config(emqx_bridge_schema, ?CONF_DEFAULT),
|
|
Config.
|
|
|
|
end_per_suite(_Config) ->
|
|
emqx_mgmt_api_test_util:end_suite([emqx_rule_engine, emqx_bridge, emqx_authn]),
|
|
mria:clear_table(emqx_authn_mnesia),
|
|
ok.
|
|
|
|
init_per_testcase(t_broken_bpapi_vsn, Config) ->
|
|
meck:new(emqx_bpapi, [passthrough]),
|
|
meck:expect(emqx_bpapi, supported_version, 1, -1),
|
|
meck:expect(emqx_bpapi, supported_version, 2, -1),
|
|
init_per_testcase(commong, Config);
|
|
init_per_testcase(t_old_bpapi_vsn, Config) ->
|
|
meck:new(emqx_bpapi, [passthrough]),
|
|
meck:expect(emqx_bpapi, supported_version, 1, 1),
|
|
meck:expect(emqx_bpapi, supported_version, 2, 1),
|
|
init_per_testcase(common, Config);
|
|
init_per_testcase(_, Config) ->
|
|
{ok, _} = emqx_cluster_rpc:start_link(node(), emqx_cluster_rpc, 1000),
|
|
{Port, Sock, Acceptor} = start_http_server(fun handle_fun_200_ok/2),
|
|
[{port, Port}, {sock, Sock}, {acceptor, Acceptor} | Config].
|
|
|
|
end_per_testcase(t_broken_bpapi_vsn, Config) ->
|
|
meck:unload([emqx_bpapi]),
|
|
end_per_testcase(common, Config);
|
|
end_per_testcase(t_old_bpapi_vsn, Config) ->
|
|
meck:unload([emqx_bpapi]),
|
|
end_per_testcase(common, Config);
|
|
end_per_testcase(_, Config) ->
|
|
Sock = ?config(sock, Config),
|
|
Acceptor = ?config(acceptor, Config),
|
|
stop_http_server(Sock, Acceptor),
|
|
clear_resources(),
|
|
ok.
|
|
|
|
clear_resources() ->
|
|
lists:foreach(
|
|
fun(#{type := Type, name := Name}) ->
|
|
{ok, _} = emqx_bridge:remove(Type, Name)
|
|
end,
|
|
emqx_bridge:list()
|
|
).
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% HTTP server for testing
|
|
%%------------------------------------------------------------------------------
|
|
start_http_server(HandleFun) ->
|
|
process_flag(trap_exit, true),
|
|
Parent = self(),
|
|
{Port, Sock} = listen_on_random_port(),
|
|
Acceptor = spawn_link(fun() ->
|
|
accept_loop(Sock, HandleFun, Parent)
|
|
end),
|
|
timer:sleep(100),
|
|
{Port, Sock, Acceptor}.
|
|
|
|
stop_http_server(Sock, Acceptor) ->
|
|
exit(Acceptor, kill),
|
|
gen_tcp:close(Sock).
|
|
|
|
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.
|
|
|
|
accept_loop(Sock, HandleFun, Parent) ->
|
|
process_flag(trap_exit, true),
|
|
{ok, Conn} = gen_tcp:accept(Sock),
|
|
Handler = spawn_link(fun() -> HandleFun(Conn, Parent) end),
|
|
gen_tcp:controlling_process(Conn, Handler),
|
|
accept_loop(Sock, HandleFun, Parent).
|
|
|
|
make_response(CodeStr, Str) ->
|
|
B = iolist_to_binary(Str),
|
|
iolist_to_binary(
|
|
io_lib:fwrite(
|
|
"HTTP/1.0 ~s\r\nContent-Type: text/html\r\nContent-Length: ~p\r\n\r\n~s",
|
|
[CodeStr, size(B), B]
|
|
)
|
|
).
|
|
|
|
handle_fun_200_ok(Conn, Parent) ->
|
|
case gen_tcp:recv(Conn, 0) of
|
|
{ok, ReqStr} ->
|
|
ct:pal("the http handler got request: ~p", [ReqStr]),
|
|
Req = parse_http_request(ReqStr),
|
|
Parent ! {http_server, received, Req},
|
|
gen_tcp:send(Conn, make_response("200 OK", "Request OK")),
|
|
handle_fun_200_ok(Conn, Parent);
|
|
{error, Reason} ->
|
|
ct:pal("the http handler recv error: ~p", [Reason]),
|
|
timer:sleep(100),
|
|
gen_tcp:close(Conn)
|
|
end.
|
|
|
|
parse_http_request(ReqStr0) ->
|
|
[Method, ReqStr1] = string:split(ReqStr0, " ", leading),
|
|
[Path, ReqStr2] = string:split(ReqStr1, " ", leading),
|
|
[_ProtoVsn, ReqStr3] = string:split(ReqStr2, "\r\n", leading),
|
|
[_HeaderStr, Body] = string:split(ReqStr3, "\r\n\r\n", leading),
|
|
#{method => Method, path => Path, body => Body}.
|
|
|
|
%%------------------------------------------------------------------------------
|
|
%% Testcases
|
|
%%------------------------------------------------------------------------------
|
|
|
|
t_http_crud_apis(Config) ->
|
|
Port = ?config(port, Config),
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
{ok, 404, _} = request(get, uri(["bridges", "foo"]), []),
|
|
{ok, 404, _} = request(get, uri(["bridges", "webhook:foo"]), []),
|
|
|
|
%% then we add a webhook bridge, using POST
|
|
%% POST /bridges/ will create a bridge
|
|
URL1 = ?URL(Port, "path1"),
|
|
Name = ?BRIDGE_NAME,
|
|
{ok, 201, Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
|
|
%ct:pal("---bridge: ~p", [Bridge]),
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := _,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL1
|
|
} = emqx_json:decode(Bridge, [return_maps]),
|
|
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
%% send an message to emqx and the message should be forwarded to the HTTP server
|
|
Body = <<"my msg">>,
|
|
emqx:publish(emqx_message:make(<<"emqx_webhook/1">>, Body)),
|
|
?assert(
|
|
receive
|
|
{http_server, received, #{
|
|
method := <<"POST">>,
|
|
path := <<"/path1">>,
|
|
body := Body
|
|
}} ->
|
|
true;
|
|
Msg ->
|
|
ct:pal("error: http got unexpected request: ~p", [Msg]),
|
|
false
|
|
after 100 ->
|
|
false
|
|
end
|
|
),
|
|
%% update the request-path of the bridge
|
|
URL2 = ?URL(Port, "path2"),
|
|
{ok, 200, Bridge2} = request(
|
|
put,
|
|
uri(["bridges", BridgeID]),
|
|
?HTTP_BRIDGE(URL2, ?BRIDGE_TYPE, Name)
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := _,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL2
|
|
},
|
|
emqx_json:decode(Bridge2, [return_maps])
|
|
),
|
|
|
|
%% list all bridges again, assert Bridge2 is in it
|
|
{ok, 200, Bridge2Str} = request(get, uri(["bridges"]), []),
|
|
?assertMatch(
|
|
[
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := _,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL2
|
|
}
|
|
],
|
|
emqx_json:decode(Bridge2Str, [return_maps])
|
|
),
|
|
|
|
%% get the bridge by id
|
|
{ok, 200, Bridge3Str} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := _,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL2
|
|
},
|
|
emqx_json:decode(Bridge3Str, [return_maps])
|
|
),
|
|
|
|
%% send an message to emqx again, check the path has been changed
|
|
emqx:publish(emqx_message:make(<<"emqx_webhook/1">>, Body)),
|
|
?assert(
|
|
receive
|
|
{http_server, received, #{path := <<"/path2">>}} ->
|
|
true;
|
|
Msg2 ->
|
|
ct:pal("error: http got unexpected request: ~p", [Msg2]),
|
|
false
|
|
after 100 ->
|
|
false
|
|
end
|
|
),
|
|
|
|
%% delete the bridge
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
%% update a deleted bridge returns an error
|
|
{ok, 404, ErrMsg2} = request(
|
|
put,
|
|
uri(["bridges", BridgeID]),
|
|
?HTTP_BRIDGE(URL2, ?BRIDGE_TYPE, Name)
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"NOT_FOUND">>,
|
|
<<"message">> := _
|
|
},
|
|
emqx_json:decode(ErrMsg2, [return_maps])
|
|
),
|
|
|
|
%% try delete bad bridge id
|
|
{ok, 404, BadId} = request(delete, uri(["bridges", "foo"]), []),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"NOT_FOUND">>,
|
|
<<"message">> := <<"Invalid bridge ID", _/binary>>
|
|
},
|
|
emqx_json:decode(BadId, [return_maps])
|
|
),
|
|
|
|
%% Deleting a non-existing bridge should result in an error
|
|
{ok, 404, ErrMsg3} = request(delete, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"NOT_FOUND">>,
|
|
<<"message">> := _
|
|
},
|
|
emqx_json:decode(ErrMsg3, [return_maps])
|
|
),
|
|
ok.
|
|
|
|
t_http_bridges_local_topic(Config) ->
|
|
Port = ?config(port, Config),
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
%% then we add a webhook bridge, using POST
|
|
%% POST /bridges/ will create a bridge
|
|
URL1 = ?URL(Port, "path1"),
|
|
Name1 = <<"t_http_bridges_with_local_topic1">>,
|
|
Name2 = <<"t_http_bridges_without_local_topic1">>,
|
|
%% create one http bridge with local_topic
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name1)
|
|
),
|
|
%% and we create another one without local_topic
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
maps:remove(<<"local_topic">>, ?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name2))
|
|
),
|
|
BridgeID1 = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name1),
|
|
BridgeID2 = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name2),
|
|
%% Send an message to emqx and the message should be forwarded to the HTTP server.
|
|
%% This is to verify we can have 2 bridges with and without local_topic fields
|
|
%% at the same time.
|
|
Body = <<"my msg">>,
|
|
emqx:publish(emqx_message:make(<<"emqx_webhook/1">>, Body)),
|
|
?assert(
|
|
receive
|
|
{http_server, received, #{
|
|
method := <<"POST">>,
|
|
path := <<"/path1">>,
|
|
body := Body
|
|
}} ->
|
|
true;
|
|
Msg ->
|
|
ct:pal("error: http got unexpected request: ~p", [Msg]),
|
|
false
|
|
after 100 ->
|
|
false
|
|
end
|
|
),
|
|
%% delete the bridge
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID1]), []),
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID2]), []),
|
|
ok.
|
|
|
|
t_check_dependent_actions_on_delete(Config) ->
|
|
Port = ?config(port, Config),
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
%% then we add a webhook bridge, using POST
|
|
%% POST /bridges/ will create a bridge
|
|
URL1 = ?URL(Port, "path1"),
|
|
Name = <<"t_http_crud_apis">>,
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
{ok, 201, Rule} = request(
|
|
post,
|
|
uri(["rules"]),
|
|
#{
|
|
<<"name">> => <<"t_http_crud_apis">>,
|
|
<<"enable">> => true,
|
|
<<"actions">> => [BridgeID],
|
|
<<"sql">> => <<"SELECT * from \"t\"">>
|
|
}
|
|
),
|
|
#{<<"id">> := RuleId} = emqx_json:decode(Rule, [return_maps]),
|
|
%% deleting the bridge should fail because there is a rule that depends on it
|
|
{ok, 400, _} = request(
|
|
delete, uri(["bridges", BridgeID]) ++ "?also_delete_dep_actions=false", []
|
|
),
|
|
%% delete the rule first
|
|
{ok, 204, <<>>} = request(delete, uri(["rules", RuleId]), []),
|
|
%% then delete the bridge is OK
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
ok.
|
|
|
|
t_cascade_delete_actions(Config) ->
|
|
Port = ?config(port, Config),
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
%% then we add a webhook bridge, using POST
|
|
%% POST /bridges/ will create a bridge
|
|
URL1 = ?URL(Port, "path1"),
|
|
Name = <<"t_http_crud_apis">>,
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
{ok, 201, Rule} = request(
|
|
post,
|
|
uri(["rules"]),
|
|
#{
|
|
<<"name">> => <<"t_http_crud_apis">>,
|
|
<<"enable">> => true,
|
|
<<"actions">> => [BridgeID],
|
|
<<"sql">> => <<"SELECT * from \"t\"">>
|
|
}
|
|
),
|
|
#{<<"id">> := RuleId} = emqx_json:decode(Rule, [return_maps]),
|
|
%% delete the bridge will also delete the actions from the rules
|
|
{ok, 204, _} = request(
|
|
delete, uri(["bridges", BridgeID]) ++ "?also_delete_dep_actions=true", []
|
|
),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
{ok, 200, Rule1} = request(get, uri(["rules", RuleId]), []),
|
|
?assertMatch(
|
|
#{
|
|
<<"actions">> := []
|
|
},
|
|
emqx_json:decode(Rule1, [return_maps])
|
|
),
|
|
{ok, 204, <<>>} = request(delete, uri(["rules", RuleId]), []),
|
|
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["rules"]),
|
|
#{
|
|
<<"name">> => <<"t_http_crud_apis">>,
|
|
<<"enable">> => true,
|
|
<<"actions">> => [BridgeID],
|
|
<<"sql">> => <<"SELECT * from \"t\"">>
|
|
}
|
|
),
|
|
|
|
{ok, 204, _} = request(delete, uri(["bridges", BridgeID]) ++ "?also_delete_dep_actions", []),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
ok.
|
|
|
|
t_broken_bpapi_vsn(Config) ->
|
|
Port = ?config(port, Config),
|
|
URL1 = ?URL(Port, "abc"),
|
|
Name = <<"t_bad_bpapi_vsn">>,
|
|
{ok, 201, _Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
%% still works since we redirect to 'restart'
|
|
{ok, 501, <<>>} = request(post, operation_path(cluster, start, BridgeID), <<"">>),
|
|
{ok, 501, <<>>} = request(post, operation_path(node, start, BridgeID), <<"">>),
|
|
ok.
|
|
|
|
t_old_bpapi_vsn(Config) ->
|
|
Port = ?config(port, Config),
|
|
URL1 = ?URL(Port, "abc"),
|
|
Name = <<"t_bad_bpapi_vsn">>,
|
|
{ok, 201, _Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
{ok, 204, <<>>} = request(post, operation_path(cluster, stop, BridgeID), <<"">>),
|
|
{ok, 204, <<>>} = request(post, operation_path(node, stop, BridgeID), <<"">>),
|
|
%% still works since we redirect to 'restart'
|
|
{ok, 204, <<>>} = request(post, operation_path(cluster, start, BridgeID), <<"">>),
|
|
{ok, 204, <<>>} = request(post, operation_path(node, start, BridgeID), <<"">>),
|
|
{ok, 204, <<>>} = request(post, operation_path(cluster, restart, BridgeID), <<"">>),
|
|
{ok, 204, <<>>} = request(post, operation_path(node, restart, BridgeID), <<"">>),
|
|
ok.
|
|
|
|
t_start_stop_bridges_node(Config) ->
|
|
{ok, 404, _} =
|
|
request(
|
|
post,
|
|
uri(["nodes", "thisbetterbenotanatomyet", "bridges", "webhook:foo", start]),
|
|
<<"">>
|
|
),
|
|
{ok, 404, _} =
|
|
request(
|
|
post,
|
|
uri(["nodes", "undefined", "bridges", "webhook:foo", start]),
|
|
<<"">>
|
|
),
|
|
do_start_stop_bridges(node, Config).
|
|
|
|
t_start_stop_bridges_cluster(Config) ->
|
|
do_start_stop_bridges(cluster, Config).
|
|
|
|
do_start_stop_bridges(Type, Config) ->
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
Port = ?config(port, Config),
|
|
URL1 = ?URL(Port, "abc"),
|
|
Name = atom_to_binary(Type),
|
|
{ok, 201, Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
%ct:pal("the bridge ==== ~p", [Bridge]),
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := <<"connected">>,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL1
|
|
} = emqx_json:decode(Bridge, [return_maps]),
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
%% stop it
|
|
{ok, 204, <<>>} = request(post, operation_path(Type, stop, BridgeID), <<"">>),
|
|
{ok, 200, Bridge2} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"stopped">>}, emqx_json:decode(Bridge2, [return_maps])),
|
|
%% start again
|
|
{ok, 204, <<>>} = request(post, operation_path(Type, start, BridgeID), <<"">>),
|
|
{ok, 200, Bridge3} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge3, [return_maps])),
|
|
%% start a started bridge
|
|
{ok, 204, <<>>} = request(post, operation_path(Type, start, BridgeID), <<"">>),
|
|
{ok, 200, Bridge3_1} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge3_1, [return_maps])),
|
|
%% restart an already started bridge
|
|
{ok, 204, <<>>} = request(post, operation_path(Type, restart, BridgeID), <<"">>),
|
|
{ok, 200, Bridge3} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge3, [return_maps])),
|
|
%% stop it again
|
|
{ok, 204, <<>>} = request(post, operation_path(Type, stop, BridgeID), <<"">>),
|
|
%% restart a stopped bridge
|
|
{ok, 204, <<>>} = request(post, operation_path(Type, restart, BridgeID), <<"">>),
|
|
{ok, 200, Bridge4} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge4, [return_maps])),
|
|
|
|
{ok, 404, _} = request(post, operation_path(Type, invalidop, BridgeID), <<"">>),
|
|
|
|
%% delete the bridge
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
%% Fail parse-id check
|
|
{ok, 404, _} = request(post, operation_path(Type, start, <<"wreckbook_fugazi">>), <<"">>),
|
|
%% Looks ok but doesn't exist
|
|
{ok, 404, _} = request(post, operation_path(Type, start, <<"webhook:cptn_hook">>), <<"">>),
|
|
|
|
%% Create broken bridge
|
|
{ListenPort, Sock} = listen_on_random_port(),
|
|
%% Connecting to this endpoint should always timeout
|
|
BadServer = iolist_to_binary(io_lib:format("localhost:~B", [ListenPort])),
|
|
BadName = <<"bad_", (atom_to_binary(Type))/binary>>,
|
|
{ok, 201, BadBridge1} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?MQTT_BRIDGE(BadServer, BadName)
|
|
),
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE_MQTT,
|
|
<<"name">> := BadName,
|
|
<<"enable">> := true,
|
|
<<"server">> := BadServer,
|
|
<<"status">> := <<"connecting">>,
|
|
<<"node_status">> := [_ | _]
|
|
} = emqx_json:decode(BadBridge1, [return_maps]),
|
|
BadBridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE_MQTT, BadName),
|
|
?assertMatch(
|
|
{ok, SC, _} when SC == 500 orelse SC == 503,
|
|
request(post, operation_path(Type, start, BadBridgeID), <<"">>)
|
|
),
|
|
ok = gen_tcp:close(Sock),
|
|
ok.
|
|
|
|
t_enable_disable_bridges(Config) ->
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
Name = ?BRIDGE_NAME,
|
|
Port = ?config(port, Config),
|
|
URL1 = ?URL(Port, "abc"),
|
|
{ok, 201, Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
%ct:pal("the bridge ==== ~p", [Bridge]),
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := <<"connected">>,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL1
|
|
} = emqx_json:decode(Bridge, [return_maps]),
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
%% disable it
|
|
{ok, 204, <<>>} = request(put, enable_path(false, BridgeID), <<"">>),
|
|
{ok, 200, Bridge2} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"stopped">>}, emqx_json:decode(Bridge2, [return_maps])),
|
|
%% enable again
|
|
{ok, 204, <<>>} = request(put, enable_path(true, BridgeID), <<"">>),
|
|
{ok, 200, Bridge3} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge3, [return_maps])),
|
|
%% enable an already started bridge
|
|
{ok, 204, <<>>} = request(put, enable_path(true, BridgeID), <<"">>),
|
|
{ok, 200, Bridge3} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge3, [return_maps])),
|
|
%% disable it again
|
|
{ok, 204, <<>>} = request(put, enable_path(false, BridgeID), <<"">>),
|
|
|
|
%% bad param
|
|
{ok, 404, _} = request(put, enable_path(foo, BridgeID), <<"">>),
|
|
{ok, 404, _} = request(put, enable_path(true, "foo"), <<"">>),
|
|
{ok, 404, _} = request(put, enable_path(true, "webhook:foo"), <<"">>),
|
|
|
|
{ok, 400, Res} = request(post, operation_path(node, start, BridgeID), <<"">>),
|
|
?assertEqual(
|
|
<<"{\"code\":\"BAD_REQUEST\",\"message\":\"Forbidden operation, bridge not enabled\"}">>,
|
|
Res
|
|
),
|
|
{ok, 400, Res} = request(post, operation_path(cluster, start, BridgeID), <<"">>),
|
|
|
|
%% enable a stopped bridge
|
|
{ok, 204, <<>>} = request(put, enable_path(true, BridgeID), <<"">>),
|
|
{ok, 200, Bridge4} = request(get, uri(["bridges", BridgeID]), []),
|
|
?assertMatch(#{<<"status">> := <<"connected">>}, emqx_json:decode(Bridge4, [return_maps])),
|
|
%% delete the bridge
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []).
|
|
|
|
t_reset_bridges(Config) ->
|
|
%% assert there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
Name = ?BRIDGE_NAME,
|
|
Port = ?config(port, Config),
|
|
URL1 = ?URL(Port, "abc"),
|
|
{ok, 201, Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
%ct:pal("the bridge ==== ~p", [Bridge]),
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := <<"connected">>,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL1
|
|
} = emqx_json:decode(Bridge, [return_maps]),
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
{ok, 204, <<>>} = request(put, uri(["bridges", BridgeID, "metrics/reset"]), []),
|
|
|
|
%% delete the bridge
|
|
{ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []).
|
|
|
|
t_with_redact_update(_Config) ->
|
|
Name = <<"redact_update">>,
|
|
Type = <<"mqtt">>,
|
|
Password = <<"123456">>,
|
|
Template = #{
|
|
<<"type">> => Type,
|
|
<<"name">> => Name,
|
|
<<"server">> => <<"127.0.0.1:1883">>,
|
|
<<"username">> => <<"test">>,
|
|
<<"password">> => Password,
|
|
<<"ingress">> =>
|
|
#{<<"remote">> => #{<<"topic">> => <<"t/#">>}}
|
|
},
|
|
|
|
{ok, 201, _} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
Template
|
|
),
|
|
|
|
%% update with redacted config
|
|
Conf = emqx_misc:redact(Template),
|
|
BridgeID = emqx_bridge_resource:bridge_id(Type, Name),
|
|
{ok, 200, _ResBin} = request(
|
|
put,
|
|
uri(["bridges", BridgeID]),
|
|
Conf
|
|
),
|
|
RawConf = emqx:get_raw_config([bridges, Type, Name]),
|
|
Value = maps:get(<<"password">>, RawConf),
|
|
?assertEqual(Password, Value),
|
|
ok.
|
|
|
|
t_bridges_probe(Config) ->
|
|
Port = ?config(port, Config),
|
|
URL = ?URL(Port, "some_path"),
|
|
|
|
{ok, 204, <<>>} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?HTTP_BRIDGE(URL, ?BRIDGE_TYPE, ?BRIDGE_NAME)
|
|
),
|
|
|
|
%% second time with same name is ok since no real bridge created
|
|
{ok, 204, <<>>} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?HTTP_BRIDGE(URL, ?BRIDGE_TYPE, ?BRIDGE_NAME)
|
|
),
|
|
|
|
{ok, 400, NxDomain} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?HTTP_BRIDGE(<<"http://203.0.113.3:1234/foo">>, ?BRIDGE_TYPE, ?BRIDGE_NAME)
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"TEST_FAILED">>,
|
|
<<"message">> := _
|
|
},
|
|
emqx_json:decode(NxDomain, [return_maps])
|
|
),
|
|
|
|
{ok, 204, _} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?MQTT_BRIDGE(<<"127.0.0.1:1883">>)
|
|
),
|
|
|
|
{ok, 400, ConnRefused} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?MQTT_BRIDGE(<<"127.0.0.1:2883">>)
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"TEST_FAILED">>,
|
|
<<"message">> := <<"Connection refused">>
|
|
},
|
|
emqx_json:decode(ConnRefused, [return_maps])
|
|
),
|
|
|
|
{ok, 400, HostNotFound} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?MQTT_BRIDGE(<<"nohost:2883">>)
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"TEST_FAILED">>,
|
|
<<"message">> := <<"Host not found">>
|
|
},
|
|
emqx_json:decode(HostNotFound, [return_maps])
|
|
),
|
|
|
|
AuthnConfig = #{
|
|
<<"mechanism">> => <<"password_based">>,
|
|
<<"backend">> => <<"built_in_database">>,
|
|
<<"user_id_type">> => <<"username">>
|
|
},
|
|
Chain = 'mqtt:global',
|
|
emqx:update_config(
|
|
[authentication],
|
|
{create_authenticator, Chain, AuthnConfig}
|
|
),
|
|
User = #{user_id => <<"u">>, password => <<"p">>},
|
|
AuthenticatorID = <<"password_based:built_in_database">>,
|
|
{ok, _} = emqx_authentication:add_user(
|
|
Chain,
|
|
AuthenticatorID,
|
|
User
|
|
),
|
|
|
|
{ok, 400, Unauthorized} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?MQTT_BRIDGE(<<"127.0.0.1:1883">>)#{<<"proto_ver">> => <<"v4">>}
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"TEST_FAILED">>,
|
|
<<"message">> := <<"Unauthorized client">>
|
|
},
|
|
emqx_json:decode(Unauthorized, [return_maps])
|
|
),
|
|
|
|
{ok, 400, Malformed} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?MQTT_BRIDGE(<<"127.0.0.1:1883">>)#{
|
|
<<"proto_ver">> => <<"v4">>, <<"password">> => <<"mySecret">>, <<"username">> => <<"u">>
|
|
}
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"TEST_FAILED">>,
|
|
<<"message">> := <<"Malformed username or password">>
|
|
},
|
|
emqx_json:decode(Malformed, [return_maps])
|
|
),
|
|
|
|
{ok, 400, NotAuthorized} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?MQTT_BRIDGE(<<"127.0.0.1:1883">>)
|
|
),
|
|
?assertMatch(
|
|
#{
|
|
<<"code">> := <<"TEST_FAILED">>,
|
|
<<"message">> := <<"Not authorized">>
|
|
},
|
|
emqx_json:decode(NotAuthorized, [return_maps])
|
|
),
|
|
|
|
{ok, 400, BadReq} = request(
|
|
post,
|
|
uri(["bridges_probe"]),
|
|
?BRIDGE(<<"bad_bridge">>, <<"unknown_type">>)
|
|
),
|
|
?assertMatch(#{<<"code">> := <<"BAD_REQUEST">>}, emqx_json:decode(BadReq, [return_maps])),
|
|
ok.
|
|
|
|
t_metrics(Config) ->
|
|
Port = ?config(port, Config),
|
|
%% assert we there's no bridges at first
|
|
{ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
|
|
|
|
%% then we add a webhook bridge, using POST
|
|
%% POST /bridges/ will create a bridge
|
|
URL1 = ?URL(Port, "path1"),
|
|
Name = ?BRIDGE_NAME,
|
|
{ok, 201, Bridge} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name)
|
|
),
|
|
|
|
%ct:pal("---bridge: ~p", [Bridge]),
|
|
Decoded = emqx_json:decode(Bridge, [return_maps]),
|
|
#{
|
|
<<"type">> := ?BRIDGE_TYPE,
|
|
<<"name">> := Name,
|
|
<<"enable">> := true,
|
|
<<"status">> := _,
|
|
<<"node_status">> := [_ | _],
|
|
<<"url">> := URL1
|
|
} = Decoded,
|
|
|
|
%% assert that the bridge return doesn't contain metrics anymore
|
|
?assertNot(maps:is_key(<<"metrics">>, Decoded)),
|
|
?assertNot(maps:is_key(<<"node_metrics">>, Decoded)),
|
|
|
|
BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE, Name),
|
|
|
|
%% check for empty bridge metrics
|
|
{ok, 200, Bridge1Str} = request(get, uri(["bridges", BridgeID, "metrics"]), []),
|
|
?assertMatch(
|
|
#{
|
|
<<"metrics">> := #{<<"success">> := 0},
|
|
<<"node_metrics">> := [_ | _]
|
|
},
|
|
emqx_json:decode(Bridge1Str, [return_maps])
|
|
),
|
|
|
|
%% check that the bridge doesn't contain metrics anymore
|
|
{ok, 200, Bridge2Str} = request(get, uri(["bridges", BridgeID]), []),
|
|
Decoded2 = emqx_json:decode(Bridge2Str, [return_maps]),
|
|
?assertNot(maps:is_key(<<"metrics">>, Decoded2)),
|
|
?assertNot(maps:is_key(<<"node_metrics">>, Decoded2)),
|
|
|
|
%% send an message to emqx and the message should be forwarded to the HTTP server
|
|
Body = <<"my msg">>,
|
|
emqx:publish(emqx_message:make(<<"emqx_webhook/1">>, Body)),
|
|
?assert(
|
|
receive
|
|
{http_server, received, #{
|
|
method := <<"POST">>,
|
|
path := <<"/path1">>,
|
|
body := Body
|
|
}} ->
|
|
true;
|
|
Msg ->
|
|
ct:pal("error: http got unexpected request: ~p", [Msg]),
|
|
false
|
|
after 100 ->
|
|
false
|
|
end
|
|
),
|
|
|
|
%% check for non-empty bridge metrics
|
|
{ok, 200, Bridge3Str} = request(get, uri(["bridges", BridgeID, "metrics"]), []),
|
|
?assertMatch(
|
|
#{
|
|
<<"metrics">> := #{<<"success">> := _},
|
|
<<"node_metrics">> := [_ | _]
|
|
},
|
|
emqx_json:decode(Bridge3Str, [return_maps])
|
|
),
|
|
|
|
%% check that metrics isn't returned when listing all bridges
|
|
{ok, 200, BridgesStr} = request(get, uri(["bridges"]), []),
|
|
?assert(
|
|
lists:all(
|
|
fun(E) -> not maps:is_key(<<"metrics">>, E) end,
|
|
emqx_json:decode(BridgesStr, [return_maps])
|
|
)
|
|
),
|
|
ok.
|
|
|
|
%% request_timeout in bridge root should match request_timeout in
|
|
%% resource_opts.
|
|
t_inconsistent_webhook_request_timeouts(Config) ->
|
|
Port = ?config(port, Config),
|
|
URL1 = ?URL(Port, "path1"),
|
|
Name = ?BRIDGE_NAME,
|
|
BadBridgeParams =
|
|
emqx_map_lib:deep_merge(
|
|
?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, Name),
|
|
#{
|
|
<<"request_timeout">> => <<"1s">>,
|
|
<<"resource_opts">> => #{<<"request_timeout">> => <<"2s">>}
|
|
}
|
|
),
|
|
{ok, 201, RawResponse} = request(
|
|
post,
|
|
uri(["bridges"]),
|
|
BadBridgeParams
|
|
),
|
|
%% note: same value on both fields
|
|
?assertMatch(
|
|
#{
|
|
<<"request_timeout">> := <<"2s">>,
|
|
<<"resource_opts">> := #{<<"request_timeout">> := <<"2s">>}
|
|
},
|
|
emqx_json:decode(RawResponse, [return_maps])
|
|
),
|
|
ok.
|
|
|
|
operation_path(node, Oper, BridgeID) ->
|
|
uri(["nodes", node(), "bridges", BridgeID, Oper]);
|
|
operation_path(cluster, Oper, BridgeID) ->
|
|
uri(["bridges", BridgeID, Oper]).
|
|
|
|
enable_path(Enable, BridgeID) ->
|
|
uri(["bridges", BridgeID, "enable", Enable]).
|
|
|
|
str(S) when is_list(S) -> S;
|
|
str(S) when is_binary(S) -> binary_to_list(S).
|