refactor: update postgres connector tests to use resource apis

apps/emqx_connector/include/emqx_connector.hrl

@@ -33,3 +33,5 @@ The """ ++ TYPE ++ " default port " ++ DEFAULT_PORT ++ " is used if '[:Port]' is
 ).
 
 -define(THROW_ERROR(Str), erlang:throw({error, Str})).
+
+-define(CONNECTOR_RESOURCE_GROUP, <<"emqx_connector">>).

apps/emqx_connector/test/emqx_connector_pgsql_SUITE.erl

@@ -24,6 +24,7 @@
 -include_lib("stdlib/include/assert.hrl").
 
 -define(PGSQL_HOST, "pgsql").
+-define(PGSQL_RESOURCE_MOD, emqx_connector_pgsql).
 
 all() ->
     emqx_common_test_helpers:all(?MODULE).
@@ -34,111 +35,89 @@ groups() ->
 init_per_suite(Config) ->
     case emqx_common_test_helpers:is_tcp_server_available(?PGSQL_HOST, ?PGSQL_DEFAULT_PORT) of
         true ->
-            ok = emqx_connector_test_helpers:start_apps([ecpool, pgsql]),
+            ok = emqx_common_test_helpers:start_apps([emqx_conf]),
+            ok = emqx_connector_test_helpers:start_apps([emqx_resource, emqx_connector]),
             Config;
         false ->
             {skip, no_pgsql}
     end.
 
 end_per_suite(_Config) ->
-    ok = emqx_connector_test_helpers:stop_apps([ecpool, pgsql]).
+    ok = emqx_common_test_helpers:stop_apps([emqx_conf]),
+    ok = emqx_connector_test_helpers:stop_apps([emqx_resource, emqx_connector]).
 
 init_per_testcase(_, Config) ->
-    ?assertEqual(
-        {ok, #{poolname => emqx_connector_pgsql}},
-        emqx_connector_pgsql:on_start(<<"emqx_connector_pgsql">>, pgsql_config())
-    ),
     Config.
 
 end_per_testcase(_, _Config) ->
-    ?assertEqual(
+    ok.
-        ok,
-        emqx_connector_pgsql:on_stop(<<"emqx_connector_pgsql">>, #{poolname => emqx_connector_pgsql})
-    ).
 
 % %%------------------------------------------------------------------------------
 % %% Testcases
 % %%------------------------------------------------------------------------------
 
-% Simple test to make sure the proper reference to the module is returned.
+t_lifecycle(_Config) ->
-t_roots(_Config) ->
+    perform_lifecycle_check(
-    ExpectedRoots = [{config, #{type => {ref, emqx_connector_pgsql, config}}}],
+        <<"emqx_connector_pgsql_SUITE">>,
-    ActualRoots = emqx_connector_pgsql:roots(),
+        pgsql_config()
-    ?assertEqual(ExpectedRoots, ActualRoots).
-
-% Not sure if this level of testing is appropriate for this function.
-% Checking the actual values/types of the returned term starts getting
-% into checking the emqx_connector_schema_lib.erl returns and the shape
-% of expected data elsewhere.
-t_fields(_Config) ->
-    Fields = emqx_connector_pgsql:fields(config),
-    lists:foreach(
-        fun({FieldName, FieldValue}) ->
-            ?assert(is_atom(FieldName)),
-            if
-                is_map(FieldValue) ->
-                    ?assert(maps:is_key(type, FieldValue) and maps:is_key(default, FieldValue));
-                true ->
-                    ?assert(is_function(FieldValue))
-            end
-        end,
-        Fields
     ).
 
-% Execute a minimal query to validate connection.
+perform_lifecycle_check(PoolName, InitialConfig) ->
-t_basic_query(_Config) ->
+    {ok, #{config := CheckedConfig}} = emqx_resource:check_config(?PGSQL_RESOURCE_MOD, InitialConfig),
-    ?assertMatch(
+    {ok, #{state := #{poolname := ReturnedPoolName} = State, status := InitialStatus}} = emqx_resource:create_local(
-        {ok, _, [{1}]},
+        PoolName,
-        emqx_connector_pgsql:on_query(
+        ?CONNECTOR_RESOURCE_GROUP,
-            <<"emqx_connector_pgsql">>, {query, test_query()}, undefined, #{
+        ?PGSQL_RESOURCE_MOD,
-                poolname => emqx_connector_pgsql
+        CheckedConfig
-            }
+    ),
-        )
+    ?assertEqual(InitialStatus, started),
-    ).
+    % Instance should match the state and status of the just started resource
-
+    {ok, ?CONNECTOR_RESOURCE_GROUP, #{state := State, status := InitialStatus}} = emqx_resource:get_instance(PoolName),
-% Perform health check.
+    ?assertEqual(ok, emqx_resource:health_check(PoolName)),
-t_do_healthcheck(_Config) ->
+    % % Perform query as further check that the resource is working as expected
-    ?assertEqual(
+    ?assertMatch({ok, _, [{1}]}, emqx_resource:query(PoolName, test_query_no_params())),
-        {ok, #{poolname => emqx_connector_pgsql}},
+    ?assertMatch({ok, _, [{1}]}, emqx_resource:query(PoolName, test_query_with_params())),
-        emqx_connector_pgsql:on_health_check(<<"emqx_connector_pgsql">>, #{
+    ?assertEqual(ok, emqx_resource:stop(PoolName)),
-            poolname => emqx_connector_pgsql
+    % Resource will be listed still, but state will be changed and healthcheck will fail
-        })
+    % as the worker no longer exists.
-    ).
+    {ok, ?CONNECTOR_RESOURCE_GROUP, #{state := State, status := StoppedStatus}} = emqx_resource:get_instance(PoolName),
-
+    ?assertEqual(StoppedStatus, stopped),
-% Perform healthcheck on a connector that does not exist.
+    ?assertEqual({error,health_check_failed}, emqx_resource:health_check(PoolName)),
-t_healthceck_when_connector_does_not_exist(_Config) ->
+    % Resource healthcheck shortcuts things by checking ets. Go deeper by checking pool itself.
-    ?assertEqual(
+    ?assertEqual({error, not_found}, ecpool:stop_sup_pool(ReturnedPoolName)),
-        {error, health_check_failed, #{poolname => emqx_connector_pgsql_does_not_exist}},
+    % Can call stop/1 again on an already stopped instance
-        emqx_connector_pgsql:on_health_check(<<"emqx_connector_pgsql_does_not_exist">>, #{
+    ?assertEqual(ok, emqx_resource:stop(PoolName)),
-            poolname => emqx_connector_pgsql_does_not_exist
+    % Make sure it can be restarted and the healthchecks and queries work properly
-        })
+    ?assertEqual(ok, emqx_resource:restart(PoolName)),
-    ).
+    {ok, ?CONNECTOR_RESOURCE_GROUP, #{status := InitialStatus}} = emqx_resource:get_instance(PoolName),
+    ?assertEqual(ok, emqx_resource:health_check(PoolName)),
+    ?assertMatch({ok, _, [{1}]}, emqx_resource:query(PoolName, test_query_no_params())),
+    ?assertMatch({ok, _, [{1}]}, emqx_resource:query(PoolName, test_query_with_params())),
+    % Stop and remove the resource in one go.
+    ?assertEqual(ok, emqx_resource:remove_local(PoolName)),
+    ?assertEqual({error, not_found}, ecpool:stop_sup_pool(ReturnedPoolName)),
+    % Should not even be able to get the resource data out of ets now unlike just stopping.
+    ?assertEqual({error, not_found}, emqx_resource:get_instance(PoolName)).
 
 % %%------------------------------------------------------------------------------
 % %% Helpers
 % %%------------------------------------------------------------------------------
 
 pgsql_config() ->
-    #{
+    RawConfig = list_to_binary(io_lib:format("""
-        auto_reconnect => true,
+    auto_reconnect = true
-        database => <<"mqtt">>,
+    database = mqtt
-        username => <<"root">>,
+    username= root
-        password => <<"public">>,
+    password = public
-        pool_size => 8,
+    pool_size = 8
-        server => {?PGSQL_HOST, ?PGSQL_DEFAULT_PORT},
+    server = \"~s:~b\"
-        ssl => #{enable => false}
+    """, [?PGSQL_HOST, ?PGSQL_DEFAULT_PORT])),
-    }.
 
-pgsql_bad_config() ->
+    {ok, Config} = hocon:binary(RawConfig),
-    #{
+    #{<<"config">> => Config}.
-        auto_reconnect => true,
-        database => <<"bad_mqtt">>,
-        username => <<"bad_root">>,
-        password => <<"bad_public">>,
-        pool_size => 8,
-        server => {?PGSQL_HOST, ?PGSQL_DEFAULT_PORT},
-        ssl => #{enable => false}
-    }.
 
-test_query() ->
+test_query_no_params() ->
-    <<"SELECT 1">>.
+    {query, <<"SELECT 1">>}.
+
+test_query_with_params() ->
+    {query, <<"SELECT $1::integer">>, [1]}.