164 lines
4.9 KiB
Erlang
164 lines
4.9 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2021 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.
|
|
%%--------------------------------------------------------------------n
|
|
|
|
-module(emqx_slow_subs_api_SUITE).
|
|
|
|
-compile(export_all).
|
|
-compile(nowarn_export_all).
|
|
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
|
|
-include_lib("emqx/include/emqx.hrl").
|
|
-include_lib("emqx/include/emqx_mqtt.hrl").
|
|
-include_lib("emqx_management/include/emqx_mgmt.hrl").
|
|
-include_lib("emqx_plugin_libs/include/emqx_slow_subs.hrl").
|
|
|
|
-define(CONTENT_TYPE, "application/x-www-form-urlencoded").
|
|
|
|
-define(HOST, "http://127.0.0.1:18083/").
|
|
|
|
-define(API_VERSION, "v4").
|
|
|
|
-define(BASE_PATH, "api").
|
|
-define(NOW, erlang:system_time(millisecond)).
|
|
|
|
all() ->
|
|
emqx_ct:all(?MODULE).
|
|
|
|
init_per_suite(Config) ->
|
|
emqx_ct_helpers:boot_modules(all),
|
|
application:load(emqx_plugin_libs),
|
|
emqx_ct_helpers:start_apps([emqx_modules, emqx_management, emqx_dashboard]),
|
|
Config.
|
|
|
|
end_per_suite(Config) ->
|
|
emqx_ct_helpers:stop_apps([emqx_management]),
|
|
Config.
|
|
|
|
init_per_testcase(_, Config) ->
|
|
emqx_mod_slow_subs:load(base_conf()),
|
|
Config.
|
|
|
|
end_per_testcase(_, Config) ->
|
|
emqx_mod_slow_subs:unload([]),
|
|
Config.
|
|
|
|
base_conf() ->
|
|
[ {threshold, 500}
|
|
, {top_k_num, 5}
|
|
, {expire_interval, timer:seconds(60)}
|
|
, {notice_interval, 0}
|
|
, {notice_qos, 0}
|
|
, {notice_batch_size, 3}
|
|
].
|
|
|
|
t_get_history(_) ->
|
|
Now = ?NOW,
|
|
Each = fun(I) ->
|
|
ClientId = erlang:list_to_binary(io_lib:format("test_~p", [I])),
|
|
ets:insert(?TOPK_TAB, #top_k{index = ?INDEX(I, ClientId),
|
|
type = average,
|
|
last_update_time = Now})
|
|
end,
|
|
|
|
lists:foreach(Each, lists:seq(1, 5)),
|
|
|
|
{ok, Data} = request_api(get, api_path(["slow_subscriptions"]), "_page=1&_limit=10",
|
|
auth_header_()),
|
|
#{meta := Meta, data := [First | _]} = decode(Data),
|
|
|
|
RMeta = #{page => 1, limit => 10, count => 5},
|
|
?assertEqual(RMeta, Meta),
|
|
|
|
RFirst = #{clientid => <<"test_5">>,
|
|
latency => 5,
|
|
type => <<"average">>,
|
|
last_update_time => Now},
|
|
|
|
?assertEqual(RFirst, First).
|
|
|
|
t_clear(_) ->
|
|
ets:insert(?TOPK_TAB, #top_k{index = ?INDEX(1, <<"test">>),
|
|
type = average,
|
|
last_update_time = ?NOW}),
|
|
|
|
{ok, _} = request_api(delete, api_path(["slow_subscriptions"]), [],
|
|
auth_header_()),
|
|
|
|
?assertEqual(0, ets:info(?TOPK_TAB, size)).
|
|
|
|
decode(Data) ->
|
|
Pairs = emqx_json:decode(Data),
|
|
to_maps(Pairs).
|
|
|
|
to_maps([H | _] = List) when is_tuple(H) ->
|
|
to_maps(List, #{});
|
|
|
|
to_maps([_ | _] = List) ->
|
|
[to_maps(X) || X <- List];
|
|
|
|
to_maps(V) -> V.
|
|
|
|
to_maps([{K, V} | T], Map) ->
|
|
AtomKey = erlang:binary_to_atom(K),
|
|
to_maps(T, Map#{AtomKey => to_maps(V)});
|
|
|
|
to_maps([], Map) ->
|
|
Map.
|
|
|
|
request_api(Method, Url, Auth) ->
|
|
request_api(Method, Url, [], Auth, []).
|
|
|
|
request_api(Method, Url, QueryParams, Auth) ->
|
|
request_api(Method, Url, QueryParams, Auth, []).
|
|
|
|
request_api(Method, Url, QueryParams, Auth, []) ->
|
|
NewUrl = case QueryParams of
|
|
"" -> Url;
|
|
_ -> Url ++ "?" ++ QueryParams
|
|
end,
|
|
do_request_api(Method, {NewUrl, [Auth]});
|
|
request_api(Method, Url, QueryParams, Auth, Body) ->
|
|
NewUrl = case QueryParams of
|
|
"" -> Url;
|
|
_ -> Url ++ "?" ++ QueryParams
|
|
end,
|
|
do_request_api(Method, {NewUrl, [Auth], "application/json", emqx_json:encode(Body)}).
|
|
|
|
do_request_api(Method, Request)->
|
|
ct:pal("Method: ~p, Request: ~p", [Method, Request]),
|
|
case httpc:request(Method, Request, [], []) of
|
|
{error, socket_closed_remotely} ->
|
|
{error, socket_closed_remotely};
|
|
{ok, {{"HTTP/1.1", Code, _}, _, Return} }
|
|
when Code =:= 200 orelse Code =:= 201 ->
|
|
{ok, Return};
|
|
{ok, {Reason, _, _}} ->
|
|
{error, Reason}
|
|
end.
|
|
|
|
auth_header_() ->
|
|
AppId = <<"admin">>,
|
|
AppSecret = <<"public">>,
|
|
auth_header_(binary_to_list(AppId), binary_to_list(AppSecret)).
|
|
|
|
auth_header_(User, Pass) ->
|
|
Encoded = base64:encode_to_string(lists:append([User,":",Pass])),
|
|
{"Authorization","Basic " ++ Encoded}.
|
|
|
|
api_path(Parts)->
|
|
?HOST ++ filename:join([?BASE_PATH, ?API_VERSION] ++ Parts).
|