Merge pull request #8694 from rabbitmq/test-resilience

Tests: more resilient time-dependent tests

Author: dumbbell

deps/rabbit/BUILD.bazel

@@ -377,7 +377,6 @@ rabbitmq_integration_suite(
     additional_beam = [
         ":test_quorum_queue_utils_beam",
     ],
-    flaky = True,
     shard_count = 7,
 )
 
@@ -565,7 +564,6 @@ rabbitmq_integration_suite(
 rabbitmq_integration_suite(
     name = "per_user_connection_channel_limit_partitions_SUITE",
     size = "large",
-    flaky = True,
 )
 
 rabbitmq_integration_suite(
@@ -1045,7 +1043,6 @@ rabbitmq_integration_suite(
     additional_beam = [
         "test/test_rabbit_event_handler.beam",
     ],
-    flaky = True,
 )
 
 rabbitmq_suite(

deps/rabbit/Makefile

@@ -149,7 +149,7 @@ TEST_DEPS = rabbitmq_ct_helpers rabbitmq_ct_client_helpers amqp_client meck prop
 PLT_APPS += mnesia
 
 dep_syslog = git https://github.com/schlagert/syslog 4.0.0
-dep_osiris = git https://github.com/rabbitmq/osiris v1.6.0
+dep_osiris = git https://github.com/rabbitmq/osiris v1.6.1
 dep_systemd = hex 0.6.1
 dep_seshat = hex 0.4.0
 

deps/rabbit/app.bzl

@@ -880,7 +880,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/definition_import_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/rabbit_common:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "deprecated_features_SUITE_beam_files",
@@ -932,7 +932,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/dynamic_qq_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "eager_sync_SUITE_beam_files",
@@ -1111,7 +1111,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/per_user_connection_tracking_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "per_vhost_connection_limit_SUITE_beam_files",
@@ -1147,7 +1147,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/per_vhost_queue_limit_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "policy_SUITE_beam_files",
@@ -1218,7 +1218,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/queue_master_location_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "queue_parallel_SUITE_beam_files",
@@ -1726,7 +1726,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/unit_log_management_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "unit_operator_policy_SUITE_beam_files",
@@ -1828,7 +1828,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/vhost_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "rabbit_cuttlefish_SUITE_beam_files",

deps/rabbit/test/dead_lettering_SUITE.erl

@@ -188,14 +188,15 @@ init_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_started(Config1, Testcase).
 
 end_per_testcase(Testcase, Config) ->
+    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_queues, []),
     {_, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
-    amqp_channel:call(Ch, #'queue.delete'{queue = ?config(queue_name, Config)}),
-    amqp_channel:call(Ch, #'queue.delete'{queue = ?config(queue_name_dlx, Config)}),
-    amqp_channel:call(Ch, #'queue.delete'{queue = ?config(queue_name_dlx_2, Config)}),
     amqp_channel:call(Ch, #'exchange.delete'{exchange = ?config(dlx_exchange, Config)}),
     _ = rabbit_ct_broker_helpers:clear_policy(Config, 0, ?config(policy, Config)),
     rabbit_ct_helpers:testcase_finished(Config, Testcase).
 
+delete_queues() ->
+    [rabbit_amqqueue:delete(Q, false, false, <<"tests">>) || Q <- rabbit_amqqueue:list()].
+
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% Dead letter exchanges
 %%
@@ -850,7 +851,15 @@ dead_letter_policy(Config) ->
                                              <<"queues">>,
                                              [{<<"dead-letter-exchange">>, DLXExchange},
                                               {<<"dead-letter-routing-key">>, DLXQName}]),
-    timer:sleep(1000),
+    ?awaitMatch([_ | _],
+                begin
+                    {ok, Q0} = rabbit_ct_broker_helpers:rpc(
+                                 Config, 0,
+                                 rabbit_amqqueue, lookup,
+                                 [rabbit_misc:r(<<"/">>, queue, QName)], infinity),
+                    amqqueue:get_policy(Q0)
+                end,
+                30000),
     %% Nack the second message
     amqp_channel:cast(Ch, #'basic.nack'{delivery_tag = DTag2,
                                         multiple     = false,
@@ -1319,7 +1328,12 @@ dead_letter_headers_first_death_route(Config) ->
     %% Send and reject the 3rd message.
     P3 = <<"msg3">>,
     publish(Ch, QName2, [P3]),
-    timer:sleep(1000),
+    case group_name(Config) of
+        at_most_once ->
+            wait_for_messages(Config, [[QName2, <<"1">>, <<"1">>, <<"0">>]]);
+        at_least_once ->
+            wait_for_messages(Config, [[QName2, <<"2">>, <<"1">>, <<"0">>]])
+    end,
     [DTag] = consume(Ch, QName2, [P3]),
     amqp_channel:cast(Ch, #'basic.reject'{delivery_tag = DTag,
                                           requeue      = false}),

deps/rabbit/test/definition_import_SUITE.erl

@@ -7,6 +7,7 @@
 
 -module(definition_import_SUITE).
 
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include_lib("eunit/include/eunit.hrl").
@@ -245,8 +246,7 @@ import_case13a(Config) ->
     %% We expect that importing an existing queue (i.e. same vhost and name)
     %% but with different arguments and different properties is a no-op.
     import_file_case(Config, "case13a"),
-    timer:sleep(1000),
-    ?assertMatch({ok, Q}, queue_lookup(Config, VHost, QueueName)).
+    ?awaitMatch({ok, Q}, queue_lookup(Config, VHost, QueueName), 30000).
 
 import_case14(Config) -> import_file_case(Config, "case14").
 %% contains a user with tags as a list

deps/rabbit/test/dynamic_qq_SUITE.erl

@@ -9,6 +9,7 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
 -import(quorum_queue_utils, [wait_for_messages_ready/3,
                              ra_name/1]).
@@ -179,21 +180,29 @@ quorum_unaffected_after_vhost_failure(Config) ->
                                             arguments = Args,
                                             durable = true
                                            }),
-    timer:sleep(300),
-
-    Info0 = rpc:call(A, rabbit_quorum_queue, infos,
-                     [rabbit_misc:r(<<"/">>, queue, QName)]),
-    ?assertEqual(Servers, lists:sort(proplists:get_value(online, Info0, []))),
+    ?awaitMatch(
+       Servers,
+       begin
+           Info0 = rpc:call(A, rabbit_quorum_queue, infos,
+                            [rabbit_misc:r(<<"/">>, queue, QName)]),
+           lists:sort(proplists:get_value(online, Info0, []))
+       end,
+       60000),
 
     %% Crash vhost on both nodes
     {ok, SupA} = rabbit_ct_broker_helpers:rpc(Config, A, rabbit_vhost_sup_sup, get_vhost_sup, [<<"/">>]),
     exit(SupA, foo),
     {ok, SupB} = rabbit_ct_broker_helpers:rpc(Config, B, rabbit_vhost_sup_sup, get_vhost_sup, [<<"/">>]),
     exit(SupB, foo),
 
-    Info = rpc:call(A, rabbit_quorum_queue, infos,
-                    [rabbit_misc:r(<<"/">>, queue, QName)]),
-    ?assertEqual(Servers, lists:sort(proplists:get_value(online, Info, []))).
+    ?awaitMatch(
+       Servers,
+       begin
+           Info = rpc:call(A, rabbit_quorum_queue, infos,
+                            [rabbit_misc:r(<<"/">>, queue, QName)]),
+           lists:sort(proplists:get_value(online, Info, []))
+       end,
+       60000).
 
 recover_follower_after_standalone_restart(Config) ->
     case rabbit_ct_helpers:is_mixed_versions() of

deps/rabbit/test/per_user_connection_channel_limit_partitions_SUITE.erl

@@ -105,7 +105,10 @@ cluster_full_partition_with_autoheal(Config) ->
     %% B drops off the network, non-reachable by either A or C
     rabbit_ct_broker_helpers:block_traffic_between(A, B),
     rabbit_ct_broker_helpers:block_traffic_between(B, C),
-    timer:sleep(?DELAY),
+    LargeCluster = lists:sort([A, C]),
+    ?awaitMatch(LargeCluster, list_running(Config, A), 60000, 3000),
+    ?awaitMatch([B], list_running(Config, B), 60000, 3000),
+    ?awaitMatch(LargeCluster, list_running(Config, C), 60000, 3000),
 
     %% A and C are still connected, so 4 connections are tracked
     %% All connections to B are dropped
@@ -116,7 +119,10 @@ cluster_full_partition_with_autoheal(Config) ->
 
     rabbit_ct_broker_helpers:allow_traffic_between(A, B),
     rabbit_ct_broker_helpers:allow_traffic_between(B, C),
-    timer:sleep(?DELAY),
+    All = lists:sort([A, B, C]),
+    ?awaitMatch(All, list_running(Config, A), 60000, 3000),
+    ?awaitMatch(All, list_running(Config, B), 60000, 3000),
+    ?awaitMatch(All, list_running(Config, C), 60000, 3000),
 
     %% during autoheal B's connections were dropped
     ?awaitMatch({4, 10},
@@ -167,3 +173,18 @@ tracked_list_of_user(Config, NodeIndex, TrackingMod, Username) ->
     rabbit_ct_broker_helpers:rpc(Config, NodeIndex,
                                  TrackingMod,
                                  list_of_user, [Username]).
+
+list_running(Config, NodeIndex) ->
+    Ret = (catch rabbit_ct_broker_helpers:rpc(Config, NodeIndex, rabbit_nodes, list_running, [])),
+    Running = case Ret of
+                  {'EXIT', {{exception, undef, _}, _}} ->
+                      rabbit_ct_broker_helpers:rpc(Config, NodeIndex, rabbit_mnesia, cluster_nodes, [running]);
+                  _ ->
+                      Ret
+              end,
+    case Running of
+        List when is_list(List) ->
+            lists:sort(List);
+        Any ->
+            Any
+    end.