Skip to content

Commit 059813a

Browse files
dcorbachodcorbacho
committed
Shovel: more common testcases
1 parent 629cf77 commit 059813a

File tree

4 files changed

+85
-89
lines changed

4 files changed

+85
-89
lines changed

‎deps/rabbitmq_shovel/test/amqp091_dynamic_SUITE.erl‎

Lines changed: 0 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,6 @@ groups() ->
4040
restart,
4141
change_definition,
4242
autodelete,
43-
autodelete_with_rejections,
4443
validation,
4544
security_validation,
4645
get_connection_name,
@@ -517,35 +516,6 @@ autodelete_do(Config, {AckMode, After, ExpSrc, ExpDest}) ->
517516
expect_count(Ch, <<"src">>, <<"hello">>, ExpSrc)
518517
end.
519518

520-
autodelete_with_rejections(Config) ->
521-
Src = <<"src">>,
522-
Dest = <<"dst">>,
523-
Args = [{<<"x-max-length">>, long, 5},
524-
{<<"x-overflow">>, longstr, <<"reject-publish">>}],
525-
with_ch(Config,
526-
fun (Ch) ->
527-
amqp_channel:call(Ch, #'queue.declare'{queue = Dest,
528-
durable = true,
529-
arguments = Args}),
530-
shovel_test_utils:set_param(Config, <<"test">>,
531-
[{<<"src-protocol">>, <<"local">>},
532-
{<<"src-queue">>, Src},
533-
{<<"src-delete-after">>, 10},
534-
{<<"dest-protocol">>, <<"local">>},
535-
{<<"dest-predeclared">>, true},
536-
{<<"dest-queue">>, Dest}
537-
]),
538-
publish_count(Ch, <<>>, Src, <<"hello">>, 10),
539-
await_autodelete(Config, <<"test">>),
540-
Expected = lists:sort([[Src, <<"5">>], [Dest, <<"5">>]]),
541-
eventually(
542-
?_assertMatch(
543-
Expected,
544-
lists:sort(rabbit_ct_broker_helpers:rabbitmqctl_list(
545-
Config, 0,
546-
["list_queues", "name", "messages_ready", "--no-table-headers"]))))
547-
end).
548-
549519
validation(Config) ->
550520
URIs = [{<<"src-uri">>, <<"amqp://">>},
551521
{<<"dest-uri">>, <<"amqp://">>}],

‎deps/rabbitmq_shovel/test/amqp10_dynamic_SUITE.erl‎

Lines changed: 0 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@ groups() ->
3333
{non_parallel_tests, [], [
3434
simple,
3535
change_definition,
36-
autodelete_with_rejections,
3736
simple_amqp10_dest,
3837
simple_amqp10_src,
3938
amqp091_to_amqp10_with_dead_lettering,
@@ -266,34 +265,6 @@ change_definition(Config) ->
266265
amqp10_expect_empty(Sess, Dest2)
267266
end).
268267

269-
autodelete_with_rejections(Config) ->
270-
Src = ?config(srcq, Config),
271-
Dest = ?config(destq, Config),
272-
with_amqp10_session(
273-
Config,
274-
fun (Sess) ->
275-
amqp10_declare_queue(Sess, Dest, #{<<"x-max-length">> => {uint, 5},
276-
<<"x-overflow">> => {utf8, <<"reject-publish">>}}),
277-
278-
shovel_test_utils:set_param(Config, ?PARAM,
279-
[{<<"src-protocol">>, <<"local">>},
280-
{<<"src-queue">>, Src},
281-
{<<"src-delete-after">>, 10},
282-
{<<"dest-protocol">>, <<"local">>},
283-
{<<"dest-predeclared">>, true},
284-
{<<"dest-queue">>, Dest}
285-
]),
286-
amqp10_publish(Sess, Src, <<"hello">>, 10),
287-
await_autodelete(Config, <<"test">>),
288-
Expected = lists:sort([[Src, <<"5">>], [Dest, <<"5">>]]),
289-
?awaitMatch(
290-
Expected,
291-
lists:sort(rabbit_ct_broker_helpers:rabbitmqctl_list(
292-
Config, 0,
293-
["list_queues", "name", "messages_ready", "--no-table-headers"])),
294-
30_000)
295-
end).
296-
297268
test_amqp10_delete_after_queue_length(Config) ->
298269
Src = ?config(srcq, Config),
299270
Dest = ?config(destq, Config),

‎deps/rabbitmq_shovel/test/local_dynamic_SUITE.erl‎

Lines changed: 0 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,6 @@ groups() ->
5353
local_to_local_delete_after_never,
5454
local_to_local_delete_after_queue_length,
5555
local_to_local_delete_after_queue_length_zero,
56-
local_to_local_delete_after_with_rejections,
5756
local_to_local_no_ack,
5857
local_to_local_quorum_no_ack,
5958
local_to_local_stream_no_ack,
@@ -559,34 +558,6 @@ local_to_local_delete_after_queue_length(Config) ->
559558
amqp10_expect_empty(Sess, Dest)
560559
end).
561560

562-
local_to_local_delete_after_with_rejections(Config) ->
563-
Src = ?config(srcq, Config),
564-
Dest = ?config(destq, Config),
565-
VHost = <<"/">>,
566-
declare_queue(Config, VHost, Dest, [{<<"x-max-length">>, long, 5},
567-
{<<"x-overflow">>, longstr, <<"reject-publish">>}]),
568-
with_amqp10_session(Config,
569-
fun (Sess) ->
570-
shovel_test_utils:set_param(Config, ?PARAM,
571-
[{<<"src-protocol">>, <<"local">>},
572-
{<<"src-queue">>, Src},
573-
{<<"src-delete-after">>, 10},
574-
{<<"dest-protocol">>, <<"local">>},
575-
{<<"dest-predeclared">>, true},
576-
{<<"dest-queue">>, Dest}
577-
]),
578-
amqp10_publish(Sess, Src, <<"tag1">>, 10),
579-
?awaitMatch(not_found, rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_runtime_parameters, lookup, [<<"/">>, <<"shovel">>, ?PARAM]), 30_000),
580-
Expected = lists:sort([[Src, <<"5">>], [Dest, <<"5">>]]),
581-
?awaitMatch(
582-
Expected,
583-
lists:sort(rabbit_ct_broker_helpers:rabbitmqctl_list(
584-
Config, 0,
585-
["list_queues", "name", "messages_ready", "--no-table-headers"])),
586-
30_000)
587-
588-
end).
589-
590561
local_to_local_no_ack(Config) ->
591562
Src = ?config(srcq, Config),
592563
Dest = ?config(destq, Config),

‎deps/rabbitmq_shovel/test/shovel_dynamic_SUITE.erl‎

Lines changed: 85 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
-compile(export_all).
1616

17+
-import(rabbit_ct_helpers, [eventually/3]).
1718
-import(shovel_test_utils, [await_autodelete/2,
1819
set_param/3,
1920
set_param_nowait/3,
@@ -69,7 +70,11 @@ tests() ->
6970
autodelete_classic_on_confirm_no_transfer,
7071
autodelete_quorum_on_confirm_no_transfer,
7172
autodelete_classic_on_publish_no_transfer,
72-
autodelete_quorum_on_publish_no_transfer
73+
autodelete_quorum_on_publish_no_transfer,
74+
autodelete_classic_on_confirm_with_rejections,
75+
autodelete_quorum_on_confirm_with_rejections,
76+
autodelete_classic_on_publish_with_rejections,
77+
autodelete_quorum_on_publish_with_rejections
7378
].
7479

7580
%% -------------------------------------------------------------------
@@ -304,6 +309,76 @@ autodelete(Config, Type, AckMode, After, ExpSrc, ExpDest) ->
304309
amqp10_expect_count(Sess, Dest, ExpDest)
305310
end).
306311

312+
autodelete_classic_on_confirm_with_rejections(Config) ->
313+
autodelete_with_rejections(Config, <<"classic">>, <<"on-confirm">>, 5, 5).
314+
315+
autodelete_quorum_on_confirm_with_rejections(Config) ->
316+
ExpSrc = fun(ExpDest) -> 100 - ExpDest end,
317+
autodelete_with_quorum_rejections(Config, <<"on-confirm">>, ExpSrc).
318+
319+
autodelete_classic_on_publish_with_rejections(Config) ->
320+
autodelete_with_rejections(Config, <<"classic">>, <<"on-publish">>, 0, 5).
321+
322+
autodelete_quorum_on_publish_with_rejections(Config) ->
323+
ExpSrc = fun(_) -> 0 end,
324+
autodelete_with_quorum_rejections(Config, <<"on-publish">>, ExpSrc).
325+
326+
autodelete_with_rejections(Config, Type, AckMode, ExpSrc, ExpDest) ->
327+
Src = ?config(srcq, Config),
328+
Dest = ?config(destq, Config),
329+
with_amqp10_session(
330+
Config,
331+
fun (Sess) ->
332+
amqp10_declare_queue(Sess, Src, #{<<"x-queue-type">> => {utf8, Type}}),
333+
amqp10_declare_queue(Sess, Dest, #{<<"x-queue-type">> => {utf8, Type},
334+
<<"x-overflow">> => {utf8, <<"reject-publish">>},
335+
<<"x-max-length">> => {ulong, 5}
336+
}),
337+
amqp10_publish(Sess, Src, <<"hello">>, 10),
338+
ExtraArgs = [{<<"ack-mode">>, AckMode},
339+
{<<"src-delete-after">>, 10}],
340+
ShovelArgs = ?config(shovel_args, Config) ++ ExtraArgs,
341+
set_param_nowait(Config, ?PARAM, ShovelArgs),
342+
await_autodelete(Config, ?PARAM),
343+
Expected = lists:sort([[Src, integer_to_binary(ExpSrc)],
344+
[Dest, integer_to_binary(ExpDest)]]),
345+
?awaitMatch(
346+
Expected,
347+
lists:sort(rabbit_ct_broker_helpers:rabbitmqctl_list(
348+
Config, 0,
349+
["list_queues", "name", "messages", "--no-table-headers"])),
350+
45_000),
351+
amqp10_expect_count(Sess, Src, ExpSrc),
352+
amqp10_expect_count(Sess, Dest, ExpDest)
353+
end).
354+
355+
autodelete_with_quorum_rejections(Config, AckMode, ExpSrcFun) ->
356+
Src = ?config(srcq, Config),
357+
Dest = ?config(destq, Config),
358+
Type = <<"quorum">>,
359+
with_amqp10_session(
360+
Config,
361+
fun (Sess) ->
362+
amqp10_declare_queue(Sess, Src, #{<<"x-queue-type">> => {utf8, Type}}),
363+
amqp10_declare_queue(Sess, Dest, #{<<"x-queue-type">> => {utf8, Type},
364+
<<"x-overflow">> => {utf8, <<"reject-publish">>},
365+
<<"x-max-length">> => {ulong, 5}
366+
}),
367+
amqp10_publish(Sess, Src, <<"hello">>, 100),
368+
ExtraArgs = [{<<"ack-mode">>, AckMode},
369+
{<<"src-delete-after">>, 50}],
370+
ShovelArgs = ?config(shovel_args, Config) ++ ExtraArgs,
371+
set_param_nowait(Config, ?PARAM, ShovelArgs),
372+
await_autodelete(Config, ?PARAM),
373+
eventually(
374+
?_assert(
375+
list_queue_messages(Config, Dest) >= 5),
376+
1000, 45),
377+
ExpDest = list_queue_messages(Config, Dest),
378+
amqp10_expect_count(Sess, Src, ExpSrcFun(ExpDest)),
379+
amqp10_expect_count(Sess, Dest, ExpDest)
380+
end).
381+
307382
%%----------------------------------------------------------------------------
308383
maybe_skip_local_protocol(Config) ->
309384
[Node] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
@@ -314,3 +389,12 @@ maybe_skip_local_protocol(Config) ->
314389
_ ->
315390
{skip, "This group requires rabbitmq_4.0.0 feature flag"}
316391
end.
392+
393+
list_queue_messages(Config, QName) ->
394+
List = rabbit_ct_broker_helpers:rabbitmqctl_list(
395+
Config, 0,
396+
["list_queues", "name", "messages", "--no-table-headers"]),
397+
[[_, Messages]] = lists:filter(fun([Q, _]) ->
398+
Q == QName
399+
end, List),
400+
binary_to_integer(Messages).

0 commit comments

Comments
 (0)