Share tests between MQTT and Web MQTT

New test suite deps/rabbitmq_mqtt/test/shared_SUITE contains tests that
are executed against both MQTT and Web MQTT.

This has two major advantages:
1. Eliminates test code duplication between rabbitmq_mqtt and
rabbitmq_web_mqtt making the tests easier to maintain and to understand.
2. Increases test coverage of Web MQTT.

It's acceptable to add a **test** dependency from rabbitmq_mqtt to
rabbitmq_web_mqtt. Obviously, there should be no such dependency
for non-test code.

Author: ansd

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -38,7 +38,7 @@
 
     add_code_path_to_node/2,
     add_code_path_to_all_nodes/2,
-    rpc/5, rpc/6,
+    rpc/4, rpc/5, rpc/6,
     rpc_all/4, rpc_all/5,
 
     start_node/2,
@@ -1558,6 +1558,9 @@ add_code_path_to_all_nodes(Config, Module) ->
       || Nodename <- Nodenames],
     ok.
 
+rpc(Config, Module, Function, Args) ->
+    rpc(Config, 0, Module, Function, Args).
+
 rpc(Config, Node, Module, Function, Args)
 when is_atom(Node) andalso Node =/= undefined ->
     rpc(Config, Node, Module, Function, Args, infinity);

deps/rabbitmq_mqtt/BUILD.bazel

@@ -86,6 +86,7 @@ dialyze(
 broker_for_integration_suites(
     extra_plugins = [
         "//deps/rabbitmq_management:erlang_app",
+        "//deps/rabbitmq_web_mqtt:erlang_app",
     ],
 )
 
@@ -193,6 +194,7 @@ rabbitmq_integration_suite(
         "@emqtt//:erlang_app",
     ],
     additional_beam = [
+        ":event_recorder",
         ":util",
     ],
 )
@@ -226,16 +228,18 @@ rabbitmq_integration_suite(
 )
 
 rabbitmq_integration_suite(
-    name = "integration_SUITE",
-    size = "large",
-    runtime_deps = [
-        "@emqtt//:erlang_app",
-         "//deps/rabbitmq_management_agent:erlang_app",
-    ],
-    additional_beam = [
-        ":event_recorder",
-        ":util",
-    ],
+   name = "shared_SUITE",
+   size = "large",
+   timeout = "eternal",
+   runtime_deps = [
+       "@emqtt//:erlang_app",
+       "//deps/rabbitmq_management_agent:erlang_app",
+       "@gun//:erlang_app",
+   ],
+   additional_beam = [
+       ":event_recorder",
+       ":util",
+   ],
 )
 
 assert_suites()

deps/rabbitmq_mqtt/Makefile

@@ -40,7 +40,7 @@ BUILD_WITHOUT_QUIC=1
 export BUILD_WITHOUT_QUIC
 
 DEPS = ranch rabbit_common rabbit amqp_client ra
-TEST_DEPS = emqtt ct_helper rabbitmq_ct_helpers rabbitmq_ct_client_helpers rabbitmq_management
+TEST_DEPS = emqtt ct_helper rabbitmq_ct_helpers rabbitmq_ct_client_helpers rabbitmq_management rabbitmq_web_mqtt
 
 dep_ct_helper = git https://github.com/extend/ct_helper.git master
 dep_emqtt = git https://github.com/emqx/emqtt.git 1.7.0-rc.2

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -76,7 +76,7 @@
          %% quorum queues and streams whose soft limit has been exceeded
          soft_limit_exceeded = sets:new([{version, 2}]) :: sets:set(),
          qos0_messages_dropped = 0 :: non_neg_integer()
-         }).
+        }).
 
 -opaque state() :: #state{}.
 

deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl

@@ -412,14 +412,14 @@ run_socket(State = #state{ socket = Sock }) ->
     rabbit_net:setopts(Sock, [{active, once}]),
     State#state{ await_recv = true }.
 
-control_throttle(State = #state{connection_state = Flow,
+control_throttle(State = #state{connection_state = ConnState,
                                 conserve = Conserve,
                                 keepalive = KState,
                                 proc_state = PState}) ->
     Throttle = Conserve orelse
                rabbit_mqtt_processor:soft_limit_exceeded(PState) orelse
                credit_flow:blocked(),
-    case {Flow, Throttle} of
+    case {ConnState, Throttle} of
         {running, true} ->
             State#state{connection_state = blocked,
                         keepalive = rabbit_mqtt_keepalive:cancel_timer(KState)};

deps/rabbitmq_mqtt/test/cluster_SUITE.erl

@@ -9,54 +9,52 @@
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
--import(util, [expect_publishes/2,
+-import(util, [expect_publishes/3,
                connect/3,
-               connect/4]).
+               connect/4,
+               await_exit/1]).
 
 -import(rabbit_ct_broker_helpers,
         [setup_steps/0,
          teardown_steps/0,
          get_node_config/3,
          rabbitmqctl/3,
-         rpc/5,
-         stop_node/2,
-         drain_node/2,
-         revive_node/2]).
+         rpc/4,
+         stop_node/2
+        ]).
 
 -define(OPTS, [{connect_timeout, 1},
                {ack_timeout, 1}]).
 
 all() ->
     [
-     {group, cluster_size_3},
      {group, cluster_size_5}
     ].
 
 groups() ->
     [
-     {cluster_size_3, [], [
-                           maintenance
-                          ]},
-     {cluster_size_5, [], [
-                           connection_id_tracking,
-                           connection_id_tracking_on_nodedown,
-                           connection_id_tracking_with_decommissioned_node
-                          ]}
+     {cluster_size_5, [],
+      [
+       connection_id_tracking,
+       connection_id_tracking_on_nodedown,
+       connection_id_tracking_with_decommissioned_node
+      ]}
     ].
 
 suite() ->
-    [{timetrap, {minutes, 5}}].
+    [{timetrap, {minutes, 3}}].
 
 %% -------------------------------------------------------------------
 %% Testsuite setup/teardown.
 %% -------------------------------------------------------------------
 
 merge_app_env(Config) ->
-    rabbit_ct_helpers:merge_app_env(Config,
-                                    {rabbit, [
-                                              {collect_statistics, basic},
-                                              {collect_statistics_interval, 100}
-                                             ]}).
+    rabbit_ct_helpers:merge_app_env(
+      Config,
+      {rabbit, [
+                {collect_statistics, basic},
+                {collect_statistics_interval, 100}
+               ]}).
 
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
@@ -65,20 +63,9 @@ init_per_suite(Config) ->
 end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config).
 
-init_per_group(cluster_size_3, Config) ->
-    case rabbit_ct_helpers:is_mixed_versions() of
-        true ->
-            {skip, "maintenance mode wrongly closes cluster-wide MQTT connections "
-             " in RMQ < 3.11.2 and < 3.10.10"};
-        false ->
-            set_cluster_size(3, Config)
-    end;
 init_per_group(cluster_size_5, Config) ->
-    set_cluster_size(5, Config).
-
-set_cluster_size(NodesCount, Config) ->
     rabbit_ct_helpers:set_config(
-      Config, [{rmq_nodes_count, NodesCount}]).
+      Config, [{rmq_nodes_count, 5}]).
 
 end_per_group(_, Config) ->
     Config.
@@ -107,28 +94,6 @@ end_per_testcase(Testcase, Config) ->
 %% Test cases
 %% -------------------------------------------------------------------
 
-maintenance(Config) ->
-    C0 = connect(<<"client-0">>, Config, 0, ?OPTS),
-    C1a = connect(<<"client-1a">>, Config, 1, ?OPTS),
-    C1b = connect(<<"client-1b">>, Config, 1, ?OPTS),
-
-    timer:sleep(500),
-
-    ok = drain_node(Config, 2),
-    ok = revive_node(Config, 2),
-    timer:sleep(500),
-    [?assert(erlang:is_process_alive(C)) || C <- [C0, C1a, C1b]],
-
-    process_flag(trap_exit, true),
-    ok = drain_node(Config, 1),
-    [await_disconnection(Pid) || Pid <- [C1a, C1b]],
-    ok = revive_node(Config, 1),
-    ?assert(erlang:is_process_alive(C0)),
-
-    ok = drain_node(Config, 0),
-    await_disconnection(C0),
-    ok = revive_node(Config, 0).
-
 %% Note about running this testsuite in a mixed-versions cluster:
 %% All even-numbered nodes will use the same code base when using a
 %% secondary Umbrella. Odd-numbered nodes might use an incompatible code
@@ -147,20 +112,20 @@ connection_id_tracking(Config) ->
     C1 = connect(Id, Config, 0, ?OPTS),
     {ok, _, _} = emqtt:subscribe(C1, <<"TopicA">>, qos0),
     ok = emqtt:publish(C1, <<"TopicA">>, <<"Payload">>),
-    ok = expect_publishes(<<"TopicA">>, [<<"Payload">>]),
+    ok = expect_publishes(C1, <<"TopicA">>, [<<"Payload">>]),
 
     %% there's one connection
     assert_connection_count(Config, 4, 2, 1),
 
     %% connect to the same node (A or 0)
     process_flag(trap_exit, true),
     C2 = connect(Id, Config, 0, ?OPTS),
-    await_disconnection(C1),
+    await_exit(C1),
     assert_connection_count(Config, 4, 2, 1),
 
     %% connect to a different node (C or 2)
     C3 = connect(Id, Config, 2, ?OPTS),
-    await_disconnection(C2),
+    await_exit(C2),
     assert_connection_count(Config, 4, 2, 1),
     ok = emqtt:disconnect(C3).
 
@@ -169,29 +134,29 @@ connection_id_tracking_on_nodedown(Config) ->
     C = connect(<<"simpleClient">>, Config, ?OPTS),
     {ok, _, _} = emqtt:subscribe(C, <<"TopicA">>, qos0),
     ok = emqtt:publish(C, <<"TopicA">>, <<"Payload">>),
-    ok = expect_publishes(<<"TopicA">>, [<<"Payload">>]),
+    ok = expect_publishes(C, <<"TopicA">>, [<<"Payload">>]),
     assert_connection_count(Config, 4, 2, 1),
     process_flag(trap_exit, true),
     ok = stop_node(Config, Server),
-    await_disconnection(C),
+    await_exit(C),
     assert_connection_count(Config, 4, 2, 0),
     ok.
 
 connection_id_tracking_with_decommissioned_node(Config) ->
-    case rpc(Config, 0, rabbit_mqtt_ff, track_client_id_in_ra, []) of
+    case rpc(Config, rabbit_mqtt_ff, track_client_id_in_ra, []) of
         false ->
             {skip, "This test requires client ID tracking in Ra"};
         true ->
             Server = get_node_config(Config, 0, nodename),
             C = connect(<<"simpleClient">>, Config, ?OPTS),
             {ok, _, _} = emqtt:subscribe(C, <<"TopicA">>, qos0),
             ok = emqtt:publish(C, <<"TopicA">>, <<"Payload">>),
-            ok = expect_publishes(<<"TopicA">>, [<<"Payload">>]),
+            ok = expect_publishes(C, <<"TopicA">>, [<<"Payload">>]),
 
             assert_connection_count(Config, 4, 2, 1),
             process_flag(trap_exit, true),
             {ok, _} = rabbitmqctl(Config, 0, ["decommission_mqtt_node", Server]),
-            await_disconnection(C),
+            await_exit(C),
             assert_connection_count(Config, 4, 2, 0),
             ok
     end.
@@ -211,10 +176,3 @@ assert_connection_count(Config, Retries, NodeId, NumElements) ->
             timer:sleep(500),
             assert_connection_count(Config, Retries-1, NodeId, NumElements)
     end.
-
-await_disconnection(Client) ->
-    receive
-        {'EXIT', Client, _} -> ok
-    after
-        20_000 -> ct:fail({missing_exit, Client})
-    end.

deps/rabbitmq_mqtt/test/event_recorder.erl

@@ -15,9 +15,10 @@
 init(_) ->
     {ok, ?INIT_STATE}.
 
-handle_event(#event{type = node_stats}, State) ->
-    {ok, State};
-handle_event(#event{type = node_node_stats}, State) ->
+handle_event(#event{type = T}, State)
+  when T =:= node_stats orelse
+       T =:= node_node_stats orelse
+       T =:= node_node_deleted ->
     {ok, State};
 handle_event(Event, State) ->
     {ok, [Event | State]}.

deps/rabbitmq_mqtt/test/ff_SUITE.erl

@@ -13,7 +13,7 @@
 
 -import(rabbit_ct_broker_helpers, [rpc/5]).
 -import(rabbit_ct_helpers, [eventually/1]).
--import(util, [expect_publishes/2,
+-import(util, [expect_publishes/3,
                get_global_counters/4,
                connect/2,
                connect/4]).
@@ -98,7 +98,7 @@ rabbit_mqtt_qos0_queue(Config) ->
     C1 = connect(ClientId, Config),
     {ok, _, [0]} = emqtt:subscribe(C1, Topic, qos0),
     ok = emqtt:publish(C1, Topic, Msg, qos0),
-    ok = expect_publishes(Topic, [Msg]),
+    ok = expect_publishes(C1, Topic, [Msg]),
     ?assertEqual(1,
                  length(rpc(Config, 0, rabbit_amqqueue, list_by_type, [rabbit_classic_queue]))),
 
@@ -109,7 +109,7 @@ rabbit_mqtt_qos0_queue(Config) ->
     ?assertEqual(1,
                  length(rpc(Config, 0, rabbit_amqqueue, list_by_type, [rabbit_classic_queue]))),
     ok = emqtt:publish(C1, Topic, Msg, qos0),
-    ok = expect_publishes(Topic, [Msg]),
+    ok = expect_publishes(C1, Topic, [Msg]),
     ?assertMatch(#{messages_delivered_total := 2,
                    messages_delivered_consume_auto_ack_total := 2},
                  get_global_counters(Config, ?PROTO_VER, 0, [{queue_type, rabbit_classic_queue}])),
@@ -125,7 +125,7 @@ rabbit_mqtt_qos0_queue(Config) ->
     ?assertEqual(1,
                  length(rpc(Config, 0, rabbit_amqqueue, list_by_type, [FeatureFlag]))),
     ok = emqtt:publish(C2, Topic, Msg, qos0),
-    ok = expect_publishes(Topic, [Msg]),
+    ok = expect_publishes(C2, Topic, [Msg]),
     ?assertMatch(#{messages_delivered_total := 1,
                    messages_delivered_consume_auto_ack_total := 1},
                  get_global_counters(Config, ?PROTO_VER, 0, [{queue_type, FeatureFlag}])),