diff --git a/apps/emqx/src/emqx_schema.erl b/apps/emqx/src/emqx_schema.erl index e319dbe15..3e3318475 100644 --- a/apps/emqx/src/emqx_schema.erl +++ b/apps/emqx/src/emqx_schema.erl @@ -75,6 +75,8 @@ -export([server_ssl_opts_schema/2, client_ssl_opts_schema/1, ciphers_schema/1, default_ciphers/1]). -export([sc/2, map/2]). +-elvis([{elvis_style, god_modules, disable}]). + namespace() -> undefined. roots() -> @@ -898,7 +900,8 @@ fields("alarm") -> Currently supports two actions, 'log' and 'publish'. 'log' is to write the alarm to log (console or file). 'publish' is to publish the alarm as an MQTT message to the system topics: -$SYS/brokers/emqx@xx.xx.xx.x/alarms/activate and $SYS/brokers/emqx@xx.xx.xx.x/alarms/deactivate""" +$SYS/brokers/emqx@xx.xx.xx.x/alarms/activate and +$SYS/brokers/emqx@xx.xx.xx.x/alarms/deactivate""" }) } , {"size_limit", @@ -916,8 +919,9 @@ When this limit is exceeded, the oldest deactivated alarms are deleted to cap th #{ default => "24h", example => "24h", desc => - """Retention time of deactivated alarms. Alarms are not deleted immediately when deactivated, but after the retention time. - """ +"""Retention time of deactivated alarms. Alarms are not deleted immediately +when deactivated, but after the retention time. +""" }) } ]. @@ -1181,7 +1185,8 @@ default_tls_vsns(dtls_all_available) -> default_tls_vsns(tls_all_available) -> emqx_tls_lib:default_versions(). --spec ciphers_schema(quic | dtls_all_available | tls_all_available | undefined) -> hocon_schema:field_schema(). +-spec ciphers_schema(quic | dtls_all_available | tls_all_available | undefined) + -> hocon_schema:field_schema(). ciphers_schema(Default) -> sc(hoconsc:array(string()), #{ default => default_ciphers(Default) diff --git a/apps/emqx_authz/test/emqx_authz_SUITE.erl b/apps/emqx_authz/test/emqx_authz_SUITE.erl index 5f0371c49..df4c904c3 100644 --- a/apps/emqx_authz/test/emqx_authz_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_SUITE.erl @@ -149,7 +149,9 @@ t_update_source(_) -> {ok, _} = emqx_authz:update(?CMD_REPLACE, []). t_move_source(_) -> - {ok, _} = emqx_authz:update(?CMD_REPLACE, [?SOURCE1, ?SOURCE2, ?SOURCE3, ?SOURCE4, ?SOURCE5, ?SOURCE6]), + {ok, _} = emqx_authz:update(?CMD_REPLACE, + [?SOURCE1, ?SOURCE2, ?SOURCE3, + ?SOURCE4, ?SOURCE5, ?SOURCE6]), ?assertMatch([ #{type := http} , #{type := mongodb} , #{type := mysql} diff --git a/apps/emqx_authz/test/emqx_authz_api_mnesia_SUITE.erl b/apps/emqx_authz/test/emqx_authz_api_mnesia_SUITE.erl index 1613df2d1..737a089a2 100644 --- a/apps/emqx_authz/test/emqx_authz_api_mnesia_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_api_mnesia_SUITE.erl @@ -26,7 +26,6 @@ -define(API_VERSION, "v5"). -define(BASE_PATH, "api"). - all() -> emqx_common_test_helpers:all(?MODULE). @@ -82,7 +81,9 @@ t_api(_) -> , uri(["authorization", "sources", "built-in-database", "username", "user1"]) , []), #{<<"data">> := [#{<<"username">> := <<"user1">>, <<"rules">> := Rules1}], - <<"meta">> := #{<<"count">> := 1,<<"limit">> := 100,<<"page">> := 1}} = jsx:decode(Request1), + <<"meta">> := #{<<"count">> := 1, + <<"limit">> := 100, + <<"page">> := 1}} = jsx:decode(Request1), #{<<"username">> := <<"user1">>, <<"rules">> := Rules1} = jsx:decode(Request2), ?assertEqual(3, length(Rules1)), diff --git a/apps/emqx_authz/test/emqx_authz_api_settings_SUITE.erl b/apps/emqx_authz/test/emqx_authz_api_settings_SUITE.erl index f8d83eee9..5ad8f1f29 100644 --- a/apps/emqx_authz/test/emqx_authz_api_settings_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_api_settings_SUITE.erl @@ -36,11 +36,11 @@ init_per_suite(Config) -> ok = emqx_common_test_helpers:start_apps( [emqx_conf, emqx_authz, emqx_dashboard], fun set_special_configs/1), - Config. end_per_suite(_Config) -> - emqx_common_test_helpers:stop_apps([emqx_authz, emqx_dashboard]), + {ok, _} = emqx_authz:update(replace, []), + emqx_common_test_helpers:stop_apps([emqx_dashboard, emqx_authz, emqx_conf]), ok. set_special_configs(emqx_dashboard) -> diff --git a/apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl b/apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl index d5a245819..4fbbed84d 100644 --- a/apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_api_sources_SUITE.erl @@ -81,7 +81,9 @@ }). -define(SOURCE6, #{<<"type">> => <<"file">>, <<"enable">> => true, - <<"rules">> => <<"{allow,{username,\"^dashboard?\"},subscribe,[\"$SYS/#\"]}.\n{allow,{ipaddr,\"127.0.0.1\"},all,[\"$SYS/#\",\"#\"]}.">> + <<"rules">> => +<<"{allow,{username,\"^dashboard?\"},subscribe,[\"$SYS/#\"]}." + "\n{allow,{ipaddr,\"127.0.0.1\"},all,[\"$SYS/#\",\"#\"]}.">> }). all() -> @@ -94,9 +96,10 @@ groups() -> init_per_suite(Config) -> meck:new(emqx_resource, [non_strict, passthrough, no_history, no_link]), meck:expect(emqx_resource, create, fun(_, _, _) -> {ok, meck_data} end), - meck:expect(emqx_resource, create_dry_run, fun(emqx_connector_mysql, _) -> {ok, meck_data}; - (T, C) -> meck:passthrough([T, C]) - end), + meck:expect(emqx_resource, create_dry_run, + fun(emqx_connector_mysql, _) -> {ok, meck_data}; + (T, C) -> meck:passthrough([T, C]) + end), meck:expect(emqx_resource, update, fun(_, _, _, _) -> {ok, meck_data} end), meck:expect(emqx_resource, health_check, fun(_) -> ok end), meck:expect(emqx_resource, remove, fun(_) -> ok end ), @@ -136,10 +139,11 @@ init_per_testcase(t_api, Config) -> meck:expect(emqx_misc, gen_id, fun() -> "fake" end), meck:new(emqx, [non_strict, passthrough, no_history, no_link]), - meck:expect(emqx, data_dir, fun() -> - {data_dir, Data} = lists:keyfind(data_dir, 1, Config), - Data - end), + meck:expect(emqx, data_dir, + fun() -> + {data_dir, Data} = lists:keyfind(data_dir, 1, Config), + Data + end), Config; init_per_testcase(_, Config) -> Config. @@ -157,7 +161,8 @@ t_api(_) -> {ok, 200, Result1} = request(get, uri(["authorization", "sources"]), []), ?assertEqual([], get_sources(Result1)), - {ok, 204, _} = request(put, uri(["authorization", "sources"]), [?SOURCE2, ?SOURCE3, ?SOURCE4, ?SOURCE5, ?SOURCE6]), + {ok, 204, _} = request(put, uri(["authorization", "sources"]), + [?SOURCE2, ?SOURCE3, ?SOURCE4, ?SOURCE5, ?SOURCE6]), {ok, 204, _} = request(post, uri(["authorization", "sources"]), ?SOURCE1), {ok, 200, Result2} = request(get, uri(["authorization", "sources"]), []), @@ -171,7 +176,8 @@ t_api(_) -> ], Sources), ?assert(filelib:is_file(emqx_authz:acl_conf_file())), - {ok, 204, _} = request(put, uri(["authorization", "sources", "http"]), ?SOURCE1#{<<"enable">> := false}), + {ok, 204, _} = request(put, uri(["authorization", "sources", "http"]), + ?SOURCE1#{<<"enable">> := false}), {ok, 200, Result3} = request(get, uri(["authorization", "sources", "http"]), []), ?assertMatch(#{<<"type">> := <<"http">>, <<"enable">> := false}, jsx:decode(Result3)), @@ -196,14 +202,28 @@ t_api(_) -> ?assert(filelib:is_file(filename:join([data_dir(), "certs", "cert-fake.pem"]))), ?assert(filelib:is_file(filename:join([data_dir(), "certs", "key-fake.pem"]))), - {ok, 204, _} = request(put, uri(["authorization", "sources", "mysql"]), ?SOURCE3#{<<"server">> := <<"192.168.1.100:3306">>}), + {ok, 204, _} = request( + put, + uri(["authorization", "sources", "mysql"]), + ?SOURCE3#{<<"server">> := <<"192.168.1.100:3306">>}), - {ok, 400, _} = request(put, uri(["authorization", "sources", "postgresql"]), ?SOURCE4#{<<"server">> := <<"fake">>}), - {ok, 400, _} = request(put, uri(["authorization", "sources", "redis"]), ?SOURCE5#{<<"servers">> := [<<"192.168.1.100:6379">>, <<"192.168.1.100:6380">>]}), + {ok, 400, _} = request( + put, + uri(["authorization", "sources", "postgresql"]), + ?SOURCE4#{<<"server">> := <<"fake">>}), + {ok, 400, _} = request( + put, + uri(["authorization", "sources", "redis"]), + ?SOURCE5#{<<"servers">> := [<<"192.168.1.100:6379">>, + <<"192.168.1.100:6380">>]}), - lists:foreach(fun(#{<<"type">> := Type}) -> - {ok, 204, _} = request(delete, uri(["authorization", "sources", binary_to_list(Type)]), []) - end, Sources), + lists:foreach( + fun(#{<<"type">> := Type}) -> + {ok, 204, _} = request( + delete, + uri(["authorization", "sources", binary_to_list(Type)]), + []) + end, Sources), {ok, 200, Result5} = request(get, uri(["authorization", "sources"]), []), ?assertEqual([], get_sources(Result5)), ?assertEqual([], emqx:get_config([authorization, sources])), diff --git a/apps/emqx_authz/test/emqx_authz_mnesia_SUITE.erl b/apps/emqx_authz/test/emqx_authz_mnesia_SUITE.erl index 70e7b6608..875ff3d09 100644 --- a/apps/emqx_authz/test/emqx_authz_mnesia_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_mnesia_SUITE.erl @@ -92,13 +92,19 @@ t_authz(_) -> listener => {tcp, default} }, - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, subscribe, <<"#">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, publish, <<"#">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo1, subscribe, <<"#">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo1, publish, <<"#">>)), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_username">>)), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, subscribe, <<"#">>)), + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_username">>)), + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"#">>)), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo3, publish, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, subscribe, <<"#">>)), + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo3, publish, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, subscribe, <<"#">>)), ok. diff --git a/apps/emqx_authz/test/emqx_authz_mongodb_SUITE.erl b/apps/emqx_authz/test/emqx_authz_mongodb_SUITE.erl index 682a193a7..2e3edd42a 100644 --- a/apps/emqx_authz/test/emqx_authz_mongodb_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_mongodb_SUITE.erl @@ -114,11 +114,17 @@ t_authz(_) -> ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, subscribe, <<"+">>)), meck:expect(emqx_resource, query, fun(_, _) -> ?SOURCE3 ++ ?SOURCE4 end), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, subscribe, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, subscribe, <<"test/test_username">>)), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_username">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, subscribe, <<"test">>)), % nomatch - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, publish, <<"test">>)), % nomatch + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"test/test_username">>)), + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_username">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, subscribe, <<"test">>)), % nomatch + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, publish, <<"test">>)), % nomatch ok. diff --git a/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl b/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl index 1db5374d3..a2d264fc9 100644 --- a/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl @@ -112,11 +112,17 @@ t_authz(_) -> ?assertEqual(deny, emqx_access_control:authorize(ClientInfo1, subscribe, <<"+">>)), meck:expect(emqx_resource, query, fun(_, _) -> {ok, ?COLUMNS, ?SOURCE3 ++ ?SOURCE4} end), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, subscribe, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, subscribe, <<"test/test_username">>)), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_username">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, subscribe, <<"test">>)), % nomatch - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, publish, <<"test">>)), % nomatch + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"test/test_username">>)), + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_username">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, subscribe, <<"test">>)), % nomatch + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, publish, <<"test">>)), % nomatch ok. diff --git a/apps/emqx_authz/test/emqx_authz_postgresql_SUITE.erl b/apps/emqx_authz/test/emqx_authz_postgresql_SUITE.erl index 0f88b6440..233701b34 100644 --- a/apps/emqx_authz/test/emqx_authz_postgresql_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_postgresql_SUITE.erl @@ -110,11 +110,17 @@ t_authz(_) -> ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, subscribe, <<"+">>)), meck:expect(emqx_resource, query, fun(_, _) -> {ok, ?COLUMNS, ?SOURCE3 ++ ?SOURCE4} end), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, subscribe, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_clientid">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo2, subscribe, <<"test/test_username">>)), - ?assertEqual(allow, emqx_access_control:authorize(ClientInfo2, publish, <<"test/test_username">>)), - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, subscribe, <<"test">>)), % nomatch - ?assertEqual(deny, emqx_access_control:authorize(ClientInfo3, publish, <<"test">>)), % nomatch + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_clientid">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo2, subscribe, <<"test/test_username">>)), + ?assertEqual(allow, emqx_access_control:authorize( + ClientInfo2, publish, <<"test/test_username">>)), + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, subscribe, <<"test">>)), % nomatch + ?assertEqual(deny, emqx_access_control:authorize( + ClientInfo3, publish, <<"test">>)), % nomatch ok. diff --git a/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl b/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl index 601958609..630d42dce 100644 --- a/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl +++ b/apps/emqx_authz/test/emqx_authz_rule_SUITE.erl @@ -26,7 +26,10 @@ -define(SOURCE2, {allow, {ipaddr, "127.0.0.1"}, all, [{eq, "#"}, {eq, "+"}]}). -define(SOURCE3, {allow, {ipaddrs, ["127.0.0.1", "192.168.1.0/24"]}, subscribe, ["%c"]}). -define(SOURCE4, {allow, {'and', [{client, "test"}, {user, "test"}]}, publish, ["topic/test"]}). --define(SOURCE5, {allow, {'or', [{username, {re, "^test"}}, {clientid, {re, "test?"}}]}, publish, ["%u", "%c"]}). +-define(SOURCE5, {allow, {'or', + [{username, {re, "^test"}}, + {clientid, {re, "test?"}}]}, + publish, ["%u", "%c"]}). all() -> emqx_common_test_helpers:all(?MODULE). @@ -53,7 +56,8 @@ set_special_configs(_App) -> t_compile(_) -> ?assertEqual({deny, all, all, [['#']]}, emqx_authz_rule:compile(?SOURCE1)), - ?assertEqual({allow, {ipaddr, {{127,0,0,1}, {127,0,0,1}, 32}}, all, [{eq, ['#']}, {eq, ['+']}]}, emqx_authz_rule:compile(?SOURCE2)), + ?assertEqual({allow, {ipaddr, {{127,0,0,1}, {127,0,0,1}, 32}}, + all, [{eq, ['#']}, {eq, ['+']}]}, emqx_authz_rule:compile(?SOURCE2)), ?assertEqual({allow, {ipaddrs,[{{127,0,0,1},{127,0,0,1},32}, @@ -69,9 +73,9 @@ t_compile(_) -> }, emqx_authz_rule:compile(?SOURCE4)), ?assertMatch({allow, - {'or', [{username, {re_pattern, _, _, _, _}}, {clientid, {re_pattern, _, _, _, _}}]}, - publish, - [{pattern, [<<"%u">>]}, {pattern, [<<"%c">>]}] + {'or', [{username, {re_pattern, _, _, _, _}}, + {clientid, {re_pattern, _, _, _, _}}]}, + publish, [{pattern, [<<"%u">>]}, {pattern, [<<"%c">>]}] }, emqx_authz_rule:compile(?SOURCE5)), ok. @@ -103,47 +107,64 @@ t_match(_) -> }, ?assertEqual({matched, deny}, - emqx_authz_rule:match(ClientInfo1, subscribe, <<"#">>, emqx_authz_rule:compile(?SOURCE1))), + emqx_authz_rule:match(ClientInfo1, subscribe, <<"#">>, + emqx_authz_rule:compile(?SOURCE1))), ?assertEqual({matched, deny}, - emqx_authz_rule:match(ClientInfo2, subscribe, <<"+">>, emqx_authz_rule:compile(?SOURCE1))), + emqx_authz_rule:match(ClientInfo2, subscribe, <<"+">>, + emqx_authz_rule:compile(?SOURCE1))), ?assertEqual({matched, deny}, - emqx_authz_rule:match(ClientInfo3, subscribe, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE1))), + emqx_authz_rule:match(ClientInfo3, subscribe, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE1))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo1, subscribe, <<"#">>, emqx_authz_rule:compile(?SOURCE2))), + emqx_authz_rule:match(ClientInfo1, subscribe, <<"#">>, + emqx_authz_rule:compile(?SOURCE2))), ?assertEqual(nomatch, - emqx_authz_rule:match(ClientInfo1, subscribe, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE2))), + emqx_authz_rule:match(ClientInfo1, subscribe, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE2))), ?assertEqual(nomatch, - emqx_authz_rule:match(ClientInfo2, subscribe, <<"#">>, emqx_authz_rule:compile(?SOURCE2))), + emqx_authz_rule:match(ClientInfo2, subscribe, <<"#">>, + emqx_authz_rule:compile(?SOURCE2))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo1, subscribe, <<"test">>, emqx_authz_rule:compile(?SOURCE3))), + emqx_authz_rule:match(ClientInfo1, subscribe, <<"test">>, + emqx_authz_rule:compile(?SOURCE3))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo2, subscribe, <<"test">>, emqx_authz_rule:compile(?SOURCE3))), + emqx_authz_rule:match(ClientInfo2, subscribe, <<"test">>, + emqx_authz_rule:compile(?SOURCE3))), ?assertEqual(nomatch, - emqx_authz_rule:match(ClientInfo2, subscribe, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE3))), + emqx_authz_rule:match(ClientInfo2, subscribe, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE3))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo1, publish, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE4))), + emqx_authz_rule:match(ClientInfo1, publish, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE4))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo2, publish, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE4))), + emqx_authz_rule:match(ClientInfo2, publish, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE4))), ?assertEqual(nomatch, - emqx_authz_rule:match(ClientInfo3, publish, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE4))), + emqx_authz_rule:match(ClientInfo3, publish, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE4))), ?assertEqual(nomatch, - emqx_authz_rule:match(ClientInfo4, publish, <<"topic/test">>, emqx_authz_rule:compile(?SOURCE4))), + emqx_authz_rule:match(ClientInfo4, publish, <<"topic/test">>, + emqx_authz_rule:compile(?SOURCE4))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo1, publish, <<"test">>, emqx_authz_rule:compile(?SOURCE5))), + emqx_authz_rule:match(ClientInfo1, publish, <<"test">>, + emqx_authz_rule:compile(?SOURCE5))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo2, publish, <<"test">>, emqx_authz_rule:compile(?SOURCE5))), + emqx_authz_rule:match(ClientInfo2, publish, <<"test">>, + emqx_authz_rule:compile(?SOURCE5))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo3, publish, <<"test">>, emqx_authz_rule:compile(?SOURCE5))), + emqx_authz_rule:match(ClientInfo3, publish, <<"test">>, + emqx_authz_rule:compile(?SOURCE5))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo3, publish, <<"fake">>, emqx_authz_rule:compile(?SOURCE5))), + emqx_authz_rule:match(ClientInfo3, publish, <<"fake">>, + emqx_authz_rule:compile(?SOURCE5))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo4, publish, <<"test">>, emqx_authz_rule:compile(?SOURCE5))), + emqx_authz_rule:match(ClientInfo4, publish, <<"test">>, + emqx_authz_rule:compile(?SOURCE5))), ?assertEqual({matched, allow}, - emqx_authz_rule:match(ClientInfo4, publish, <<"fake">>, emqx_authz_rule:compile(?SOURCE5))), - + emqx_authz_rule:match(ClientInfo4, publish, <<"fake">>, + emqx_authz_rule:compile(?SOURCE5))), ok. - diff --git a/apps/emqx_conf/src/emqx_conf_schema.erl b/apps/emqx_conf/src/emqx_conf_schema.erl index a187bc58c..ba3ac31b6 100644 --- a/apps/emqx_conf/src/emqx_conf_schema.erl +++ b/apps/emqx_conf/src/emqx_conf_schema.erl @@ -341,8 +341,9 @@ fields("cluster_call") -> })} , {"cleanup_interval", sc(emqx_schema:duration(), - #{ desc => "Time interval to clear completed but stale transactions. - Ensure that the number of completed transactions is less than the max_history." + #{ desc => +"Time interval to clear completed but stale transactions. +Ensure that the number of completed transactions is less than the max_history." , default => "5m" })} ]; @@ -505,7 +506,7 @@ fields("authorization") -> translations() -> ["ekka", "kernel", "emqx"]. translation("ekka") -> - [ {"cluster_discovery", fun tr_cluster__discovery/1}]; + [ {"cluster_discovery", fun tr_cluster_discovery/1}]; translation("kernel") -> [ {"logger_level", fun tr_logger_level/1} , {"logger", fun tr_logger/1}]; @@ -540,7 +541,7 @@ tr_override_conf_file(Conf, Filename) -> [_ | _] = DataDir, filename:join([DataDir, "configs", Filename]). -tr_cluster__discovery(Conf) -> +tr_cluster_discovery(Conf) -> Strategy = conf_get("cluster.discovery_strategy", Conf), {Strategy, filter(options(Strategy, Conf))}. diff --git a/apps/emqx_gateway/src/emqx_gateway_schema.erl b/apps/emqx_gateway/src/emqx_gateway_schema.erl index 71fac2743..9f35225b7 100644 --- a/apps/emqx_gateway/src/emqx_gateway_schema.erl +++ b/apps/emqx_gateway/src/emqx_gateway_schema.erl @@ -45,6 +45,7 @@ , comma_separated_list/0 , ip_port/0 ]). +-elvis([{elvis_style, dont_repeat_yourself, disable}]). -export([namespace/0, roots/0 , fields/1]). @@ -324,7 +325,8 @@ the LwM2M client" sc(ref(translator), #{ desc => "The topic for gateway to publish the notify events from LwM2M client.
-After succeed observe a resource of LwM2M client, Gateway will send the notifyevents via this topic, if the client reports any resource changes" +After succeed observe a resource of LwM2M client, Gateway will send the +notifyevents via this topic, if the client reports any resource changes" })} , {register, sc(ref(translator), diff --git a/apps/emqx_gateway/test/emqx_lwm2m_SUITE.erl b/apps/emqx_gateway/test/emqx_lwm2m_SUITE.erl index 49079d579..c3331eccd 100644 --- a/apps/emqx_gateway/test/emqx_lwm2m_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_lwm2m_SUITE.erl @@ -80,7 +80,8 @@ groups() -> [ case01_register, case01_register_additional_opts, - %% case01_register_incorrect_opts, %% TODO now we can't handle partial decode packet + %% TODO now we can't handle partial decode packet + %% case01_register_incorrect_opts, case01_register_report, case02_update_deregister, case03_register_wrong_version, @@ -164,7 +165,9 @@ init_per_testcase(_AllTestCase, Config) -> {ok, _} = application:ensure_all_started(emqx_gateway), {ok, ClientUdpSock} = gen_udp:open(0, [binary, {active, false}]), - {ok, C} = emqtt:start_link([{host, "localhost"},{port, 1883},{clientid, <<"c1">>}]), + {ok, C} = emqtt:start_link([{host, "localhost"}, + {port, 1883}, + {clientid, <<"c1">>}]), {ok, _} = emqtt:connect(C), timer:sleep(100), @@ -189,12 +192,14 @@ case01_register(Config) -> MsgId = 12, SubTopic = list_to_binary("lwm2m/"++Epn++"/dn/#"), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), %% checkpoint 1 - response #coap_message{type = Type, method = Method, id = RspId, options = Opts} = @@ -215,13 +220,16 @@ case01_register(Config) -> %%---------------------------------------- ?LOGT("start to send DE-REGISTER command", []), MsgId3 = 52, - test_send_coap_request( UdpSock, - delete, - sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), - #coap_content{payload = <<>>}, - [], - MsgId3), - #coap_message{type = ack, id = RspId3, method = Method3} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + delete, + sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), + #coap_content{payload = <<>>}, + [], + MsgId3), + #coap_message{type = ack, + id = RspId3, + method = Method3} = test_recv_coap_response(UdpSock), {ok,deleted} = Method3, MsgId3 = RspId3, timer:sleep(50), @@ -236,13 +244,16 @@ case01_register_additional_opts(Config) -> MsgId = 12, SubTopic = list_to_binary("lwm2m/"++Epn++"/dn/#"), - AddOpts = "ep=~ts<=345&lwm2m=1&apn=psmA.eDRX0.ctnb&cust_opt=shawn&im=123&ct=1.4&mt=mdm9620&mv=1.2", - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?" ++ AddOpts, [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + AddOpts = "ep=~ts<=345&lwm2m=1&apn=psmA.eDRX0.ctnb&cust_opt=shawn&" + "im=123&ct=1.4&mt=mdm9620&mv=1.2", + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?" ++ AddOpts, [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), %% checkpoint 1 - response #coap_message{type = Type, method = Method, id = RspId, options = Opts} = @@ -263,13 +274,16 @@ case01_register_additional_opts(Config) -> %%---------------------------------------- ?LOGT("start to send DE-REGISTER command", []), MsgId3 = 52, - test_send_coap_request( UdpSock, - delete, - sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), - #coap_content{payload = <<>>}, - [], - MsgId3), - #coap_message{type = ack, id = RspId3, method = Method3} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + delete, + sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), + #coap_content{payload = <<>>}, + [], + MsgId3), + #coap_message{type = ack, + id = RspId3, + method = Method3} = test_recv_coap_response(UdpSock), {ok,deleted} = Method3, MsgId3 = RspId3, timer:sleep(50), @@ -285,12 +299,14 @@ case01_register_incorrect_opts(Config) -> AddOpts = "ep=~ts<=345&lwm2m=1&incorrect_opt", - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?" ++ AddOpts, [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?" ++ AddOpts, [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), %% checkpoint 1 - response #coap_message{type = ack, method = Method, id = MsgId} = @@ -309,12 +325,14 @@ case01_register_report(Config) -> emqtt:subscribe(?config(emqx_c, Config), ReportTopic, qos0), timer:sleep(200), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), #coap_message{type = Type, method = Method, id = RspId, options = Opts} = test_recv_coap_response(UdpSock), @@ -327,16 +345,16 @@ case01_register_report(Config) -> timer:sleep(50), true = lists:member(SubTopic, test_mqtt_broker:get_subscrbied_topics()), - ReadResult = emqx_json:encode(#{ - <<"msgType">> => <<"register">>, - <<"data">> => #{ - <<"alternatePath">> => <<"/">>, - <<"ep">> => list_to_binary(Epn), - <<"lt">> => 345, - <<"lwm2m">> => <<"1">>, - <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, <<"/4">>, <<"/5">>] - } - }), + ReadResult = emqx_json:encode( + #{<<"msgType">> => <<"register">>, + <<"data">> => #{ + <<"alternatePath">> => <<"/">>, + <<"ep">> => list_to_binary(Epn), + <<"lt">> => 345, + <<"lwm2m">> => <<"1">>, + <<"objectList">> => [<<"/1">>, <<"/2">>, + <<"/3">>, <<"/4">>, <<"/5">>] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(ReportTopic)), %%---------------------------------------- @@ -344,13 +362,16 @@ case01_register_report(Config) -> %%---------------------------------------- ?LOGT("start to send DE-REGISTER command", []), MsgId3 = 52, - test_send_coap_request( UdpSock, - delete, - sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), - #coap_content{payload = <<>>}, - [], - MsgId3), - #coap_message{type = ack, id = RspId3, method = Method3} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + delete, + sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), + #coap_content{payload = <<>>}, + [], + MsgId3), + #coap_message{type = ack, + id = RspId3, + method = Method3} = test_recv_coap_response(UdpSock), {ok,deleted} = Method3, MsgId3 = RspId3, timer:sleep(50), @@ -368,28 +389,32 @@ case02_update_deregister(Config) -> emqtt:subscribe(?config(emqx_c, Config), ReportTopic, qos0), timer:sleep(200), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), timer:sleep(100), - #coap_message{type = ack, method = Method, options = Opts} = test_recv_coap_response(UdpSock), + #coap_message{type = ack, + method = Method, + options = Opts} = test_recv_coap_response(UdpSock), ?assertEqual({ok,created}, Method), ?LOGT("Options got: ~p", [Opts]), Location = maps:get(location_path, Opts), - Register = emqx_json:encode(#{ - <<"msgType">> => <<"register">>, - <<"data">> => #{ - <<"alternatePath">> => <<"/">>, - <<"ep">> => list_to_binary(Epn), - <<"lt">> => 345, - <<"lwm2m">> => <<"1">>, - <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, <<"/4">>, <<"/5">>] - } - }), + Register = emqx_json:encode( + #{<<"msgType">> => <<"register">>, + <<"data">> => #{ + <<"alternatePath">> => <<"/">>, + <<"ep">> => list_to_binary(Epn), + <<"lt">> => 345, + <<"lwm2m">> => <<"1">>, + <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, + <<"/4">>, <<"/5">>] + }}), ?assertEqual(Register, test_recv_mqtt_response(ReportTopic)), %%---------------------------------------- @@ -397,25 +422,29 @@ case02_update_deregister(Config) -> %%---------------------------------------- ?LOGT("start to send UPDATE command", []), MsgId2 = 27, - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b~ts?lt=789", [?PORT, join_path(Location, <<>>)]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , , ">>}, - [], - MsgId2), - #coap_message{type = ack, id = RspId2, method = Method2} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b~ts?lt=789", [?PORT, join_path(Location, <<>>)]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , , ">>}, + [], + MsgId2), + #coap_message{type = ack, + id = RspId2, + method = Method2} = test_recv_coap_response(UdpSock), {ok,changed} = Method2, MsgId2 = RspId2, - Update = emqx_json:encode(#{ - <<"msgType">> => <<"update">>, - <<"data">> => #{ - <<"alternatePath">> => <<"/">>, - <<"ep">> => list_to_binary(Epn), - <<"lt">> => 789, - <<"lwm2m">> => <<"1">>, - <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, <<"/4">>, <<"/5">>, <<"/6">>] - } - }), + Update = emqx_json:encode( + #{<<"msgType">> => <<"update">>, + <<"data">> => #{ + <<"alternatePath">> => <<"/">>, + <<"ep">> => list_to_binary(Epn), + <<"lt">> => 789, + <<"lwm2m">> => <<"1">>, + <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, + <<"/4">>, <<"/5">>, <<"/6">>] + }}), ?assertEqual(Update, test_recv_mqtt_response(ReportTopic)), %%---------------------------------------- @@ -423,13 +452,16 @@ case02_update_deregister(Config) -> %%---------------------------------------- ?LOGT("start to send DE-REGISTER command", []), MsgId3 = 52, - test_send_coap_request( UdpSock, - delete, - sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), - #coap_content{payload = <<>>}, - [], - MsgId3), - #coap_message{type = ack, id = RspId3, method = Method3} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + delete, + sprintf("coap://127.0.0.1:~b~ts", [?PORT, join_path(Location, <<>>)]), + #coap_content{payload = <<>>}, + [], + MsgId3), + #coap_message{type = ack, + id = RspId3, + method = Method3} = test_recv_coap_response(UdpSock), {ok,deleted} = Method3, MsgId3 = RspId3, @@ -444,12 +476,14 @@ case03_register_wrong_version(Config) -> Epn = "urn:oma:lwm2m:oma:3", MsgId = 12, SubTopic = list_to_binary("lwm2m/"++Epn++"/dn/#"), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=8.3", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=8.3", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), #coap_message{type = ack, method = Method} = test_recv_coap_response(UdpSock), ?assertEqual({error, bad_request}, Method), timer:sleep(50), @@ -465,12 +499,14 @@ case04_register_and_lifetime_timeout(Config) -> MsgId = 12, SubTopic = list_to_binary("lwm2m/"++Epn++"/dn/#"), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=2&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=2&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), timer:sleep(100), #coap_message{type = ack, method = Method} = test_recv_coap_response(UdpSock), ?assertEqual({ok,created}, Method), @@ -491,12 +527,14 @@ case05_register_wrong_epn(Config) -> MsgId = 12, UdpSock = ?config(sock, Config), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?lt=345&lwm2m=1.0", [?PORT]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?lt=345&lwm2m=1.0", [?PORT]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId), #coap_message{type = ack, method = Method} = test_recv_coap_response(UdpSock), ?assertEqual({error,bad_request}, Method). @@ -508,13 +546,16 @@ case05_register_wrong_epn(Config) -> %% Epn = "urn:oma:lwm2m:oma:3", %% MsgId = 12, -%% test_send_coap_request( UdpSock, -%% post, -%% sprintf("coap://127.0.0.1:~b/rd?ep=~ts&lwm2m=1", [?PORT, Epn]), -%% #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, -%% [], -%% MsgId), -%% #coap_message{type = ack, method = Method} = test_recv_coap_response(UdpSock), +%% test_send_coap_request( +%% UdpSock, +%% post, +%% sprintf("coap://127.0.0.1:~b/rd?ep=~ts&lwm2m=1", [?PORT, Epn]), +%% #coap_content{content_format = <<"text/plain">>, +%% payload = <<", , , , ">>}, +%% [], +%% MsgId), +%% #coap_message{type = ack, +%% method = Method} = test_recv_coap_response(UdpSock), %% ?assertEqual({error,bad_request}, Method), %% timer:sleep(50), %% ?assertEqual([], test_mqtt_broker:get_subscrbied_topics()). @@ -531,13 +572,15 @@ case07_register_alternate_path_01(Config) -> emqtt:subscribe(?config(emqx_c, Config), ReportTopic, qos0), timer:sleep(200), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, - payload = <<";rt=\"oma.lwm2m\";ct=11543,,,">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<";rt=\"oma.lwm2m\";ct=11543," + ",,">>}, + [], + MsgId), timer:sleep(50), true = lists:member(SubTopic, test_mqtt_broker:get_subscrbied_topics()). @@ -553,13 +596,15 @@ case07_register_alternate_path_02(Config) -> emqtt:subscribe(?config(emqx_c, Config), ReportTopic, qos0), timer:sleep(200), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, - payload = <<";rt=\"oma.lwm2m\";ct=11543,,,">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<";rt=\"oma.lwm2m\";ct=11543," + ",,">>}, + [], + MsgId), timer:sleep(50), true = lists:member(SubTopic, test_mqtt_broker:get_subscrbied_topics()). @@ -575,39 +620,40 @@ case08_reregister(Config) -> emqtt:subscribe(?config(emqx_c, Config), ReportTopic, qos0), timer:sleep(200), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, - payload = <<";rt=\"oma.lwm2m\";ct=11543,,,">>}, - [], - MsgId), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<";rt=\"oma.lwm2m\";ct=11543," + ",,">>}, + [], + MsgId), timer:sleep(50), true = lists:member(SubTopic, test_mqtt_broker:get_subscrbied_topics()), ReadResult = emqx_json:encode( - #{ - <<"msgType">> => <<"register">>, + #{<<"msgType">> => <<"register">>, <<"data">> => #{ - <<"alternatePath">> => <<"/lwm2m">>, - <<"ep">> => list_to_binary(Epn), - <<"lt">> => 345, - <<"lwm2m">> => <<"1">>, - <<"objectList">> => [<<"/1/0">>, <<"/2/0">>, <<"/3/0">>] - } - } - ), + <<"alternatePath">> => <<"/lwm2m">>, + <<"ep">> => list_to_binary(Epn), + <<"lt">> => 345, + <<"lwm2m">> => <<"1">>, + <<"objectList">> => [<<"/1/0">>, <<"/2/0">>, <<"/3/0">>] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(ReportTopic)), timer:sleep(1000), %% the same lwm2mc client registers to server again - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, - payload = <<";rt=\"oma.lwm2m\";ct=11543,,,">>}, - [], - MsgId + 1), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<";rt=\"oma.lwm2m\";ct=11543," + ",,">>}, + [], + MsgId + 1), %% verify the lwm2m client is still online ?assertEqual(ReadResult, test_recv_mqtt_response(ReportTopic)). @@ -620,13 +666,15 @@ case10_read(Config) -> emqtt:subscribe(?config(emqx_c, Config), RespTopic, qos0), timer:sleep(200), %% step 1, device register ... - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, - payload = <<";rt=\"oma.lwm2m\";ct=11543,,,">>}, - [], - MsgId1), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<";rt=\"oma.lwm2m\";ct=11543," + ",,">>}, + [], + MsgId1), #coap_message{method = Method1} = test_recv_coap_response(UdpSock), ?assertEqual({ok,created}, Method1), test_recv_mqtt_response(RespTopic), @@ -646,7 +694,9 @@ case10_read(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options=Options2, + payload=Payload2} = Request2, ?LOGT("LwM2M client got ~p", [Request2]), ?assertEqual(get, Method2), @@ -654,21 +704,29 @@ case10_read(Config) -> ?assertEqual(<<>>, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, Request2, true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, + payload = <<"EMQ">>}, + Request2, + true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"reqPath">> => <<"/3/0/0">>, - <<"content">> => [#{ - path => <<"/3/0/0">>, - value => <<"EMQ">> - }] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"reqPath">> => <<"/3/0/0">>, + <<"content">> => [#{path => <<"/3/0/0">>, + value => <<"EMQ">>} + ] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case10_read_separate_ack(Config) -> @@ -699,7 +757,8 @@ case10_read_separate_ack(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, payload = Payload2} = Request2, ?LOGT("LwM2M client got ~p", [Request2]), ?assertEqual(get, Method2), @@ -707,31 +766,36 @@ case10_read_separate_ack(Config) -> ?assertEqual(<<>>, Payload2), test_send_empty_ack(UdpSock, "127.0.0.1", ?PORT, Request2), - ReadResultACK = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"ack">>, - <<"data">> => #{ - <<"path">> => <<"/3/0/0">> - } - }), + ReadResultACK = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"ack">>, + <<"data">> => #{ <<"path">> => <<"/3/0/0">> } + }), ?assertEqual(ReadResultACK, test_recv_mqtt_response(RespTopic)), timer:sleep(100), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, Request2, false), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, + payload = <<"EMQ">>}, + Request2, + false), timer:sleep(100), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"reqPath">> => <<"/3/0/0">>, - <<"content">> => [#{ - path => <<"/3/0/0">>, - value => <<"EMQ">> - }] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"reqPath">> => <<"/3/0/0">>, + <<"content">> => [#{path => <<"/3/0/0">>, + value => <<"EMQ">>}] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case11_read_object_tlv(Config) -> @@ -767,32 +831,41 @@ case11_read_object_tlv(Config) -> ?assertEqual(get, Method2), timer:sleep(50), - Tlv = <<16#08, 16#00, 16#3C, 16#C8, 16#00, 16#14, 16#4F, 16#70, 16#65, 16#6E, 16#20, 16#4D, 16#6F, 16#62, 16#69, 16#6C, 16#65, 16#20, 16#41, 16#6C, 16#6C, 16#69, 16#61, 16#6E, 16#63, 16#65, 16#C8, 16#01, 16#16, 16#4C, 16#69, 16#67, 16#68, 16#74, 16#77, 16#65, 16#69, 16#67, 16#68, 16#74, 16#20, 16#4D, 16#32, 16#4D, 16#20, 16#43, 16#6C, 16#69, 16#65, 16#6E, 16#74, 16#C8, 16#02, 16#09, 16#33, 16#34, 16#35, 16#30, 16#30, 16#30, 16#31, 16#32, 16#33>>, - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"application/vnd.oma.lwm2m+tlv">>, payload = Tlv}, Request2, true), + Tlv = <<16#08, 16#00, 16#3C, 16#C8, 16#00, 16#14, 16#4F, 16#70, 16#65, + 16#6E, 16#20, 16#4D, 16#6F, 16#62, 16#69, 16#6C, 16#65, 16#20, + 16#41, 16#6C, 16#6C, 16#69, 16#61, 16#6E, 16#63, 16#65, 16#C8, + 16#01, 16#16, 16#4C, 16#69, 16#67, 16#68, 16#74, 16#77, 16#65, + 16#69, 16#67, 16#68, 16#74, 16#20, 16#4D, 16#32, 16#4D, 16#20, + 16#43, 16#6C, 16#69, 16#65, 16#6E, 16#74, 16#C8, 16#02, 16#09, + 16#33, 16#34, 16#35, 16#30, 16#30, 16#30, 16#31, 16#32, 16#33>>, + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"application/vnd.oma.lwm2m+tlv">>, + payload = Tlv}, + Request2, + true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"reqPath">> => <<"/3/0">>, - <<"content">> => [ - #{ - path => <<"/3/0/0">>, - value => <<"Open Mobile Alliance">> - }, - #{ - path => <<"/3/0/1">>, - value => <<"Lightweight M2M Client">> - }, - #{ - path => <<"/3/0/2">>, - value => <<"345000123">> - } - ] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"reqPath">> => <<"/3/0">>, + <<"content">> => + [#{path => <<"/3/0/0">>, + value => <<"Open Mobile Alliance">>}, + #{path => <<"/3/0/1">>, + value => <<"Lightweight M2M Client">>}, + #{path => <<"/3/0/2">>, + value => <<"345000123">>} + ] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case11_read_object_json(Config) -> @@ -829,32 +902,37 @@ case11_read_object_json(Config) -> ?assertEqual(get, Method2), timer:sleep(50), - Json = <<"{\"bn\":\"/3/0\",\"e\":[{\"n\":\"0\",\"sv\":\"Open Mobile Alliance\"},{\"n\":\"1\",\"sv\":\"Lightweight M2M Client\"},{\"n\":\"2\",\"sv\":\"345000123\"}]}">>, - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"application/vnd.oma.lwm2m+json">>, payload = Json}, Request2, true), + Json = <<"{\"bn\":\"/3/0\",\"e\":[{\"n\":\"0\",\"sv\":\"Open Mobile " + "Alliance\"},{\"n\":\"1\",\"sv\":\"Lightweight M2M Client\"}," + "{\"n\":\"2\",\"sv\":\"345000123\"}]}">>, + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"application/vnd.oma.lwm2m+json">>, + payload = Json}, + Request2, + true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"reqPath">> => <<"/3/0">>, - <<"content">> => [ - #{ - path => <<"/3/0/0">>, - value => <<"Open Mobile Alliance">> - }, - #{ - path => <<"/3/0/1">>, - value => <<"Lightweight M2M Client">> - }, - #{ - path => <<"/3/0/2">>, - value => <<"345000123">> - } - ] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"reqPath">> => <<"/3/0">>, + <<"content">> => + [#{path => <<"/3/0/0">>, + value => <<"Open Mobile Alliance">>}, + #{path => <<"/3/0/1">>, + value => <<"Lightweight M2M Client">>}, + #{path => <<"/3/0/2">>, + value => <<"345000123">>} + ] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case12_read_resource_opaque(Config) -> @@ -891,23 +969,29 @@ case12_read_resource_opaque(Config) -> timer:sleep(50), Opaque = <<20, 21, 22, 23>>, - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"application/octet-stream">>, payload = Opaque}, Request2, true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"application/octet-stream">>, + payload = Opaque}, + Request2, + true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"reqPath">> => <<"/3/0/8">>, - <<"content">> => [ - #{ - path => <<"/3/0/8">>, - value => base64:encode(Opaque) - } - ] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"reqPath">> => <<"/3/0/8">>, + <<"content">> => + [#{path => <<"/3/0/8">>, + value => base64:encode(Opaque)} + ] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case13_read_no_xml(Config) -> @@ -925,12 +1009,10 @@ case13_read_no_xml(Config) -> %% step2, send a READ command to device CmdId = 206, CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, - Command = #{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, + Command = #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"path">> => <<"/9723/0/0">> - } + <<"data">> => #{ <<"path">> => <<"/9723/0/0">> } }, CommandJson = emqx_json:encode(Command), ?LOGT("CommandJson=~p", [CommandJson]), @@ -943,17 +1025,26 @@ case13_read_no_xml(Config) -> ?assertEqual(get, Method2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, Request2, true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, + payload = <<"EMQ">>}, + Request2, + true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"reqPath">> => <<"/9723/0/0">>, - <<"code">> => <<"4.00">>, - <<"codeMsg">> => <<"bad_request">> - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"reqPath">> => <<"/9723/0/0">>, + <<"code">> => <<"4.00">>, + <<"codeMsg">> => <<"bad_request">> + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case20_single_write(Config) -> @@ -983,7 +1074,8 @@ case20_single_write(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(put, Method2), ?assertEqual(<<"/3/0/13">>, Path2), @@ -991,18 +1083,19 @@ case20_single_write(Config) -> ?assertEqual(Tlv_Value, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, changed}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, changed}, #coap_content{}, Request2, true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/13">>, - <<"code">> => <<"2.04">>, - <<"codeMsg">> => <<"changed">> - }, - <<"msgType">> => <<"write">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/13">>, + <<"code">> => <<"2.04">>, + <<"codeMsg">> => <<"changed">>}, + <<"msgType">> => <<"write">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case20_write(Config) -> @@ -1020,21 +1113,22 @@ case20_write(Config) -> %% step2, send a WRITE command to device CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, CmdId = 307, - Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + Command = #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, <<"msgType">> => <<"write">>, <<"data">> => #{ - <<"basePath">> => <<"/3/0/13">>, - <<"content">> => [#{ - type => <<"Float">>, - value => <<"12345.0">> - }] - } - }, + <<"basePath">> => <<"/3/0/13">>, + <<"content">> => + [#{type => <<"Float">>, + value => <<"12345.0">>}] + }}, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(put, Method2), ?assertEqual(<<"/3/0/13">>, Path2), @@ -1042,18 +1136,18 @@ case20_write(Config) -> ?assertEqual(Tlv_Value, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, changed}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, changed}, #coap_content{}, Request2, true), timer:sleep(100), - WriteResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/13">>, - <<"code">> => <<"2.04">>, - <<"codeMsg">> => <<"changed">> - }, - <<"msgType">> => <<"write">> - }), + WriteResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/13">>, + <<"code">> => <<"2.04">>, + <<"codeMsg">> => <<"changed">> }, + <<"msgType">> => <<"write">>}), ?assertEqual(WriteResult, test_recv_mqtt_response(RespTopic)). case21_write_object(Config) -> @@ -1071,26 +1165,26 @@ case21_write_object(Config) -> %% step2, send a WRITE command to device CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, CmdId = 307, - Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + Command = #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, <<"msgType">> => <<"write">>, <<"data">> => #{ - <<"basePath">> => <<"/3/0/">>, - <<"content">> => [#{ - path => <<"13">>, - type => <<"Integer">>, - value => <<"12345">> - },#{ - path => <<"14">>, - type => <<"String">>, - value => <<"87x">> - }] - } - }, + <<"basePath">> => <<"/3/0/">>, + <<"content">> => + [#{path => <<"13">>, + type => <<"Integer">>, + value => <<"12345">>}, + #{path => <<"14">>, + type => <<"String">>, + value => <<"87x">>}] + }}, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload=Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(post, Method2), ?assertEqual(<<"/3/0">>, Path2), @@ -1099,19 +1193,19 @@ case21_write_object(Config) -> ?assertEqual(Tlv_Value, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, changed}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, changed}, #coap_content{}, Request2, true), timer:sleep(100), - - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"write">>, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/">>, - <<"code">> => <<"2.04">>, - <<"codeMsg">> => <<"changed">> - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"msgType">> => <<"write">>, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/">>, + <<"code">> => <<"2.04">>, + <<"codeMsg">> => <<"changed">> + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case22_write_error(Config) -> @@ -1132,15 +1226,11 @@ case22_write_error(Config) -> Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, <<"msgType">> => <<"write">>, <<"data">> => #{ - <<"basePath">> => <<"/3/0/1">>, - <<"content">> => [ - #{ - type => <<"Integer">>, - value => <<"12345">> - } - ] - } - }, + <<"basePath">> => <<"/3/0/1">>, + <<"content">> => + [#{type => <<"Integer">>, + value => <<"12345">>}] + }}, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), @@ -1151,18 +1241,20 @@ case22_write_error(Config) -> ?assertEqual(<<"/3/0/1">>, Path2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {error, bad_request}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {error, bad_request}, #coap_content{}, + Request2, true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/1">>, - <<"code">> => <<"4.00">>, - <<"codeMsg">> => <<"bad_request">> - }, - <<"msgType">> => <<"write">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/1">>, + <<"code">> => <<"4.00">>, + <<"codeMsg">> => <<"bad_request">>}, + <<"msgType">> => <<"write">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case_create_basic(Config) -> @@ -1189,25 +1281,28 @@ case_create_basic(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(post, Method2), ?assertEqual(<<"/5">>, Path2), ?assertEqual(<<"">>, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, created}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, created}, #coap_content{}, Request2, true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/5">>, - <<"code">> => <<"2.01">>, - <<"codeMsg">> => <<"created">> - }, - <<"msgType">> => <<"create">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/5">>, + <<"code">> => <<"2.01">>, + <<"codeMsg">> => <<"created">>}, + <<"msgType">> => <<"create">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case_delete_basic(Config) -> @@ -1227,33 +1322,34 @@ case_delete_basic(Config) -> CmdId = 307, Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, <<"msgType">> => <<"delete">>, - <<"data">> => #{ - <<"path">> => <<"/5/0">> - } + <<"data">> => #{ <<"path">> => <<"/5/0">> } }, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(delete, Method2), ?assertEqual(<<"/5/0">>, Path2), ?assertEqual(<<"">>, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, deleted}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, deleted}, #coap_content{}, Request2, true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/5/0">>, - <<"code">> => <<"2.02">>, - <<"codeMsg">> => <<"deleted">> - }, - <<"msgType">> => <<"delete">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/5/0">>, + <<"code">> => <<"2.02">>, + <<"codeMsg">> => <<"deleted">>}, + <<"msgType">> => <<"delete">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case30_execute(Config) -> @@ -1271,37 +1367,41 @@ case30_execute(Config) -> %% step2, send a WRITE command to device CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, CmdId = 307, - Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + Command = #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, <<"msgType">> => <<"execute">>, <<"data">> => #{ - <<"path">> => <<"/3/0/4">>, - %% "args" should not be present for "/3/0/4", only for testing the encoding here - <<"args">> => <<"2,7">> - } + <<"path">> => <<"/3/0/4">>, + %% "args" should not be present for "/3/0/4", only for + %% testing the encoding here + <<"args">> => <<"2,7">>} }, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(post, Method2), ?assertEqual(<<"/3/0/4">>, Path2), ?assertEqual(<<"2,7">>, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, changed}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, changed}, #coap_content{}, Request2, true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/4">>, - <<"code">> => <<"2.04">>, - <<"codeMsg">> => <<"changed">> - }, - <<"msgType">> => <<"execute">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/4">>, + <<"code">> => <<"2.04">>, + <<"codeMsg">> => <<"changed">>}, + <<"msgType">> => <<"execute">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case31_execute_error(Config) -> @@ -1322,33 +1422,36 @@ case31_execute_error(Config) -> Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, <<"msgType">> => <<"execute">>, <<"data">> => #{ - <<"path">> => <<"/3/0/4">>, - <<"args">> => <<"2,7">> - } + <<"path">> => <<"/3/0/4">>, + <<"args">> => <<"2,7">>} }, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(post, Method2), ?assertEqual(<<"/3/0/4">>, Path2), ?assertEqual(<<"2,7">>, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {error, unauthorized}, #coap_content{}, Request2, true), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {error, unauthorized}, #coap_content{}, + Request2, true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/4">>, - <<"code">> => <<"4.01">>, - <<"codeMsg">> => <<"unauthorized">> - }, - <<"msgType">> => <<"execute">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, + <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/4">>, + <<"code">> => <<"4.01">>, + <<"codeMsg">> => <<"unauthorized">>}, + <<"msgType">> => <<"execute">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case40_discover(Config) -> @@ -1375,7 +1478,9 @@ case40_discover(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), ?assertEqual(get, Method2), ?assertEqual(<<"/3/0/7">>, Path2), @@ -1383,26 +1488,28 @@ case40_discover(Config) -> timer:sleep(50), PayloadDiscover = <<";dim=8;pmin=10;pmax=60;gt=50;lt=42.2,">>, - test_send_coap_response(UdpSock, - "127.0.0.1", - ?PORT, - {ok, content}, - #coap_content{content_format = <<"application/link-format">>, payload = PayloadDiscover}, - Request2, - true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"application/link-format">>, + payload = PayloadDiscover}, + Request2, + true), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"discover">>, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/7">>, - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"content">> => - [<<";dim=8;pmin=10;pmax=60;gt=50;lt=42.2">>, <<"">>] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"discover">>, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/7">>, + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"content">> => + [<<";dim=8;pmin=10;pmax=60;gt=50;lt=42.2">>, + <<"">>] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case50_write_attribute(Config) -> @@ -1432,34 +1539,33 @@ case50_write_attribute(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(100), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, ?LOGT("got options: ~p", [Options2]), Path2 = get_coap_path(Options2), Query2 = lists:sort(maps:to_list(get_coap_query(Options2))), ?assertEqual(put, Method2), ?assertEqual(<<"/3/0/9">>, Path2), - ?assertEqual(lists:sort([{<<"pmax">>, <<"5">>},{<<"lt">>, <<"5">>},{<<"pmin">>,<<"1">>}]), Query2), + ?assertEqual(lists:sort([{<<"pmax">>, <<"5">>}, + {<<"lt">>, <<"5">>}, + {<<"pmin">>,<<"1">>}]), Query2), ?assertEqual(<<>>, Payload2), timer:sleep(50), - test_send_coap_response(UdpSock, - "127.0.0.1", - ?PORT, - {ok, changed}, - #coap_content{}, - Request2, - true), - timer:sleep(100), + test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, + {ok, changed}, #coap_content{}, + Request2, true), + timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/9">>, - <<"code">> => <<"2.04">>, - <<"codeMsg">> => <<"changed">> - }, - <<"msgType">> => <<"write-attr">> - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/9">>, + <<"code">> => <<"2.04">>, + <<"codeMsg">> => <<"changed">>}, + <<"msgType">> => <<"write-attr">> + }), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case60_observe(Config) -> @@ -1481,15 +1587,15 @@ case60_observe(Config) -> CmdId = 307, Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, <<"msgType">> => <<"observe">>, - <<"data">> => #{ - <<"path">> => <<"/3/0/10">> - } + <<"data">> => #{<<"path">> => <<"/3/0/10">>} }, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50), Request2 = test_recv_coap_request(UdpSock), - #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, + #coap_message{method = Method2, + options = Options2, + payload = Payload2} = Request2, Path2 = get_coap_path(Options2), Observe = get_coap_observe(Options2), ?assertEqual(get, Method2), @@ -1498,55 +1604,53 @@ case60_observe(Config) -> ?assertEqual(<<>>, Payload2), timer:sleep(50), - test_send_coap_observe_ack( UdpSock, - "127.0.0.1", - ?PORT, - {ok, content}, - #coap_content{content_format = <<"text/plain">>, payload = <<"2048">>}, - Request2), + test_send_coap_observe_ack( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, payload = <<"2048">>}, + Request2), timer:sleep(100), - ReadResult = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"observe">>, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/10">>, - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"content">> => [#{ - path => <<"/3/0/10">>, - value => 2048 - }] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"observe">>, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/10">>, + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"content">> => + [#{path => <<"/3/0/10">>, + value => 2048}] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)), %% step3 the notifications timer:sleep(200), ObSeq = 3, - test_send_coap_notif( UdpSock, - "127.0.0.1", - ?PORT, - #coap_content{content_format = <<"text/plain">>, payload = <<"4096">>}, - ObSeq, - Request2), + test_send_coap_notif( + UdpSock, + "127.0.0.1", + ?PORT, + #coap_content{content_format = <<"text/plain">>, payload = <<"4096">>}, + ObSeq, + Request2), timer:sleep(100), #coap_message{} = test_recv_coap_response(UdpSock), - ReadResult2 = emqx_json:encode(#{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"notify">>, - <<"seqNum">> => ObSeq, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/10">>, - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"content">> => [#{ - path => <<"/3/0/10">>, - value => 4096 - }] - } - }), + ReadResult2 = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"notify">>, + <<"seqNum">> => ObSeq, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/10">>, + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"content">> => + [#{path => <<"/3/0/10">>, + value => 4096}] + }}), ?assertEqual(ReadResult2, test_recv_mqtt_response(RespTopicAD)), %% Step3. cancel observe @@ -1561,7 +1665,9 @@ case60_observe(Config) -> test_mqtt_broker:publish(CommandTopic, CommandJson3, 0), timer:sleep(50), Request3 = test_recv_coap_request(UdpSock), - #coap_message{method = Method3, options=Options3, payload=Payload3} = Request3, + #coap_message{method = Method3, + options = Options3, + payload = Payload3} = Request3, Path3 = get_coap_path(Options3), Observe3 = get_coap_observe(Options3), ?assertEqual(get, Method3), @@ -1570,31 +1676,31 @@ case60_observe(Config) -> ?assertEqual(<<>>, Payload3), timer:sleep(50), - test_send_coap_observe_ack( UdpSock, - "127.0.0.1", - ?PORT, - {ok, content}, - #coap_content{content_format = <<"text/plain">>, payload = <<"1150">>}, - Request3), + test_send_coap_observe_ack( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, payload = <<"1150">>}, + Request3), timer:sleep(100), - ReadResult3 = emqx_json:encode(#{ - <<"requestID">> => CmdId3, <<"cacheID">> => CmdId3, - <<"msgType">> => <<"cancel-observe">>, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/10">>, - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"content">> => [#{ - path => <<"/3/0/10">>, - value => 1150 - }] - } - }), + ReadResult3 = emqx_json:encode( + #{<<"requestID">> => CmdId3, + <<"cacheID">> => CmdId3, + <<"msgType">> => <<"cancel-observe">>, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/10">>, + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"content">> => + [#{path => <<"/3/0/10">>, + value => 1150}] + }}), ?assertEqual(ReadResult3, test_recv_mqtt_response(RespTopic)). %% case80_specail_object_19_0_0_notify(Config) -> -%% %% step 1, device register, with extra register options +%% %% step 1, device register, with extra register options %% Epn = "urn:oma:lwm2m:oma:3", %% RegOptionWangYi = "&apn=psmA.eDRX0.ctnb&im=13456&ct=2.0&mt=MDM9206&mv=4.0", %% MsgId1 = 15, @@ -1603,64 +1709,67 @@ case60_observe(Config) -> %% RespTopic = list_to_binary("lwm2m/"++Epn++"/up/resp"), %% emqtt:subscribe(?config(emqx_c, Config), RespTopic, qos0), %% timer:sleep(200), - -%% test_send_coap_request( UdpSock, -%% post, -%% sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1"++RegOptionWangYi, [?PORT, Epn]), -%% #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, -%% [], -%% MsgId1), -%% #coap_message{method = Method1} = test_recv_coap_response(UdpSock), -%% ?assertEqual({ok,created}, Method1), -%% ReadResult = emqx_json:encode(#{ -%% <<"msgType">> => <<"register">>, -%% <<"data">> => #{ -%% <<"alternatePath">> => <<"/">>, -%% <<"ep">> => list_to_binary(Epn), -%% <<"lt">> => 345, -%% <<"lwm2m">> => <<"1">>, -%% <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, <<"/4">>, <<"/5">>], -%% <<"apn">> => <<"psmA.eDRX0.ctnb">>, -%% <<"im">> => <<"13456">>, -%% <<"ct">> => <<"2.0">>, -%% <<"mt">> => <<"MDM9206">>, -%% <<"mv">> => <<"4.0">> -%% } -%% }), -%% ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)), - -%% %% step2, send a OBSERVE command to device -%% CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, -%% CmdId = 307, -%% Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, -%% <<"msgType">> => <<"observe">>, -%% <<"data">> => #{ -%% <<"path">> => <<"/19/0/0">> -%% } -%% }, -%% CommandJson = emqx_json:encode(Command), -%% test_mqtt_broker:publish(CommandTopic, CommandJson, 0), -%% timer:sleep(50), -%% Request2 = test_recv_coap_request(UdpSock), -%% #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, -%% Path2 = get_coap_path(Options2), -%% Observe = get_coap_observe(Options2), -%% ?assertEqual(get, Method2), -%% ?assertEqual(<<"/19/0/0">>, Path2), -%% ?assertEqual(Observe, 0), -%% ?assertEqual(<<>>, Payload2), -%% timer:sleep(50), - -%% test_send_coap_observe_ack( UdpSock, -%% "127.0.0.1", -%% ?PORT, -%% {ok, content}, -%% #coap_content{content_format = <<"text/plain">>, payload = <<"2048">>}, -%% Request2), -%% timer:sleep(100). - -%% step 3, device send uplink data notifications - +%% +%% test_send_coap_request( +%% UdpSock, +%% post, +%% sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1"++RegOptionWangYi, [?PORT, Epn]), +%% #coap_content{content_format = <<"text/plain">>, +%% payload = <<", , , , ">>}, +%% [], +%% MsgId1), +%% #coap_message{method = Method1} = test_recv_coap_response(UdpSock), +%% ?assertEqual({ok,created}, Method1), +%% ReadResult = emqx_json:encode( +%% #{<<"msgType">> => <<"register">>, +%% <<"data">> => #{ +%% <<"alternatePath">> => <<"/">>, +%% <<"ep">> => list_to_binary(Epn), +%% <<"lt">> => 345, +%% <<"lwm2m">> => <<"1">>, +%% <<"objectList">> => [<<"/1">>, <<"/2">>, <<"/3">>, +%% <<"/4">>, <<"/5">>], +%% <<"apn">> => <<"psmA.eDRX0.ctnb">>, +%% <<"im">> => <<"13456">>, +%% <<"ct">> => <<"2.0">>, +%% <<"mt">> => <<"MDM9206">>, +%% <<"mv">> => <<"4.0">>} +%% }), +%% ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)), +%% +%% %% step2, send a OBSERVE command to device +%% CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, +%% CmdId = 307, +%% Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, +%% <<"msgType">> => <<"observe">>, +%% <<"data">> => #{ +%% <<"path">> => <<"/19/0/0">> +%% } +%% }, +%% CommandJson = emqx_json:encode(Command), +%% test_mqtt_broker:publish(CommandTopic, CommandJson, 0), +%% timer:sleep(50), +%% Request2 = test_recv_coap_request(UdpSock), +%% #coap_message{method = Method2, +%% options = Options2, +%% payload = Payload2} = Request2, +%% Path2 = get_coap_path(Options2), +%% Observe = get_coap_observe(Options2), +%% ?assertEqual(get, Method2), +%% ?assertEqual(<<"/19/0/0">>, Path2), +%% ?assertEqual(Observe, 0), +%% ?assertEqual(<<>>, Payload2), +%% timer:sleep(50), +%% +%% test_send_coap_observe_ack( +%% UdpSock, +%% "127.0.0.1", +%% ?PORT, +%% {ok, content}, +%% #coap_content{content_format = <<"text/plain">>, payload = <<"2048">>}, +%% Request2), +%% timer:sleep(100). +%% %% case80_specail_object_19_1_0_write(Config) -> %% Epn = "urn:oma:lwm2m:oma:3", %% RegOptionWangYi = "&apn=psmA.eDRX0.ctnb&im=13456&ct=2.0&mt=MDM9206&mv=4.0", @@ -1669,52 +1778,57 @@ case60_observe(Config) -> %% RespTopic = list_to_binary("lwm2m/"++Epn++"/up/resp"), %% emqtt:subscribe(?config(emqx_c, Config), RespTopic, qos0), %% timer:sleep(200), - -%% test_send_coap_request( UdpSock, -%% post, -%% sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1"++RegOptionWangYi, [?PORT, Epn]), -%% #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, -%% [], -%% MsgId1), +%% +%% test_send_coap_request( +%% UdpSock, +%% post, +%% sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1"++RegOptionWangYi, [?PORT, Epn]), +%% #coap_content{content_format = <<"text/plain">>, +%% payload = <<", , , , ">>}, +%% [], +%% MsgId1), %% #coap_message{method = Method1} = test_recv_coap_response(UdpSock), %% ?assertEqual({ok,created}, Method1), %% test_recv_mqtt_response(RespTopic), - -%% %% step2, send a WRITE command to device +%% +%% %% step2, send a WRITE command to device %% CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, %% CmdId = 307, -%% Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, +%% Command = #{<<"requestID">> => CmdId, +%% <<"cacheID">> => CmdId, %% <<"msgType">> => <<"write">>, %% <<"data">> => #{ -%% <<"path">> => <<"/19/1/0">>, -%% <<"type">> => <<"Opaque">>, -%% <<"value">> => base64:encode(<<12345:32>>) -%% } -%% }, - +%% <<"path">> => <<"/19/1/0">>, +%% <<"type">> => <<"Opaque">>, +%% <<"value">> => base64:encode(<<12345:32>>) +%% }}, +%% %% CommandJson = emqx_json:encode(Command), %% test_mqtt_broker:publish(CommandTopic, CommandJson, 0), %% timer:sleep(50), %% Request2 = test_recv_coap_request(UdpSock), -%% #coap_message{method = Method2, options=Options2, payload=Payload2} = Request2, +%% #coap_message{method = Method2, +%% options = Options2, +%% payload = Payload2} = Request2, %% Path2 = get_coap_path(Options2), %% ?assertEqual(put, Method2), %% ?assertEqual(<<"/19/1/0">>, Path2), %% ?assertEqual(<<3:2, 0:1, 0:2, 4:3, 0, 12345:32>>, Payload2), %% timer:sleep(50), - -%% test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, changed}, #coap_content{}, Request2, true), +%% +%% test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, +%% {ok, changed}, #coap_content{}, Request2, true), %% timer:sleep(100), - -%% ReadResult = emqx_json:encode(#{ -%% <<"requestID">> => CmdId, <<"cacheID">> => CmdId, -%% <<"data">> => #{ -%% <<"reqPath">> => <<"/19/1/0">>, -%% <<"code">> => <<"2.04">>, -%% <<"codeMsg">> => <<"changed">> -%% }, -%% <<"msgType">> => <<"write">> -%% }), +%% +%% ReadResult = emqx_json:encode( +%% #{<<"requestID">> => CmdId, +%% <<"cacheID">> => CmdId, +%% <<"data">> => #{ +%% <<"reqPath">> => <<"/19/1/0">>, +%% <<"code">> => <<"2.04">>, +%% <<"codeMsg">> => <<"changed">>}, +%% <<"msgType">> => <<"write">> +%% }), %% ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). case90_psm_mode(Config) -> @@ -1736,13 +1850,17 @@ server_cache_mode(Config, RegOption) -> emqtt:subscribe(?config(emqx_c, Config), RespTopic, qos0), timer:sleep(200), - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?"++RegOption, [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = <<", , , , ">>}, - [], - MsgId1), - #coap_message{type = ack, method = Method1, options = Opts} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?"++RegOption, [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, + payload = <<", , , , ">>}, + [], + MsgId1), + #coap_message{type = ack, + method = Method1, + options = Opts} = test_recv_coap_response(UdpSock), ?assertEqual({ok,created}, Method1), ?LOGT("Options got: ~p", [Opts]), Location = maps:get(location_path, Opts), @@ -1760,7 +1878,8 @@ server_cache_mode(Config, RegOption) -> send_read_command_1(2, UdpSock), send_read_command_1(3, UdpSock), - ?assertEqual(timeout_test_recv_coap_request, test_recv_coap_request(UdpSock)), + ?assertEqual(timeout_test_recv_coap_request, + test_recv_coap_request(UdpSock)), device_update_1(UdpSock, Location), @@ -1775,13 +1894,10 @@ server_cache_mode(Config, RegOption) -> send_read_command_1(CmdId, _UdpSock) -> Epn = "urn:oma:lwm2m:oma:3", CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>, - Command = #{ - <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"path">> => <<"/3/0/0">> - } - }, + Command = #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{<<"path">> => <<"/3/0/0">>} + }, CommandJson = emqx_json:encode(Command), test_mqtt_broker:publish(CommandTopic, CommandJson, 0), timer:sleep(50). @@ -1795,20 +1911,28 @@ verify_read_response_1(CmdId, UdpSock) -> ?LOGT("LwM2M client got ~p", [Request]), %% device replies the commond - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, Request, true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, + payload = <<"EMQ">>}, + Request, + true), - ReadResult = emqx_json:encode(#{ <<"requestID">> => CmdId, <<"cacheID">> => CmdId, - <<"msgType">> => <<"read">>, - <<"data">> => #{ - <<"reqPath">> => <<"/3/0/0">>, - <<"code">> => <<"2.05">>, - <<"codeMsg">> => <<"content">>, - <<"content">> => [#{ - path => <<"/3/0/0">>, - value => <<"EMQ">> - }] - } - }), + ReadResult = emqx_json:encode( + #{<<"requestID">> => CmdId, <<"cacheID">> => CmdId, + <<"msgType">> => <<"read">>, + <<"data">> => #{ + <<"reqPath">> => <<"/3/0/0">>, + <<"code">> => <<"2.05">>, + <<"codeMsg">> => <<"content">>, + <<"content">> => + [#{path => <<"/3/0/0">>, + value => <<"EMQ">> + }] + }}), ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)). device_update_1(UdpSock, Location) -> @@ -1816,13 +1940,16 @@ device_update_1(UdpSock, Location) -> RespTopic = list_to_binary("lwm2m/"++Epn++"/up/resp"), ?LOGT("send UPDATE command", []), MsgId2 = 27, - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b~ts?lt=789", [?PORT, join_path(Location, <<>>)]), - #coap_content{payload = <<>>}, - [], - MsgId2), - #coap_message{type = ack, id = MsgId2, method = Method2} = test_recv_coap_response(UdpSock), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b~ts?lt=789", [?PORT, join_path(Location, <<>>)]), + #coap_content{payload = <<>>}, + [], + MsgId2), + #coap_message{type = ack, + id = MsgId2, + method = Method2} = test_recv_coap_response(UdpSock), {ok,changed} = Method2, test_recv_mqtt_response(RespTopic). @@ -1835,34 +1962,56 @@ test_recv_mqtt_response(RespTopic) -> end. test_send_coap_request(UdpSock, Method, Uri, Content, Options, MsgId) -> - is_record(Content, coap_content) orelse error("Content must be a #coap_content!"), - is_list(Options) orelse error("Options must be a list"), + is_record(Content, coap_content) orelse + error("Content must be a #coap_content!"), + is_list(Options) orelse + error("Options must be a list"), case resolve_uri(Uri) of {coap, {IpAddr, Port}, Path, Query} -> - Request0 = request(con, Method, Content, [{uri_path, Path}, {uri_query, Query} | Options]), + Request0 = request( + con, Method, Content, + [{uri_path, Path}, {uri_query, Query} | Options] + ), Request = Request0#coap_message{id = MsgId}, ?LOGT("send_coap_request Request=~p", [Request]), + RequestBinary = emqx_coap_frame:serialize_pkt(Request, undefined), - ?LOGT("test udp socket send to ~p:~p, data=~p", [IpAddr, Port, RequestBinary]), + ?LOGT("test udp socket send to ~p:~p, data=~p", + [IpAddr, Port, RequestBinary]), ok = gen_udp:send(UdpSock, IpAddr, Port, RequestBinary); {SchemeDiff, ChIdDiff, _, _} -> - error(lists:flatten(io_lib:format("scheme ~ts or ChId ~ts does not match with socket", [SchemeDiff, ChIdDiff]))) + error( + lists:flatten( + io_lib:format( + "scheme ~ts or ChId ~ts does not match with socket", + [SchemeDiff, ChIdDiff]))) end. test_recv_coap_response(UdpSock) -> {ok, {Address, Port, Packet}} = gen_udp:recv(UdpSock, 0, 2000), {ok, Response, _, _} = emqx_coap_frame:parse(Packet, undefined), - ?LOGT("test udp receive from ~p:~p, data1=~p, Response=~p", [Address, Port, Packet, Response]), - #coap_message{type = ack, method = Method, id=Id, token = Token, options = Options, payload = Payload} = Response, - ?LOGT("receive coap response Method=~p, Id=~p, Token=~p, Options=~p, Payload=~p", [Method, Id, Token, Options, Payload]), + ?LOGT("test udp receive from ~p:~p, data1=~p, Response=~p", + [Address, Port, Packet, Response]), + #coap_message{type = ack, method = Method, id = Id, + token = Token, options = Options, payload = Payload} = Response, + ?LOGT("receive coap response Method=~p, Id=~p, Token=~p, " + "Options=~p, Payload=~p", [Method, Id, Token, Options, Payload]), Response. test_recv_coap_request(UdpSock) -> case gen_udp:recv(UdpSock, 0, 2000) of {ok, {_Address, _Port, Packet}} -> {ok, Request, _, _} = emqx_coap_frame:parse(Packet, undefined), - #coap_message{type = con, method = Method, id=Id, token = Token, payload = Payload, options = Options} = Request, - ?LOGT("receive coap request Method=~p, Id=~p, Token=~p, Options=~p, Payload=~p", [Method, Id, Token, Options, Payload]), + #coap_message{ + type = con, + id = Id, + method = Method, + token = Token, + payload = Payload, + options = Options} = Request, + ?LOGT("receive coap request Method=~p, Id=~p, Token=~p, " + "Options=~p, Payload=~p", + [Method, Id, Token, Options, Payload]), Request; {error, Reason} -> ?LOGT("test_recv_coap_request failed, Reason=~p", [Reason]), @@ -1870,7 +2019,8 @@ test_recv_coap_request(UdpSock) -> end. test_send_coap_response(UdpSock, Host, Port, Code, Content, Request, Ack) -> - is_record(Content, coap_content) orelse error("Content must be a #coap_content!"), + is_record(Content, coap_content) orelse + error("Content must be a #coap_content!"), is_list(Host) orelse error("Host is not a string"), {ok, IpAddr} = inet:getaddr(Host, inet), @@ -1880,17 +2030,20 @@ test_send_coap_response(UdpSock, Host, Port, Code, Content, Request, Ack) -> false -> Response end, ?LOGT("test_send_coap_response Response=~p", [Response2]), - ok = gen_udp:send(UdpSock, IpAddr, Port, emqx_coap_frame:serialize_pkt(Response2, undefined)). + ok = gen_udp:send(UdpSock, IpAddr, Port, + emqx_coap_frame:serialize_pkt(Response2, undefined)). test_send_empty_ack(UdpSock, Host, Port, Request) -> is_list(Host) orelse error("Host is not a string"), {ok, IpAddr} = inet:getaddr(Host, inet), EmptyACK = emqx_coap_message:ack(Request), ?LOGT("test_send_empty_ack EmptyACK=~p", [EmptyACK]), - ok = gen_udp:send(UdpSock, IpAddr, Port, emqx_coap_frame:serialize_pkt(EmptyACK, undefined)). + ok = gen_udp:send(UdpSock, IpAddr, Port, + emqx_coap_frame:serialize_pkt(EmptyACK, undefined)). test_send_coap_observe_ack(UdpSock, Host, Port, Code, Content, Request) -> - is_record(Content, coap_content) orelse error("Content must be a #coap_content!"), + is_record(Content, coap_content) orelse + error("Content must be a #coap_content!"), is_list(Host) orelse error("Host is not a string"), {ok, IpAddr} = inet:getaddr(Host, inet), @@ -1903,7 +2056,8 @@ test_send_coap_observe_ack(UdpSock, Host, Port, Code, Content, Request) -> ok = gen_udp:send(UdpSock, IpAddr, Port, ResponseBinary). test_send_coap_notif(UdpSock, Host, Port, Content, ObSeq, Request) -> - is_record(Content, coap_content) orelse error("Content must be a #coap_content!"), + is_record(Content, coap_content) orelse + error("Content must be a #coap_content!"), is_list(Host) orelse error("Host is not a string"), {ok, IpAddr} = inet:getaddr(Host, inet), @@ -1915,12 +2069,13 @@ test_send_coap_notif(UdpSock, Host, Port, Content, ObSeq, Request) -> ok = gen_udp:send(UdpSock, IpAddr, Port, NotifBinary). std_register(UdpSock, Epn, ObjectList, MsgId1, RespTopic) -> - test_send_coap_request( UdpSock, - post, - sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, payload = ObjectList}, - [], - MsgId1), + test_send_coap_request( + UdpSock, + post, + sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=345&lwm2m=1", [?PORT, Epn]), + #coap_content{content_format = <<"text/plain">>, payload = ObjectList}, + [], + MsgId1), #coap_message{method = {ok,created}} = test_recv_coap_response(UdpSock), test_recv_mqtt_response(RespTopic), timer:sleep(100). @@ -1978,8 +2133,11 @@ sprintf(Format, Args) -> lists:flatten(io_lib:format(Format, Args)). response(Code, #coap_content{content_format = Format, payload = Payload}, Req) -> - #coap_message{options = Opts} = Msg = emqx_coap_message:response(Code, Payload, Req), + Msg = #coap_message{options = Opts} + = emqx_coap_message:response(Code, Payload, Req), Msg#coap_message{options = Opts#{content_format => Format}}. -request(Type, Method, #coap_content{content_format = Format, payload = Payload}, Opts) -> - emqx_coap_message:request(Type, Method, Payload, [{content_format, Format} | Opts]). +request(Type, Method, #coap_content{content_format = Format, + payload = Payload}, Opts) -> + emqx_coap_message:request(Type, Method, + Payload, [{content_format, Format} | Opts]). diff --git a/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl b/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl index 8f11bca78..babf9d36c 100644 --- a/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_lwm2m_api_SUITE.erl @@ -110,8 +110,11 @@ t_lookup_cmd_read(Config) -> test_send_coap_request( UdpSock, post, sprintf("coap://127.0.0.1:~b/rd?ep=~ts<=600&lwm2m=1", [?PORT, Epn]), - #coap_content{content_format = <<"text/plain">>, - payload = <<";rt=\"oma.lwm2m\";ct=11543,,,">>}, + #coap_content{ + content_format = <<"text/plain">>, + payload = <<";rt=\"oma.lwm2m\";ct=11543," + ",,">> + }, [], MsgId1), #coap_message{method = Method1} = test_recv_coap_response(UdpSock), @@ -141,7 +144,14 @@ t_lookup_cmd_read(Config) -> ?LOGT("LwM2M client got ~p", [Request2]), timer:sleep(50), - test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, Request2, true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, + Request2, + true), timer:sleep(200), normal_received_request(Epn, <<"/3/0/0">>, <<"read">>). @@ -177,13 +187,14 @@ t_lookup_cmd_discover(Config) -> timer:sleep(50), PayloadDiscover = <<";dim=8;pmin=10;pmax=60;gt=50;lt=42.2,">>, - test_send_coap_response(UdpSock, - "127.0.0.1", - ?PORT, - {ok, content}, - #coap_content{content_format = <<"application/link-format">>, payload = PayloadDiscover}, - Request2, - true), + test_send_coap_response( + UdpSock, + "127.0.0.1", + ?PORT, + {ok, content}, + #coap_content{content_format = <<"application/link-format">>, payload = PayloadDiscover}, + Request2, + true), timer:sleep(200), discover_received_request(Epn, <<"/3/0/7">>, <<"discover">>). diff --git a/apps/emqx_gateway/test/emqx_stomp_SUITE.erl b/apps/emqx_gateway/test/emqx_stomp_SUITE.erl index d51eaaad6..179471127 100644 --- a/apps/emqx_gateway/test/emqx_stomp_SUITE.erl +++ b/apps/emqx_gateway/test/emqx_stomp_SUITE.erl @@ -87,35 +87,36 @@ t_connect(_) -> %% Connect will be failed, because of bad login or passcode %% FIXME: Waiting for authentication works - %with_connection(fun(Sock) -> - % gen_tcp:send(Sock, serialize(<<"CONNECT">>, - % [{<<"accept-version">>, ?STOMP_VER}, - % {<<"host">>, <<"127.0.0.1:61613">>}, - % {<<"login">>, <<"admin">>}, - % {<<"passcode">>, <<"admin">>}, - % {<<"heart-beat">>, <<"1000,2000">>}])), - % {ok, Data} = gen_tcp:recv(Sock, 0), - % {ok, #stomp_frame{command = <<"ERROR">>, - % headers = _, - % body = <<"Login or passcode error!">>}, _, _} = - % parse(Data) - % end), + %with_connection( + % fun(Sock) -> + % gen_tcp:send(Sock, serialize(<<"CONNECT">>, + % [{<<"accept-version">>, ?STOMP_VER}, + % {<<"host">>, <<"127.0.0.1:61613">>}, + % {<<"login">>, <<"admin">>}, + % {<<"passcode">>, <<"admin">>}, + % {<<"heart-beat">>, <<"1000,2000">>}])), + % {ok, Data} = gen_tcp:recv(Sock, 0), + % {ok, Frame, _, _} = parse(Data), + % #stomp_frame{command = <<"ERROR">>, + % headers = _, + % body = <<"Login or passcode error!">>} = Frame + % end), %% Connect will be failed, because of bad version with_connection(fun(Sock) -> - gen_tcp:send(Sock, serialize(<<"CONNECT">>, - [{<<"accept-version">>, <<"2.0,2.1">>}, - {<<"host">>, <<"127.0.0.1:61613">>}, - {<<"login">>, <<"guest">>}, - {<<"passcode">>, <<"guest">>}, - {<<"heart-beat">>, <<"1000,2000">>}])), + gen_tcp:send(Sock, + serialize(<<"CONNECT">>, + [{<<"accept-version">>, <<"2.0,2.1">>}, + {<<"host">>, <<"127.0.0.1:61613">>}, + {<<"login">>, <<"guest">>}, + {<<"passcode">>, <<"guest">>}, + {<<"heart-beat">>, <<"1000,2000">>}])), {ok, Data} = gen_tcp:recv(Sock, 0), - {ok, - #stomp_frame{command = <<"ERROR">>, - headers = _, - body = <<"Login Failed: Supported protocol versions < 1.2">>}, - _, - _ } = parse(Data) + {ok, Frame, _, _} = parse(Data), + #stomp_frame{ + command = <<"ERROR">>, + headers = _, + body = <<"Login Failed: Supported protocol versions < 1.2">>} = Frame end). t_heartbeat(_) -> @@ -407,7 +408,7 @@ t_rest_clienit_info(_) -> %% kickout {204, _} = request(delete, ClientPath), - ignored = gen_server:call(?CM, ignore, infinity). % sync + ignored = gen_server:call(emqx_cm, ignore, infinity), % sync ok = emqx_pool:flush_async_tasks(), {200, Clients2} = request(get, "/gateway/stomp/clients"), ?assertEqual(0, length(maps:get(data, Clients2))) diff --git a/elvis.config b/elvis.config index 30178622a..fb8ee9df1 100644 --- a/elvis.config +++ b/elvis.config @@ -17,7 +17,8 @@ {elvis_style, operator_spaces, #{rules => [{right, "|"}, {left, "|"}, {right, "||"}, - {left, "||"}]}} + {left, "||"}]}}, + {elvis_style, dont_repeat_yourself, #{ min_complexity => 20 }} ] }, #{dirs => ["test", "apps/**/test"],