test(emqx_coap): add some test case

apps/emqx_gateway/src/coap/emqx_coap_channel.erl

@@ -263,16 +263,7 @@ check_token(true,
     end;
 
 check_token(false, Msg, Channel) ->
-    case emqx_coap_message:get_option(uri_query, Msg) of
+    call_session(handle_request, Msg, Channel).
-        #{<<"clientid">> := _} ->
-            Reply = emqx_coap_message:piggyback({error, unauthorized}, Msg),
-            {ok, {outgoing, Reply}, Channel};
-        #{<<"token">> := _} ->
-            Reply = emqx_coap_message:piggyback({error, unauthorized}, Msg),
-            {ok, {outgoing, Reply}, Channel};
-        _ ->
-            call_session(handle_request, Msg, Channel)
-    end.
 
 try_takeover(idle, DesireId, Msg, Channel) ->
     case emqx_coap_message:get_option(uri_path, Msg, []) of

apps/emqx_gateway/src/coap/test/emqx_coap_SUITE.erl

@@ -1,319 +0,0 @@
-%%--------------------------------------------------------------------
-%% Copyright (c) 2020-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.
-%%--------------------------------------------------------------------
-
--module(emqx_coap_SUITE).
-
-% -compile(export_all).
-% -compile(nowarn_export_all).
-
-% -include_lib("gen_coap/include/coap.hrl").
-% -include_lib("eunit/include/eunit.hrl").
-% -include_lib("emqx/include/emqx.hrl").
-
-% -define(LOGT(Format, Args), ct:pal(Format, Args)).
-
-% all() -> emqx_ct:all(?MODULE).
-
-% init_per_suite(Config) ->
-%     emqx_ct_helpers:start_apps([emqx_coap], fun set_special_cfg/1),
-%     Config.
-
-% set_special_cfg(emqx_coap) ->
-%     Opts = application:get_env(emqx_coap, dtls_opts,[]),
-%     Opts2 = [{keyfile, emqx_ct_helpers:deps_path(emqx, "etc/certs/key.pem")},
-%              {certfile, emqx_ct_helpers:deps_path(emqx, "etc/certs/cert.pem")}],
-%     application:set_env(emqx_coap, dtls_opts, emqx_misc:merge_opts(Opts, Opts2)),
-%     application:set_env(emqx_coap, enable_stats, true);
-% set_special_cfg(_) ->
-%     ok.
-
-% end_per_suite(Config) ->
-%     emqx_ct_helpers:stop_apps([emqx_coap]),
-%     Config.
-
-% %%--------------------------------------------------------------------
-% %% Test Cases
-% %%--------------------------------------------------------------------
-
-% t_publish(_Config) ->
-%     Topic = <<"abc">>, Payload = <<"123">>,
-%     TopicStr = binary_to_list(Topic),
-%     URI = "coap://127.0.0.1/mqtt/"++TopicStr++"?c=client1&u=tom&p=secret",
-
-%     %% Sub topic first
-%     emqx:subscribe(Topic),
-
-%     Reply = er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     {ok, changed, _} = Reply,
-
-%     receive
-%         {deliver, Topic, Msg} ->
-%             ?assertEqual(Topic, Msg#message.topic),
-%             ?assertEqual(Payload, Msg#message.payload)
-%     after
-%         500 ->
-%             ?assert(false)
-%     end.
-
-% t_publish_authz_deny(_Config) ->
-%     Topic = <<"abc">>, Payload = <<"123">>,
-%     TopicStr = binary_to_list(Topic),
-%     URI = "coap://127.0.0.1/mqtt/"++TopicStr++"?c=client1&u=tom&p=secret",
-
-%     %% Sub topic first
-%     emqx:subscribe(Topic),
-
-%     ok = meck:new(emqx_access_control, [non_strict, passthrough, no_history]),
-%     ok = meck:expect(emqx_access_control, authorize, 3, deny),
-%     Reply = er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?assertEqual({error,forbidden}, Reply),
-%     ok = meck:unload(emqx_access_control),
-%     receive
-%         {deliver, Topic, Msg} -> ct:fail({unexpected, {Topic, Msg}})
-%     after
-%         500 -> ok
-%     end.
-
-% t_observe(_Config) ->
-%     Topic = <<"abc">>, TopicStr = binary_to_list(Topic),
-%     Payload = <<"123">>,
-%     Uri = "coap://127.0.0.1/mqtt/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     {ok, Pid, N, Code, Content} = er_coap_observer:observe(Uri), 
-%     ?LOGT("observer Pid=~p, N=~p, Code=~p, Content=~p", [Pid, N, Code, Content]),
-
-%     [SubPid] = emqx:subscribers(Topic),
-%     ?assert(is_pid(SubPid)),
-
-%     %% Publish a message
-%     emqx:publish(emqx_message:make(Topic, Payload)),
-
-%     Notif = receive_notification(),
-%     ?LOGT("observer get Notif=~p", [Notif]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv}} = Notif,
-%     ?assertEqual(Payload, PayloadRecv),
-
-%     er_coap_observer:stop(Pid),
-%     timer:sleep(100),
-
-%     [] = emqx:subscribers(Topic).
-
-% t_observe_authz_deny(_Config) ->
-%     Topic = <<"abc">>, TopicStr = binary_to_list(Topic),
-%     Uri = "coap://127.0.0.1/mqtt/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     ok = meck:new(emqx_access_control, [non_strict, passthrough, no_history]),
-%     ok = meck:expect(emqx_access_control, authorize, 3, deny),
-%     ?assertEqual({error,forbidden}, er_coap_observer:observe(Uri)),
-%     [] = emqx:subscribers(Topic),
-%     ok = meck:unload(emqx_access_control).
-
-% t_observe_wildcard(_Config) ->
-%     Topic = <<"+/b">>, TopicStr = emqx_http_lib:uri_encode(binary_to_list(Topic)),
-%     Payload = <<"123">>,
-%     Uri = "coap://127.0.0.1/mqtt/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     {ok, Pid, N, Code, Content} = er_coap_observer:observe(Uri),
-%     ?LOGT("observer Uri=~p, Pid=~p, N=~p, Code=~p, Content=~p", [Uri, Pid, N, Code, Content]),
-
-%     [SubPid] = emqx:subscribers(Topic),
-%     ?assert(is_pid(SubPid)),
-
-%     %% Publish a message
-%     emqx:publish(emqx_message:make(<<"a/b">>, Payload)),
-
-%     Notif = receive_notification(),
-%     ?LOGT("observer get Notif=~p", [Notif]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv}} = Notif,
-%     ?assertEqual(Payload, PayloadRecv),
-
-%     er_coap_observer:stop(Pid),
-%     timer:sleep(100),
-
-%     [] = emqx:subscribers(Topic).
-
-% t_observe_pub(_Config) ->
-%     Topic = <<"+/b">>, TopicStr = emqx_http_lib:uri_encode(binary_to_list(Topic)),
-%     Uri = "coap://127.0.0.1/mqtt/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     {ok, Pid, N, Code, Content} = er_coap_observer:observe(Uri),
-%     ?LOGT("observer Pid=~p, N=~p, Code=~p, Content=~p", [Pid, N, Code, Content]),
-
-%     [SubPid] = emqx:subscribers(Topic),
-%     ?assert(is_pid(SubPid)),
-
-%     Topic2 = <<"a/b">>, Payload2 = <<"UFO">>,
-%     TopicStr2 = emqx_http_lib:uri_encode(binary_to_list(Topic2)),
-%     URI2 = "coap://127.0.0.1/mqtt/"++TopicStr2++"?c=client1&u=tom&p=secret",
-
-%     Reply2 = er_coap_client:request(put, URI2, #coap_content{format = <<"application/octet-stream">>, payload = Payload2}),
-%     {ok,changed, _} = Reply2,
-
-%     Notif2 = receive_notification(),
-%     ?LOGT("observer get Notif2=~p", [Notif2]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv2}} = Notif2,
-%     ?assertEqual(Payload2, PayloadRecv2),
-
-%     Topic3 = <<"j/b">>, Payload3 = <<"ET629">>,
-%     TopicStr3 = emqx_http_lib:uri_encode(binary_to_list(Topic3)),
-%     URI3 = "coap://127.0.0.1/mqtt/"++TopicStr3++"?c=client2&u=mike&p=guess",
-%     Reply3 = er_coap_client:request(put, URI3, #coap_content{format = <<"application/octet-stream">>, payload = Payload3}),
-%     {ok,changed, _} = Reply3,
-
-%     Notif3 = receive_notification(),
-%     ?LOGT("observer get Notif3=~p", [Notif3]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv3}} = Notif3,
-%     ?assertEqual(Payload3, PayloadRecv3),
-
-%     er_coap_observer:stop(Pid).
-
-% t_one_clientid_sub_2_topics(_Config) ->
-%     Topic1 = <<"abc">>, TopicStr1 = binary_to_list(Topic1),
-%     Payload1 = <<"123">>,
-%     Uri1 = "coap://127.0.0.1/mqtt/"++TopicStr1++"?c=client1&u=tom&p=secret",
-%     {ok, Pid1, N1, Code1, Content1} = er_coap_observer:observe(Uri1),
-%     ?LOGT("observer 1 Pid=~p, N=~p, Code=~p, Content=~p", [Pid1, N1, Code1, Content1]),
-
-%     [SubPid] = emqx:subscribers(Topic1),
-%     ?assert(is_pid(SubPid)),
-
-%     Topic2 = <<"x/y">>, TopicStr2 = emqx_http_lib:uri_encode(binary_to_list(Topic2)),
-%     Payload2 = <<"456">>,
-%     Uri2 = "coap://127.0.0.1/mqtt/"++TopicStr2++"?c=client1&u=tom&p=secret",
-%     {ok, Pid2, N2, Code2, Content2} = er_coap_observer:observe(Uri2),
-%     ?LOGT("observer 2 Pid=~p, N=~p, Code=~p, Content=~p", [Pid2, N2, Code2, Content2]),
-
-%     [SubPid] = emqx:subscribers(Topic2),
-%     ?assert(is_pid(SubPid)),
-
-%     emqx:publish(emqx_message:make(Topic1, Payload1)),
-
-%     Notif1 = receive_notification(),
-%     ?LOGT("observer 1 get Notif=~p", [Notif1]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv1}} = Notif1,
-%     ?assertEqual(Payload1, PayloadRecv1),
-
-%     emqx:publish(emqx_message:make(Topic2, Payload2)),
-
-%     Notif2 = receive_notification(),
-%     ?LOGT("observer 2 get Notif=~p", [Notif2]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv2}} = Notif2,
-%     ?assertEqual(Payload2, PayloadRecv2),
-
-%     er_coap_observer:stop(Pid1),
-%     er_coap_observer:stop(Pid2).
-
-% t_invalid_parameter(_Config) ->
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     %% "cid=client2" is invaid
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     Topic3 = <<"a/b">>, Payload3 = <<"ET629">>,
-%     TopicStr3 = emqx_http_lib:uri_encode(binary_to_list(Topic3)),
-%     URI3 = "coap://127.0.0.1/mqtt/"++TopicStr3++"?cid=client2&u=tom&p=simple",
-%     Reply3 = er_coap_client:request(put, URI3, #coap_content{format = <<"application/octet-stream">>, payload = Payload3}),
-%     ?assertMatch({error,bad_request}, Reply3),
-
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     %% "what=hello" is invaid
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     URI4 = "coap://127.0.0.1/mqtt/"++TopicStr3++"?what=hello",
-%     Reply4 = er_coap_client:request(put, URI4, #coap_content{format = <<"application/octet-stream">>, payload = Payload3}),
-%     ?assertMatch({error, bad_request}, Reply4).
-
-% t_invalid_topic(_Config) ->
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     %% "a/b" is a valid topic string
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     Topic3 = <<"a/b">>, Payload3 = <<"ET629">>,
-%     TopicStr3 = binary_to_list(Topic3),
-%     URI3 = "coap://127.0.0.1/mqtt/"++TopicStr3++"?c=client2&u=tom&p=simple",
-%     Reply3 = er_coap_client:request(put, URI3, #coap_content{format = <<"application/octet-stream">>, payload = Payload3}),
-%     ?assertMatch({ok,changed,_Content}, Reply3),
-
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     %% "+?#" is invaid topic string
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-%     URI4 = "coap://127.0.0.1/mqtt/"++"+?#"++"?what=hello",
-%     Reply4 = er_coap_client:request(put, URI4, #coap_content{format = <<"application/octet-stream">>, payload = Payload3}),
-%     ?assertMatch({error,bad_request}, Reply4).
-
-% % mqtt connection kicked by coap with same client id
-% t_kick_1(_Config) ->
-%     URI = "coap://127.0.0.1/mqtt/abc?c=clientid&u=tom&p=secret",
-%     % workaround: emqx:subscribe does not kick same client id.
-%     spawn_monitor(fun() ->
-%         {ok, C} = emqtt:start_link([{host,      "localhost"},
-%                                     {clientid, <<"clientid">>},
-%                                     {username,  <<"plain">>},
-%                                     {password,  <<"plain">>}]),
-%         {ok, _} = emqtt:connect(C) end),
-%     er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>,
-%                                                    payload = <<"123">>}),
-%     receive
-%         {'DOWN', _, _, _, _} -> ok
-%     after 2000 ->
-%         ?assert(false)
-%     end.
-
-% % mqtt connection kicked by coap with same client id
-% t_authz(_Config) ->
-%     OldPath = emqx:get_env(plugins_etc_dir),
-%     application:set_env(emqx, plugins_etc_dir,
-%                         emqx_ct_helpers:deps_path(emqx_authz, "test")),
-%     Conf = #{<<"authz">> =>
-%              #{<<"rules">> =>
-%                [#{<<"principal">> =>#{<<"username">> => <<"coap">>},
-%                   <<"permission">> => deny,
-%                   <<"topics">> => [<<"abc">>],
-%                   <<"action">> => <<"publish">>}
-%                ]}},
-%     ok = file:write_file(filename:join(emqx:get_env(plugins_etc_dir), 'authz.conf'), jsx:encode(Conf)),
-%     application:ensure_all_started(emqx_authz),
-
-%     emqx:subscribe(<<"abc">>),
-%     URI = "coap://127.0.0.1/mqtt/adbc?c=client1&u=coap&p=secret",
-%     er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>,
-%                                                    payload = <<"123">>}),
-%     receive
-%         _Something -> ?assert(false)
-%     after 2000 ->
-%         ok
-%     end,
-
-%     ok = emqx_hooks:del('client.authorize', {emqx_authz, authorize}),
-%     file:delete(filename:join(emqx:get_env(plugins_etc_dir), 'authz.conf')),
-%     application:set_env(emqx, plugins_etc_dir, OldPath),
-%     application:stop(emqx_authz).
-
-% t_stats(_) ->
-%     ok.
-
-% t_auth_failure(_) ->
-%     ok.
-
-% t_qos_supprot(_) ->
-%     ok.
-
-% %%--------------------------------------------------------------------
-% %% Helpers
-
-% receive_notification() ->
-%     receive
-%         {coap_notify, Pid, N2, Code2, Content2} ->
-%             {coap_notify, Pid, N2, Code2, Content2}
-%     after 2000 ->
-%         receive_notification_timeout
-%     end.
-
-% testdir(DataPath) ->
-%     Ls = filename:split(DataPath),
-%     filename:join(lists:sublist(Ls, 1, length(Ls) - 1)).

apps/emqx_gateway/src/coap/test/emqx_coap_pubsub_SUITE.erl

@@ -1,678 +0,0 @@
-%%--------------------------------------------------------------------
-%% Copyright (c) 2020-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.
-%%--------------------------------------------------------------------
-
--module(emqx_coap_pubsub_SUITE).
-
-% -compile(export_all).
-% -compile(nowarn_export_all).
-
-
-% -include_lib("gen_coap/include/coap.hrl").
-% -include_lib("eunit/include/eunit.hrl").
-% -include_lib("emqx/include/emqx.hrl").
-
-% -define(LOGT(Format, Args), ct:pal(Format, Args)).
-
-% all() -> emqx_ct:all(?MODULE).
-
-% init_per_suite(Config) ->
-%     emqx_ct_helpers:start_apps([emqx_coap], fun set_special_cfg/1),
-%     Config.
-
-% set_special_cfg(emqx_coap) ->
-%     application:set_env(emqx_coap, enable_stats, true);
-% set_special_cfg(_) ->
-%     ok.
-
-% end_per_suite(Config) ->
-%     emqx_ct_helpers:stop_apps([emqx_coap]),
-%     Config.
-
-% %%--------------------------------------------------------------------
-% %% Test Cases
-% %%--------------------------------------------------------------------
-
-% t_update_max_age(_Config) ->
-%     TopicInPayload = <<"topic1">>,
-%     Payload = <<"<topic1>;ct=42">>,
-%     Payload1 = <<"<topic1>;ct=50">>,
-%     URI = "coap://127.0.0.1/ps/"++"?c=client1&u=tom&p=secret",
-%     URI2 = "coap://127.0.0.1/ps/topic1"++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     TopicInfo = [{TopicInPayload, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(TopicInPayload),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     timer:sleep(50),
-
-%     %% post to create the same topic but with different max age and ct value in payload
-%     Reply1 = er_coap_client:request(post, URI, #coap_content{max_age = 70, format = <<"application/link-format">>, payload = Payload1}),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply1,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{TopicInPayload, MaxAge2, CT2, _ResPayload, _TimeStamp1}] = emqx_coap_pubsub_topics:lookup_topic_info(TopicInPayload),
-%     ?assertEqual(70, MaxAge2),
-%     ?assertEqual(<<"50">>, CT2),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI2).
-
-% t_create_subtopic(_Config) ->
-%     TopicInPayload = <<"topic1">>,
-%     TopicInPayloadStr = "topic1",
-%     Payload = <<"<topic1>;ct=42">>,
-%     URI = "coap://127.0.0.1/ps/"++"?c=client1&u=tom&p=secret",
-%     RealURI = "coap://127.0.0.1/ps/topic1"++"?c=client1&u=tom&p=secret",
-
-%     Reply = er_coap_client:request(post, URI, #coap_content{format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     TopicInfo = [{TopicInPayload, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(TopicInPayload),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     timer:sleep(50),
-
-%     %% post to create the a sub topic
-%     SubPayload = <<"<subtopic>;ct=42">>,
-%     SubTopicInPayloadStr = "subtopic",
-%     SubURI = "coap://127.0.0.1/ps/"++TopicInPayloadStr++"?c=client1&u=tom&p=secret",
-%     SubRealURI = "coap://127.0.0.1/ps/"++TopicInPayloadStr++"/"++SubTopicInPayloadStr++"?c=client1&u=tom&p=secret",
-%     FullTopic = list_to_binary(TopicInPayloadStr++"/"++SubTopicInPayloadStr),
-%     Reply1 = er_coap_client:request(post, SubURI, #coap_content{format = <<"application/link-format">>, payload = SubPayload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     {ok,created, #coap_content{location_path = LocPath1}} = Reply1,
-%     ?assertEqual([<<"/ps/topic1/subtopic">>] ,LocPath1),
-%     [{FullTopic, MaxAge2, CT2, _ResPayload, _}] = emqx_coap_pubsub_topics:lookup_topic_info(FullTopic),
-%     ?assertEqual(60, MaxAge2),
-%     ?assertEqual(<<"42">>, CT2),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, SubRealURI),
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, RealURI).
-
-% t_over_max_age(_Config) ->
-%     TopicInPayload = <<"topic1">>,
-%     Payload = <<"<topic1>;ct=42">>,
-%     URI = "coap://127.0.0.1/ps/"++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{max_age = 2, format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     TopicInfo = [{TopicInPayload, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(TopicInPayload),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(2, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     timer:sleep(3000),
-%     ?assertEqual(true, emqx_coap_pubsub_topics:is_topic_timeout(TopicInPayload)).
-
-% t_refreash_max_age(_Config) ->
-%     TopicInPayload = <<"topic1">>,
-%     Payload = <<"<topic1>;ct=42">>,
-%     Payload1 = <<"<topic1>;ct=50">>,
-%     URI = "coap://127.0.0.1/ps/"++"?c=client1&u=tom&p=secret",
-%     RealURI = "coap://127.0.0.1/ps/topic1"++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{max_age = 5, format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     TopicInfo = [{TopicInPayload, MaxAge1, CT1, _ResPayload, TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(TopicInPayload),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?LOGT("TimeStamp=~p", [TimeStamp]),
-%     ?assertEqual(5, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     timer:sleep(3000),
-
-%     %% post to create the same topic, the max age timer will be restarted with the new max age value
-%     Reply1 = er_coap_client:request(post, URI, #coap_content{max_age = 5, format = <<"application/link-format">>, payload = Payload1}),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply1,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{TopicInPayload, MaxAge2, CT2, _ResPayload, TimeStamp1}] = emqx_coap_pubsub_topics:lookup_topic_info(TopicInPayload),
-%     ?LOGT("TimeStamp1=~p", [TimeStamp1]),
-%     ?assertEqual(5, MaxAge2),
-%     ?assertEqual(<<"50">>, CT2),
-
-%     timer:sleep(3000),
-%     ?assertEqual(false, emqx_coap_pubsub_topics:is_topic_timeout(TopicInPayload)),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, RealURI).
-
-% t_case01_publish_post(_Config) ->
-%     timer:sleep(100),
-%     MainTopic = <<"maintopic">>,
-%     TopicInPayload = <<"topic1">>,
-%     Payload = <<"<topic1>;ct=42">>,
-%     MainTopicStr = binary_to_list(MainTopic),
-
-%     %% post to create topic maintopic/topic1
-%     URI1 = "coap://127.0.0.1/ps/"++MainTopicStr++"?c=client1&u=tom&p=secret",
-%     FullTopic = list_to_binary(MainTopicStr++"/"++binary_to_list(TopicInPayload)),
-%     Reply1 = er_coap_client:request(post, URI1, #coap_content{format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     {ok,created, #coap_content{location_path = LocPath1}} = Reply1,
-%     ?assertEqual([<<"/ps/maintopic/topic1">>] ,LocPath1),
-%     [{FullTopic, MaxAge, CT2, <<>>, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(FullTopic),
-%     ?assertEqual(60, MaxAge),
-%     ?assertEqual(<<"42">>, CT2),
-
-%     %% post to publish message to topic maintopic/topic1
-%     FullTopicStr = emqx_http_lib:uri_encode(binary_to_list(FullTopic)),
-%     URI2 = "coap://127.0.0.1/ps/"++FullTopicStr++"?c=client1&u=tom&p=secret",
-%     PubPayload = <<"PUBLISH">>,
-
-%     %% Sub topic first
-%     emqx:subscribe(FullTopic),
-
-%     Reply2 = er_coap_client:request(post, URI2, #coap_content{format = <<"application/octet-stream">>, payload = PubPayload}),
-%     ?LOGT("Reply =~p", [Reply2]),
-%     {ok,changed, _} = Reply2,
-%     TopicInfo = [{FullTopic, MaxAge, CT2, PubPayload, _TimeStamp1}] = emqx_coap_pubsub_topics:lookup_topic_info(FullTopic),
-%     ?LOGT("the topic info =~p", [TopicInfo]),
-
-%     assert_recv(FullTopic, PubPayload),
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI2).
-
-% t_case02_publish_post(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"payload">>,
-
-%     %% Sub topic first
-%     emqx:subscribe(Topic),
-
-%     %% post to publish a new topic "topic1", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(60, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     assert_recv(Topic, Payload),
-
-%     %% post to publish a new message to the same topic "topic1" with different payload
-%     NewPayload = <<"newpayload">>,
-%     Reply1 = er_coap_client:request(post, URI, #coap_content{format = <<"application/octet-stream">>, payload = NewPayload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     {ok,changed, _} = Reply1,
-%     [{Topic, MaxAge, CT, NewPayload, _TimeStamp1}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-
-%     assert_recv(Topic, NewPayload),
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case03_publish_post(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"payload">>,
-
-%     %% Sub topic first
-%     emqx:subscribe(Topic),
-
-%     %% post to publish a new topic "topic1", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(60, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     assert_recv(Topic, Payload),
-
-%     %% post to publish a new message to the same topic "topic1", but the ct is not same as created
-%     NewPayload = <<"newpayload">>,
-%     Reply1 = er_coap_client:request(post, URI, #coap_content{format = <<"application/exi">>, payload = NewPayload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     ?assertEqual({error,bad_request}, Reply1),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case04_publish_post(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"payload">>,
-
-%     %% post to publish a new topic "topic1", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{max_age = 5, format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(5, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     %% after max age timeout, the topic still exists but the status is timeout
-%     timer:sleep(6000),
-%     ?assertEqual(true, emqx_coap_pubsub_topics:is_topic_timeout(Topic)),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case01_publish_put(_Config) ->
-%     MainTopic = <<"maintopic">>,
-%     TopicInPayload = <<"topic1">>,
-%     Payload = <<"<topic1>;ct=42">>,
-%     MainTopicStr = binary_to_list(MainTopic),
-
-%     %% post to create topic maintopic/topic1
-%     URI1 = "coap://127.0.0.1/ps/"++MainTopicStr++"?c=client1&u=tom&p=secret",
-%     FullTopic = list_to_binary(MainTopicStr++"/"++binary_to_list(TopicInPayload)),
-%     Reply1 = er_coap_client:request(post, URI1, #coap_content{format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     {ok,created, #coap_content{location_path = LocPath1}} = Reply1,
-%     ?assertEqual([<<"/ps/maintopic/topic1">>] ,LocPath1),
-%     [{FullTopic, MaxAge, CT2, <<>>, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(FullTopic),
-%     ?assertEqual(60, MaxAge),
-%     ?assertEqual(<<"42">>, CT2),
-
-%     %% put to publish message to topic maintopic/topic1
-%     FullTopicStr = emqx_http_lib:uri_encode(binary_to_list(FullTopic)),
-%     URI2 = "coap://127.0.0.1/ps/"++FullTopicStr++"?c=client1&u=tom&p=secret",
-%     PubPayload = <<"PUBLISH">>,
-
-%     %% Sub topic first
-%     emqx:subscribe(FullTopic),
-
-%     Reply2 = er_coap_client:request(put, URI2, #coap_content{format = <<"application/octet-stream">>, payload = PubPayload}),
-%     ?LOGT("Reply =~p", [Reply2]),
-%     {ok,changed, _} = Reply2,
-%     [{FullTopic, MaxAge, CT2, PubPayload, _TimeStamp1}] = emqx_coap_pubsub_topics:lookup_topic_info(FullTopic),
-
-%     assert_recv(FullTopic, PubPayload),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI2).
-
-% t_case02_publish_put(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"payload">>,
-
-%     %% Sub topic first
-%     emqx:subscribe(Topic),
-
-%     %% put to publish a new topic "topic1", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(60, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     assert_recv(Topic, Payload),
-
-%     %% put to publish a new message to the same topic "topic1" with different payload
-%     NewPayload = <<"newpayload">>,
-%     Reply1 = er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>, payload = NewPayload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     {ok,changed, _} = Reply1,
-%     [{Topic, MaxAge, CT, NewPayload, _TimeStamp1}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-
-%     assert_recv(Topic, NewPayload),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case03_publish_put(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"payload">>,
-
-%     %% Sub topic first
-%     emqx:subscribe(Topic),
-
-%     %% put to publish a new topic "topic1", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(60, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     assert_recv(Topic, Payload),
-
-%     %% put to publish a new message to the same topic "topic1", but the ct is not same as created
-%     NewPayload = <<"newpayload">>,
-%     Reply1 = er_coap_client:request(put, URI, #coap_content{format = <<"application/exi">>, payload = NewPayload}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     ?assertEqual({error,bad_request}, Reply1),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case04_publish_put(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"payload">>,
-
-%     %% put to publish a new topic "topic1", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(put, URI, #coap_content{max_age = 5, format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/topic1">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(5, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     %% after max age timeout, no publish message to the same topic, the topic info will be deleted
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%
-%     % but there is one thing to do is we don't count in the publish message received from emqx(from other node).TBD!!!!!!!!!!!!!
-%     %%%%%%%%%%%%%%%%%%%%%%%%%%
-%     timer:sleep(6000),
-%     ?assertEqual(true, emqx_coap_pubsub_topics:is_topic_timeout(Topic)),
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case01_subscribe(_Config) ->
-%     Topic = <<"topic1">>,
-%     Payload1 = <<"<topic1>;ct=42">>,
-%     timer:sleep(100),
-
-%     %% First post to create a topic "topic1"
-%     Uri = "coap://127.0.0.1/ps/"++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, Uri, #coap_content{format = <<"application/link-format">>, payload = Payload1}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = [LocPath]}} = Reply,
-%     ?assertEqual(<<"/ps/topic1">> ,LocPath),
-%     TopicInfo = [{Topic, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     %% Subscribe the topic
-%     Uri1 = "coap://127.0.0.1"++binary_to_list(LocPath)++"?c=client1&u=tom&p=secret",
-%     {ok, Pid, N, Code, Content} = er_coap_observer:observe(Uri1),
-%     ?LOGT("observer Pid=~p, N=~p, Code=~p, Content=~p", [Pid, N, Code, Content]),
-
-%     [SubPid] = emqx:subscribers(Topic),
-%     ?assert(is_pid(SubPid)),
-
-%     %% Publish a message
-%     Payload = <<"123">>,
-%     emqx:publish(emqx_message:make(Topic, Payload)),
-
-%     Notif = receive_notification(),
-%     ?LOGT("observer get Notif=~p", [Notif]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = PayloadRecv}} = Notif,
-
-%     ?assertEqual(Payload, PayloadRecv),
-
-%     %% GET to read the publish message of the topic
-%     Reply1 = er_coap_client:request(get, Uri1),
-%     ?LOGT("Reply=~p", [Reply1]),
-%     {ok,content, #coap_content{payload = <<"123">>}} = Reply1,
-
-%     er_coap_observer:stop(Pid),
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, Uri1).
-
-% t_case02_subscribe(_Config) ->
-%     Topic = <<"a/b">>,
-%     TopicStr = binary_to_list(Topic),
-%     PercentEncodedTopic = emqx_http_lib:uri_encode(TopicStr),
-%     Payload = <<"payload">>,
-
-%     %% post to publish a new topic "a/b", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{max_age = 5, format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/a/b">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(5, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     %% Wait for the max age of the timer expires
-%     timer:sleep(6000),
-%     ?assertEqual(true, emqx_coap_pubsub_topics:is_topic_timeout(Topic)),
-
-%     %% Subscribe to the timeout topic "a/b", still successfully，got {ok, nocontent} Method
-%     Uri = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     Reply1 = {ok, Pid, _N, nocontent, _} = er_coap_observer:observe(Uri),
-%     ?LOGT("Subscribe Reply=~p", [Reply1]), 
-
-%     [SubPid] = emqx:subscribers(Topic),
-%     ?assert(is_pid(SubPid)),
-
-%     %% put to publish to topic "a/b"
-%     Reply2 = er_coap_client:request(put, URI, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     {ok,changed, #coap_content{}} = Reply2,
-%     [{Topic, MaxAge1, CT, Payload, TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT),
-%     ?assertEqual(false, TimeStamp =:= timeout),
-
-%     %% Publish a message
-%     emqx:publish(emqx_message:make(Topic, Payload)),
-
-%     Notif = receive_notification(),
-%     ?LOGT("observer get Notif=~p", [Notif]),
-%     {coap_notify, _, _, {ok,content}, #coap_content{payload = Payload}} = Notif,
-
-%     er_coap_observer:stop(Pid),
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case03_subscribe(_Config) ->
-%     %% Subscribe to the unexisted topic "a/b", got not_found
-%     Topic = <<"a/b">>,
-%     TopicStr = binary_to_list(Topic),
-%     PercentEncodedTopic = emqx_http_lib:uri_encode(TopicStr),
-%     Uri = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     {error, not_found} = er_coap_observer:observe(Uri),
-
-%     [] = emqx:subscribers(Topic).
-
-% t_case04_subscribe(_Config) ->
-%     %% Subscribe to the wildcad topic "+/b", got bad_request
-%     Topic = <<"+/b">>,
-%     TopicStr = binary_to_list(Topic),
-%     PercentEncodedTopic = emqx_http_lib:uri_encode(TopicStr),
-%     Uri = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     {error, bad_request} = er_coap_observer:observe(Uri),
-
-%     [] = emqx:subscribers(Topic).
-
-% t_case01_read(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"PubPayload">>,
-%     timer:sleep(100),
-
-%     %% First post to create a topic "topic1"
-%     Uri = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, Uri, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = [LocPath]}} = Reply,
-%     ?assertEqual(<<"/ps/topic1">> ,LocPath),
-%     TopicInfo = [{Topic, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     %% GET to read the publish message of the topic
-%     timer:sleep(1000),
-%     Reply1 = er_coap_client:request(get, Uri),
-%     ?LOGT("Reply=~p", [Reply1]),
-%     {ok,content, #coap_content{payload = Payload}} = Reply1,
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, Uri).
-
-% t_case02_read(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"PubPayload">>,
-%     timer:sleep(100),
-
-%     %% First post to publish a topic "topic1"
-%     Uri = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, Uri, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = [LocPath]}} = Reply,
-%     ?assertEqual(<<"/ps/topic1">> ,LocPath),
-%     TopicInfo = [{Topic, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     %% GET to read the publish message of unmatched format, got bad_request
-%     Reply1 = er_coap_client:request(get, Uri, #coap_content{format = <<"application/json">>}),
-%     ?LOGT("Reply=~p", [Reply1]),
-%     {error, bad_request} = Reply1,
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, Uri).
-
-% t_case03_read(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Uri = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     timer:sleep(100),
-
-%     %% GET to read the nexisted topic "topic1", got not_found
-%     Reply = er_coap_client:request(get, Uri),
-%     ?LOGT("Reply=~p", [Reply]),
-%     {error, not_found} = Reply.
-
-% t_case04_read(_Config) ->
-%     Topic = <<"topic1">>,
-%     TopicStr = binary_to_list(Topic),
-%     Payload = <<"PubPayload">>,
-%     timer:sleep(100),
-
-%     %% First post to publish a topic "topic1"
-%     Uri = "coap://127.0.0.1/ps/"++TopicStr++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, Uri, #coap_content{format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = [LocPath]}} = Reply,
-%     ?assertEqual(<<"/ps/topic1">> ,LocPath),
-%     TopicInfo = [{Topic, MaxAge1, CT1, _ResPayload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?LOGT("lookup topic info=~p", [TopicInfo]),
-%     ?assertEqual(60, MaxAge1),
-%     ?assertEqual(<<"42">>, CT1),
-
-%     %% GET to read the publish message of wildcard topic, got bad_request
-%     WildTopic = binary_to_list(<<"+/topic1">>),
-%     Uri1 = "coap://127.0.0.1/ps/"++WildTopic++"?c=client1&u=tom&p=secret",
-%     Reply1 = er_coap_client:request(get, Uri1, #coap_content{format = <<"application/json">>}),
-%     ?LOGT("Reply=~p", [Reply1]),
-%     {error, bad_request} = Reply1,
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, Uri).
-
-% t_case05_read(_Config) ->
-%     Topic = <<"a/b">>,
-%     TopicStr = binary_to_list(Topic),
-%     PercentEncodedTopic = emqx_http_lib:uri_encode(TopicStr),
-%     Payload = <<"payload">>,
-
-%     %% post to publish a new topic "a/b", and the topic is created
-%     URI = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     Reply = er_coap_client:request(post, URI, #coap_content{max_age = 5, format = <<"application/octet-stream">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/a/b">>] ,LocPath),
-%     [{Topic, MaxAge, CT, Payload, _TimeStamp}] = emqx_coap_pubsub_topics:lookup_topic_info(Topic),
-%     ?assertEqual(5, MaxAge),
-%     ?assertEqual(<<"42">>, CT),
-
-%     %% Wait for the max age of the timer expires
-%     timer:sleep(6000),
-%     ?assertEqual(true, emqx_coap_pubsub_topics:is_topic_timeout(Topic)),
-
-%     %% GET to read the expired publish message, supposed to get {ok, nocontent}, but now got {ok, content}
-%     Reply1 = er_coap_client:request(get, URI),
-%     ?LOGT("Reply=~p", [Reply1]),
-%     {ok, content, #coap_content{payload = <<>>}}= Reply1,
-
-%     {ok, deleted, #coap_content{}} = er_coap_client:request(delete, URI).
-
-% t_case01_delete(_Config) ->
-%     TopicInPayload = <<"a/b">>,
-%     TopicStr = binary_to_list(TopicInPayload),
-%     PercentEncodedTopic = emqx_http_lib:uri_encode(TopicStr),
-%     Payload = list_to_binary("<"++PercentEncodedTopic++">;ct=42"),
-%     URI = "coap://127.0.0.1/ps/"++"?c=client1&u=tom&p=secret",
-
-%     %% Client post to CREATE topic "a/b"
-%     Reply = er_coap_client:request(post, URI, #coap_content{format = <<"application/link-format">>, payload = Payload}),
-%     ?LOGT("Reply =~p", [Reply]),
-%     {ok,created, #coap_content{location_path = LocPath}} = Reply,
-%     ?assertEqual([<<"/ps/a/b">>] ,LocPath),
-
-%     %% Client post to CREATE topic "a/b/c"
-%     TopicInPayload1 = <<"a/b/c">>,
-%     PercentEncodedTopic1 = emqx_http_lib:uri_encode(binary_to_list(TopicInPayload1)),
-%     Payload1 = list_to_binary("<"++PercentEncodedTopic1++">;ct=42"),
-%     Reply1 = er_coap_client:request(post, URI, #coap_content{format = <<"application/link-format">>, payload = Payload1}),
-%     ?LOGT("Reply =~p", [Reply1]),
-%     {ok,created, #coap_content{location_path = LocPath1}} = Reply1,
-%     ?assertEqual([<<"/ps/a/b/c">>] ,LocPath1),
-
-%     timer:sleep(50),
-
-%     %% DELETE the topic "a/b"
-%     UriD = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     ReplyD = er_coap_client:request(delete, UriD),
-%     ?LOGT("Reply=~p", [ReplyD]),
-%     {ok, deleted, #coap_content{}}= ReplyD,
-
-%     timer:sleep(300), %% Waiting gen_server:cast/2 for deleting operation
-%     ?assertEqual(false, emqx_coap_pubsub_topics:is_topic_existed(TopicInPayload)),
-%     ?assertEqual(false, emqx_coap_pubsub_topics:is_topic_existed(TopicInPayload1)).
-
-% t_case02_delete(_Config) ->
-%     TopicInPayload = <<"a/b">>,
-%     TopicStr = binary_to_list(TopicInPayload),
-%     PercentEncodedTopic = emqx_http_lib:uri_encode(TopicStr),
-
-%     %% DELETE the unexisted topic "a/b"
-%     Uri1 = "coap://127.0.0.1/ps/"++PercentEncodedTopic++"?c=client1&u=tom&p=secret",
-%     Reply1 = er_coap_client:request(delete, Uri1),
-%     ?LOGT("Reply=~p", [Reply1]),
-%     {error, not_found} = Reply1.
-
-% t_case13_emit_stats_test(_Config) ->
-%     ok.
-
-% %%--------------------------------------------------------------------
-% %% Internal functions
-
-% receive_notification() ->
-%     receive
-%         {coap_notify, Pid, N2, Code2, Content2} ->
-%             {coap_notify, Pid, N2, Code2, Content2}
-%     after 2000 ->
-%         receive_notification_timeout
-%     end.
-
-% assert_recv(Topic, Payload) ->
-%     receive
-%         {deliver, _, Msg} ->
-%             ?assertEqual(Topic, Msg#message.topic),
-%             ?assertEqual(Payload, Msg#message.payload)
-%     after
-%         500 ->
-%             ?assert(false)
-%     end.
-

apps/emqx_gateway/test/emqx_coap_SUITE.erl

@@ -0,0 +1,289 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-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.
+%%--------------------------------------------------------------------
+
+-module(emqx_coap_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("er_coap_client/include/coap.hrl").
+-include_lib("emqx/include/emqx.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+
+-define(CONF_DEFAULT, <<"
+gateway.coap
+{
+    idle_timeout = 30s
+    enable_stats = false
+    mountpoint = \"\"
+    notify_type = qos
+    connection_required = true
+    subscribe_qos = qos1
+    publish_qos = qos1
+    authentication = undefined
+
+    listeners.udp.default
+    {bind = 5683}
+}
+">>).
+
+-define(LOGT(Format, Args), ct:pal("TEST_SUITE: " ++ Format, Args)).
+-define(PS_PREFIX, "coap://127.0.0.1/ps").
+-define(MQTT_PREFIX, "coap://127.0.0.1/mqtt").
+
+
+all() -> emqx_ct:all(?MODULE).
+
+init_per_suite(Config) ->
+    emqx_ct_helpers:start_apps([emqx_gateway], fun set_special_cfg/1),
+    Config.
+
+set_special_cfg(emqx_gateway) ->
+    ok = emqx_config:init_load(emqx_gateway_schema, ?CONF_DEFAULT);
+
+set_special_cfg(_) ->
+    ok.
+
+end_per_suite(Config) ->
+    emqx_ct_helpers:stop_apps([emqx_gateway]),
+    Config.
+
+%%--------------------------------------------------------------------
+%% Test Cases
+%%--------------------------------------------------------------------
+t_connection(_Config) ->
+    Action = fun(Channel) ->
+                     %% connection
+                     Token = connection(Channel),
+
+                     timer:sleep(100),
+                     ?assertNotEqual([], emqx_gateway_cm_registry:lookup_channels(coap, <<"client1">>)),
+
+                     %% heartbeat
+                     HeartURI = ?MQTT_PREFIX ++ "/connection?clientid=client1&token=" ++ Token,
+                     ?LOGT("send heartbeat request:~s~n", [HeartURI]),
+                     {ok, changed, _} = er_coap_client:request(put, HeartURI),
+
+                     disconnection(Channel, Token),
+
+                     timer:sleep(100),
+                     ?assertEqual([], emqx_gateway_cm_registry:lookup_channels(coap, <<"client1">>))
+             end,
+    do(Action).
+
+
+t_publish(_Config) ->
+    Action = fun(Channel, Token) ->
+                     Topic = <<"/abc">>,
+                     Payload = <<"123">>,
+
+                     TopicStr = binary_to_list(Topic),
+                     URI = ?PS_PREFIX ++ TopicStr ++ "?clientid=client1&token=" ++ Token,
+
+                     %% Sub topic first
+                     emqx:subscribe(Topic),
+
+                     Req = make_req(post, Payload),
+                     {ok, changed, _} = do_request(Channel, URI, Req),
+
+                     receive
+                         {deliver, Topic, Msg} ->
+                             ?assertEqual(Topic, Msg#message.topic),
+                             ?assertEqual(Payload, Msg#message.payload)
+                     after
+                         500 ->
+                             ?assert(false)
+                     end
+             end,
+
+    with_connection(Action).
+
+
+t_publish_authz_deny(_Config) ->
+    Action = fun(Channel, Token) ->
+                     Topic = <<"/abc">>,
+                     Payload = <<"123">>,
+                     InvalidToken = lists:reverse(Token),
+
+                     TopicStr = binary_to_list(Topic),
+                     URI = ?PS_PREFIX ++ TopicStr ++ "?clientid=client1&token=" ++ InvalidToken,
+
+                     %% Sub topic first
+                     emqx:subscribe(Topic),
+
+                     Req = make_req(post, Payload),
+                     Result = do_request(Channel, URI, Req),
+                     ?assertEqual({error, reset}, Result)
+             end,
+
+    with_connection(Action).
+
+t_subscribe(_Config) ->
+    Topic = <<"/abc">>,
+    Fun = fun(Channel, Token) ->
+                  TopicStr = binary_to_list(Topic),
+                  Payload = <<"123">>,
+
+                  URI = ?PS_PREFIX ++ TopicStr ++ "?clientid=client1&token=" ++ Token,
+                  Req = make_req(get, Payload, [{observe, 0}]),
+                  {ok, content, _} = do_request(Channel, URI, Req),
+                  ?LOGT("observer topic:~s~n", [Topic]),
+
+                  timer:sleep(100),
+                  [SubPid] = emqx:subscribers(Topic),
+                  ?assert(is_pid(SubPid)),
+
+                  %% Publish a message
+                  emqx:publish(emqx_message:make(Topic, Payload)),
+                  {ok, content, Notify} = with_response(Channel),
+                  ?LOGT("observer get Notif=~p", [Notify]),
+
+                  #coap_content{payload = PayloadRecv} = Notify,
+
+                  ?assertEqual(Payload, PayloadRecv)
+          end,
+
+    with_connection(Fun),
+    timer:sleep(100),
+
+    ?assertEqual([], emqx:subscribers(Topic)).
+
+
+t_un_subscribe(_Config) ->
+    Topic = <<"/abc">>,
+    Fun = fun(Channel, Token) ->
+                  TopicStr = binary_to_list(Topic),
+                  Payload = <<"123">>,
+
+                  URI = ?PS_PREFIX ++ TopicStr ++ "?clientid=client1&token=" ++ Token,
+
+                  Req = make_req(get, Payload, [{observe, 0}]),
+                  {ok, content, _} = do_request(Channel, URI, Req),
+                  ?LOGT("observer topic:~s~n", [Topic]),
+
+                  timer:sleep(100),
+                  [SubPid] = emqx:subscribers(Topic),
+                  ?assert(is_pid(SubPid)),
+
+                  UnReq = make_req(get, Payload, [{observe, 1}]),
+                  {ok, nocontent, _} = do_request(Channel, URI, UnReq),
+                  ?LOGT("un observer topic:~s~n", [Topic]),
+                  timer:sleep(100),
+                  ?assertEqual([], emqx:subscribers(Topic))
+          end,
+
+    with_connection(Fun).
+
+t_observe_wildcard(_Config) ->
+    Fun = fun(Channel, Token) ->
+                  %% resolve_url can't process wildcard with #
+                  Topic = <<"/abc/+">>,
+                  TopicStr = binary_to_list(Topic),
+                  Payload = <<"123">>,
+
+                  URI = ?PS_PREFIX ++ TopicStr ++ "?clientid=client1&token=" ++ Token,
+                  Req = make_req(get, Payload, [{observe, 0}]),
+                  {ok, content, _} = do_request(Channel, URI, Req),
+                  ?LOGT("observer topic:~s~n", [Topic]),
+
+                  timer:sleep(100),
+                  [SubPid] = emqx:subscribers(Topic),
+                  ?assert(is_pid(SubPid)),
+
+                  %% Publish a message
+                  PubTopic = <<"/abc/def">>,
+                  emqx:publish(emqx_message:make(PubTopic, Payload)),
+                  {ok, content, Notify} = with_response(Channel),
+
+                  ?LOGT("observer get Notif=~p", [Notify]),
+
+                  #coap_content{payload = PayloadRecv} = Notify,
+
+                  ?assertEqual(Payload, PayloadRecv)
+          end,
+
+    with_connection(Fun).
+
+connection(Channel) ->
+    URI = ?MQTT_PREFIX ++ "/connection?clientid=client1&username=admin&password=public",
+    Req = make_req(post),
+    {ok, created, Data} = do_request(Channel, URI, Req),
+    #coap_content{payload = BinToken} = Data,
+    binary_to_list(BinToken).
+
+disconnection(Channel, Token) ->
+    %% delete
+    URI = ?MQTT_PREFIX ++ "/connection?clientid=client1&token=" ++ Token,
+    Req = make_req(delete),
+    {ok, deleted, _} = do_request(Channel, URI, Req).
+
+make_req(Method) ->
+    make_req(Method, <<>>).
+
+make_req(Method, Payload) ->
+    make_req(Method, Payload, []).
+
+make_req(Method, Payload, Opts) ->
+    er_coap_message:request(con, Method, Payload, Opts).
+
+do_request(Channel, URI, #coap_message{options = Opts} = Req) ->
+
+    {_, _, Path, Query} = er_coap_client:resolve_uri(URI),
+    Opts2 = [{uri_path, Path}, {uri_query, Query} | Opts],
+    Req2 = Req#coap_message{options = Opts2},
+    ?LOGT("send request:~s~nReq:~p~n", [URI, Req2]),
+
+    {ok, _} = er_coap_channel:send(Channel, Req2),
+    with_response(Channel).
+
+with_response(Channel) ->
+    receive
+        {coap_response, _ChId, Channel, _Ref, Message=#coap_message{method=Code}} ->
+            return_response(Code, Message);
+        {coap_error, _ChId, Channel, _Ref, reset} ->
+            {error, reset}
+    after 2000 ->
+            {error, timeout}
+    end.
+
+return_response({ok, Code}, Message) ->
+    {ok, Code, er_coap_message:get_content(Message)};
+return_response({error, Code}, #coap_message{payload= <<>>}) ->
+    {error, Code};
+return_response({error, Code}, Message) ->
+    {error, Code, er_coap_message:get_content(Message)}.
+
+do(Fun) ->
+    ChId = {{127, 0, 0, 1}, 5683},
+    {ok, Sock} = er_coap_udp_socket:start_link(),
+    {ok, Channel} = er_coap_udp_socket:get_channel(Sock, ChId),
+    %% send and receive
+    Res = Fun(Channel),
+    %% terminate the processes
+    er_coap_channel:close(Channel),
+    er_coap_udp_socket:close(Sock),
+    Res.
+
+with_connection(Action) ->
+    Fun = fun(Channel) ->
+                  Token = connection(Channel),
+                  timer:sleep(100),
+                  Action(Channel, Token),
+                  disconnection(Channel, Token),
+                  timer:sleep(100)
+          end,
+    do(Fun).