test: test message deduplication with no duplication header

Signed-off-by: Matteo Cafasso <[email protected]>

Author: noxdafox

test/exchange_SUITE.erl

@@ -90,20 +90,30 @@ declare_exchanges(Config) ->
     ?assertExit(_, amqp_channel:call(Channel, DeclareErr)).
 
 deduplicate_message(Config) ->
+    Get = #'basic.get'{queue = <<"test">>},
     Channel = rabbit_ct_client_helpers:open_channel(Config),
 
     #'exchange.declare_ok'{} = amqp_channel:call(
                                  Channel, make_exchange(<<"test">>, 10, 10000)),
     bind_new_queue(Channel, <<"test">>, <<"test">>),
 
+    %% Deduplication header present
     publish_message(Channel, <<"test">>, "deduplicate-this"),
     publish_message(Channel, <<"test">>, "deduplicate-this"),
 
-    timer:sleep(2000),
+    timer:sleep(1000),
+
+    {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
+    #'basic.get_empty'{} = amqp_channel:call(Channel, Get),
+
+    %% Deduplication header absent
+    publish_message(Channel, <<"test">>),
+    publish_message(Channel, <<"test">>),
+
+    timer:sleep(1000),
 
-    Get = #'basic.get'{queue = <<"test">>},
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
-    #'basic.get_empty'{} = amqp_channel:call(Channel, Get).
+    {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
 
 deduplicate_message_ttl(Config) ->
     Channel = rabbit_ct_client_helpers:open_channel(Config),
@@ -131,7 +141,7 @@ deduplicate_message_cache_overflow(Config) ->
     publish_message(Channel, <<"test">>, "deduplicate-that"),
     publish_message(Channel, <<"test">>, "deduplicate-this"),
 
-    timer:sleep(2000),
+    timer:sleep(1000),
 
     Get = #'basic.get'{queue = <<"test">>},
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
@@ -156,6 +166,11 @@ bind_new_queue(Ch, Ex, Q) ->
     Binding = #'queue.bind'{queue = Q, exchange = Ex, routing_key = <<"#">>},
     #'queue.bind_ok'{} = amqp_channel:call(Ch, Binding).
 
+publish_message(Ch, Ex) ->
+    Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
+    Msg = #amqp_msg{payload = <<"payload">>},
+    amqp_channel:cast(Ch, Publish, Msg).
+
 publish_message(Ch, Ex, D) ->
     Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
     Props = #'P_basic'{headers = [{<<"x-deduplication-header">>, longstr, D}]},

test/queue_SUITE.erl

@@ -65,19 +65,29 @@ end_per_testcase(Testcase, Config) ->
 %% -------------------------------------------------------------------
 
 deduplicate_message(Config) ->
+    Get = #'basic.get'{queue = <<"test">>},
     Channel = rabbit_ct_client_helpers:open_channel(Config),
 
     #'queue.declare_ok'{} = amqp_channel:call(Channel, make_queue(<<"test">>)),
     bind_new_exchange(Channel, <<"test">>, <<"test">>),
 
+    %% Deduplication header present
     publish_message(Channel, <<"test">>, "deduplicate-this"),
     publish_message(Channel, <<"test">>, "deduplicate-this"),
 
-    timer:sleep(2000),
+    timer:sleep(1000),
 
-    Get = #'basic.get'{queue = <<"test">>},
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
-    #'basic.get_empty'{} = amqp_channel:call(Channel, Get).
+    #'basic.get_empty'{} = amqp_channel:call(Channel, Get),
+
+    %% Deduplication header absent
+    publish_message(Channel, <<"test">>),
+    publish_message(Channel, <<"test">>),
+
+    timer:sleep(1000),
+
+    {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
+    {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
 
 deduplicate_message_ttl(Config) ->
     Channel = rabbit_ct_client_helpers:open_channel(Config),
@@ -161,6 +171,11 @@ bind_new_exchange(Ch, Ex, Q) ->
     Binding = #'queue.bind'{queue = Q, exchange = Ex, routing_key = <<"#">>},
     #'queue.bind_ok'{} = amqp_channel:call(Ch, Binding).
 
+publish_message(Ch, Ex) ->
+    Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
+    Msg = #amqp_msg{payload = <<"payload">>},
+    amqp_channel:cast(Ch, Publish, Msg).
+
 publish_message(Ch, Ex, D) ->
     Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
     Props = #'P_basic'{headers = [{<<"x-deduplication-header">>, longstr, D}]},