Packages
Erlang/Elixir Cloud Framework elasticsearch Service
Current section
Files
Jump to
Current section
Files
test/erlasticsearch_SUITE.erl
%%%-------------------------------------------------------------------
%%% @author Mahesh Paolini-Subramanya <mahesh@dieswaytoofast.com>
%%% @copyright (C) 2013 Mahesh Paolini-Subramanya
%%% @doc type definitions and records.
%%% @end
%%%
%%% This source file is subject to the New BSD License. You should have received
%%% a copy of the New BSD license with this software. If not, it can be
%%% retrieved from: http://www.opensource.org/licenses/bsd-license.php
%%%-------------------------------------------------------------------
-module(erlasticsearch_SUITE).
-author('Mahesh Paolini-Subramanya <mahesh@dieswaytoofast.com>').
-include_lib("proper/include/proper.hrl").
-include_lib("common_test/include/ct.hrl").
-compile(export_all).
-define(CHECKSPEC(M,F,N), true = proper:check_spec({M,F,N})).
-define(PROPTEST(A), true = proper:quickcheck(A())).
-define(NUMTESTS, 500).
-define(DOCUMENT_DEPTH, 5).
-define(THREE_SHARDS, <<"{\"settings\":{\"number_of_shards\":3}}">>).
-define(MAPPING_KEY, <<"some_type">>).
-define(MAPPING_VALUE, <<"boolean">>).
-define(MAPPING_DOC(Type), [{Type, [{<<"properties">>, [{?MAPPING_KEY, [{<<"type">>, ?MAPPING_VALUE}]}]}]}]).
-define(ALIASES_DOC(Index, Alias), [{<<"actions">>, [[{<<"add">>, [{<<"index">>, Index}, {<<"alias">>, Alias}]}]]}]).
-define(UPDATE_KEY, <<"udpate_key">>).
-define(UPDATE_VALUE, <<"udpate_value">>).
-define(UPDATE_DOC, [{<<"doc">>, [{?UPDATE_KEY, ?UPDATE_VALUE}]}]).
suite() ->
[{ct_hooks,[cth_surefire]}, {timetrap,{seconds,320}}].
init_per_suite(Config) ->
% setup_lager(),
setup_environment(),
Config.
end_per_suite(_Config) ->
ok.
% Send in either the triplet or the name, at random
pool_name(Name) ->
case random:uniform(2) of
1 -> {"localhost", 9500, Name};
2 -> Name
end.
connection_options(1) ->
[];
connection_options(2) ->
[{thrift_host, "localhost"}];
connection_options(3) ->
[{thrift_host, "localhost"},
{thrift_port, 9500}];
connection_options(4) ->
[{thrift_host, "localhost"},
{thrift_port, 9500},
{thrift_options, [{framed, false}]}];
connection_options(_) ->
[{thrift_host, "localhost"},
{thrift_port, 9500},
{thrift_options, [{framed, false}]},
{binary_response, false}].
pool_options(1) ->
[];
pool_options(2) ->
[{size, 7}];
pool_options(_) ->
[{size, 7},
{max_overflow, 14}].
retry_options(1) ->
[];
retry_options(2) ->
[{retry_interval, 500}];
retry_options(_) ->
[{retry_interval, 500},
{retry_amount, 5}].
update_config(Config) ->
Version = es_test_version(Config),
Config1 = lists:foldl(fun(X, Acc) ->
proplists:delete(X, Acc)
end, Config, [es_test_version,
index,
type,
index_with_shards,
connection_options,
pool_options,
pool,
retry_options]),
[{es_test_version, Version + 1} | Config1].
es_test_version(Config) ->
case proplists:get_value(es_test_version, Config) of
undefined -> 1;
Val -> Val
end.
init_per_group(_GroupName, Config) ->
Config1 =
case ?config(saved_config, Config) of
{_, Config0} -> Config0;
undefined -> Config
end,
Version = es_test_version(Config1),
PoolName = pool_name(random_name(<<"pool_">>)),
PoolOptions = pool_options(Version),
ConnectionOptions = connection_options(Version),
RetryOptions = retry_options(Version),
Config2 = [{pool_options, PoolOptions},
{connection_options, ConnectionOptions ++ RetryOptions},
{pool, PoolName} | Config1],
start(Config2),
Index = random_name(<<"index_">>),
IndexWithShards = erlasticsearch:join([Index, <<"with_shards">>], <<"_">>),
Config3 = [{index, Index}, {index_with_shards, IndexWithShards}]
++ Config2,
% Clear out any existing indices w/ this name
delete_all_indices(PoolName, Config3),
Type = random_name(<<"type_">>),
[{type, Type}] ++ Config3.
end_per_group(_GroupName, Config) ->
PoolName = ?config(pool, Config),
Index = ?config(index, Config),
IndexWithShards = ?config(index_with_shards, Config),
delete_all_indices(PoolName, Index),
delete_all_indices(PoolName, IndexWithShards),
stop(Config),
Config1 = update_config(Config),
{save_config, Config1}.
init_per_testcase(_TestCase, Config) ->
Config.
end_per_testcase(_TestCase, _Config) ->
ok.
groups() ->
[{crud_index, [{repeat, 5}],
[t_is_index_1,
t_is_index_all,
t_is_type_1,
t_is_type_all,
t_create_index,
t_create_index_with_shards,
t_open_index
]},
{crud_mapping, [{repeat, 5}],
[t_put_mapping,
t_get_mapping,
t_delete_mapping
]},
{aliases, [{repeat, 5}],
[t_aliases,
t_insert_alias_1,
t_insert_alias_2,
t_delete_alias,
t_is_alias,
t_get_alias
]},
{index_helpers, [{repeat, 5}],
[t_flush_1,
t_flush_list,
t_flush_all,
t_refresh_1,
t_refresh_list,
t_refresh_all,
t_optimize_1,
t_optimize_list,
t_optimize_all,
t_segments_1,
t_segments_list,
t_segments_all,
t_status_1,
t_indices_stats,
t_status_all,
t_clear_cache_1,
t_clear_cache_list,
t_clear_cache_all
]},
% These three _MUST_ be in this sequence, and by themselves
{crud_doc, [{repeat, 5}],
[ t_insert_doc,
t_get_doc,
t_update_doc,
t_delete_doc,
t_bulk
]},
{doc_helpers, [{repeat, 5}],
[t_is_doc,
t_mget_index,
t_mget_type,
t_mget_id
]},
{test, [{repeat, 5}],
[
t_delete_by_query_param
]},
{cluster_helpers, [{repeat, 5}],
[t_health,
t_cluster_state,
t_state,
t_nodes_info,
t_nodes_stats
]},
{search, [{repeat, 5}],
[t_search,
t_count,
t_delete_by_query_param,
t_delete_by_query_doc
]}
].
all() ->
test_condition([
% {group, test}
{group, crud_index},
{group, crud_doc},
{group, search},
{group, index_helpers},
{group, cluster_helpers},
{group, doc_helpers},
{group, crud_mapping},
{group, aliases}
]).
t_health(Config) ->
ServerRef = ?config(pool, Config),
Response = erlasticsearch:health(ServerRef),
true = erlasticsearch:is_200(Response).
t_cluster_state(Config) ->
t_state(Config).
t_state(Config) ->
ServerRef = ?config(pool, Config),
Response1 = erlasticsearch:state(ServerRef),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:state(ServerRef, [{filter_nodes, true}]),
true = erlasticsearch:is_200(Response2).
t_nodes_info(Config) ->
ServerRef = ?config(pool, Config),
Response1 = erlasticsearch:nodes_info(ServerRef),
true = erlasticsearch:is_200(Response1).
t_nodes_stats(Config) ->
ServerRef = ?config(pool, Config),
Response1 = erlasticsearch:nodes_stats(ServerRef),
true = erlasticsearch:is_200(Response1).
t_indices_stats(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
check_indices_stats(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_status_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
check_status_1(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_status_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
check_status_all(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_clear_cache_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
clear_cache_1(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_clear_cache_list(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
clear_cache_list(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_clear_cache_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
clear_cache_all(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_is_index_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
are_indices_1(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_is_index_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
are_indices_all(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_is_type_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
build_data(ServerRef, Index, Type),
are_types_1(ServerRef, Index, Type),
clear_data(ServerRef, Index).
t_is_type_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
build_data(ServerRef, Index, Type),
are_types_all(ServerRef, Index, Type).
% clear_data(ServerRef, Index).
check_status_1(ServerRef, Index) ->
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
Response = erlasticsearch:status(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)).
check_indices_stats(ServerRef, Index) ->
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
Response = erlasticsearch:indices_stats(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)).
check_status_all(ServerRef, Index) ->
FullIndexList =
lists:map(fun(X) ->
enumerated(Index, X)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
Response = erlasticsearch:status(ServerRef, FullIndexList),
true = erlasticsearch:is_200(Response).
clear_cache_1(ServerRef, Index) ->
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
Response1 = erlasticsearch:clear_cache(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:clear_cache(ServerRef, FullIndex, [{filter, true}]),
true = erlasticsearch:is_200(Response2)
end, lists:seq(1, ?DOCUMENT_DEPTH)).
clear_cache_list(ServerRef, Index) ->
FullIndexList =
lists:map(fun(X) ->
enumerated(Index, X)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
Response1 = erlasticsearch:clear_cache(ServerRef, FullIndexList),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:clear_cache(ServerRef, FullIndexList, [{filter, true}]),
true = erlasticsearch:is_200(Response2).
clear_cache_all(ServerRef, _Index) ->
Response1 = erlasticsearch:clear_cache(ServerRef),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:clear_cache(ServerRef, [], [{filter, true}]),
true = erlasticsearch:is_200(Response2).
are_types_1(ServerRef, Index, Type) ->
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
lists:foreach(fun(Y) ->
FullType = enumerated(Type, Y),
true = true_response(erlasticsearch:is_type(ServerRef, FullIndex, FullType))
end, lists:seq(1, ?DOCUMENT_DEPTH))
end, lists:seq(1, ?DOCUMENT_DEPTH)).
are_types_all(ServerRef, Index, Type) ->
FullIndexList =
lists:map(fun(X) ->
enumerated(Index, X)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
FullTypeList =
lists:map(fun(X) ->
enumerated(Type, X)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
% List of indices
lists:foreach(fun(X) ->
FullType = enumerated(Type, X),
true = true_response(erlasticsearch:is_type(ServerRef, FullIndexList, FullType))
end, lists:seq(1, ?DOCUMENT_DEPTH)),
% List of types
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
true = true_response(erlasticsearch:is_type(ServerRef, FullIndex, FullTypeList))
end, lists:seq(1, ?DOCUMENT_DEPTH)),
% List of indices and types
true = true_response(erlasticsearch:is_type(ServerRef, FullIndexList, FullTypeList)).
build_data(ServerRef, Index, Type) ->
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
lists:foreach(fun(Y) ->
FullType = enumerated(Type, Y),
BX = list_to_binary(integer_to_list(X)),
erlasticsearch:insert_doc(ServerRef, FullIndex,
FullType, BX, json_document(X)),
erlasticsearch:flush(ServerRef)
end, lists:seq(1, ?DOCUMENT_DEPTH))
end, lists:seq(1, ?DOCUMENT_DEPTH)).
clear_data(ServerRef, Index) ->
lists:foreach(fun(X) ->
FullIndex = enumerated(Index, X),
erlasticsearch:delete_index(ServerRef, FullIndex)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
erlasticsearch:flush(ServerRef).
% Also deletes indices
t_create_index(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
delete_all_indices(ServerRef, Index).
% Also deletes indices
t_put_mapping(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
erlasticsearch:create_index(ServerRef, Index),
MappingDoc = jsx:encode(?MAPPING_DOC(Type)),
Response = erlasticsearch:put_mapping(ServerRef, Index, Type, MappingDoc),
true = erlasticsearch:is_200(Response),
delete_this_index(ServerRef, Index).
t_get_mapping(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
erlasticsearch:create_index(ServerRef, Index),
MappingDoc = jsx:encode(?MAPPING_DOC(Type)),
Response1 = erlasticsearch:put_mapping(ServerRef, Index, Type, MappingDoc),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:get_mapping(ServerRef, Index, Type),
validate_mapping(Index, Type, Response2),
delete_this_index(ServerRef, Index).
validate_mapping(Index, Type, Response) ->
{body, Data1} = lists:keyfind(body, 1, Response),
DataB = case is_binary(Data1) of
true ->
jsx:decode(Data1);
false ->
Data1
end,
{Index, DataI} = lists:keyfind(Index, 1, DataB),
{<<"mappings">>, DataM} = lists:keyfind(<<"mappings">>, 1, DataI),
{Type, DataT} = lists:keyfind(Type, 1, DataM),
{<<"properties">>, DataP} = lists:keyfind(<<"properties">>, 1, DataT),
{?MAPPING_KEY, [{<<"type">>, ?MAPPING_VALUE}]} = lists:keyfind(?MAPPING_KEY, 1, DataP).
t_delete_mapping(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
erlasticsearch:create_index(ServerRef, Index),
MappingDoc = jsx:encode(?MAPPING_DOC(Type)),
Response1 = erlasticsearch:put_mapping(ServerRef, Index, Type, MappingDoc),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:delete_mapping(ServerRef, Index, Type),
true = erlasticsearch:is_200(Response2),
delete_this_index(ServerRef, Index).
t_aliases(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Alias = random_name(Index),
erlasticsearch:create_index(ServerRef, Index),
AliasesDoc = ?ALIASES_DOC(Index, Alias),
Response = erlasticsearch:aliases(ServerRef, AliasesDoc),
true = erlasticsearch:is_200(Response),
delete_this_index(ServerRef, Index).
t_insert_alias_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Alias = random_name(Index),
erlasticsearch:create_index(ServerRef, Index),
Response = erlasticsearch:insert_alias(ServerRef, Index, Alias),
true = erlasticsearch:is_200(Response),
delete_this_index(ServerRef, Index).
t_insert_alias_2(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Alias = random_name(Index),
erlasticsearch:create_index(ServerRef, Index),
Params = [{<<"routing">>, <<"1">>}],
Response = erlasticsearch:insert_alias(ServerRef, Index, Alias, Params),
true = erlasticsearch:is_200(Response),
delete_this_index(ServerRef, Index).
t_delete_alias(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Alias = random_name(Index),
erlasticsearch:create_index(ServerRef, Index),
Response1 = erlasticsearch:insert_alias(ServerRef, Index, Alias),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:delete_alias(ServerRef, Index, Alias),
true = erlasticsearch:is_200(Response2),
delete_this_index(ServerRef, Index).
t_is_alias(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Alias = random_name(Index),
erlasticsearch:create_index(ServerRef, Index),
Response1 = erlasticsearch:insert_alias(ServerRef, Index, Alias),
true = erlasticsearch:is_200(Response1),
true = true_response(erlasticsearch:is_alias(ServerRef, Index, Alias)),
delete_this_index(ServerRef, Index).
t_get_alias(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Alias = random_name(Index),
erlasticsearch:create_index(ServerRef, Index),
Response1 = erlasticsearch:insert_alias(ServerRef, Index, Alias),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:get_alias(ServerRef, Index, Alias),
validate_alias(Index, Alias, Response2),
delete_this_index(ServerRef, Index).
validate_alias(Index, Alias, Response) ->
{body, Data1} = lists:keyfind(body, 1, Response),
Data2 = case is_binary(Data1) of
true ->
jsx:decode(Data1);
false ->
Data1
end,
{Index, Data3} = lists:keyfind(Index, 1, Data2),
{<<"aliases">>, Data4} = lists:keyfind(<<"aliases">>, 1, Data3),
{Alias, _} = lists:keyfind(Alias, 1, Data4).
t_create_index_with_shards(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index_with_shards, Config),
create_indices(ServerRef, Index),
delete_all_indices(ServerRef, Index).
t_flush_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
Response = erlasticsearch:flush(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
delete_all_indices(ServerRef, Index).
t_flush_list(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Indexes =
lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
erlasticsearch:join([Index, BX], <<"_">>)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
Response = erlasticsearch:flush(ServerRef, Indexes),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_flush_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Response = erlasticsearch:flush(ServerRef),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_refresh_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
Response = erlasticsearch:refresh(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
delete_all_indices(ServerRef, Index).
t_refresh_list(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Indexes =
lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
erlasticsearch:join([Index, BX], <<"_">>)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
Response = erlasticsearch:refresh(ServerRef, Indexes),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_refresh_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Response = erlasticsearch:refresh(ServerRef),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_optimize_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
Response = erlasticsearch:optimize(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
delete_all_indices(ServerRef, Index).
t_optimize_list(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Indexes =
lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
erlasticsearch:join([Index, BX], <<"_">>)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
Response = erlasticsearch:optimize(ServerRef, Indexes),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_optimize_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Response = erlasticsearch:optimize(ServerRef),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_segments_1(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
Response = erlasticsearch:segments(ServerRef, FullIndex),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
delete_all_indices(ServerRef, Index).
t_segments_list(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Indexes =
lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
erlasticsearch:join([Index, BX], <<"_">>)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
Response = erlasticsearch:segments(ServerRef, Indexes),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_segments_all(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
create_indices(ServerRef, Index),
Response = erlasticsearch:segments(ServerRef),
true = erlasticsearch:is_200(Response),
delete_all_indices(ServerRef, Index).
t_open_index(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
process_t_insert_doc(ServerRef, Config),
Response = erlasticsearch:close_index(ServerRef, Index),
true = erlasticsearch:is_200(Response),
Response1 = erlasticsearch:open_index(ServerRef, Index),
true = erlasticsearch:is_200(Response1),
process_t_delete_doc(ServerRef, Config).
t_mget_id(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
Query = id_query(),
process_t_insert_doc(ServerRef, Config),
Result = erlasticsearch:mget_doc(ServerRef, Index, Type, Query),
?DOCUMENT_DEPTH = docs_from_result(Result),
process_t_delete_doc(ServerRef, Config).
t_mget_type(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
Query = id_query(Type),
process_t_insert_doc(ServerRef, Config),
Result = erlasticsearch:mget_doc(ServerRef, Index, Query),
?DOCUMENT_DEPTH = docs_from_result(Result),
process_t_delete_doc(ServerRef, Config).
t_mget_index(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
Query = id_query(Index, Type),
process_t_insert_doc(ServerRef, Config),
Result = erlasticsearch:mget_doc(ServerRef, Query),
?DOCUMENT_DEPTH = docs_from_result(Result),
process_t_delete_doc(ServerRef, Config).
t_search(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
process_t_insert_doc(ServerRef, Config),
lists:foreach(fun(X) ->
Query = param_query(X),
Result = erlasticsearch:search(ServerRef, Index, Type, <<>>, [{q, Query}]),
% The document is structured so that the number of top level
% keys is as (?DOCUMENT_DEPTH + 1 - X)
?DOCUMENT_DEPTH = hits_from_result(Result) + X - 1
end, lists:seq(1, ?DOCUMENT_DEPTH)),
process_t_delete_doc(ServerRef, Config).
t_count(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
process_t_insert_doc(ServerRef, Config),
lists:foreach(fun(X) ->
Query1 = param_query(X),
Query2 = json_query(X),
% query as parameter
Result1 = erlasticsearch:count(ServerRef, Index, Type, <<>>, [{q, Query1}]),
% The document is structured so that the number of top level
% keys is as (?DOCUMENT_DEPTH + 1 - X)
?DOCUMENT_DEPTH = count_from_result(Result1) + X - 1,
% query as doc
Result2 = erlasticsearch:count(ServerRef, Index, Type, Query2, []),
% The document is structured so that the number of top level
% keys is as (?DOCUMENT_DEPTH + 1 - X)
?DOCUMENT_DEPTH = count_from_result(Result2) + X - 1
end, lists:seq(1, ?DOCUMENT_DEPTH)),
process_t_delete_doc(ServerRef, Config).
t_delete_by_query_param(Config) ->
ServerRef = ?config(pool, Config),
% One Index
Index = ?config(index, Config),
Type = ?config(type, Config),
Query1 = param_query(1),
process_t_insert_doc(ServerRef, Config),
Result1 = erlasticsearch:count(ServerRef, Index, Type, <<>>, [{q, Query1}]),
5 = count_from_result(Result1),
DResult1 = erlasticsearch:delete_by_query(ServerRef, Index, Type, <<>>, [{q, Query1}]),
true = erlasticsearch:is_200(DResult1),
erlasticsearch:flush(ServerRef, Index),
DResult1a = erlasticsearch:count(ServerRef, Index, Type, <<>>, [{q, Query1}]),
0 = count_from_result(DResult1a),
% All Indices
process_t_insert_doc(ServerRef, Config),
ADResult1 = erlasticsearch:delete_by_query(ServerRef, <<>>, [{q, Query1}]),
true = erlasticsearch:is_200(ADResult1),
erlasticsearch:flush(ServerRef, Index),
ADResult1a = erlasticsearch:count(ServerRef, <<>>, [{q, Query1}]),
0 = count_from_result(ADResult1a).
% Don't need to delete docs, 'cos they are already deleted
% process_t_delete_doc(ServerRef, Config).
t_delete_by_query_doc(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
Query1 = param_query(1),
Query2 = json_query(1),
process_t_insert_doc(ServerRef, Config),
Result1 = erlasticsearch:count(ServerRef, Index, Type, <<>>, [{q, Query1}]),
5 = count_from_result(Result1),
DResult1 = erlasticsearch:delete_by_query(ServerRef, Index, Type, Query2, []),
true = erlasticsearch:is_200(DResult1),
erlasticsearch:flush(ServerRef, Index),
DResult1a = erlasticsearch:count(ServerRef, Index, Type, <<>>, [{q, Query1}]),
0 = count_from_result(DResult1a),
% process_t_delete_doc(ServerRef, Config).
% All Indices
process_t_insert_doc(ServerRef, Config),
ADResult1 = erlasticsearch:delete_by_query(ServerRef, Query2),
true = erlasticsearch:is_200(ADResult1),
erlasticsearch:flush(ServerRef, Index),
ADResult1a = erlasticsearch:count(ServerRef, <<>>, [{q, Query1}]),
0 = count_from_result(ADResult1a).
% Don't need to delete docs, 'cos they are already deleted
% process_t_delete_doc(ServerRef, Config).
id_query() ->
Ids = lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
[{<<"_id">>, BX}]
end, lists:seq(1, ?DOCUMENT_DEPTH)),
jsx:encode([{docs, Ids}]).
id_query(Type) ->
Ids = lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
[{<<"_type">>, Type}, {<<"_id">>, BX}]
end, lists:seq(1, ?DOCUMENT_DEPTH)),
jsx:encode([{docs, Ids}]).
id_query(Index, Type) ->
Ids = lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
[{<<"_index">>, Index}, {<<"_type">>, Type}, {<<"_id">>, BX}]
end, lists:seq(1, ?DOCUMENT_DEPTH)),
jsx:encode([{docs, Ids}]).
param_query(X) ->
Key = key(X),
Value = value(X),
erlasticsearch:join([Key, Value], <<":">>).
json_query(X) ->
Key = key(X),
Value = value(X),
jsx:encode([{<<"query">>, [{<<"term">>, [{Key, Value}]}]}]).
decode_body(Body) when is_binary(Body) ->
jsx:decode(Body);
decode_body(Body) -> Body.
hits_from_result(Result) ->
{body, Body} = lists:keyfind(body, 1, Result),
DBody = decode_body(Body),
case lists:keyfind(<<"hits">>, 1, DBody) of
false -> throw(false);
{_, Value} ->
case lists:keyfind(<<"total">>, 1, Value) of
false -> throw(false);
{_, Data} -> Data
end
end.
docs_from_result(Result) ->
{body, Body} = lists:keyfind(body, 1, Result),
DBody = decode_body(Body),
case lists:keyfind(<<"docs">>, 1, DBody) of
false -> throw(false);
{_, Value} ->
length(Value)
end.
count_from_result(Result) ->
{body, Body} = lists:keyfind(body, 1, Result),
DBody = decode_body(Body),
case lists:keyfind(<<"count">>, 1, DBody) of
false -> throw(false);
{_, Value} -> Value
end.
t_bulk(Config) ->
ServerRef = ?config(pool, Config),
process_t_bulk(ServerRef, Config).
t_insert_doc(Config) ->
ServerRef = ?config(pool, Config),
process_t_insert_doc(ServerRef, Config),
process_t_delete_doc(ServerRef, Config).
t_is_doc(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
process_t_insert_doc(ServerRef, Config),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
true = true_response(erlasticsearch:is_doc(ServerRef, Index, Type, BX))
end, lists:seq(1, ?DOCUMENT_DEPTH)),
process_t_delete_doc(ServerRef, Config).
t_get_doc(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
process_t_insert_doc(ServerRef, Config),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
Response = erlasticsearch:get_doc(ServerRef, Index, Type, BX),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
process_t_delete_doc(ServerRef, Config).
t_update_doc(Config) ->
ServerRef = ?config(pool, Config),
Index = ?config(index, Config),
Type = ?config(type, Config),
process_t_insert_doc(ServerRef, Config),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
Doc = ?UPDATE_DOC,
Response1 = erlasticsearch:update_doc(ServerRef, Index, Type, BX, Doc),
true = erlasticsearch:is_200(Response1),
Response2 = erlasticsearch:get_doc(ServerRef, Index, Type, BX),
validate_update(Response2)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
process_t_delete_doc(ServerRef, Config).
validate_update(Response) ->
{body, Data1} = lists:keyfind(body, 1, Response),
Data2 = case is_binary(Data1) of
true ->
jsx:decode(Data1);
false ->
Data1
end,
{_, Data3} = lists:keyfind(<<"_source">>, 1, Data2),
{?UPDATE_KEY, ?UPDATE_VALUE} = lists:keyfind(?UPDATE_KEY, 1, Data3).
t_delete_doc(Config) ->
ServerRef = ?config(pool, Config),
process_t_insert_doc(ServerRef, Config),
process_t_delete_doc(ServerRef, Config).
process_t_insert_doc(ServerRef, Config) ->
Index = ?config(index, Config),
Type = ?config(type, Config),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
Response = erlasticsearch:insert_doc(ServerRef, Index,
Type, BX, json_document(X)),
true = erlasticsearch:is_200_or_201(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
erlasticsearch:flush(ServerRef, Index).
process_t_bulk(ServerRef, Config) ->
Index = ?config(index, Config),
Type = ?config(type, Config),
lists:foreach(fun(X) ->
Response1 = erlasticsearch:bulk(ServerRef, <<>>,
<<>>, bulk_document(<<>>, <<>>, X)),
true = erlasticsearch:is_200_or_201(Response1),
Response2 = erlasticsearch:bulk(ServerRef, Index,
Type, bulk_document(Index, <<>>, X)),
true = erlasticsearch:is_200_or_201(Response2),
Response3 = erlasticsearch:bulk(ServerRef, Index,
Type, bulk_document(Index, Type, X)),
true = erlasticsearch:is_200_or_201(Response3)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
erlasticsearch:flush(ServerRef, Index).
process_t_delete_doc(ServerRef, Config) ->
Index = ?config(index, Config),
Type = ?config(type, Config),
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
Response = erlasticsearch:delete_doc(ServerRef, Index, Type, BX),
true = erlasticsearch:is_200(Response)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
erlasticsearch:flush(ServerRef, Index).
%% Test helpers
% Create a bunch-a indices
create_indices(ServerRef, Index) ->
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
erlasticsearch:create_index(ServerRef, FullIndex),
erlasticsearch:flush(ServerRef, Index)
end, lists:seq(1, ?DOCUMENT_DEPTH)).
are_indices_1(ServerRef, Index) ->
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
true = true_response(erlasticsearch:is_index(ServerRef, FullIndex))
end, lists:seq(1, ?DOCUMENT_DEPTH)).
are_indices_all(ServerRef, Index) ->
FullIndexList =
lists:map(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
erlasticsearch:join([Index, BX], <<"_">>)
end, lists:seq(1, ?DOCUMENT_DEPTH)),
true = true_response(erlasticsearch:is_index(ServerRef, FullIndexList)).
delete_all_indices(ServerRef, Config) when is_list(Config) ->
Index = ?config(index, Config),
IndexWithShards = erlasticsearch:join([Index, <<"with_shards">>], <<"_">>),
delete_all_indices(ServerRef, Index),
delete_all_indices(ServerRef, IndexWithShards),
erlasticsearch:flush(ServerRef);
% Optionally check to see if the indices exist before trying to delete
delete_all_indices(ServerRef, Index) when is_binary(Index) ->
lists:foreach(fun(X) ->
BX = list_to_binary(integer_to_list(X)),
FullIndex = erlasticsearch:join([Index, BX], <<"_">>),
delete_this_index(ServerRef, FullIndex)
end, lists:seq(1, ?DOCUMENT_DEPTH)).
delete_this_index(ServerRef, Index) ->
case true_response(erlasticsearch:is_index(ServerRef, Index)) of
true ->
Response = erlasticsearch:delete_index(ServerRef, Index),
true = erlasticsearch:is_200(Response);
false -> true
end.
json_document(N) ->
jsx:encode(document(N)).
document(N) ->
lists:foldl(fun(X, Acc) ->
Tuple = case X of
1 ->
[{key(1), value(1)}];
X ->
[{key(X), value(X)},
{sub(X), document(N-1)}]
end,
Tuple ++ Acc
end, [], lists:seq(1, N)).
bulk_document(Index, Type, N) ->
case N of
1 -> bulk_line(Index, Type, 1);
X -> erlasticsearch:join([bulk_line(Index, Type, X), bulk_document(Index, Type, X-1)], <<>>)
end.
bulk_line(<<>>, <<>>, N) ->
Index = index(N),
Type = type(N),
Id = id(N),
Key = key(N),
Value = value(N),
<<"{ \"index\" : { \"_index\" : \"", Index/binary, "\", \"_type\" : \"", Type/binary, "\", \"_id\" : \"", Id/binary, "\" } }\n{ \"", Key/binary, "\" : \"", Value/binary, "\" }\n">>;
% Pre-existing Index
bulk_line(_Index, <<>>, N) ->
Type = type(N),
Id = id(N),
Key = key(N),
Value = value(N),
<<"{ \"index\" : { \"_type\" : \"", Type/binary, "\", \"_id\" : \"", Id/binary, "\" } }\n{ \"", Key/binary, "\" : \"", Value/binary, "\" }\n">>;
% Pre-existing Index and Type
bulk_line(_Index, _Type, N) ->
Id = id(N),
Key = key(N),
Value = value(N),
<<"{ \"index\" : { \"_id\" : \"", Id/binary, "\" } }\n{ \"", Key/binary, "\" : \"", Value/binary, "\" }\n">>.
index(N) -> data_index(index, N).
type(N) -> data_index(type, N).
id(N) -> data_index(id, N).
key(N) -> data_index(key, N).
value(N) -> data_index(value, N).
sub(N) -> data_index(sub, N).
enumerated(Item, N) when is_binary(Item) ->
data_index(list_to_atom(binary_to_list(Item)), N).
-spec data_index(atom(), integer()) -> binary().
data_index(Data, Index) ->
BData = list_to_binary(atom_to_list(Data)),
BIndex = list_to_binary(integer_to_list(Index)),
erlasticsearch:join([BData, BIndex], <<"_">>).
random_name(Name) ->
random:seed(erlang:now()),
Id = list_to_binary(integer_to_list(random:uniform(999999999))),
<<Name/binary, Id/binary>>.
true_response(Response) ->
case lists:keyfind(result, 1, Response) of
{result, true} -> true;
{result, <<"true">>} -> true;
_ -> false
end.
setup_environment() ->
random:seed(erlang:now()).
setup_lager() ->
reltool_util:application_start(lager),
lager:set_loglevel(lager_console_backend, debug),
lager:set_loglevel(lager_file_backend, "console.log", debug).
start(Config) ->
PoolName = ?config(pool, Config),
ConnectionOptions = ?config(connection_options, Config),
PoolOptions = ?config(pool_options, Config),
reltool_util:application_start(erlasticsearch),
erlasticsearch:start_pool(PoolName, PoolOptions, ConnectionOptions)
.
stop(Config) ->
PoolName = ?config(pool, Config),
erlasticsearch:stop_pool(PoolName),
reltool_util:application_stop(erlasticsearch),
ok.
-define(DEFAULT_THRIFT_HOST, "localhost").
-define(DEFAULT_THRIFT_PORT, 9500).
test_condition(L) ->
case gen_tcp:connect(?DEFAULT_THRIFT_HOST, ?DEFAULT_THRIFT_PORT, []) of
{ok, Socket} ->
catch gen_tcp:close(Socket),
L;
{error, econnrefused} ->
error_logger:error_msg("unable to test ~p",
[{?DEFAULT_THRIFT_HOST,
?DEFAULT_THRIFT_PORT}]),
{skip, elasticsearch_dead};
{error, Reason} ->
error_logger:error_msg("unable to test ~p: ~p",
[{?DEFAULT_THRIFT_HOST,
?DEFAULT_THRIFT_PORT}, Reason]),
{skip, elasticsearch_dead}
end.