src/couch_replicator/test/eunit/couch_replicator_error_reporting_tests.erl - couchdb - Git at Google

 % 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_replicator_error_reporting_tests).

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

 error_reporting_test_() ->
     {
         foreach,
         fun couch_replicator_test_helper:test_setup/0,
         fun couch_replicator_test_helper:test_teardown/1,
         [
             ?TDEF_FE(t_fail_bulk_docs),
             ?TDEF_FE(t_fail_changes_reader),
             ?TDEF_FE(t_fail_doc_put_4xx_well_formed_json_error),
             ?TDEF_FE(t_fail_doc_put_4xx_unexpected_json_error),
             ?TDEF_FE(t_fail_doc_put_4xx_invalid_json_error),
             ?TDEF_FE(t_skip_doc_put_401_errors),
             ?TDEF_FE(t_skip_doc_put_403_errors),
             ?TDEF_FE(t_skip_doc_put_413_errors),
             ?TDEF_FE(t_skip_doc_put_415_errors),
             ?TDEF_FE(t_skip_doc_put_invalid_attachment_name),
             ?TDEF_FE(t_fail_revs_diff),
             ?TDEF_FE(t_fail_bulk_get, 15),
             ?TDEF_FE(t_fail_open_docs_get, 15),
             ?TDEF_FE(t_fail_changes_queue),
             ?TDEF_FE(t_fail_changes_manager),
             ?TDEF_FE(t_fail_changes_reader_proc),
             ?TDEF_FE(t_dont_start_duplicate_job),
             ?TDEF_FE(t_can_start_multiple_jobs),
             ?TDEF_FE(t_stop_duplicate_job)
         ]
     }.

 t_fail_bulk_docs({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     mock_fail_req(post, "/_bulk_docs", {ok, "403", [], [<<"{\"x\":\"y\"}">>]}),
     populate_db(Source, 6, 6),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({bulk_docs_failed, 403, {[{<<"x">>, <<"y">>}]}}, Result),

     couch_replicator_notifier:stop(Listener).

 t_fail_changes_reader({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     mock_fail_req(get, "/_changes", {ok, "418", [], [<<"{\"x\":\"y\"}">>]}),
     populate_db(Source, 6, 6),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({changes_req_failed, 418, {[{<<"x">>, <<"y">>}]}}, Result),

     couch_replicator_notifier:stop(Listener).

 t_fail_doc_put_4xx_well_formed_json_error({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     ErrBody = [<<"{\"error\":\"x\", \"reason\":\"y\"}">>],
     mock_fail_req(put, "/6", {ok, "400", [], ErrBody}),
     populate_db(Source, 6, 6, _WithAttachments = true),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({doc_write_failed, {<<"x">>, <<"y">>}}, Result),

     couch_replicator_notifier:stop(Listener).

 t_fail_doc_put_4xx_unexpected_json_error({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     ErrBody = [<<"{\"a\":\"b\"}">>],
     mock_fail_req(put, "/6", {ok, "400", [], ErrBody}),
     populate_db(Source, 6, 6, _WithAttachments = true),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({doc_write_failed, {400, [{<<"a">>, <<"b">>}]}}, Result),

     couch_replicator_notifier:stop(Listener).

 t_fail_doc_put_4xx_invalid_json_error({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     mock_fail_req(put, "/6", {ok, "400", [], [<<"potato">>]}),
     populate_db(Source, 6, 6, _WithAttachments = true),

     {error, Result} = wait_rep_result(RepId),
     ?assertMatch({doc_write_failed, {invalid_json, _}}, Result),

     couch_replicator_notifier:stop(Listener).

 t_skip_doc_put_401_errors({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     populate_db(Source, 6, 6, _WithAttachments = true),
     ErrBody = [<<"{\"error\":\"unauthorized\", \"reason\":\"vdu\"}">>],
     mock_fail_req(put, "/6", {ok, "401", [], ErrBody}),
     {ok, RepId} = replicate(Source, Target, #{continuous => false}),
     {ok, Listener} = rep_result_listener(RepId),
     Res = wait_rep_result(RepId),
     % Replication job should succeed
     ?assertMatch({ok, {[_ | _]}}, Res),
     {ok, {Props}} = Res,
     History = proplists:get_value(<<"history">>, Props),
     ?assertMatch([{[_ | _]}], History),
     [{HistProps}] = History,
     DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
     DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
     ?assertEqual(5, DocsWritten),
     ?assertEqual(1, DocWriteFailures),
     couch_replicator_notifier:stop(Listener).

 t_skip_doc_put_403_errors({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     populate_db(Source, 6, 6, _WithAttachments = true),
     ErrBody = [<<"{\"error\":\"forbidden\", \"reason\":\"vdu\"}">>],
     mock_fail_req(put, "/6", {ok, "403", [], ErrBody}),
     {ok, RepId} = replicate(Source, Target, #{continuous => false}),
     {ok, Listener} = rep_result_listener(RepId),
     Res = wait_rep_result(RepId),
     % Replication job should succeed
     ?assertMatch({ok, {[_ | _]}}, Res),
     {ok, {Props}} = Res,
     History = proplists:get_value(<<"history">>, Props),
     ?assertMatch([{[_ | _]}], History),
     [{HistProps}] = History,
     DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
     DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
     ?assertEqual(5, DocsWritten),
     ?assertEqual(1, DocWriteFailures),
     couch_replicator_notifier:stop(Listener).

 t_skip_doc_put_413_errors({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     populate_db(Source, 6, 6, _WithAttachments = true),
     ErrBody = [<<"{\"error\":\"too_large\", \"reason\":\"too_large\"}">>],
     mock_fail_req(put, "/6", {ok, "413", [], ErrBody}),
     {ok, RepId} = replicate(Source, Target, #{continuous => false}),
     {ok, Listener} = rep_result_listener(RepId),
     Res = wait_rep_result(RepId),
     % Replication job should succeed
     ?assertMatch({ok, {[_ | _]}}, Res),
     {ok, {Props}} = Res,
     History = proplists:get_value(<<"history">>, Props),
     ?assertMatch([{[_ | _]}], History),
     [{HistProps}] = History,
     DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
     DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
     ?assertEqual(5, DocsWritten),
     ?assertEqual(1, DocWriteFailures),
     couch_replicator_notifier:stop(Listener).

 t_skip_doc_put_415_errors({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     populate_db(Source, 6, 6, _WithAttachments = true),
     ErrBody = [<<"{\"error\":\"unsupported_media_type\", \"reason\":\"bad_media\"}">>],
     mock_fail_req(put, "/6", {ok, "415", [], ErrBody}),
     {ok, RepId} = replicate(Source, Target, #{continuous => false}),
     {ok, Listener} = rep_result_listener(RepId),
     Res = wait_rep_result(RepId),
     % Replication job should succeed
     ?assertMatch({ok, {[_ | _]}}, Res),
     {ok, {Props}} = Res,
     History = proplists:get_value(<<"history">>, Props),
     ?assertMatch([{[_ | _]}], History),
     [{HistProps}] = History,
     DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
     DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
     ?assertEqual(5, DocsWritten),
     ?assertEqual(1, DocWriteFailures),
     couch_replicator_notifier:stop(Listener).

 t_skip_doc_put_invalid_attachment_name({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     populate_db(Source, 6, 6, _WithAttachments = true),
     ErrBody = [
         <<"{\"error\":\"bad_request\", \"reason\":\"Attachment name '_foo' starts with prohibited character '_'\"}">>
     ],
     mock_fail_req(put, "/6", {ok, "400", [], ErrBody}),
     {ok, RepId} = replicate(Source, Target, #{continuous => false}),
     {ok, Listener} = rep_result_listener(RepId),
     Res = wait_rep_result(RepId),
     % Replication job should succeed
     ?assertMatch({ok, {[_ | _]}}, Res),
     {ok, {Props}} = Res,
     History = proplists:get_value(<<"history">>, Props),
     ?assertMatch([{[_ | _]}], History),
     [{HistProps}] = History,
     DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
     DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
     ?assertEqual(5, DocsWritten),
     ?assertEqual(1, DocWriteFailures),
     couch_replicator_notifier:stop(Listener).

 t_fail_revs_diff({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     mock_fail_req(post, "/_revs_diff", {ok, "407", [], [<<"{\"x\":\"y\"}">>]}),
     populate_db(Source, 6, 6),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({revs_diff_failed, 407, {[{<<"x">>, <<"y">>}]}}, Result),

     couch_replicator_notifier:stop(Listener).

 t_fail_bulk_get({_Ctx, {Source, Target}}) ->
     % For _bulk_get the expectation is that the replication job will fallback
     % to a plain GET so the shape of the test is a bit different than the other
     % tests here.
     meck:new(couch_replicator_api_wrap, [passthrough]),
     populate_db(Source, 1, 5),
     {ok, _} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     % Tolerate a 500 error
     mock_fail_req(post, "/_bulk_get", {ok, "501", [], [<<"not_implemented">>]}),
     meck:reset(couch_replicator_api_wrap),
     populate_db(Source, 6, 6),
     wait_target_in_sync(Source, Target),
     % Check that there was a fallback to a plain GET
     ?assertEqual(1, meck:num_calls(couch_replicator_api_wrap, open_doc_revs, 6)),

     % Tolerate a 400 error
     mock_fail_req(post, "/_bulk_get", {ok, "418", [], [<<"{\"x\":\"y\"}">>]}),
     meck:reset(couch_replicator_api_wrap),
     populate_db(Source, 7, 7),
     wait_target_in_sync(Source, Target),
     % Check that there was a falback to a plain GET
     ?assertEqual(1, meck:num_calls(couch_replicator_api_wrap, open_doc_revs, 6)).

 t_fail_open_docs_get({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     Opts = #{
         % We're testing the case of individual doc rev GETs
         use_bulk_get => false,
         % Perform at least one retry before giving up (for extra coverage)
         retries_per_request => 2
     },
     {ok, RepId} = replicate(Source, Target, Opts),
     wait_target_in_sync(Source, Target),

     {ok, Listener} = rep_result_listener(RepId),
     % Break open_doc_revs on the server side and see what happens
     meck:new(fabric_doc_open_revs, [passthrough]),
     meck:expect(fabric_doc_open_revs, go, fun
         (Src, <<"6">>, _, _) when Src =:= Source ->
             % This is a random error, no particular reason for a 404
             meck:exception(throw, not_found);
         (ArgDb, ArgDocId, ArgRevs, ArgOpts) ->
             meck:passthrough([ArgDb, ArgDocId, ArgRevs, ArgOpts])
     end),
     populate_db(Source, 6, 6),
     {error, Result} = wait_rep_result(RepId),
     ?assertMatch({worker_died, _, {process_died, _, open_doc_revs_failed}}, Result),
     ?assert(meck:num_calls(fabric_doc_open_revs, go, 4) >= 2),
     couch_replicator_notifier:stop(Listener).

 t_fail_changes_queue({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     RepPid = couch_replicator_test_helper:get_pid(RepId),
     State = sys:get_state(RepPid),
     ChangesQueue = element(20, State),
     ?assert(is_process_alive(ChangesQueue)),

     {ok, Listener} = rep_result_listener(RepId),
     exit(ChangesQueue, boom),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({changes_queue_died, boom}, Result),
     couch_replicator_notifier:stop(Listener).

 t_fail_changes_manager({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     RepPid = couch_replicator_test_helper:get_pid(RepId),
     State = sys:get_state(RepPid),
     ChangesManager = element(21, State),
     ?assert(is_process_alive(ChangesManager)),

     {ok, Listener} = rep_result_listener(RepId),
     exit(ChangesManager, bam),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({changes_manager_died, bam}, Result),
     couch_replicator_notifier:stop(Listener).

 t_fail_changes_reader_proc({_Ctx, {Source, Target}}) ->
     populate_db(Source, 1, 5),
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),

     RepPid = couch_replicator_test_helper:get_pid(RepId),
     State = sys:get_state(RepPid),
     ChangesReader = element(22, State),
     ?assert(is_process_alive(ChangesReader)),

     {ok, Listener} = rep_result_listener(RepId),
     exit(ChangesReader, kapow),

     {error, Result} = wait_rep_result(RepId),
     ?assertEqual({changes_reader_died, kapow}, Result),
     couch_replicator_notifier:stop(Listener).

 t_dont_start_duplicate_job({_Ctx, {Source, Target}}) ->
     meck:new(couch_replicator_pg, [passthrough]),
     Pid = pid_from_another_node(),
     meck:expect(couch_replicator_pg, should_start, fun(_, _) -> {no, Pid} end),
     Rep = make_rep(Source, Target, #{continuous => true}),
     ExpectErr = {error, {already_started, Pid}},
     ?assertEqual(ExpectErr, couch_replicator_scheduler_job:start_link(Rep)).

 t_can_start_multiple_jobs({_Ctx, {Source, Target1}}) ->
     Target2 = couch_replicator_test_helper:setup_db(),
     populate_db(Source, 1, 5),

     {ok, RepId1} = replicate(Source, Target1),
     {ok, RepId2} = replicate(Source, Target2),
     RepPid1 = couch_replicator_test_helper:get_pid(RepId1),
     RepPid2 = couch_replicator_test_helper:get_pid(RepId2),
     ?assert(is_pid(RepPid1)),
     ?assert(is_pid(RepPid2)),

     ?assert(is_process_alive(RepPid1)),
     ?assert(is_process_alive(RepPid2)),

     wait_target_in_sync(Source, Target1),
     wait_target_in_sync(Source, Target2),

     ?assert(is_process_alive(RepPid1)),
     ?assert(is_process_alive(RepPid2)),

     exit(RepPid1, kill),
     exit(RepPid2, kill),
     couch_replicator_test_helper:teardown_db(Target2).

 t_stop_duplicate_job({_Ctx, {Source, Target}}) ->
     {ok, RepId} = replicate(Source, Target),
     wait_target_in_sync(Source, Target),
     RepPid = couch_replicator_test_helper:get_pid(RepId),
     {ok, Listener} = rep_result_listener(RepId),
     Pid = pid_from_another_node(),
     meck:expect(couch_replicator_pg, should_run, fun(_, _) -> {no, Pid} end),
     RepPid ! {'$gen_cast', checkpoint},
     {error, Result} = wait_rep_result(RepId),
     ?assertEqual(duplicate_job, Result),
     couch_replicator_notifier:stop(Listener).

 pid_from_another_node() ->
     % Use a Pid serialized from a node named A@1
     % (A@1)1> term_to_binary(self()).
     Bin = <<131, 88, 100, 0, 3, 65, 64, 49, 0, 0, 0, 89, 0, 0, 0, 0, 99, 137, 147, 218>>,
     Pid = binary_to_term(Bin),
     ?assertEqual('A@1', node(Pid)),
     Pid.

 mock_fail_req(Method, Path, Return) ->
     meck:expect(
         ibrowse,
         send_req_direct,
         fun(W, Url, Headers, Meth, Body, Opts, TOut) ->
             Args = [W, Url, Headers, Meth, Body, Opts, TOut],
             #{path := UPath} = uri_string:parse(Url),
             case {lists:suffix(Path, UPath), Method == Meth} of
                 {true, true} ->
                     _ = meck:passthrough(Args),
                     Return;
                 {_, _} ->
                     meck:passthrough(Args)
             end
         end
     ).

 rep_result_listener(RepId) ->
     ReplyTo = self(),
     {ok, _Listener} = couch_replicator_notifier:start_link(
         fun
             ({_, RepId2, _} = Ev) when RepId2 =:= RepId ->
                 ReplyTo ! Ev;
             (_) ->
                 ok
         end
     ).

 wait_rep_result(RepId) ->
     receive
         {finished, RepId, RepResult} -> {ok, RepResult};
         {error, RepId, Reason} -> {error, Reason}
     end.

 populate_db(DbName, Start, End) ->
     populate_db(DbName, Start, End, false).

 populate_db(DbName, Start, End, WithAttachments) ->
     Docs = lists:foldl(
         fun(DocIdCounter, Acc) ->
             Id = integer_to_binary(DocIdCounter),
             Atts =
                 case WithAttachments of
                     true -> [att(<<"att1">>, 1024, <<"app/binary">>)];
                     false -> []
                 end,
             Doc = #doc{id = Id, body = {[]}, atts = Atts},
             [Doc | Acc]
         end,
         [],
         lists:seq(Start, End)
     ),
     {ok, [_ | _]} = fabric:update_docs(DbName, Docs, [?ADMIN_CTX]).

 att(Name, Size, Type) ->
     couch_att:new([
         {name, Name},
         {type, Type},
         {att_len, Size},
         {data, fun(Count) -> crypto:strong_rand_bytes(Count) end}
     ]).

 wait_target_in_sync(Source, Target) ->
     {ok, SourceDocCount} = fabric:get_doc_count(Source),
     wait_target_in_sync_loop(SourceDocCount, Target, 300).

 wait_target_in_sync_loop(_DocCount, _TargetName, 0) ->
     erlang:error(
         {assertion_failed, [
             {module, ?MODULE},
             {line, ?LINE},
             {reason, "Could not get source and target databases in sync"}
         ]}
     );
 wait_target_in_sync_loop(DocCount, TargetName, RetriesLeft) ->
     {ok, TargetDocCount} = fabric:get_doc_count(TargetName),
     case TargetDocCount == DocCount of
         true ->
             true;
         false ->
             ok = timer:sleep(500),
             wait_target_in_sync_loop(DocCount, TargetName, RetriesLeft - 1)
     end.

 replicate(Source, Target) ->
     replicate(Source, Target, #{}).

 replicate(Source, Target, #{} = Opts) ->
     Rep = make_rep(Source, Target, Opts),
     ok = couch_replicator_scheduler:add_job(Rep),
     couch_replicator_scheduler:reschedule(),
     {ok, Rep#rep.id}.

 make_rep(Source, Target, #{} = OverrideOpts) ->
     Opts0 = #{
         source => url(Source),
         target => url(Target),
         continuous => true,
         worker_processes => 1,
         retries_per_request => 1,
         % Low connection timeout so _changes feed gets restarted quicker
         connection_timeout => 3000
     },
     RepMap = maps:merge(Opts0, OverrideOpts),
     % parse_rep_doc accepts {[...]} ejson format
     RepEJson = couch_util:json_decode(couch_util:json_encode(RepMap)),
     {ok, Rep} = couch_replicator_parse:parse_rep_doc(RepEJson, ?ADMIN_USER),
     Rep.

 url(DbName) ->
     couch_replicator_test_helper:cluster_db_url(DbName).
	% 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_replicator_error_reporting_tests).

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

	error_reporting_test_() ->
	{
	foreach,
	fun couch_replicator_test_helper:test_setup/0,
	fun couch_replicator_test_helper:test_teardown/1,
	[
	?TDEF_FE(t_fail_bulk_docs),
	?TDEF_FE(t_fail_changes_reader),
	?TDEF_FE(t_fail_doc_put_4xx_well_formed_json_error),
	?TDEF_FE(t_fail_doc_put_4xx_unexpected_json_error),
	?TDEF_FE(t_fail_doc_put_4xx_invalid_json_error),
	?TDEF_FE(t_skip_doc_put_401_errors),
	?TDEF_FE(t_skip_doc_put_403_errors),
	?TDEF_FE(t_skip_doc_put_413_errors),
	?TDEF_FE(t_skip_doc_put_415_errors),
	?TDEF_FE(t_skip_doc_put_invalid_attachment_name),
	?TDEF_FE(t_fail_revs_diff),
	?TDEF_FE(t_fail_bulk_get, 15),
	?TDEF_FE(t_fail_open_docs_get, 15),
	?TDEF_FE(t_fail_changes_queue),
	?TDEF_FE(t_fail_changes_manager),
	?TDEF_FE(t_fail_changes_reader_proc),
	?TDEF_FE(t_dont_start_duplicate_job),
	?TDEF_FE(t_can_start_multiple_jobs),
	?TDEF_FE(t_stop_duplicate_job)
	]
	}.

	t_fail_bulk_docs({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	mock_fail_req(post, "/_bulk_docs", {ok, "403", [], [<<"{\"x\":\"y\"}">>]}),
	populate_db(Source, 6, 6),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({bulk_docs_failed, 403, {[{<<"x">>, <<"y">>}]}}, Result),

	couch_replicator_notifier:stop(Listener).

	t_fail_changes_reader({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	mock_fail_req(get, "/_changes", {ok, "418", [], [<<"{\"x\":\"y\"}">>]}),
	populate_db(Source, 6, 6),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({changes_req_failed, 418, {[{<<"x">>, <<"y">>}]}}, Result),

	couch_replicator_notifier:stop(Listener).

	t_fail_doc_put_4xx_well_formed_json_error({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	ErrBody = [<<"{\"error\":\"x\", \"reason\":\"y\"}">>],
	mock_fail_req(put, "/6", {ok, "400", [], ErrBody}),
	populate_db(Source, 6, 6, _WithAttachments = true),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({doc_write_failed, {<<"x">>, <<"y">>}}, Result),

	couch_replicator_notifier:stop(Listener).

	t_fail_doc_put_4xx_unexpected_json_error({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	ErrBody = [<<"{\"a\":\"b\"}">>],
	mock_fail_req(put, "/6", {ok, "400", [], ErrBody}),
	populate_db(Source, 6, 6, _WithAttachments = true),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({doc_write_failed, {400, [{<<"a">>, <<"b">>}]}}, Result),

	couch_replicator_notifier:stop(Listener).

	t_fail_doc_put_4xx_invalid_json_error({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	mock_fail_req(put, "/6", {ok, "400", [], [<<"potato">>]}),
	populate_db(Source, 6, 6, _WithAttachments = true),

	{error, Result} = wait_rep_result(RepId),
	?assertMatch({doc_write_failed, {invalid_json, _}}, Result),

	couch_replicator_notifier:stop(Listener).

	t_skip_doc_put_401_errors({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	populate_db(Source, 6, 6, _WithAttachments = true),
	ErrBody = [<<"{\"error\":\"unauthorized\", \"reason\":\"vdu\"}">>],
	mock_fail_req(put, "/6", {ok, "401", [], ErrBody}),
	{ok, RepId} = replicate(Source, Target, #{continuous => false}),
	{ok, Listener} = rep_result_listener(RepId),
	Res = wait_rep_result(RepId),
	% Replication job should succeed
	?assertMatch({ok, {[_ \| _]}}, Res),
	{ok, {Props}} = Res,
	History = proplists:get_value(<<"history">>, Props),
	?assertMatch([{[_ \| _]}], History),
	[{HistProps}] = History,
	DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
	DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
	?assertEqual(5, DocsWritten),
	?assertEqual(1, DocWriteFailures),
	couch_replicator_notifier:stop(Listener).

	t_skip_doc_put_403_errors({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	populate_db(Source, 6, 6, _WithAttachments = true),
	ErrBody = [<<"{\"error\":\"forbidden\", \"reason\":\"vdu\"}">>],
	mock_fail_req(put, "/6", {ok, "403", [], ErrBody}),
	{ok, RepId} = replicate(Source, Target, #{continuous => false}),
	{ok, Listener} = rep_result_listener(RepId),
	Res = wait_rep_result(RepId),
	% Replication job should succeed
	?assertMatch({ok, {[_ \| _]}}, Res),
	{ok, {Props}} = Res,
	History = proplists:get_value(<<"history">>, Props),
	?assertMatch([{[_ \| _]}], History),
	[{HistProps}] = History,
	DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
	DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
	?assertEqual(5, DocsWritten),
	?assertEqual(1, DocWriteFailures),
	couch_replicator_notifier:stop(Listener).

	t_skip_doc_put_413_errors({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	populate_db(Source, 6, 6, _WithAttachments = true),
	ErrBody = [<<"{\"error\":\"too_large\", \"reason\":\"too_large\"}">>],
	mock_fail_req(put, "/6", {ok, "413", [], ErrBody}),
	{ok, RepId} = replicate(Source, Target, #{continuous => false}),
	{ok, Listener} = rep_result_listener(RepId),
	Res = wait_rep_result(RepId),
	% Replication job should succeed
	?assertMatch({ok, {[_ \| _]}}, Res),
	{ok, {Props}} = Res,
	History = proplists:get_value(<<"history">>, Props),
	?assertMatch([{[_ \| _]}], History),
	[{HistProps}] = History,
	DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
	DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
	?assertEqual(5, DocsWritten),
	?assertEqual(1, DocWriteFailures),
	couch_replicator_notifier:stop(Listener).

	t_skip_doc_put_415_errors({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	populate_db(Source, 6, 6, _WithAttachments = true),
	ErrBody = [<<"{\"error\":\"unsupported_media_type\", \"reason\":\"bad_media\"}">>],
	mock_fail_req(put, "/6", {ok, "415", [], ErrBody}),
	{ok, RepId} = replicate(Source, Target, #{continuous => false}),
	{ok, Listener} = rep_result_listener(RepId),
	Res = wait_rep_result(RepId),
	% Replication job should succeed
	?assertMatch({ok, {[_ \| _]}}, Res),
	{ok, {Props}} = Res,
	History = proplists:get_value(<<"history">>, Props),
	?assertMatch([{[_ \| _]}], History),
	[{HistProps}] = History,
	DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
	DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
	?assertEqual(5, DocsWritten),
	?assertEqual(1, DocWriteFailures),
	couch_replicator_notifier:stop(Listener).

	t_skip_doc_put_invalid_attachment_name({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	populate_db(Source, 6, 6, _WithAttachments = true),
	ErrBody = [
	<<"{\"error\":\"bad_request\", \"reason\":\"Attachment name '_foo' starts with prohibited character '_'\"}">>
	],
	mock_fail_req(put, "/6", {ok, "400", [], ErrBody}),
	{ok, RepId} = replicate(Source, Target, #{continuous => false}),
	{ok, Listener} = rep_result_listener(RepId),
	Res = wait_rep_result(RepId),
	% Replication job should succeed
	?assertMatch({ok, {[_ \| _]}}, Res),
	{ok, {Props}} = Res,
	History = proplists:get_value(<<"history">>, Props),
	?assertMatch([{[_ \| _]}], History),
	[{HistProps}] = History,
	DocsWritten = proplists:get_value(<<"docs_written">>, HistProps),
	DocWriteFailures = proplists:get_value(<<"doc_write_failures">>, HistProps),
	?assertEqual(5, DocsWritten),
	?assertEqual(1, DocWriteFailures),
	couch_replicator_notifier:stop(Listener).

	t_fail_revs_diff({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	mock_fail_req(post, "/_revs_diff", {ok, "407", [], [<<"{\"x\":\"y\"}">>]}),
	populate_db(Source, 6, 6),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({revs_diff_failed, 407, {[{<<"x">>, <<"y">>}]}}, Result),

	couch_replicator_notifier:stop(Listener).

	t_fail_bulk_get({_Ctx, {Source, Target}}) ->
	% For _bulk_get the expectation is that the replication job will fallback
	% to a plain GET so the shape of the test is a bit different than the other
	% tests here.
	meck:new(couch_replicator_api_wrap, [passthrough]),
	populate_db(Source, 1, 5),
	{ok, _} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	% Tolerate a 500 error
	mock_fail_req(post, "/_bulk_get", {ok, "501", [], [<<"not_implemented">>]}),
	meck:reset(couch_replicator_api_wrap),
	populate_db(Source, 6, 6),
	wait_target_in_sync(Source, Target),
	% Check that there was a fallback to a plain GET
	?assertEqual(1, meck:num_calls(couch_replicator_api_wrap, open_doc_revs, 6)),

	% Tolerate a 400 error
	mock_fail_req(post, "/_bulk_get", {ok, "418", [], [<<"{\"x\":\"y\"}">>]}),
	meck:reset(couch_replicator_api_wrap),
	populate_db(Source, 7, 7),
	wait_target_in_sync(Source, Target),
	% Check that there was a falback to a plain GET
	?assertEqual(1, meck:num_calls(couch_replicator_api_wrap, open_doc_revs, 6)).

	t_fail_open_docs_get({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	Opts = #{
	% We're testing the case of individual doc rev GETs
	use_bulk_get => false,
	% Perform at least one retry before giving up (for extra coverage)
	retries_per_request => 2
	},
	{ok, RepId} = replicate(Source, Target, Opts),
	wait_target_in_sync(Source, Target),

	{ok, Listener} = rep_result_listener(RepId),
	% Break open_doc_revs on the server side and see what happens
	meck:new(fabric_doc_open_revs, [passthrough]),
	meck:expect(fabric_doc_open_revs, go, fun
	(Src, <<"6">>, _, _) when Src =:= Source ->
	% This is a random error, no particular reason for a 404
	meck:exception(throw, not_found);
	(ArgDb, ArgDocId, ArgRevs, ArgOpts) ->
	meck:passthrough([ArgDb, ArgDocId, ArgRevs, ArgOpts])
	end),
	populate_db(Source, 6, 6),
	{error, Result} = wait_rep_result(RepId),
	?assertMatch({worker_died, _, {process_died, _, open_doc_revs_failed}}, Result),
	?assert(meck:num_calls(fabric_doc_open_revs, go, 4) >= 2),
	couch_replicator_notifier:stop(Listener).

	t_fail_changes_queue({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	RepPid = couch_replicator_test_helper:get_pid(RepId),
	State = sys:get_state(RepPid),
	ChangesQueue = element(20, State),
	?assert(is_process_alive(ChangesQueue)),

	{ok, Listener} = rep_result_listener(RepId),
	exit(ChangesQueue, boom),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({changes_queue_died, boom}, Result),
	couch_replicator_notifier:stop(Listener).

	t_fail_changes_manager({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	RepPid = couch_replicator_test_helper:get_pid(RepId),
	State = sys:get_state(RepPid),
	ChangesManager = element(21, State),
	?assert(is_process_alive(ChangesManager)),

	{ok, Listener} = rep_result_listener(RepId),
	exit(ChangesManager, bam),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({changes_manager_died, bam}, Result),
	couch_replicator_notifier:stop(Listener).

	t_fail_changes_reader_proc({_Ctx, {Source, Target}}) ->
	populate_db(Source, 1, 5),
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),

	RepPid = couch_replicator_test_helper:get_pid(RepId),
	State = sys:get_state(RepPid),
	ChangesReader = element(22, State),
	?assert(is_process_alive(ChangesReader)),

	{ok, Listener} = rep_result_listener(RepId),
	exit(ChangesReader, kapow),

	{error, Result} = wait_rep_result(RepId),
	?assertEqual({changes_reader_died, kapow}, Result),
	couch_replicator_notifier:stop(Listener).

	t_dont_start_duplicate_job({_Ctx, {Source, Target}}) ->
	meck:new(couch_replicator_pg, [passthrough]),
	Pid = pid_from_another_node(),
	meck:expect(couch_replicator_pg, should_start, fun(_, _) -> {no, Pid} end),
	Rep = make_rep(Source, Target, #{continuous => true}),
	ExpectErr = {error, {already_started, Pid}},
	?assertEqual(ExpectErr, couch_replicator_scheduler_job:start_link(Rep)).

	t_can_start_multiple_jobs({_Ctx, {Source, Target1}}) ->
	Target2 = couch_replicator_test_helper:setup_db(),
	populate_db(Source, 1, 5),

	{ok, RepId1} = replicate(Source, Target1),
	{ok, RepId2} = replicate(Source, Target2),
	RepPid1 = couch_replicator_test_helper:get_pid(RepId1),
	RepPid2 = couch_replicator_test_helper:get_pid(RepId2),
	?assert(is_pid(RepPid1)),
	?assert(is_pid(RepPid2)),

	?assert(is_process_alive(RepPid1)),
	?assert(is_process_alive(RepPid2)),

	wait_target_in_sync(Source, Target1),
	wait_target_in_sync(Source, Target2),

	?assert(is_process_alive(RepPid1)),
	?assert(is_process_alive(RepPid2)),

	exit(RepPid1, kill),
	exit(RepPid2, kill),
	couch_replicator_test_helper:teardown_db(Target2).

	t_stop_duplicate_job({_Ctx, {Source, Target}}) ->
	{ok, RepId} = replicate(Source, Target),
	wait_target_in_sync(Source, Target),
	RepPid = couch_replicator_test_helper:get_pid(RepId),
	{ok, Listener} = rep_result_listener(RepId),
	Pid = pid_from_another_node(),
	meck:expect(couch_replicator_pg, should_run, fun(_, _) -> {no, Pid} end),
	RepPid ! {'$gen_cast', checkpoint},
	{error, Result} = wait_rep_result(RepId),
	?assertEqual(duplicate_job, Result),
	couch_replicator_notifier:stop(Listener).

	pid_from_another_node() ->
	% Use a Pid serialized from a node named A@1
	% (A@1)1> term_to_binary(self()).
	Bin = <<131, 88, 100, 0, 3, 65, 64, 49, 0, 0, 0, 89, 0, 0, 0, 0, 99, 137, 147, 218>>,
	Pid = binary_to_term(Bin),
	?assertEqual('A@1', node(Pid)),
	Pid.

	mock_fail_req(Method, Path, Return) ->
	meck:expect(
	ibrowse,
	send_req_direct,
	fun(W, Url, Headers, Meth, Body, Opts, TOut) ->
	Args = [W, Url, Headers, Meth, Body, Opts, TOut],
	#{path := UPath} = uri_string:parse(Url),
	case {lists:suffix(Path, UPath), Method == Meth} of
	{true, true} ->
	_ = meck:passthrough(Args),
	Return;
	{_, _} ->
	meck:passthrough(Args)
	end
	end
	).

	rep_result_listener(RepId) ->
	ReplyTo = self(),
	{ok, _Listener} = couch_replicator_notifier:start_link(
	fun
	({_, RepId2, _} = Ev) when RepId2 =:= RepId ->
	ReplyTo ! Ev;
	(_) ->
	ok
	end
	).

	wait_rep_result(RepId) ->
	receive
	{finished, RepId, RepResult} -> {ok, RepResult};
	{error, RepId, Reason} -> {error, Reason}
	end.

	populate_db(DbName, Start, End) ->
	populate_db(DbName, Start, End, false).

	populate_db(DbName, Start, End, WithAttachments) ->
	Docs = lists:foldl(
	fun(DocIdCounter, Acc) ->
	Id = integer_to_binary(DocIdCounter),
	Atts =
	case WithAttachments of
	true -> [att(<<"att1">>, 1024, <<"app/binary">>)];
	false -> []
	end,
	Doc = #doc{id = Id, body = {[]}, atts = Atts},
	[Doc \| Acc]
	end,
	[],
	lists:seq(Start, End)
	),
	{ok, [_ \| _]} = fabric:update_docs(DbName, Docs, [?ADMIN_CTX]).

	att(Name, Size, Type) ->
	couch_att:new([
	{name, Name},
	{type, Type},
	{att_len, Size},
	{data, fun(Count) -> crypto:strong_rand_bytes(Count) end}
	]).

	wait_target_in_sync(Source, Target) ->
	{ok, SourceDocCount} = fabric:get_doc_count(Source),
	wait_target_in_sync_loop(SourceDocCount, Target, 300).

	wait_target_in_sync_loop(_DocCount, _TargetName, 0) ->
	erlang:error(
	{assertion_failed, [
	{module, ?MODULE},
	{line, ?LINE},
	{reason, "Could not get source and target databases in sync"}
	]}
	);
	wait_target_in_sync_loop(DocCount, TargetName, RetriesLeft) ->
	{ok, TargetDocCount} = fabric:get_doc_count(TargetName),
	case TargetDocCount == DocCount of
	true ->
	true;
	false ->
	ok = timer:sleep(500),
	wait_target_in_sync_loop(DocCount, TargetName, RetriesLeft - 1)
	end.

	replicate(Source, Target) ->
	replicate(Source, Target, #{}).

	replicate(Source, Target, #{} = Opts) ->
	Rep = make_rep(Source, Target, Opts),
	ok = couch_replicator_scheduler:add_job(Rep),
	couch_replicator_scheduler:reschedule(),
	{ok, Rep#rep.id}.

	make_rep(Source, Target, #{} = OverrideOpts) ->
	Opts0 = #{
	source => url(Source),
	target => url(Target),
	continuous => true,
	worker_processes => 1,
	retries_per_request => 1,
	% Low connection timeout so _changes feed gets restarted quicker
	connection_timeout => 3000
	},
	RepMap = maps:merge(Opts0, OverrideOpts),
	% parse_rep_doc accepts {[...]} ejson format
	RepEJson = couch_util:json_decode(couch_util:json_encode(RepMap)),
	{ok, Rep} = couch_replicator_parse:parse_rep_doc(RepEJson, ?ADMIN_USER),
	Rep.

	url(DbName) ->
	couch_replicator_test_helper:cluster_db_url(DbName).