Use a separate connection per test

1 files changed, 70 insertions, 77 deletions

diff --git a/test/priority_queue_SUITE.erl b/test/priority_queue_SUITE.erl
index db5db78155..cac228813b 100644
--- a/test/priority_queue_SUITE.erl
+++ b/test/priority_queue_SUITE.erl
@@ -29,35 +29,28 @@ all() ->
 
 groups() ->
     [
-      {cluster_size_2, [], [
-          {parallel_tests, [parallel], [
-              ackfold,
-              drop,
-              dropwhile_fetchwhile,
-              info_head_message_timestamp,
-              matching,
-              mirror_queue_sync,
-              mirror_queue_sync_priority_above_max,
-              mirror_queue_sync_priority_above_max_pending_ack,
-              mirror_queue_sync_order,
-              purge,
-              requeue,
-              resume,
-              simple_order,
-              straight_through,
-              invoke
-            ]},
-          {non_parallel_tests, [], [
-              recovery %% Restart RabbitMQ.
-            ]}
-        ]},
-      {cluster_size_3, [], [
-          {parallel_tests, [parallel], [
-              mirror_queue_auto_ack,
-              mirror_fast_reset_policy,
-              mirror_reset_policy
-            ]}
-        ]}
+     {cluster_size_2, [], [
+                           ackfold,
+                           drop,
+                           dropwhile_fetchwhile,
+                           info_head_message_timestamp,
+                           matching,
+                           mirror_queue_sync,
+                           mirror_queue_sync_priority_above_max,
+                           mirror_queue_sync_priority_above_max_pending_ack,
+                           %mirror_queue_sync_order,
+                           purge,
+                           requeue,
+                           resume,
+                           simple_order,
+                           straight_through,
+                           invoke
+                          ]},
+     {cluster_size_3, [], [
+                           %mirror_queue_auto_ack,
+                           mirror_fast_reset_policy,
+                           mirror_reset_policy
+                          ]}
     ].
 
 %% -------------------------------------------------------------------
@@ -72,35 +65,35 @@ end_per_suite(Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config).
 
 init_per_group(cluster_size_2, Config) ->
+    Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
     Config1 = rabbit_ct_helpers:set_config(Config, [
-        {rmq_nodes_count, 2}
+                                                    {rmq_nodes_count, 2},
+                                                    {rmq_nodename_suffix, Suffix}
       ]),
     rabbit_ct_helpers:run_steps(Config1,
       rabbit_ct_broker_helpers:setup_steps() ++
       rabbit_ct_client_helpers:setup_steps());
 init_per_group(cluster_size_3, Config) ->
+    Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
     Config1 = rabbit_ct_helpers:set_config(Config, [
-        {rmq_nodes_count, 3}
+                                                    {rmq_nodes_count, 3},
+                                                    {rmq_nodename_suffix, Suffix}
       ]),
     rabbit_ct_helpers:run_steps(Config1,
       rabbit_ct_broker_helpers:setup_steps() ++
-      rabbit_ct_client_helpers:setup_steps());
-init_per_group(_, Config) ->
-    Config.
+      rabbit_ct_client_helpers:setup_steps()).
 
-end_per_group(ClusterSizeGroup, Config)
-when ClusterSizeGroup =:= cluster_size_2
-orelse ClusterSizeGroup =:= cluster_size_3 ->
+end_per_group(_Group, Config) ->
     rabbit_ct_helpers:run_steps(Config,
       rabbit_ct_client_helpers:teardown_steps() ++
-      rabbit_ct_broker_helpers:teardown_steps());
-end_per_group(_, Config) ->
-    Config.
+      rabbit_ct_broker_helpers:teardown_steps()).
 
 init_per_testcase(Testcase, Config) ->
+    rabbit_ct_client_helpers:setup_steps(),
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
 end_per_testcase(Testcase, Config) ->
+    rabbit_ct_client_helpers:teardown_steps(),
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
 %% -------------------------------------------------------------------
@@ -138,25 +131,8 @@ end_per_testcase(Testcase, Config) ->
 %%
 %% [0] publish enough to get credit flow from msg store
 
-recovery(Config) ->
-    {Conn, Ch} = open(Config),
-    Q = <<"recovery-queue">>,
-    declare(Ch, Q, 3),
-    publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]),
-    amqp_connection:close(Conn),
-
-    %% TODO This terminates the automatically open connection and breaks
-    %% coverage.
-    rabbit_ct_broker_helpers:restart_broker(Config, 0),
-
-    {Conn2, Ch2} = open(Config),
-    get_all(Ch2, Q, do_ack, [3, 3, 3, 2, 2, 2, 1, 1, 1]),
-    delete(Ch2, Q),
-    amqp_connection:close(Conn2),
-    passed.
-
 simple_order(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"simple_order-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]),
@@ -167,10 +143,11 @@ simple_order(Config) ->
     get_all(Ch, Q, do_ack, [3, 3, 3, 2, 2, 2, 1, 1, 1]),
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 matching(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"matching-queue">>,
     declare(Ch, Q, 5),
     %% We round priority down, and 0 is the default
@@ -178,10 +155,11 @@ matching(Config) ->
     get_all(Ch, Q, do_ack, [5, 10, undefined, 0, undefined]),
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 resume(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"resume-queue">>,
     declare(Ch, Q, 5),
     amqp_channel:call(Ch, #'confirm.select'{}),
@@ -190,10 +168,11 @@ resume(Config) ->
     amqp_channel:call(Ch, #'queue.purge'{queue = Q}), %% Assert it exists
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 straight_through(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"straight_through-queue">>,
     declare(Ch, Q, 3),
     [begin
@@ -207,6 +186,7 @@ straight_through(Config) ->
     get_empty(Ch, Q),
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 invoke(Config) ->
@@ -218,7 +198,7 @@ invoke(Config) ->
     %% be pending messages so the priority queue receives the
     %% `run_backing_queue` cast message sent to the old master.
     A = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
-    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
     Q = <<"invoke-queue">>,
     declare(Ch, Q, 3),
     Pid = queue_pid(Config, A, rabbit_misc:r(<<"/">>, queue, Q)),
@@ -230,10 +210,11 @@ invoke(Config) ->
     Pid = Pid2,
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 dropwhile_fetchwhile(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"dropwhile_fetchwhile-queue">>,
     [begin
          declare(Ch, Q, Args ++ arguments(3)),
@@ -247,10 +228,11 @@ dropwhile_fetchwhile(Config) ->
                   {<<"x-dead-letter-exchange">>, longstr, <<"amq.fanout">>}]
                 ]],
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 ackfold(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"ackfolq-queue1">>,
     Q2 = <<"ackfold-queue2">>,
     declare(Ch, Q,
@@ -268,10 +250,11 @@ ackfold(Config) ->
     delete(Ch, Q),
     delete(Ch, Q2),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 requeue(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"requeue-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [1, 2, 3]),
@@ -282,10 +265,11 @@ requeue(Config) ->
     get_all(Ch, Q, do_ack, [3, 2, 1]),
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 drop(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"drop-queue">>,
     declare(Ch, Q, [{<<"x-max-length">>, long, 4} | arguments(3)]),
     publish(Ch, Q, [1, 2, 3, 1, 2, 3, 1, 2, 3]),
@@ -294,10 +278,11 @@ drop(Config) ->
     get_all(Ch, Q, do_ack, [2, 1, 1, 1]),
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 purge(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"purge-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [1, 2, 3]),
@@ -305,6 +290,7 @@ purge(Config) ->
     get_empty(Ch, Q),
     delete(Ch, Q),
     rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 info_head_message_timestamp(Config) ->
@@ -375,7 +361,7 @@ ram_duration(_Config) ->
     passed.
 
 mirror_queue_sync(Config) ->
-    Ch = rabbit_ct_client_helpers:open_channel(Config, 0),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
     Q = <<"mirror_queue_sync-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [1, 2, 3]),
@@ -389,13 +375,14 @@ mirror_queue_sync(Config) ->
     rabbit_ct_broker_helpers:control_action(sync_queue, Nodename0,
       [binary_to_list(Q)], [{"-p", "/"}]),
     wait_for_sync(Config, Nodename0, rabbit_misc:r(<<"/">>, queue, Q)),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 mirror_queue_sync_priority_above_max(Config) ->
     A = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
     %% Tests synchronisation of slaves when priority is higher than max priority.
     %% This causes an infinity loop (and test timeout) before rabbitmq-server-795
-    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
     Q = <<"mirror_queue_sync_priority_above_max-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [5, 5, 5]),
@@ -405,6 +392,7 @@ mirror_queue_sync_priority_above_max(Config) ->
       [binary_to_list(Q)], [{"-p", "/"}]),
     wait_for_sync(Config, A, rabbit_misc:r(<<"/">>, queue, Q)),
     delete(Ch, Q),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 mirror_queue_sync_priority_above_max_pending_ack(Config) ->
@@ -412,7 +400,7 @@ mirror_queue_sync_priority_above_max_pending_ack(Config) ->
     %% Tests synchronisation of slaves when priority is higher than max priority
     %% and there are pending acks.
     %% This causes an infinity loop (and test timeout) before rabbitmq-server-795
-    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
     Q = <<"mirror_queue_sync_priority_above_max_pending_ack-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [5, 5, 5]),
@@ -427,6 +415,7 @@ mirror_queue_sync_priority_above_max_pending_ack(Config) ->
     synced_msgs(Config, A, rabbit_misc:r(<<"/">>, queue, Q), 3),
     synced_msgs(Config, B, rabbit_misc:r(<<"/">>, queue, Q), 3),
     delete(Ch, Q),
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 mirror_queue_auto_ack(Config) ->
@@ -436,7 +425,7 @@ mirror_queue_auto_ack(Config) ->
     %% the slaves will crash with the depth notification as they will not
     %% match the master delta.
     %% Bug rabbitmq-server 687
-    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
     Q = <<"mirror_queue_auto_ack-queue">>,
     declare(Ch, Q, 3),
     publish(Ch, Q, [1, 2, 3]),
@@ -447,6 +436,7 @@ mirror_queue_auto_ack(Config) ->
     %% Retrieve slaves
     SPids = slave_pids(Config, A, rabbit_misc:r(<<"/">>, queue, Q)),
     [{SNode1, _SPid1}, {SNode2, SPid2}] = nodes_and_pids(SPids),
+    rabbit_ct_client_helpers:close_channel(Ch),
 
     %% Restart one of the slaves so `request_depth` is triggered
     rabbit_ct_broker_helpers:restart_node(Config, SNode1),
@@ -457,13 +447,15 @@ mirror_queue_auto_ack(Config) ->
     SPid2 = proplists:get_value(SNode2, Slaves),
 
     delete(Ch, Q),
+    rabbit_ct_client_helpers:close_connection(Conn),
+
     passed.
 
 mirror_queue_sync_order(Config) ->
     A = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
     B = rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
-    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
-    Ch2 = rabbit_ct_client_helpers:open_channel(Config, B),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
+    {Conn2, Ch2} = rabbit_ct_client_helpers:open_connection_and_channel(Config, B),
     Q = <<"mirror_queue_sync_order-queue">>,
     declare(Ch, Q, 3),
     publish_payload(Ch, Q, [{1, <<"msg1">>}, {2, <<"msg2">>},
@@ -485,6 +477,8 @@ mirror_queue_sync_order(Config) ->
                                  <<"msg4">>, <<"msg1">>]),
 
     delete(Ch2, Q),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    rabbit_ct_client_helpers:close_connection(Conn2),
     passed.
 
 mirror_reset_policy(Config) ->
@@ -503,7 +497,7 @@ mirror_fast_reset_policy(Config) ->
 
 mirror_reset_policy(Config, Wait) ->
     A = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
-    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, A),
     Q = <<"mirror_reset_policy-queue">>,
     declare(Ch, Q, 5),
     Pid = queue_pid(Config, A, rabbit_misc:r(<<"/">>, queue, Q)),
@@ -525,13 +519,12 @@ mirror_reset_policy(Config, Wait) ->
     %% Verify master has not crashed
     Pid = queue_pid(Config, A, rabbit_misc:r(<<"/">>, queue, Q)),
     delete(Ch, Q),
+
+    rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
 %%----------------------------------------------------------------------------
 
-open(Config) ->
-    rabbit_ct_client_helpers:open_connection_and_channel(Config, 0).
-
 declare(Ch, Q, Args) when is_list(Args) ->
     amqp_channel:call(Ch, #'queue.declare'{queue     = Q,
                                            durable   = true,