|
| 1 | +%% This Source Code Form is subject to the terms of the Mozilla Public |
| 2 | +%% License, v. 2.0. If a copy of the MPL was not distributed with this |
| 3 | +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. |
| 4 | +%% |
| 5 | +%% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. All rights reserved. |
| 6 | + |
| 7 | +-module(unit_stream_arg_validation_SUITE). |
| 8 | + |
| 9 | +-compile(nowarn_export_all). |
| 10 | +-compile(export_all). |
| 11 | + |
| 12 | +-include_lib("common_test/include/ct.hrl"). |
| 13 | +-include_lib("eunit/include/eunit.hrl"). |
| 14 | +-include_lib("amqp_client/include/amqp_client.hrl"). |
| 15 | + |
| 16 | +suite() -> |
| 17 | + [{timetrap, {minutes, 5}}]. |
| 18 | + |
| 19 | +all() -> |
| 20 | + [ |
| 21 | + {group, unit}, |
| 22 | + {group, integration} |
| 23 | + ]. |
| 24 | + |
| 25 | +groups() -> |
| 26 | + [ |
| 27 | + {unit, [parallel], |
| 28 | + [check_max_age_empty_string, |
| 29 | + check_max_age_no_leading_digits, |
| 30 | + check_max_age_zero_count, |
| 31 | + check_max_age_valid_units, |
| 32 | + check_max_age_invalid_unit]}, |
| 33 | + {integration, [parallel], |
| 34 | + [filter_size_lower_bound, |
| 35 | + filter_size_upper_bound, |
| 36 | + filter_size_below_lower_bound, |
| 37 | + filter_size_zero, |
| 38 | + filter_size_float, |
| 39 | + filter_size_on_classic_queue]} |
| 40 | + ]. |
| 41 | + |
| 42 | +%% ------------------------------------------------------------------- |
| 43 | +%% Suite setup/teardown. |
| 44 | +%% ------------------------------------------------------------------- |
| 45 | + |
| 46 | +init_per_suite(Config) -> |
| 47 | + rabbit_ct_helpers:log_environment(), |
| 48 | + rabbit_ct_helpers:run_setup_steps(Config). |
| 49 | + |
| 50 | +end_per_suite(Config) -> |
| 51 | + rabbit_ct_helpers:run_teardown_steps(Config). |
| 52 | + |
| 53 | +init_per_group(unit, Config) -> |
| 54 | + Config; |
| 55 | +init_per_group(integration, Config) -> |
| 56 | + Config1 = rabbit_ct_helpers:set_config(Config, |
| 57 | + [{rmq_nodename_suffix, integration}, |
| 58 | + {rmq_nodes_count, 1}]), |
| 59 | + rabbit_ct_helpers:run_steps(Config1, |
| 60 | + rabbit_ct_broker_helpers:setup_steps() ++ |
| 61 | + rabbit_ct_client_helpers:setup_steps()). |
| 62 | + |
| 63 | +end_per_group(unit, _Config) -> |
| 64 | + ok; |
| 65 | +end_per_group(integration, Config) -> |
| 66 | + rabbit_ct_helpers:run_steps(Config, |
| 67 | + rabbit_ct_client_helpers:teardown_steps() ++ |
| 68 | + rabbit_ct_broker_helpers:teardown_steps()). |
| 69 | + |
| 70 | +init_per_testcase(Testcase, Config) -> |
| 71 | + Config1 = rabbit_ct_helpers:testcase_started(Config, Testcase), |
| 72 | + Q = rabbit_data_coercion:to_binary(Testcase), |
| 73 | + rabbit_ct_helpers:set_config(Config1, [{queue_name, Q}]). |
| 74 | + |
| 75 | +end_per_testcase(Testcase, Config) -> |
| 76 | + rabbit_ct_helpers:testcase_finished(Config, Testcase). |
| 77 | + |
| 78 | +%% ------------------------------------------------------------------- |
| 79 | +%% Unit tests: rabbit_amqqueue:check_max_age/1. |
| 80 | +%% Pure function; no broker required. |
| 81 | +%% ------------------------------------------------------------------- |
| 82 | + |
| 83 | +check_max_age_empty_string(_Config) -> |
| 84 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<>>)). |
| 85 | + |
| 86 | +check_max_age_no_leading_digits(_Config) -> |
| 87 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"D">>)), |
| 88 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"M">>)), |
| 89 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"Year">>)), |
| 90 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"-1D">>)). |
| 91 | + |
| 92 | +check_max_age_zero_count(_Config) -> |
| 93 | + lists:foreach( |
| 94 | + fun(Unit) -> |
| 95 | + ?assertEqual({error, invalid_max_age}, |
| 96 | + rabbit_amqqueue:check_max_age(list_to_binary("0" ++ Unit))) |
| 97 | + end, |
| 98 | + ["Y", "M", "D", "h", "m", "s"]). |
| 99 | + |
| 100 | +check_max_age_valid_units(_Config) -> |
| 101 | + lists:foreach( |
| 102 | + fun(Unit) -> |
| 103 | + Result = rabbit_amqqueue:check_max_age(list_to_binary("10" ++ Unit)), |
| 104 | + ?assert(is_integer(Result)), |
| 105 | + ?assert(Result > 0) |
| 106 | + end, |
| 107 | + ["Y", "M", "D", "h", "m", "s"]). |
| 108 | + |
| 109 | +check_max_age_invalid_unit(_Config) -> |
| 110 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"1A">>)), |
| 111 | + %% Lowercase y is not a valid unit; only uppercase Y is |
| 112 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"1y">>)), |
| 113 | + %% Multiple trailing characters are not a valid unit |
| 114 | + ?assertEqual({error, invalid_max_age}, rabbit_amqqueue:check_max_age(<<"1D2">>)). |
| 115 | + |
| 116 | +%% ------------------------------------------------------------------- |
| 117 | +%% Integration tests: x-stream-filter-size-bytes validation. |
| 118 | +%% These tests require a running broker. |
| 119 | +%% ------------------------------------------------------------------- |
| 120 | + |
| 121 | +filter_size_lower_bound(Config) -> |
| 122 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 123 | + Q = ?config(queue_name, Config), |
| 124 | + ?assertEqual({'queue.declare_ok', Q, 0, 0}, |
| 125 | + declare_stream(Config, Server, Q, |
| 126 | + [{<<"x-stream-filter-size-bytes">>, long, 16}])), |
| 127 | + delete_queue(Config, Q). |
| 128 | + |
| 129 | +filter_size_upper_bound(Config) -> |
| 130 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 131 | + Q = ?config(queue_name, Config), |
| 132 | + ?assertEqual({'queue.declare_ok', Q, 0, 0}, |
| 133 | + declare_stream(Config, Server, Q, |
| 134 | + [{<<"x-stream-filter-size-bytes">>, long, 255}])), |
| 135 | + delete_queue(Config, Q). |
| 136 | + |
| 137 | +filter_size_below_lower_bound(Config) -> |
| 138 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 139 | + Q = ?config(queue_name, Config), |
| 140 | + ExpectedError = <<"PRECONDITION_FAILED - Invalid value for x-stream-filter-size-bytes">>, |
| 141 | + ?assertExit( |
| 142 | + {{shutdown, {server_initiated_close, 406, ExpectedError}}, _}, |
| 143 | + declare_stream(Config, Server, Q, |
| 144 | + [{<<"x-stream-filter-size-bytes">>, long, 15}])). |
| 145 | + |
| 146 | +filter_size_zero(Config) -> |
| 147 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 148 | + Q = ?config(queue_name, Config), |
| 149 | + %% 0 satisfies `check_non_neg_int_arg/2` but still falls below the valid range |
| 150 | + ExpectedError = <<"PRECONDITION_FAILED - Invalid value for x-stream-filter-size-bytes">>, |
| 151 | + ?assertExit( |
| 152 | + {{shutdown, {server_initiated_close, 406, ExpectedError}}, _}, |
| 153 | + declare_stream(Config, Server, Q, |
| 154 | + [{<<"x-stream-filter-size-bytes">>, long, 0}])). |
| 155 | + |
| 156 | +filter_size_float(Config) -> |
| 157 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 158 | + Q = ?config(queue_name, Config), |
| 159 | + %% Before the fix, a float in [16.0, 255.0] was silently accepted because Erlang |
| 160 | + %% compares floats and integers numerically, bypassing the `x-stream-filter-size-bytes` |
| 161 | + %% range guard. |
| 162 | + ?assertExit( |
| 163 | + {{shutdown, {server_initiated_close, 406, _}}, _}, |
| 164 | + declare_stream(Config, Server, Q, |
| 165 | + [{<<"x-stream-filter-size-bytes">>, float, 32.0}])). |
| 166 | + |
| 167 | +filter_size_on_classic_queue(Config) -> |
| 168 | + Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename), |
| 169 | + Q = ?config(queue_name, Config), |
| 170 | + %% `x-stream-filter-size-bytes` is now in stream capabilities and therefore in the |
| 171 | + %% global queue_arguments union, so classic queues must reject it. |
| 172 | + ?assertExit( |
| 173 | + {{shutdown, {server_initiated_close, 406, _}}, _}, |
| 174 | + declare_classic(Config, Server, Q, |
| 175 | + [{<<"x-stream-filter-size-bytes">>, long, 32}])). |
| 176 | + |
| 177 | +%% ------------------------------------------------------------------- |
| 178 | +%% Helpers. |
| 179 | +%% ------------------------------------------------------------------- |
| 180 | + |
| 181 | +declare_stream(Config, Server, Q, ExtraArgs) -> |
| 182 | + call_declare(Config, Server, |
| 183 | + #'queue.declare'{queue = Q, |
| 184 | + durable = true, |
| 185 | + arguments = [{<<"x-queue-type">>, longstr, <<"stream">>} |
| 186 | + | ExtraArgs]}). |
| 187 | + |
| 188 | +declare_classic(Config, Server, Q, ExtraArgs) -> |
| 189 | + call_declare(Config, Server, |
| 190 | + #'queue.declare'{queue = Q, |
| 191 | + durable = true, |
| 192 | + arguments = [{<<"x-queue-type">>, longstr, <<"classic">>} |
| 193 | + | ExtraArgs]}). |
| 194 | + |
| 195 | +call_declare(Config, Server, Cmd) -> |
| 196 | + Ch = rabbit_ct_client_helpers:open_channel(Config, Server), |
| 197 | + Reply = amqp_channel:call(Ch, Cmd), |
| 198 | + rabbit_ct_client_helpers:close_channel(Ch), |
| 199 | + Reply. |
| 200 | + |
| 201 | +delete_queue(Config, Q) -> |
| 202 | + rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, do_delete_queue, [Q]). |
| 203 | + |
| 204 | +do_delete_queue(Name) -> |
| 205 | + QName = rabbit_misc:r(<<"/">>, queue, Name), |
| 206 | + case rabbit_amqqueue:lookup(QName) of |
| 207 | + {ok, Q} -> |
| 208 | + {ok, _} = rabbit_amqqueue:delete(Q, false, false, <<"dummy">>); |
| 209 | + _ -> |
| 210 | + ok |
| 211 | + end. |
0 commit comments