yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
emqx/apps/emqx_bridge_sqlserver/test/emqx_bridge_sqlserver_SUITE...

694 lines
22 KiB
Erlang
Raw Permalink Blame History

%%--------------------------------------------------------------------
% Copyright (c) 2022-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%--------------------------------------------------------------------
-module(emqx_bridge_sqlserver_SUITE).
-compile(nowarn_export_all).
-compile(export_all).
-include("emqx_bridge_sqlserver/include/emqx_bridge_sqlserver.hrl").
-include_lib("eunit/include/eunit.hrl").
-include_lib("common_test/include/ct.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
% SQL definitions
-define(SQL_BRIDGE,
"insert into t_mqtt_msg(msgid, topic, qos, payload) values ( ${id}, ${topic}, ${qos}, ${payload})"
).
-define(SQL_SERVER_DRIVER, "ms-sql").
-define(SQL_CREATE_DATABASE_IF_NOT_EXISTS,
" IF NOT EXISTS(SELECT name FROM sys.databases WHERE name = 'mqtt')"
" BEGIN"
" CREATE DATABASE mqtt;"
" END"
).
-define(SQL_CREATE_TABLE_IN_DB_MQTT,
" CREATE TABLE mqtt.dbo.t_mqtt_msg"
" (id int PRIMARY KEY IDENTITY(1000000001,1) NOT NULL,"
" msgid VARCHAR(64) NULL,"
" topic VARCHAR(100) NULL,"
" qos tinyint NOT NULL DEFAULT 0,"
%% use VARCHAR to use utf8 encoding
%% for default, sqlserver use utf16 encoding NVARCHAR()
" payload VARCHAR(100) NULL,"
" arrived DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP)"
).
-define(SQL_DROP_DB_MQTT, "DROP DATABASE mqtt").
-define(SQL_DROP_TABLE, "DROP TABLE mqtt.dbo.t_mqtt_msg").
-define(SQL_DELETE, "DELETE from mqtt.dbo.t_mqtt_msg").
-define(SQL_SELECT, "SELECT payload FROM mqtt.dbo.t_mqtt_msg").
-define(SQL_SELECT_COUNT, "SELECT COUNT(*) FROM mqtt.dbo.t_mqtt_msg").
% DB defaults
-define(SQL_SERVER_DATABASE, "mqtt").
-define(SQL_SERVER_USERNAME, "sa").
-define(SQL_SERVER_PASSWORD, "mqtt_public1").
-define(BATCH_SIZE, 10).
-define(REQUEST_TIMEOUT_MS, 500).
-define(WORKER_POOL_SIZE, 4).
-define(WITH_CON(Process),
Con = connect_direct_sqlserver(Config),
Process,
ok = disconnect(Con)
).
%% How to run it locally (all commands are run in $PROJ_ROOT dir):
%% A: run ct on host
%% 1. Start all deps services
%% ```bash
%% sudo docker compose -f .ci/docker-compose-file/docker-compose.yaml \
%% -f .ci/docker-compose-file/docker-compose-sqlserver.yaml \
%% -f .ci/docker-compose-file/docker-compose-toxiproxy.yaml \
%% up --build
%% ```
%%
%% 2. Run use cases with special environment variables
%% 11433 is toxiproxy exported port.
%% Local:
%% ```bash
%% SQLSERVER_HOST=toxiproxy SQLSERVER_PORT=11433 \
%% PROXY_HOST=toxiproxy PROXY_PORT=1433 \
%% ./rebar3 as test ct -c -v --readable true --name ct@127.0.0.1 \
%% --suite apps/emqx_bridge_sqlserver/test/emqx_bridge_sqlserver_SUITE.erl
%% ```
%%
%% B: run ct in docker container
%% run script:
%% ```bash
%% ./scripts/ct/run.sh --ci --app apps/emqx_bridge_sqlserver/ -- \
%% --name 'test@127.0.0.1' -c -v --readable true \
%% --suite apps/emqx_bridge_sqlserver/test/emqx_bridge_sqlserver_SUITE.erl
%% ````
%%------------------------------------------------------------------------------
%% CT boilerplate
%%------------------------------------------------------------------------------
all() ->
[
{group, async},
{group, sync}
].
groups() ->
TCs = emqx_common_test_helpers:all(?MODULE),
NonBatchCases = [t_write_timeout],
BatchingGroups = [{group, with_batch}, {group, without_batch}],
[
{async, BatchingGroups},
{sync, BatchingGroups},
{with_batch, TCs -- NonBatchCases},
{without_batch, TCs}
].
init_per_group(async, Config) ->
[{query_mode, async} | Config];
init_per_group(sync, Config) ->
[{query_mode, sync} | Config];
init_per_group(with_batch, Config0) ->
Config = [{enable_batch, true} | Config0],
common_init(Config);
init_per_group(without_batch, Config0) ->
Config = [{enable_batch, false} | Config0],
common_init(Config);
init_per_group(_Group, Config) ->
Config.
end_per_group(Group, Config) when Group =:= with_batch; Group =:= without_batch ->
connect_and_drop_table(Config),
connect_and_drop_db(Config),
ProxyHost = ?config(proxy_host, Config),
ProxyPort = ?config(proxy_port, Config),
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
ok;
end_per_group(_Group, _Config) ->
ok.
init_per_suite(Config) ->
Passfile = filename:join(?config(priv_dir, Config), "passfile"),
ok = file:write_file(Passfile, <<?SQL_SERVER_PASSWORD>>),
[{sqlserver_passfile, Passfile} | Config].
end_per_suite(_Config) ->
emqx_mgmt_api_test_util:end_suite(),
ok = emqx_common_test_helpers:stop_apps([emqx_bridge, emqx_conf]),
ok.
init_per_testcase(_Testcase, Config) ->
%% drop database and table
%% connect_and_clear_table(Config),
%% create a new one
%% TODO: create a new database for each test case
delete_bridge(Config),
snabbkaffe:start_trace(),
Config.
end_per_testcase(_Testcase, Config) ->
ProxyHost = ?config(proxy_host, Config),
ProxyPort = ?config(proxy_port, Config),
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
connect_and_clear_table(Config),
ok = snabbkaffe:stop(),
delete_bridge(Config),
ok.
%%------------------------------------------------------------------------------
%% Testcases
%%------------------------------------------------------------------------------
t_setup_via_config_and_publish(Config) ->
?assertMatch(
{ok, _},
create_bridge(Config)
),
Val = str(erlang:unique_integer()),
SentData = sent_data(Val),
?check_trace(
begin
?wait_async_action(
?assertEqual(ok, send_message(Config, SentData)),
#{?snk_kind := sqlserver_connector_query_return},
10_000
),
?assertMatch(
[{Val}],
connect_and_get_payload(Config)
),
ok
end,
fun(Trace0) ->
Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
?assertMatch([#{result := ok}], Trace),
ok
end
),
ok.
t_setup_via_http_api_and_publish(Config) ->
BridgeType = ?config(sqlserver_bridge_type, Config),
Name = ?config(sqlserver_name, Config),
SQLServerConfig0 = ?config(sqlserver_config, Config),
SQLServerConfig = SQLServerConfig0#{
<<"name">> => Name,
<<"type">> => BridgeType,
%% NOTE: using literal password with HTTP API requests.
<<"password">> => <<?SQL_SERVER_PASSWORD>>
},
?assertMatch(
{ok, _},
create_bridge_http(SQLServerConfig)
),
Val = str(erlang:unique_integer()),
SentData = sent_data(Val),
?check_trace(
begin
?wait_async_action(
?assertEqual(ok, send_message(Config, SentData)),
#{?snk_kind := sqlserver_connector_query_return},
10_000
),
?assertMatch(
[{Val}],
connect_and_get_payload(Config)
),
ok
end,
fun(Trace0) ->
Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
?assertMatch([#{result := ok}], Trace),
ok
end
),
ok.
t_get_status(Config) ->
?assertMatch(
{ok, _},
create_bridge(Config)
),
ProxyPort = ?config(proxy_port, Config),
ProxyHost = ?config(proxy_host, Config),
ProxyName = ?config(proxy_name, Config),
health_check_resource_ok(Config),
emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
health_check_resource_down(Config)
end),
ok.
t_create_disconnected(Config) ->
ProxyPort = ?config(proxy_port, Config),
ProxyHost = ?config(proxy_host, Config),
ProxyName = ?config(proxy_name, Config),
emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
?assertMatch({ok, _}, create_bridge(Config)),
health_check_resource_down(Config)
end),
health_check_resource_ok(Config),
ok.
t_create_with_invalid_password(Config) ->
BridgeType = ?config(sqlserver_bridge_type, Config),
Name = ?config(sqlserver_name, Config),
SQLServerConfig0 = ?config(sqlserver_config, Config),
SQLServerConfig = SQLServerConfig0#{
<<"name">> => Name,
<<"type">> => BridgeType,
<<"password">> => <<"wrong_password">>
},
?check_trace(
begin
?assertMatch(
{ok, _},
create_bridge_http(SQLServerConfig)
)
end,
fun(Trace) ->
?assertMatch(
[#{error := {start_pool_failed, _, _}}],
?of_kind(sqlserver_connector_start_failed, Trace)
),
ok
end
),
ok.
t_write_failure(Config) ->
ProxyName = ?config(proxy_name, Config),
ProxyPort = ?config(proxy_port, Config),
ProxyHost = ?config(proxy_host, Config),
QueryMode = ?config(query_mode, Config),
Val = str(erlang:unique_integer()),
SentData = sent_data(Val),
{{ok, _}, {ok, _}} =
?wait_async_action(
create_bridge(Config),
#{?snk_kind := resource_connected_enter},
20_000
),
emqx_common_test_helpers:with_failure(down, ProxyName, ProxyHost, ProxyPort, fun() ->
case QueryMode of
sync ->
?assertMatch(
{error, {resource_error, #{reason := timeout}}},
send_message(Config, SentData)
);
async ->
?assertMatch(
ok, send_message(Config, SentData)
)
end
end),
ok.
t_write_timeout(_Config) ->
%% msodbc driver handled all connection exceptions
%% the case is same as t_write_failure/1
ok.
t_simple_query(Config) ->
BatchSize = batch_size(Config),
?assertMatch(
{ok, _},
create_bridge(Config)
),
{Requests, Vals} = gen_batch_req(BatchSize),
?check_trace(
begin
?wait_async_action(
begin
[?assertEqual(ok, query_resource(Config, Request)) || Request <- Requests]
end,
#{?snk_kind := sqlserver_connector_query_return},
10_000
),
%% just assert the data count is correct
?assertMatch(
BatchSize,
connect_and_get_count(Config)
),
%% assert the data order is correct
?assertMatch(
Vals,
connect_and_get_payload(Config)
)
end,
fun(Trace0) ->
Trace = ?of_kind(sqlserver_connector_query_return, Trace0),
case BatchSize of
1 ->
?assertMatch([#{result := ok}], Trace);
_ ->
[?assertMatch(#{result := ok}, Trace1) || Trace1 <- Trace]
end,
ok
end
),
ok.
-define(MISSING_TINYINT_ERROR,
"[Microsoft][ODBC Driver 17 for SQL Server][SQL Server]"
"Conversion failed when converting the varchar value 'undefined' to data type tinyint. SQLSTATE IS: 22018"
).
t_missing_data(Config) ->
QueryMode = ?config(query_mode, Config),
?assertMatch(
{ok, _},
create_bridge(Config)
),
Result = send_message(Config, #{}),
case QueryMode of
sync ->
?assertMatch(
{error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
Result
);
async ->
?assertMatch(
ok, send_message(Config, #{})
)
end,
ok.
t_bad_parameter(Config) ->
QueryMode = ?config(query_mode, Config),
?assertMatch(
{ok, _},
create_bridge(Config)
),
Result = send_message(Config, #{}),
case QueryMode of
sync ->
?assertMatch(
{error, {unrecoverable_error, {invalid_request, ?MISSING_TINYINT_ERROR}}},
Result
);
async ->
?assertMatch(
ok, send_message(Config, #{})
)
end,
ok.
%%------------------------------------------------------------------------------
%% Helper fns
%%------------------------------------------------------------------------------
common_init(ConfigT) ->
Host = os:getenv("SQLSERVER_HOST", "toxiproxy"),
Port = list_to_integer(os:getenv("SQLSERVER_PORT", str(?SQLSERVER_DEFAULT_PORT))),
Config0 = [
{sqlserver_host, Host},
{sqlserver_port, Port},
%% see also for `proxy_name` : $PROJ_ROOT/.ci/docker-compose-file/toxiproxy.json
{proxy_name, "sqlserver"},
{batch_size, batch_size(ConfigT)}
| ConfigT
],
BridgeType = proplists:get_value(bridge_type, Config0, <<"sqlserver">>),
case emqx_common_test_helpers:is_tcp_server_available(Host, Port) of
true ->
% Setup toxiproxy
ProxyHost = os:getenv("PROXY_HOST", "toxiproxy"),
ProxyPort = list_to_integer(os:getenv("PROXY_PORT", "8474")),
emqx_common_test_helpers:reset_proxy(ProxyHost, ProxyPort),
% Ensure enterprise bridge module is loaded
ok = emqx_common_test_helpers:start_apps([emqx_conf, emqx_bridge, odbc]),
_ = emqx_bridge_enterprise:module_info(),
emqx_mgmt_api_test_util:init_suite(),
% Connect to sqlserver directly
% drop old db and table, and then create new ones
connect_and_create_db_and_table(Config0),
{Name, SQLServerConf} = sqlserver_config(BridgeType, Config0),
Config =
[
{sqlserver_config, SQLServerConf},
{sqlserver_bridge_type, BridgeType},
{sqlserver_name, Name},
{proxy_host, ProxyHost},
{proxy_port, ProxyPort}
| Config0
],
Config;
false ->
case os:getenv("IS_CI") of
"yes" ->
throw(no_sqlserver);
_ ->
{skip, no_sqlserver}
end
end.
sqlserver_config(BridgeType, Config) ->
Port = integer_to_list(?config(sqlserver_port, Config)),
Server = ?config(sqlserver_host, Config) ++ ":" ++ Port,
Name = atom_to_binary(?MODULE),
BatchSize = batch_size(Config),
QueryMode = ?config(query_mode, Config),
Passfile = ?config(sqlserver_passfile, Config),
ConfigString =
io_lib:format(
"bridges.~s.~s {\n"
" enable = true\n"
" server = ~p\n"
" database = ~p\n"
" username = ~p\n"
" password = ~p\n"
" sql = ~p\n"
" driver = ~p\n"
" resource_opts = {\n"
" request_ttl = 500ms\n"
" batch_size = ~b\n"
" query_mode = ~s\n"
" worker_pool_size = ~b\n"
" }\n"
"}",
[
BridgeType,
Name,
Server,
?SQL_SERVER_DATABASE,
?SQL_SERVER_USERNAME,
"file://" ++ Passfile,
?SQL_BRIDGE,
?SQL_SERVER_DRIVER,
BatchSize,
QueryMode,
?WORKER_POOL_SIZE
]
),
{Name, parse_and_check(ConfigString, BridgeType, Name)}.
parse_and_check(ConfigString, BridgeType, Name) ->
{ok, RawConf} = hocon:binary(ConfigString, #{format => map}),
hocon_tconf:check_plain(emqx_bridge_schema, RawConf, #{required => false, atom_key => false}),
#{<<"bridges">> := #{BridgeType := #{Name := Config}}} = RawConf,
Config.
create_bridge(Config) ->
create_bridge(Config, _Overrides = #{}).
create_bridge(Config, Overrides) ->
BridgeType = ?config(sqlserver_bridge_type, Config),
Name = ?config(sqlserver_name, Config),
SSConfig0 = ?config(sqlserver_config, Config),
SSConfig = emqx_utils_maps:deep_merge(SSConfig0, Overrides),
emqx_bridge:create(BridgeType, Name, SSConfig).
delete_bridge(Config) ->
BridgeType = ?config(sqlserver_bridge_type, Config),
Name = ?config(sqlserver_name, Config),
emqx_bridge:remove(BridgeType, Name).
create_bridge_http(Params) ->
Path = emqx_mgmt_api_test_util:api_path(["bridges"]),
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
case emqx_mgmt_api_test_util:request_api(post, Path, "", AuthHeader, Params) of
{ok, Res} -> {ok, emqx_utils_json:decode(Res, [return_maps])};
Error -> Error
end.
send_message(Config, Payload) ->
Name = ?config(sqlserver_name, Config),
BridgeType = ?config(sqlserver_bridge_type, Config),
BridgeID = emqx_bridge_resource:bridge_id(BridgeType, Name),
emqx_bridge:send_message(BridgeID, Payload).
query_resource(Config, Request) ->
Name = ?config(sqlserver_name, Config),
BridgeType = ?config(sqlserver_bridge_type, Config),
ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
emqx_resource:query(ResourceID, Request, #{timeout => 1_000}).
query_resource_async(Config, Request) ->
Name = ?config(sqlserver_name, Config),
BridgeType = ?config(sqlserver_bridge_type, Config),
Ref = alias([reply]),
AsyncReplyFun = fun(Result) -> Ref ! {result, Ref, Result} end,
ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name),
Return = emqx_resource:query(ResourceID, Request, #{
timeout => 500, async_reply_fun => {AsyncReplyFun, []}
}),
{Return, Ref}.
resource_id(Config) ->
Name = ?config(sqlserver_name, Config),
BridgeType = ?config(sqlserver_bridge_type, Config),
_ResourceID = emqx_bridge_resource:resource_id(BridgeType, Name).
health_check_resource_ok(Config) ->
?assertEqual({ok, connected}, emqx_resource_manager:health_check(resource_id(Config))).
health_check_resource_down(Config) ->
case emqx_resource_manager:health_check(resource_id(Config)) of
{ok, Status} when Status =:= disconnected orelse Status =:= connecting ->
ok;
{error, timeout} ->
ok;
Other ->
?assert(
false, lists:flatten(io_lib:format("invalid health check result:~p~n", [Other]))
)
end.
receive_result(Ref, Timeout) ->
receive
{result, Ref, Result} ->
{ok, Result};
{Ref, Result} ->
{ok, Result}
after Timeout ->
timeout
end.
connect_direct_sqlserver(Config) ->
Opts = [
{host, ?config(sqlserver_host, Config)},
{port, ?config(sqlserver_port, Config)},
{username, ?SQL_SERVER_USERNAME},
{password, ?SQL_SERVER_PASSWORD},
{driver, ?SQL_SERVER_DRIVER},
{pool_size, 8}
],
{ok, Con} = connect(Opts),
Con.
connect(Options) ->
ConnectStr = lists:concat(conn_str(Options, [])),
Opts = proplists:get_value(options, Options, []),
odbc:connect(ConnectStr, Opts).
disconnect(Ref) ->
odbc:disconnect(Ref).
% These funs connect and then stop the sqlserver connection
connect_and_create_db_and_table(Config) ->
?WITH_CON(begin
{updated, undefined} = directly_query(Con, ?SQL_CREATE_DATABASE_IF_NOT_EXISTS),
{updated, undefined} = directly_query(Con, ?SQL_CREATE_TABLE_IN_DB_MQTT)
end).
connect_and_drop_db(Config) ->
?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_DB_MQTT)).
connect_and_drop_table(Config) ->
?WITH_CON({updated, undefined} = directly_query(Con, ?SQL_DROP_TABLE)).
connect_and_clear_table(Config) ->
?WITH_CON({updated, _} = directly_query(Con, ?SQL_DELETE)).
connect_and_get_payload(Config) ->
?WITH_CON(
{selected, ["payload"], Rows} = directly_query(Con, ?SQL_SELECT)
),
Rows.
connect_and_get_count(Config) ->
?WITH_CON(
{selected, [[]], [{Count}]} = directly_query(Con, ?SQL_SELECT_COUNT)
),
Count.
directly_query(Con, Query) ->
directly_query(Con, Query, ?REQUEST_TIMEOUT_MS).
directly_query(Con, Query, Timeout) ->
odbc:sql_query(Con, Query, Timeout).
%%--------------------------------------------------------------------
%% help functions
%%--------------------------------------------------------------------
batch_size(Config) ->
case ?config(enable_batch, Config) of
true -> ?BATCH_SIZE;
false -> 1
end.
conn_str([], Acc) ->
%% TODO: for msodbc 18+, we need to add "Encrypt=YES;TrustServerCertificate=YES"
%% but havn't tested now
%% we should use this for msodbcsql 18+
%% lists:join(";", ["Encrypt=YES", "TrustServerCertificate=YES" | Acc]);
lists:join(";", Acc);
conn_str([{driver, Driver} | Opts], Acc) ->
conn_str(Opts, ["Driver=" ++ str(Driver) | Acc]);
conn_str([{host, Host} | Opts], Acc) ->
Port = proplists:get_value(port, Opts, str(?SQLSERVER_DEFAULT_PORT)),
NOpts = proplists:delete(port, Opts),
conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
conn_str([{port, Port} | Opts], Acc) ->
Host = proplists:get_value(host, Opts, "localhost"),
NOpts = proplists:delete(host, Opts),
conn_str(NOpts, ["Server=" ++ str(Host) ++ "," ++ str(Port) | Acc]);
conn_str([{database, Database} | Opts], Acc) ->
conn_str(Opts, ["Database=" ++ str(Database) | Acc]);
conn_str([{username, Username} | Opts], Acc) ->
conn_str(Opts, ["UID=" ++ str(Username) | Acc]);
conn_str([{password, Password} | Opts], Acc) ->
conn_str(Opts, ["PWD=" ++ str(Password) | Acc]);
conn_str([{_, _} | Opts], Acc) ->
conn_str(Opts, Acc).
sent_data(Payload) ->
#{
payload => to_bin(Payload),
id => <<"0005F8F84FFFAFB9F44200000D810002">>,
topic => <<"test/topic">>,
qos => 0
}.
gen_batch_req(Count) when
is_integer(Count) andalso Count > 0
->
Vals = [{str(erlang:unique_integer())} || _Seq <- lists:seq(1, Count)],
Requests = [{send_message, sent_data(Payload)} || {Payload} <- Vals],
{Requests, Vals};
gen_batch_req(Count) ->
ct:pal("Gen batch requests failed with unexpected Count: ~p", [Count]).
str(List) when is_list(List) ->
unicode:characters_to_list(List, utf8);
str(Bin) when is_binary(Bin) ->
unicode:characters_to_list(Bin, utf8);
str(Num) when is_number(Num) ->
number_to_list(Num).
number_to_list(Int) when is_integer(Int) ->
integer_to_list(Int);
number_to_list(Float) when is_float(Float) ->
float_to_list(Float, [{decimals, 10}, compact]).
to_bin(List) when is_list(List) ->
unicode:characters_to_binary(List, utf8);
to_bin(Bin) when is_binary(Bin) ->
Bin.
Reference in New Issue View Git Blame Copy Permalink