|
| 1 | +-module(unicode_SUITE). |
| 2 | + |
| 3 | +-include_lib("common_test/include/ct.hrl"). |
| 4 | +-include_lib("eunit/include/eunit.hrl"). |
| 5 | +-include_lib("amqp_client/include/amqp_client.hrl"). |
| 6 | + |
| 7 | +-compile(export_all). |
| 8 | + |
| 9 | +%% Unicode U+1F407 |
| 10 | +-define(UNICODE_STRING, "bunny🐇bunny"). |
| 11 | + |
| 12 | +all() -> |
| 13 | + [ |
| 14 | + {group, queues} |
| 15 | + ]. |
| 16 | + |
| 17 | +groups() -> |
| 18 | + [ |
| 19 | + {queues, [], [ |
| 20 | + classic_queue_v1, |
| 21 | + classic_queue_v2, |
| 22 | + quorum_queue, |
| 23 | + stream |
| 24 | + ]} |
| 25 | + ]. |
| 26 | + |
| 27 | +%% ------------------------------------------------------------------- |
| 28 | +%% Testsuite setup/teardown. |
| 29 | +%% ------------------------------------------------------------------- |
| 30 | + |
| 31 | +init_per_suite(Config) -> |
| 32 | + rabbit_ct_helpers:log_environment(), |
| 33 | + rabbit_ct_helpers:run_setup_steps(Config). |
| 34 | + |
| 35 | +end_per_suite(Config) -> |
| 36 | + rabbit_ct_helpers:run_teardown_steps(Config). |
| 37 | + |
| 38 | +init_per_group(Group, Config0) -> |
| 39 | + PrivDir0 = ?config(priv_dir, Config0), |
| 40 | + PrivDir = filename:join(PrivDir0, ?UNICODE_STRING), |
| 41 | + ok = file:make_dir(PrivDir), |
| 42 | + Config = rabbit_ct_helpers:set_config(Config0, [{priv_dir, PrivDir}, |
| 43 | + {rmq_nodename_suffix, Group}]), |
| 44 | + rabbit_ct_helpers:run_steps(Config, |
| 45 | + rabbit_ct_broker_helpers:setup_steps() ++ |
| 46 | + rabbit_ct_client_helpers:setup_steps() |
| 47 | + ). |
| 48 | + |
| 49 | +end_per_group(_, Config) -> |
| 50 | + rabbit_ct_helpers:run_steps(Config, |
| 51 | + rabbit_ct_client_helpers:teardown_steps() ++ |
| 52 | + rabbit_ct_broker_helpers:teardown_steps()). |
| 53 | + |
| 54 | +init_per_testcase(Testcase, Config) -> |
| 55 | + rabbit_ct_helpers:testcase_started(Config, Testcase). |
| 56 | + |
| 57 | +end_per_testcase(Testcase, Config) -> |
| 58 | + rabbit_ct_helpers:testcase_finished(Config, Testcase). |
| 59 | + |
| 60 | +classic_queue_v1(Config) -> |
| 61 | + ok = rabbit_ct_broker_helpers:rpc( |
| 62 | + Config, 0, application, set_env, [rabbit, classic_queue_default_version, 1]), |
| 63 | + ok = queue(Config, ?FUNCTION_NAME, []). |
| 64 | + |
| 65 | +classic_queue_v2(Config) -> |
| 66 | + ok = rabbit_ct_broker_helpers:rpc( |
| 67 | + Config, 0, application, set_env, [rabbit, classic_queue_default_version, 2]), |
| 68 | + ok = queue(Config, ?FUNCTION_NAME, []). |
| 69 | + |
| 70 | +quorum_queue(Config) -> |
| 71 | + ok = queue(Config, ?FUNCTION_NAME, [{<<"x-queue-type">>, longstr, <<"quorum">>}]). |
| 72 | + |
| 73 | +queue(Config, QName0, Args) -> |
| 74 | + QName1 = rabbit_data_coercion:to_binary(QName0), |
| 75 | + QName = <<QName1/binary, ?UNICODE_STRING/utf8>>, |
| 76 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 77 | + Ch = rabbit_ct_client_helpers:open_channel(Config, Server), |
| 78 | + amqp_channel:call(Ch, #'queue.declare'{queue = QName, |
| 79 | + durable = true, |
| 80 | + arguments = Args |
| 81 | + }), |
| 82 | + rabbit_ct_client_helpers:publish(Ch, QName, 1), |
| 83 | + {#'basic.get_ok'{}, #amqp_msg{payload = <<"1">>}} = |
| 84 | + amqp_channel:call(Ch, #'basic.get'{queue = QName, no_ack = false}), |
| 85 | + {'queue.delete_ok', 0} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), |
| 86 | + ok. |
| 87 | + |
| 88 | +stream(Config) -> |
| 89 | + ok = rabbit_ct_broker_helpers:enable_feature_flag(Config, stream_queue), |
| 90 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 91 | + ConsumerTag = QName0 = atom_to_binary(?FUNCTION_NAME), |
| 92 | + QName = <<QName0/binary, ?UNICODE_STRING/utf8>>, |
| 93 | + Ch = rabbit_ct_client_helpers:open_channel(Config, Server), |
| 94 | + amqp_channel:call(Ch, #'queue.declare'{queue = QName, |
| 95 | + durable = true, |
| 96 | + arguments = [{<<"x-queue-type">>, longstr, <<"stream">>}] |
| 97 | + }), |
| 98 | + rabbit_ct_client_helpers:publish(Ch, QName, 1), |
| 99 | + ?assertMatch(#'basic.qos_ok'{}, |
| 100 | + amqp_channel:call(Ch, #'basic.qos'{global = false, |
| 101 | + prefetch_count = 1})), |
| 102 | + amqp_channel:subscribe(Ch, |
| 103 | + #'basic.consume'{queue = QName, |
| 104 | + no_ack = false, |
| 105 | + consumer_tag = ConsumerTag, |
| 106 | + arguments = [{<<"x-stream-offset">>, long, 0}]}, |
| 107 | + self()), |
| 108 | + receive |
| 109 | + #'basic.consume_ok'{consumer_tag = ConsumerTag} -> |
| 110 | + ok |
| 111 | + end, |
| 112 | + DelTag = receive |
| 113 | + {#'basic.deliver'{delivery_tag = DeliveryTag}, _} -> |
| 114 | + DeliveryTag |
| 115 | + after 5000 -> |
| 116 | + ct:fail(timeout) |
| 117 | + end, |
| 118 | + ok = amqp_channel:cast(Ch, #'basic.ack'{delivery_tag = DelTag, |
| 119 | + multiple = false}), |
| 120 | + amqp_channel:call(Ch, #'basic.cancel'{consumer_tag = ConsumerTag}), |
| 121 | + {'queue.delete_ok', 0} = amqp_channel:call(Ch, #'queue.delete'{queue = QName}), |
| 122 | + ok. |
0 commit comments