Merge pull request #11193 from keynslug/fix/emqx-ft-test-flaps

fix(ft-test): use new cth tooling in emqx_ft_storage_fs_SUITE

apps/emqx/test/emqx_cth_cluster.erl

@@ -206,11 +206,6 @@ default_appspec(emqx_conf, Spec, _NodeSpecs) ->
         base_port := BasePort,
         work_dir := WorkDir
     } = Spec,
-    Listeners = [
-        #{Type => #{default => #{bind => format("127.0.0.1:~p", [Port])}}}
-     || Type <- [tcp, ssl, ws, wss],
-        Port <- [listener_port(BasePort, Type)]
-    ],
     Cluster =
         case get_cluster_seeds(Spec) of
             [_ | _] = Seeds ->
@@ -239,7 +234,7 @@ default_appspec(emqx_conf, Spec, _NodeSpecs) ->
                 tcp_server_port => gen_rpc_port(BasePort),
                 port_discovery => manual
             },
-            listeners => lists:foldl(fun maps:merge/2, #{}, Listeners)
+            listeners => allocate_listener_ports([tcp, ssl, ws, wss], Spec)
         }
     };
 default_appspec(_App, _, _) ->
@@ -252,6 +247,13 @@ get_cluster_seeds(#{join_to := Node}) ->
 get_cluster_seeds(#{core_nodes := CoreNodes}) ->
     CoreNodes.
 
+allocate_listener_port(Type, #{base_port := BasePort}) ->
+    Port = listener_port(BasePort, Type),
+    #{Type => #{default => #{bind => format("127.0.0.1:~p", [Port])}}}.
+
+allocate_listener_ports(Types, Spec) ->
+    lists:foldl(fun maps:merge/2, #{}, [allocate_listener_port(Type, Spec) || Type <- Types]).
+
 start_node_init(Spec = #{name := Node}) ->
     Node = start_bare_node(Node, Spec),
     pong = net_adm:ping(Node),

apps/emqx_ft/test/emqx_ft_storage_fs_SUITE.erl

@@ -35,10 +35,18 @@ groups() ->
     ].
 
 init_per_suite(Config) ->
-    ok = emqx_common_test_helpers:start_apps([emqx_ft], emqx_ft_test_helpers:env_handler(Config)),
+    Storage = emqx_ft_test_helpers:local_storage(Config),
-    Config.
+    WorkDir = ?config(priv_dir, Config),
-end_per_suite(_Config) ->
+    Apps = emqx_cth_suite:start(
-    ok = emqx_common_test_helpers:stop_apps([emqx_ft]),
+        [
+            {emqx_ft, #{config => emqx_ft_test_helpers:config(Storage)}}
+        ],
+        #{work_dir => WorkDir}
+    ),
+    [{suite_apps, Apps} | Config].
+
+end_per_suite(Config) ->
+    ok = emqx_cth_suite:stop(?config(suite_apps, Config)),
     ok.
 
 init_per_testcase(Case, Config) ->
@@ -46,14 +54,25 @@ init_per_testcase(Case, Config) ->
 end_per_testcase(_Case, _Config) ->
     ok.
 
-init_per_group(cluster, Config) ->
+init_per_group(Group = cluster, Config) ->
-    Node = emqx_ft_test_helpers:start_additional_node(Config, emqx_ft_storage_fs1),
+    WorkDir = filename:join(?config(priv_dir, Config), Group),
-    [{additional_node, Node} | Config];
+    Apps = [
+        {emqx_conf, #{start => false}},
+        {emqx_ft, "file_transfer { enable = true, storage.local { enable = true } }"}
+    ],
+    Nodes = emqx_cth_cluster:start(
+        [
+            {emqx_ft_storage_fs1, #{apps => Apps, join_to => node()}},
+            {emqx_ft_storage_fs2, #{apps => Apps, join_to => node()}}
+        ],
+        #{work_dir => WorkDir}
+    ),
+    [{cluster, Nodes} | Config];
 init_per_group(_Group, Config) ->
     Config.
 
 end_per_group(cluster, Config) ->
-    ok = emqx_ft_test_helpers:stop_additional_node(?config(additional_node, Config));
+    ok = emqx_cth_suite:stop(?config(cluster, Config));
 end_per_group(_Group, _Config) ->
     ok.
 
@@ -62,12 +81,9 @@ end_per_group(_Group, _Config) ->
 %%--------------------------------------------------------------------
 
 t_multinode_exports(Config) ->
-    Node1 = ?config(additional_node, Config),
+    [Node1, Node2 | _] = ?config(cluster, Config),
     ok = emqx_ft_test_helpers:upload_file(<<"c/1">>, <<"f:1">>, "fn1", <<"data">>, Node1),
-
-    Node2 = node(),
     ok = emqx_ft_test_helpers:upload_file(<<"c/2">>, <<"f:2">>, "fn2", <<"data">>, Node2),
-
     ?assertMatch(
         [
             #{transfer := {<<"c/1">>, <<"f:1">>}, name := "fn1"},

apps/emqx_ft/test/emqx_ft_test_helpers.erl

@@ -24,23 +24,6 @@
 -define(S3_HOST, <<"minio">>).
 -define(S3_PORT, 9000).
 
-start_additional_node(Config, Name) ->
-    emqx_common_test_helpers:start_slave(
-        Name,
-        [
-            {apps, [emqx_ft]},
-            {join_to, node()},
-            {configure_gen_rpc, true},
-            {env_handler, env_handler(Config)}
-        ]
-    ).
-
-stop_additional_node(Node) ->
-    _ = rpc:call(Node, ekka, leave, []),
-    ok = rpc:call(Node, emqx_common_test_helpers, stop_apps, [[emqx_ft]]),
-    ok = emqx_common_test_helpers:stop_slave(Node),
-    ok.
-
 env_handler(Config) ->
     fun
         (emqx_ft) ->