Merge pull request #12834 from rabbitmq/fix-flakes

More fixes for test flakes: AMQP 1.0 client, management clustering property-based tests

Author: michaelklishin

deps/amqp10_client/test/system_SUITE.erl

@@ -12,10 +12,10 @@
 
 -include_lib("amqp10_common/include/amqp10_framing.hrl").
 
--include("src/amqp10_client.hrl").
-
 -compile([export_all, nowarn_export_all]).
 
+-define(TIMEOUT, 30000).
+
 suite() ->
     [{timetrap, {minutes, 4}}].
 
@@ -184,7 +184,7 @@ open_close_connection(Config) ->
     {ok, Connection2} = amqp10_client:open_connection(OpnConf),
     receive
         {amqp10_event, {connection, Connection2, opened}} -> ok
-    after 5000 -> exit(connection_timeout)
+    after ?TIMEOUT -> exit(connection_timeout)
     end,
     ok = amqp10_client:close_connection(Connection2),
     ok = amqp10_client:close_connection(Connection).
@@ -201,7 +201,7 @@ open_connection_plain_sasl(Config) ->
     {ok, Connection} = amqp10_client:open_connection(OpnConf),
     receive
         {amqp10_event, {connection, Connection, opened}} -> ok
-    after 5000 -> exit(connection_timeout)
+    after ?TIMEOUT -> exit(connection_timeout)
     end,
     ok = amqp10_client:close_connection(Connection).
 
@@ -225,7 +225,7 @@ open_connection_plain_sasl_parse_uri(Config) ->
     {ok, Connection} = amqp10_client:open_connection(OpnConf),
     receive
         {amqp10_event, {connection, Connection, opened}} -> ok
-    after 5000 -> exit(connection_timeout)
+    after ?TIMEOUT -> exit(connection_timeout)
     end,
     ok = amqp10_client:close_connection(Connection).
 
@@ -245,7 +245,7 @@ open_connection_plain_sasl_failure(Config) ->
         % some implementation may simply close the tcp_connection
         {amqp10_event, {connection, Connection, {closed, shutdown}}} -> ok
 
-    after 5000 ->
+    after ?TIMEOUT ->
               ct:pal("Connection process is alive? = ~tp~n",
                      [erlang:is_process_alive(Connection)]),
               exit(connection_timeout)
@@ -469,27 +469,27 @@ notify_with_performative(Config) ->
 
     {ok, Connection} = amqp10_client:open_connection(OpenConf),
     receive {amqp10_event, {connection, Connection, {opened, #'v1_0.open'{}}}} -> ok
-    after 5000 -> ct:fail({missing_event, ?LINE})
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
     end,
 
     {ok, Session1} = amqp10_client:begin_session(Connection),
     receive {amqp10_event, {session, Session1, {begun, #'v1_0.begin'{}}}} -> ok
-    after 5000 -> ct:fail({missing_event, ?LINE})
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
     end,
 
     {ok, Sender1} = amqp10_client:attach_sender_link(Session1, <<"sender 1">>, <<"/exchanges/amq.fanout">>),
     receive {amqp10_event, {link, Sender1, {attached, #'v1_0.attach'{}}}} -> ok
-    after 5000 -> ct:fail({missing_event, ?LINE})
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
     end,
 
     ok = amqp10_client:detach_link(Sender1),
     receive {amqp10_event, {link, Sender1, {detached, #'v1_0.detach'{}}}} -> ok
-    after 5000 -> ct:fail({missing_event, ?LINE})
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
     end,
 
     ok = amqp10_client:end_session(Session1),
     receive {amqp10_event, {session, Session1, {ended, #'v1_0.end'{}}}} -> ok
-    after 5000 -> ct:fail({missing_event, ?LINE})
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
     end,
 
     %% Test that the amqp10_client:*_sync functions work.
@@ -501,7 +501,7 @@ notify_with_performative(Config) ->
 
     ok = amqp10_client:close_connection(Connection),
     receive {amqp10_event, {connection, Connection, {closed, #'v1_0.close'{}}}} -> ok
-    after 5000 -> ct:fail({missing_event, ?LINE})
+    after ?TIMEOUT -> ct:fail({missing_event, ?LINE})
     end.
 
 % a message is sent before the link attach is guaranteed to
@@ -600,13 +600,13 @@ subscribe(Config) ->
     [begin
          receive {amqp10_msg, Receiver, Msg} ->
                      ok = amqp10_client:accept_msg(Receiver, Msg)
-         after 2000 -> ct:fail(timeout)
+         after ?TIMEOUT -> ct:fail(timeout)
          end
      end || _ <- lists:seq(1, 10)],
     ok = assert_no_message(Receiver),
 
     receive {amqp10_event, {link, Receiver, credit_exhausted}} -> ok
-    after 5000 -> flush(),
+    after ?TIMEOUT -> flush(),
                   exit(credit_exhausted_assert)
     end,
 
@@ -854,7 +854,7 @@ multi_transfer_without_delivery_id(Config) ->
     receive
         {amqp10_msg, Receiver, _InMsg} ->
             ok
-    after 2000 ->
+    after ?TIMEOUT ->
               exit(delivery_timeout)
     end,
 
@@ -909,7 +909,7 @@ incoming_heartbeat(Config) ->
           {closed, _}}}
           when Connection0 =:= Connection ->
             ok
-    after 5000 ->
+    after ?TIMEOUT ->
               exit(incoming_heartbeat_assert)
     end,
     demonitor(MockRef).
@@ -926,7 +926,7 @@ await_link(Who, What, Err) ->
         {amqp10_event, {link, Who0, {detached, Why}}}
           when Who0 =:= Who ->
             ct:fail(Why)
-    after 5000 ->
+    after ?TIMEOUT ->
               flush(),
               ct:fail(Err)
     end.
@@ -943,7 +943,7 @@ await_disposition(DeliveryTag) ->
     receive
         {amqp10_disposition, {accepted, DeliveryTag0}}
           when DeliveryTag0 =:= DeliveryTag -> ok
-    after 3000 ->
+    after ?TIMEOUT ->
               flush(),
               ct:fail(dispostion_timeout)
     end.
@@ -955,7 +955,7 @@ count_received_messages0(Receiver, Count) ->
     receive
         {amqp10_msg, Receiver, _Msg} ->
             count_received_messages0(Receiver, Count + 1)
-    after 500 ->
+    after 5000 ->
               Count
     end.
 
@@ -968,7 +968,7 @@ receive_messages0(Receiver, N, Acc) ->
     receive
         {amqp10_msg, Receiver, Msg} ->
             receive_messages0(Receiver, N - 1, [Msg | Acc])
-    after 5000  ->
+    after ?TIMEOUT  ->
               LastReceivedMsg = case Acc of
                                     [] -> none;
                                     [M | _] -> M

deps/rabbit/test/feature_flags_v2_SUITE.erl

@@ -203,7 +203,7 @@ stop_slave_node(Node) ->
             persistent_term:erase({?MODULE, Node}),
 
             ct:pal("- Stopping slave node `~ts`...", [Node]),
-            ok = peer:stop(NodePid)
+            _ = peer:stop(NodePid)
     end.
 
 connect_nodes([FirstNode | OtherNodes] = Nodes) ->

deps/rabbitmq_management/test/clustering_prop_SUITE.erl

@@ -109,17 +109,19 @@ prop_connection_channel_counts(Config) ->
                                  {1, force_stats}])),
             begin
                 % ensure we begin with no connections
+                ct:pal("Init testcase"),
                 true = validate_counts(Config, []),
                 Cons = lists:foldl(fun (Op, Agg) ->
                                           execute_op(Config, Op, Agg)
                                    end, [], Ops),
                 %% TODO retry a few times
+                ct:pal("Check testcase"),
                 Res = retry_for(
                         fun() ->
                                 force_stats(Config),
                                 validate_counts(Config, Cons) end,
                         60),
-                cleanup(Cons),
+                ct:pal("Cleanup testcase"),
                 rabbit_ct_helpers:await_condition(
                   fun () ->
                           cleanup(Cons),
@@ -138,8 +140,16 @@ validate_counts(Config, Conns) ->
     Ch1 = length(http_get_from_node(Config, 0, "/channels")),
     Ch2 = length(http_get_from_node(Config, 1, "/channels")),
     Ch3 = length(http_get_from_node(Config, 2, "/channels")),
-    [Expected, Expected, Expected, ChanCount, ChanCount, ChanCount]
-    =:= [C1, C2, C3, Ch1, Ch2, Ch3].
+    Res = ([Expected, Expected, Expected, ChanCount, ChanCount, ChanCount]
+           =:= [C1, C2, C3, Ch1, Ch2, Ch3]),
+    case Res of
+        false ->
+            ct:pal("Validate counts connections: ~p channels: ~p got ~p",
+                   [Expected, ChanCount, [C1, C2, C3, Ch1, Ch2, Ch3]]);
+        true ->
+            ok
+    end,
+    Res.
 
 
 cleanup(Conns) ->