% Licensed under the Apache License, Version 2.0 (the "License"); you may not
% use this file except in compliance with the License. You may obtain a copy of
% the License at
%
%   http://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
% License for the specific language governing permissions and limitations under
% the License.

-module(couch_query_servers_tests).

-include_lib("couch/include/couch_db.hrl").
-include_lib("couch/include/couch_eunit.hrl").


setup() ->
    meck:new([config, couch_log]).


teardown(_) ->
    meck:unload().


setup_oom() ->
    test_util:start_couch([ioq]).


teardown_oom(Ctx) ->
    meck:unload(),
    test_util:stop_couch(Ctx).


sum_overflow_test_() ->
    {
        "Test overflow detection in the _sum reduce function",
        {
            setup,
            fun setup/0,
            fun teardown/1,
            [
                fun should_return_error_on_overflow/0,
                fun should_return_object_on_log/0,
                fun should_return_object_on_false/0
            ]
        }
    }.


filter_oom_test_() ->
{
    "Test recovery from oom in filters",
    {
        setup,
        fun setup_oom/0,
        fun teardown_oom/1,
        [
            fun should_split_large_batches/0
        ]
    }
}.

should_return_error_on_overflow() ->
    meck:reset([config, couch_log]),
    meck:expect(
            config, get, ["query_server_config", "reduce_limit", "true"],
            "true"
        ),
    meck:expect(couch_log, error, ['_', '_'], ok),
    KVs = gen_sum_kvs(),
    {ok, [Result]} = couch_query_servers:reduce(<<"foo">>, [<<"_sum">>], KVs),
    ?assertMatch({[{<<"error">>, <<"builtin_reduce_error">>} | _]}, Result),
    ?assert(meck:called(config, get, '_')),
    ?assert(meck:called(couch_log, error, '_')).


should_return_object_on_log() ->
    meck:reset([config, couch_log]),
    meck:expect(
            config, get, ["query_server_config", "reduce_limit", "true"],
            "log"
        ),
    meck:expect(couch_log, error, ['_', '_'], ok),
    KVs = gen_sum_kvs(),
    {ok, [Result]} = couch_query_servers:reduce(<<"foo">>, [<<"_sum">>], KVs),
    ?assertMatch({[_ | _]}, Result),
    Keys = [K || {K, _} <- element(1, Result)],
    ?assert(not lists:member(<<"error">>, Keys)),
    ?assert(meck:called(config, get, '_')),
    ?assert(meck:called(couch_log, error, '_')).


should_return_object_on_false() ->
    meck:reset([config, couch_log]),
    meck:expect(
            config, get, ["query_server_config", "reduce_limit", "true"],
            "false"
        ),
    meck:expect(couch_log, error, ['_', '_'], ok),
    KVs = gen_sum_kvs(),
    {ok, [Result]} = couch_query_servers:reduce(<<"foo">>, [<<"_sum">>], KVs),
    ?assertMatch({[_ | _]}, Result),
    Keys = [K || {K, _} <- element(1, Result)],
    ?assert(not lists:member(<<"error">>, Keys)),
    ?assert(meck:called(config, get, '_')),
    ?assertNot(meck:called(couch_log, error, '_')).


should_split_large_batches() ->
    Req = {json_req, {[]}},
    Db = <<"somedb">>,
    DDoc = #doc{
        id = <<"_design/foo">>,
        revs = {0, [<<"bork bork bork">>]},
        body = {[
            {<<"filters">>, {[
                {<<"bar">>, <<"function(req, doc) {return true;}">>}
            ]}}
        ]}
    },
    FName = <<"bar">>,
    Docs = [
        #doc{id = <<"a">>, body = {[]}},
        #doc{id = <<"b">>, body = {[]}}
    ],
    meck:new(couch_os_process, [passthrough]),
    meck:expect(couch_os_process, prompt, fun(Pid, Data) ->
        case Data of
            [<<"ddoc">>, _, [<<"filters">>, <<"bar">>], [[_, _], _]] ->
                throw({os_process_error, {exit_status, 1}});
            [<<"ddoc">>, _, [<<"filters">>, <<"bar">>], [[_], _]] ->
                [true, [split_batch]];
            _ ->
                meck:passthrough([Pid, Data])
        end
    end),
    {ok, Ret} = couch_query_servers:filter_docs(Req, Db, DDoc, FName, Docs),
    ?assertEqual([split_batch, split_batch], Ret).


gen_sum_kvs() ->
    lists:map(fun(I) ->
        Props = lists:map(fun(_) ->
            K = couch_util:encodeBase64Url(crypto:strong_rand_bytes(16)),
            {K, 1}
        end, lists:seq(1, 20)),
        [I, {Props}]
    end, lists:seq(1, 10)).