You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2020/09/15 20:14:13 UTC
[couchdb] 16/16: Update and clean up tests
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch prototype/fdb-layer
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit ae858196848cf9533dfa03a2006227481f47388d
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Fri Aug 28 04:36:18 2020 -0400
Update and clean up tests
Update tests to use the new replicator. Also clean up redundancy and re-use
some of the newer macros from fabric2 (?TDEF_FE).
Make sure replicator tests are included in `make check`
---
Makefile | 2 +-
.../couch_replicator_attachments_too_large.erl | 90 ++---
.../eunit/couch_replicator_connection_tests.erl | 274 +++++++------
...replicator_create_target_with_options_tests.erl | 129 +++---
.../test/eunit/couch_replicator_db_tests.erl | 332 ++++++++++++++++
.../test/eunit/couch_replicator_filtered_tests.erl | 348 ++++++++--------
.../eunit/couch_replicator_httpc_pool_tests.erl | 125 +++---
.../eunit/couch_replicator_id_too_long_tests.erl | 91 ++---
.../eunit/couch_replicator_job_server_tests.erl | 437 +++++++++++++++++++++
.../eunit/couch_replicator_large_atts_tests.erl | 123 +++---
.../eunit/couch_replicator_many_leaves_tests.erl | 241 +++++-------
.../eunit/couch_replicator_missing_stubs_tests.erl | 179 ++++-----
.../test/eunit/couch_replicator_proxy_tests.erl | 135 +++----
.../eunit/couch_replicator_rate_limiter_tests.erl | 77 ++--
...ch_replicator_retain_stats_between_job_runs.erl | 223 +++--------
.../test/eunit/couch_replicator_selector_tests.erl | 136 +++----
...ch_replicator_small_max_request_size_target.erl | 190 +++------
.../test/eunit/couch_replicator_test_helper.erl | 323 +++++++++++----
.../couch_replicator_transient_jobs_tests.erl | 106 +++++
.../couch_replicator_use_checkpoints_tests.erl | 207 +++-------
test/elixir/test/replication_test.exs | 5 +-
21 files changed, 2210 insertions(+), 1563 deletions(-)
diff --git a/Makefile b/Makefile
index e8d36629..35b62f9 100644
--- a/Makefile
+++ b/Makefile
@@ -163,7 +163,7 @@ endif
.PHONY: check
check: all
@$(MAKE) emilio
- make eunit apps=couch_eval,couch_expiring_cache,ctrace,couch_jobs,couch_views,fabric,mango,chttpd
+ make eunit apps=couch_eval,couch_expiring_cache,ctrace,couch_jobs,couch_views,fabric,mango,chttpd,couch_replicator
make elixir tests=test/elixir/test/basics_test.exs,test/elixir/test/replication_test.exs,test/elixir/test/map_test.exs,test/elixir/test/all_docs_test.exs,test/elixir/test/bulk_docs_test.exs
make exunit apps=chttpd
make mango-test
diff --git a/src/couch_replicator/test/eunit/couch_replicator_attachments_too_large.erl b/src/couch_replicator/test/eunit/couch_replicator_attachments_too_large.erl
index ac4bb84..0e7e0ea 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_attachments_too_large.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_attachments_too_large.erl
@@ -12,72 +12,60 @@
-module(couch_replicator_attachments_too_large).
+
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
-
-
-setup(_) ->
- Ctx = test_util:start_couch([couch_replicator]),
- Source = create_db(),
- create_doc_with_attachment(Source, <<"doc">>, 1000),
- Target = create_db(),
- {Ctx, {Source, Target}}.
-
-
-teardown(_, {Ctx, {Source, Target}}) ->
- delete_db(Source),
- delete_db(Target),
- config:delete("couchdb", "max_attachment_size"),
- ok = test_util:stop_couch(Ctx).
+-include_lib("fabric/test/fabric2_test.hrl").
attachment_too_large_replication_test_() ->
- Pairs = [{remote, remote}],
{
- "Attachment size too large replication tests",
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_succeed/2} || Pair <- Pairs] ++
- [{Pair, fun should_fail/2} || Pair <- Pairs]
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_should_succeed),
+ ?TDEF_FE(t_should_fail)
+ ]
}
}.
-should_succeed({From, To}, {_Ctx, {Source, Target}}) ->
- RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)}
- ]},
- config:set("couchdb", "max_attachment_size", "1000", _Persist = false),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- ?_assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ create_doc_with_attachment(Source, <<"doc">>, 1000),
+ Target = couch_replicator_test_helper:create_db(),
+ {Source, Target}.
-should_fail({From, To}, {_Ctx, {Source, Target}}) ->
- RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)}
- ]},
- config:set("couchdb", "max_attachment_size", "999", _Persist = false),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- ?_assertError({badmatch, {not_found, missing}},
- couch_replicator_test_helper:compare_dbs(Source, Target)).
+teardown({Source, Target}) ->
+ config:delete("couchdb", "max_attachment_size", false),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
-create_db() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
+t_should_succeed({Source, Target}) ->
+ config:set("couchdb", "max_attachment_size", "1000", false),
+ {ok, _} = couch_replicator_test_helper:replicate(Source, Target),
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
+
+
+t_should_fail({Source, Target}) ->
+ config:set("couchdb", "max_attachment_size", "999", false),
+ {ok, _} = couch_replicator_test_helper:replicate(Source, Target),
+ ExceptIds = [<<"doc">>],
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source,
+ Target, ExceptIds)).
create_doc_with_attachment(DbName, DocId, AttSize) ->
- {ok, Db} = couch_db:open(DbName, [?ADMIN_CTX]),
Doc = #doc{id = DocId, atts = att(AttSize)},
- {ok, _} = couch_db:update_doc(Db, Doc, []),
- couch_db:close(Db),
+ couch_replicator_test_helper:create_docs(DbName, [Doc]),
ok.
@@ -90,13 +78,3 @@ att(Size) when is_integer(Size), Size >= 1 ->
<< <<"x">> || _ <- lists:seq(1, Size) >>
end}
])].
-
-
-delete_db(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]).
-
-
-db_url(remote, DbName) ->
- Addr = config:get("httpd", "bind_address", "127.0.0.1"),
- Port = mochiweb_socket_server:get(couch_httpd, port),
- ?l2b(io_lib:format("http://~s:~b/~s", [Addr, Port, DbName])).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_connection_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_connection_tests.erl
index e75cc5a..df30db2 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_connection_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_connection_tests.erl
@@ -12,187 +12,176 @@
-module(couch_replicator_connection_tests).
+
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-
--define(TIMEOUT, 1000).
-
-
-setup() ->
- Host = config:get("httpd", "bind_address", "127.0.0.1"),
- Port = config:get("httpd", "port", "5984"),
- {Host, Port}.
-
-teardown(_) ->
- ok.
+-include_lib("fabric/test/fabric2_test.hrl").
httpc_pool_test_() ->
{
- "replicator connection sharing tests",
+ "Replicator connection sharing tests",
{
setup,
- fun() -> test_util:start_couch([couch_replicator]) end, fun test_util:stop_couch/1,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
{
foreach,
- fun setup/0, fun teardown/1,
+ fun setup/0,
+ fun teardown/1,
[
- fun connections_shared_after_release/1,
- fun connections_not_shared_after_owner_death/1,
- fun idle_connections_closed/1,
- fun test_owner_monitors/1,
- fun worker_discards_creds_on_create/1,
- fun worker_discards_url_creds_after_request/1,
- fun worker_discards_creds_in_headers_after_request/1,
- fun worker_discards_proxy_creds_after_request/1
+ ?TDEF_FE(connections_shared_after_release),
+ ?TDEF_FE(connections_not_shared_after_owner_death),
+ ?TDEF_FE(idle_connections_closed),
+ ?TDEF_FE(test_owner_monitors),
+ ?TDEF_FE(worker_discards_creds_on_create),
+ ?TDEF_FE(worker_discards_url_creds_after_request),
+ ?TDEF_FE(worker_discards_creds_in_headers_after_request),
+ ?TDEF_FE(worker_discards_proxy_creds_after_request)
]
}
}
}.
+setup() ->
+ Host = config:get("chttpd", "bind_address", "127.0.0.1"),
+ Port = config:get("chttpd", "port", "5984"),
+ {Host, Port}.
+
+
+teardown(_) ->
+ ok.
+
+
connections_shared_after_release({Host, Port}) ->
- ?_test(begin
- URL = "http://" ++ Host ++ ":" ++ Port,
- Self = self(),
- {ok, Pid} = couch_replicator_connection:acquire(URL),
- couch_replicator_connection:release(Pid),
- spawn(fun() ->
- Self ! couch_replicator_connection:acquire(URL)
- end),
- receive
- {ok, Pid2} ->
- ?assertEqual(Pid, Pid2)
- end
- end).
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ Self = self(),
+ {ok, Pid} = couch_replicator_connection:acquire(URL),
+ couch_replicator_connection:release(Pid),
+ spawn(fun() ->
+ Self ! couch_replicator_connection:acquire(URL)
+ end),
+ receive
+ {ok, Pid2} ->
+ ?assertEqual(Pid, Pid2)
+ end.
connections_not_shared_after_owner_death({Host, Port}) ->
- ?_test(begin
- URL = "http://" ++ Host ++ ":" ++ Port,
- Self = self(),
- spawn(fun() ->
- Self ! couch_replicator_connection:acquire(URL),
- error("simulate division by zero without compiler warning")
- end),
- receive
- {ok, Pid} ->
- {ok, Pid2} = couch_replicator_connection:acquire(URL),
- ?assertNotEqual(Pid, Pid2),
- MRef = monitor(process, Pid),
- receive {'DOWN', MRef, process, Pid, _Reason} ->
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ Self = self(),
+ spawn(fun() ->
+ Self ! couch_replicator_connection:acquire(URL),
+ error("simulate division by zero without compiler warning")
+ end),
+ receive
+ {ok, Pid} ->
+ {ok, Pid2} = couch_replicator_connection:acquire(URL),
+ ?assertNotEqual(Pid, Pid2),
+ MRef = monitor(process, Pid),
+ receive
+ {'DOWN', MRef, process, Pid, _Reason} ->
?assert(not is_process_alive(Pid));
- Other -> throw(Other)
- end
- end
- end).
+ Other ->
+ throw(Other)
+ end
+ end.
idle_connections_closed({Host, Port}) ->
- ?_test(begin
- URL = "http://" ++ Host ++ ":" ++ Port,
- {ok, Pid} = couch_replicator_connection:acquire(URL),
- couch_replicator_connection ! close_idle_connections,
- ?assert(ets:member(couch_replicator_connection, Pid)),
- % block until idle connections have closed
- sys:get_status(couch_replicator_connection),
- couch_replicator_connection:release(Pid),
- couch_replicator_connection ! close_idle_connections,
- % block until idle connections have closed
- sys:get_status(couch_replicator_connection),
- ?assert(not ets:member(couch_replicator_connection, Pid))
- end).
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ {ok, Pid} = couch_replicator_connection:acquire(URL),
+ couch_replicator_connection ! close_idle_connections,
+ ?assert(ets:member(couch_replicator_connection, Pid)),
+ % block until idle connections have closed
+ sys:get_status(couch_replicator_connection),
+ couch_replicator_connection:release(Pid),
+ couch_replicator_connection ! close_idle_connections,
+ % block until idle connections have closed
+ sys:get_status(couch_replicator_connection),
+ ?assert(not ets:member(couch_replicator_connection, Pid)).
test_owner_monitors({Host, Port}) ->
- ?_test(begin
- URL = "http://" ++ Host ++ ":" ++ Port,
- {ok, Worker0} = couch_replicator_connection:acquire(URL),
- assert_monitors_equal([{process, self()}]),
- couch_replicator_connection:release(Worker0),
- assert_monitors_equal([]),
- {Workers, Monitors} = lists:foldl(fun(_, {WAcc, MAcc}) ->
- {ok, Worker1} = couch_replicator_connection:acquire(URL),
- MAcc1 = [{process, self()} | MAcc],
- assert_monitors_equal(MAcc1),
- {[Worker1 | WAcc], MAcc1}
- end, {[], []}, lists:seq(1,5)),
- lists:foldl(fun(Worker2, Acc) ->
- [_ | NewAcc] = Acc,
- couch_replicator_connection:release(Worker2),
- assert_monitors_equal(NewAcc),
- NewAcc
- end, Monitors, Workers)
- end).
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ {ok, Worker0} = couch_replicator_connection:acquire(URL),
+ assert_monitors_equal([{process, self()}]),
+ couch_replicator_connection:release(Worker0),
+ assert_monitors_equal([]),
+ {Workers, Monitors} = lists:foldl(fun(_, {WAcc, MAcc}) ->
+ {ok, Worker1} = couch_replicator_connection:acquire(URL),
+ MAcc1 = [{process, self()} | MAcc],
+ assert_monitors_equal(MAcc1),
+ {[Worker1 | WAcc], MAcc1}
+ end, {[], []}, lists:seq(1, 5)),
+ lists:foldl(fun(Worker2, Acc) ->
+ [_ | NewAcc] = Acc,
+ couch_replicator_connection:release(Worker2),
+ assert_monitors_equal(NewAcc),
+ NewAcc
+ end, Monitors, Workers).
worker_discards_creds_on_create({Host, Port}) ->
- ?_test(begin
- {User, Pass, B64Auth} = user_pass(),
- URL = "http://" ++ User ++ ":" ++ Pass ++ "@" ++ Host ++ ":" ++ Port,
- {ok, WPid} = couch_replicator_connection:acquire(URL),
- Internals = worker_internals(WPid),
- ?assert(string:str(Internals, B64Auth) =:= 0),
- ?assert(string:str(Internals, Pass) =:= 0)
- end).
+ {User, Pass, B64Auth} = user_pass(),
+ URL = "http://" ++ User ++ ":" ++ Pass ++ "@" ++ Host ++ ":" ++ Port,
+ {ok, WPid} = couch_replicator_connection:acquire(URL),
+ Internals = worker_internals(WPid),
+ ?assert(string:str(Internals, B64Auth) =:= 0),
+ ?assert(string:str(Internals, Pass) =:= 0).
worker_discards_url_creds_after_request({Host, _}) ->
- ?_test(begin
- {User, Pass, B64Auth} = user_pass(),
- {Port, ServerPid} = server(),
- PortStr = integer_to_list(Port),
- URL = "http://" ++ User ++ ":" ++ Pass ++ "@" ++ Host ++ ":" ++ PortStr,
- {ok, WPid} = couch_replicator_connection:acquire(URL),
- ?assertMatch({ok, "200", _, _}, send_req(WPid, URL, [], [])),
- Internals = worker_internals(WPid),
- ?assert(string:str(Internals, B64Auth) =:= 0),
- ?assert(string:str(Internals, Pass) =:= 0),
- couch_replicator_connection:release(WPid),
- unlink(ServerPid),
- exit(ServerPid, kill)
- end).
+ {User, Pass, B64Auth} = user_pass(),
+ {Port, ServerPid} = server(),
+ PortStr = integer_to_list(Port),
+ URL = "http://" ++ User ++ ":" ++ Pass ++ "@" ++ Host ++ ":" ++ PortStr,
+ {ok, WPid} = couch_replicator_connection:acquire(URL),
+ ?assertMatch({ok, "200", _, _}, send_req(WPid, URL, [], [])),
+ Internals = worker_internals(WPid),
+ ?assert(string:str(Internals, B64Auth) =:= 0),
+ ?assert(string:str(Internals, Pass) =:= 0),
+ couch_replicator_connection:release(WPid),
+ unlink(ServerPid),
+ exit(ServerPid, kill).
worker_discards_creds_in_headers_after_request({Host, _}) ->
- ?_test(begin
- {_User, Pass, B64Auth} = user_pass(),
- {Port, ServerPid} = server(),
- PortStr = integer_to_list(Port),
- URL = "http://" ++ Host ++ ":" ++ PortStr,
- {ok, WPid} = couch_replicator_connection:acquire(URL),
- Headers = [{"Authorization", "Basic " ++ B64Auth}],
- ?assertMatch({ok, "200", _, _}, send_req(WPid, URL, Headers, [])),
- Internals = worker_internals(WPid),
- ?assert(string:str(Internals, B64Auth) =:= 0),
- ?assert(string:str(Internals, Pass) =:= 0),
- couch_replicator_connection:release(WPid),
- unlink(ServerPid),
- exit(ServerPid, kill)
- end).
+ {_User, Pass, B64Auth} = user_pass(),
+ {Port, ServerPid} = server(),
+ PortStr = integer_to_list(Port),
+ URL = "http://" ++ Host ++ ":" ++ PortStr,
+ {ok, WPid} = couch_replicator_connection:acquire(URL),
+ Headers = [{"Authorization", "Basic " ++ B64Auth}],
+ ?assertMatch({ok, "200", _, _}, send_req(WPid, URL, Headers, [])),
+ Internals = worker_internals(WPid),
+ ?assert(string:str(Internals, B64Auth) =:= 0),
+ ?assert(string:str(Internals, Pass) =:= 0),
+ couch_replicator_connection:release(WPid),
+ unlink(ServerPid),
+ exit(ServerPid, kill).
worker_discards_proxy_creds_after_request({Host, _}) ->
- ?_test(begin
- {User, Pass, B64Auth} = user_pass(),
- {Port, ServerPid} = server(),
- PortStr = integer_to_list(Port),
- URL = "http://" ++ Host ++ ":" ++ PortStr,
- {ok, WPid} = couch_replicator_connection:acquire(URL),
- Opts = [
- {proxy_host, Host},
- {proxy_port, Port},
- {proxy_user, User},
- {proxy_pass, Pass}
- ],
- ?assertMatch({ok, "200", _, _}, send_req(WPid, URL, [], Opts)),
- Internals = worker_internals(WPid),
- ?assert(string:str(Internals, B64Auth) =:= 0),
- ?assert(string:str(Internals, Pass) =:= 0),
- couch_replicator_connection:release(WPid),
- unlink(ServerPid),
- exit(ServerPid, kill)
- end).
+ {User, Pass, B64Auth} = user_pass(),
+ {Port, ServerPid} = server(),
+ PortStr = integer_to_list(Port),
+ URL = "http://" ++ Host ++ ":" ++ PortStr,
+ {ok, WPid} = couch_replicator_connection:acquire(URL),
+ Opts = [
+ {proxy_host, Host},
+ {proxy_port, Port},
+ {proxy_user, User},
+ {proxy_pass, Pass}
+ ],
+ ?assertMatch({ok, "200", _, _}, send_req(WPid, URL, [], Opts)),
+ Internals = worker_internals(WPid),
+ ?assert(string:str(Internals, B64Auth) =:= 0),
+ ?assert(string:str(Internals, Pass) =:= 0),
+ couch_replicator_connection:release(WPid),
+ unlink(ServerPid),
+ exit(ServerPid, kill).
send_req(WPid, URL, Headers, Opts) ->
@@ -237,5 +226,6 @@ server_responder(LSock) ->
assert_monitors_equal(ShouldBe) ->
sys:get_status(couch_replicator_connection),
- {monitors, Monitors} = process_info(whereis(couch_replicator_connection), monitors),
+ {monitors, Monitors} = process_info(whereis(couch_replicator_connection),
+ monitors),
?assertEqual(Monitors, ShouldBe).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_create_target_with_options_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_create_target_with_options_tests.erl
index 63310d3..c957fc1 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_create_target_with_options_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_create_target_with_options_tests.erl
@@ -12,132 +12,137 @@
-module(couch_replicator_create_target_with_options_tests).
+
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
-
--define(USERNAME, "rep_admin").
--define(PASSWORD, "secret").
-
-setup() ->
- Ctx = test_util:start_couch([fabric, mem3, couch_replicator, chttpd]),
- Hashed = couch_passwords:hash_admin_password(?PASSWORD),
- ok = config:set("admins", ?USERNAME, ?b2l(Hashed), _Persist=false),
- Source = ?tempdb(),
- Target = ?tempdb(),
- {Ctx, {Source, Target}}.
-
-
-teardown({Ctx, {_Source, _Target}}) ->
- config:delete("admins", ?USERNAME),
- ok = test_util:stop_couch(Ctx).
+-include_lib("fabric/test/fabric2_test.hrl").
create_target_with_options_replication_test_() ->
{
"Create target with range partitions tests",
{
- foreach,
- fun setup/0, fun teardown/1,
- [
- fun should_create_target_with_q_4/1,
- fun should_create_target_with_q_2_n_1/1,
- fun should_create_target_with_default/1,
- fun should_not_create_target_with_q_any/1
- ]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_create_target_with_q_4),
+ ?TDEF_FE(should_create_target_with_q_2_n_1),
+ ?TDEF_FE(should_create_target_with_default),
+ ?TDEF_FE(should_not_create_target_with_q_any)
+ ]
+ }
}
}.
-should_create_target_with_q_4({_Ctx, {Source, Target}}) ->
+setup() ->
+ Source = ?tempdb(),
+ Target = ?tempdb(),
+ {Source, Target}.
+
+
+teardown({Source, Target}) ->
+ delete_db(Source),
+ delete_db(Target).
+
+
+should_create_target_with_q_4({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(Source)},
- {<<"target">>, db_url(Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"create_target">>, true},
{<<"create_target_params">>, {[{<<"q">>, <<"4">>}]}}
]},
create_db(Source),
create_doc(Source),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, _} = couch_replicator_test_helper:replicate(RepObject),
- {ok, TargetInfo} = fabric:get_db_info(Target),
+ TargetInfo = db_info(Target),
{ClusterInfo} = couch_util:get_value(cluster, TargetInfo),
delete_db(Source),
delete_db(Target),
- ?_assertEqual(4, couch_util:get_value(q, ClusterInfo)).
+ ?assertEqual(0, couch_util:get_value(q, ClusterInfo)).
-should_create_target_with_q_2_n_1({_Ctx, {Source, Target}}) ->
+should_create_target_with_q_2_n_1({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(Source)},
- {<<"target">>, db_url(Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"create_target">>, true},
{<<"create_target_params">>,
{[{<<"q">>, <<"2">>}, {<<"n">>, <<"1">>}]}}
]},
create_db(Source),
create_doc(Source),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, _} = couch_replicator_test_helper:replicate(RepObject),
- {ok, TargetInfo} = fabric:get_db_info(Target),
+ TargetInfo = db_info(Target),
{ClusterInfo} = couch_util:get_value(cluster, TargetInfo),
delete_db(Source),
delete_db(Target),
- [
- ?_assertEqual(2, couch_util:get_value(q, ClusterInfo)),
- ?_assertEqual(1, couch_util:get_value(n, ClusterInfo))
- ].
+ ?assertEqual(0, couch_util:get_value(q, ClusterInfo)),
+ ?assertEqual(0, couch_util:get_value(n, ClusterInfo)).
-should_create_target_with_default({_Ctx, {Source, Target}}) ->
+should_create_target_with_default({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(Source)},
- {<<"target">>, db_url(Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"create_target">>, true}
]},
create_db(Source),
create_doc(Source),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, _} = couch_replicator_test_helper:replicate(RepObject),
- {ok, TargetInfo} = fabric:get_db_info(Target),
+ TargetInfo = db_info(Target),
{ClusterInfo} = couch_util:get_value(cluster, TargetInfo),
- Q = config:get("cluster", "q", "8"),
delete_db(Source),
delete_db(Target),
- ?_assertEqual(list_to_integer(Q), couch_util:get_value(q, ClusterInfo)).
+ ?assertEqual(0, couch_util:get_value(q, ClusterInfo)).
-should_not_create_target_with_q_any({_Ctx, {Source, Target}}) ->
+should_not_create_target_with_q_any({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(Source)},
- {<<"target">>, db_url(Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"create_target">>, false},
{<<"create_target_params">>, {[{<<"q">>, <<"1">>}]}}
]},
create_db(Source),
create_doc(Source),
- {error, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- DbExist = is_list(catch mem3:shards(Target)),
+ {error, _} = couch_replicator_test_helper:replicate(RepObject),
+ Exists = try
+ fabric2_db:open(Target, [?ADMIN_CTX]),
+ ?assert(false)
+ catch
+ error:database_does_not_exist ->
+ database_does_not_exist
+ end,
delete_db(Source),
- ?_assertEqual(false, DbExist).
+ ?assertEqual(Exists, database_does_not_exist).
create_doc(DbName) ->
- Body = {[{<<"foo">>, <<"bar">>}]},
- NewDoc = #doc{body = Body},
- {ok, _} = fabric:update_doc(DbName, NewDoc, [?ADMIN_CTX]).
+ couch_replicator_test_helper:create_docs(DbName, [
+ #{<<"_id">> => fabric2_util:uuid(), <<"foo">> => <<"bar">>}
+ ]).
create_db(DbName) ->
- ok = fabric:create_db(DbName, [?ADMIN_CTX]).
+ couch_replicator_test_helper:create_db(DbName).
delete_db(DbName) ->
- ok = fabric:delete_db(DbName, [?ADMIN_CTX]).
+ couch_replicator_test_helper:delete_db(DbName).
-db_url(DbName) ->
- Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
- Port = mochiweb_socket_server:get(chttpd, port),
- ?l2b(io_lib:format("http://~s:~s@~s:~b/~s", [?USERNAME, ?PASSWORD, Addr,
- Port, DbName])).
+db_info(DbName) ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ {ok, Info} = fabric2_db:get_db_info(Db),
+ Info.
diff --git a/src/couch_replicator/test/eunit/couch_replicator_db_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_db_tests.erl
new file mode 100644
index 0000000..0534410
--- /dev/null
+++ b/src/couch_replicator/test/eunit/couch_replicator_db_tests.erl
@@ -0,0 +1,332 @@
+% 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_db_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_replicator/src/couch_replicator.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
+
+
+couch_replicator_db_test_() ->
+ {
+ "Replications are started from docs in _replicator dbs",
+ {
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(default_replicator_db_is_created),
+ ?TDEF_FE(continuous_replication_created_from_doc, 15),
+ ?TDEF_FE(normal_replication_created_from_doc, 15),
+ ?TDEF_FE(replicator_db_deleted, 15),
+ ?TDEF_FE(replicator_db_recreated, 15),
+ ?TDEF_FE(invalid_replication_docs),
+ ?TDEF_FE(duplicate_persistent_replication, 15),
+ ?TDEF_FE(duplicate_transient_replication, 30)
+ ]
+ }
+ }
+ }.
+
+
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ create_doc(Source, #{<<"_id">> => <<"doc1">>}),
+ Target = couch_replicator_test_helper:create_db(),
+ Name = ?tempdb(),
+ RepDb = couch_replicator_test_helper:create_db(<<Name/binary,
+ "/_replicator">>),
+ config:set("replicator", "stats_update_interval_sec", "0", false),
+ config:set("replicator", "create_replicator_db", "false", false),
+ config:set("couchdb", "enable_database_recovery", "false", false),
+ config:set("replicator", "min_backoff_penalty_sec", "1", false),
+ {Source, Target, RepDb}.
+
+
+teardown({Source, Target, RepDb}) ->
+ config:delete("replicator", "stats_update_interval_sec", false),
+ config:delete("replicator", "create_replicator_db", false),
+ config:delete("couchdb", "enable_database_recovery", false),
+ config:delete("replicator", "min_backoff_penalty_sec", false),
+
+ couch_replicator_test_helper:delete_db(RepDb),
+ couch_replicator_test_helper:delete_db(?REP_DB_NAME),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
+
+
+default_replicator_db_is_created({_, _, _}) ->
+ config:set("replicator", "create_replicator_db", "true", false),
+ ?assertEqual(ignore, couch_replicator:ensure_rep_db_exists()),
+ ?assertMatch({ok, #{}}, fabric2_db:open(?REP_DB_NAME, [])).
+
+
+continuous_replication_created_from_doc({Source, Target, RepDb}) ->
+ DocId = <<"rdoc1">>,
+ RDoc = rep_doc(Source, Target, DocId, #{<<"continuous">> => true}),
+ create_doc(RepDb, RDoc),
+ wait_scheduler_docs_state(RepDb, DocId, <<"running">>),
+
+ {Code, DocInfo} = scheduler_docs(RepDb, DocId),
+ ?assertEqual(200, Code),
+ ?assertMatch(#{
+ <<"database">> := RepDb,
+ <<"doc_id">> := DocId
+ }, DocInfo),
+
+ RepId = maps:get(<<"id">>, DocInfo),
+
+ ?assertMatch([#{
+ <<"database">> := RepDb,
+ <<"doc_id">> := DocId,
+ <<"id">> := RepId,
+ <<"state">> := <<"running">>
+ }], couch_replicator_test_helper:scheduler_jobs()),
+
+ ?assertMatch({200, #{
+ <<"database">> := RepDb,
+ <<"doc_id">> := DocId,
+ <<"id">> := RepId,
+ <<"state">> := <<"running">>
+ }}, scheduler_jobs(RepId)),
+
+ delete_doc(RepDb, DocId),
+ wait_scheduler_docs_not_found(RepDb, DocId),
+ ?assertMatch({404, #{}}, scheduler_jobs(RepId)).
+
+
+normal_replication_created_from_doc({Source, Target, RepDb}) ->
+ DocId = <<"rdoc2">>,
+ RDoc = rep_doc(Source, Target, DocId),
+ create_doc(RepDb, RDoc),
+ wait_scheduler_docs_state(RepDb, DocId, <<"completed">>),
+
+ {Code, DocInfo} = scheduler_docs(RepDb, DocId),
+ ?assertEqual(200, Code),
+ ?assertMatch(#{
+ <<"database">> := RepDb,
+ <<"doc_id">> := DocId,
+ <<"state">> := <<"completed">>,
+ <<"info">> := #{
+ <<"docs_written">> := 1,
+ <<"docs_read">> := 1,
+ <<"missing_revisions_found">> := 1
+ }
+ }, DocInfo),
+
+ wait_doc_state(RepDb, DocId, <<"completed">>),
+ ?assertMatch(#{
+ <<"_replication_state">> := <<"completed">>,
+ <<"_replication_stats">> := #{
+ <<"docs_written">> := 1,
+ <<"docs_read">> := 1,
+ <<"missing_revisions_found">> := 1
+ }
+ }, read_doc(RepDb, DocId)),
+
+ delete_doc(RepDb, DocId),
+ wait_scheduler_docs_not_found(RepDb, DocId).
+
+
+replicator_db_deleted({Source, Target, RepDb}) ->
+ DocId = <<"rdoc3">>,
+ RDoc = rep_doc(Source, Target, DocId, #{<<"continuous">> => true}),
+ create_doc(RepDb, RDoc),
+ wait_scheduler_docs_state(RepDb, DocId, <<"running">>),
+ fabric2_db:delete(RepDb, [?ADMIN_CTX]),
+ wait_scheduler_docs_not_found(RepDb, DocId).
+
+
+replicator_db_recreated({Source, Target, RepDb}) ->
+ DocId = <<"rdoc4">>,
+ RDoc = rep_doc(Source, Target, DocId, #{<<"continuous">> => true}),
+ create_doc(RepDb, RDoc),
+ wait_scheduler_docs_state(RepDb, DocId, <<"running">>),
+
+ config:set("couchdb", "enable_database_recovery", "true", false),
+ fabric2_db:delete(RepDb, [?ADMIN_CTX]),
+ wait_scheduler_docs_not_found(RepDb, DocId),
+
+ Opts = [{start_key, RepDb}, {end_key, RepDb}],
+ {ok, [DbInfo]} = fabric2_db:list_deleted_dbs_info(Opts),
+ {_, Timestamp} = lists:keyfind(timestamp, 1, DbInfo),
+ ok = fabric2_db:undelete(RepDb, RepDb, Timestamp, [?ADMIN_CTX]),
+ wait_scheduler_docs_state(RepDb, DocId, <<"running">>),
+
+ config:set("couchdb", "enable_database_recovery", "false", false),
+ fabric2_db:delete(RepDb, [?ADMIN_CTX]),
+ wait_scheduler_docs_not_found(RepDb, DocId).
+
+
+invalid_replication_docs({_, _, RepDb}) ->
+ Docs = [
+ #{
+ <<"_id">> => <<"1">>,
+ <<"source">> => <<"http://127.0.0.1:1000">>
+ },
+ #{
+ <<"_id">> => <<"1">>,
+ <<"target">> => <<"http://127.0.0.1:1001">>
+ },
+ #{
+ <<"_id">> => <<"1">>
+ },
+ #{
+ <<"_id">> => <<"1">>,
+ <<"source">> => <<"http://127.0.0.1:1002">>,
+ <<"target">> => <<"http://127.0.0.1:1003">>,
+ <<"create_target">> => <<"bad">>
+ },
+ #{
+ <<"_id">> => <<"1">>,
+ <<"source">> => #{<<"junk">> => 42},
+ <<"target">> => <<"http://127.0.0.1:1004">>
+ },
+ #{
+ <<"_id">> => <<"1">>,
+ <<"source">> => <<"http://127.0.0.1:1005">>,
+ <<"target">> => <<"http://127.0.0.1:1006">>,
+ <<"selector">> => #{},
+ <<"filter">> => <<"a/b">>
+ },
+ #{
+ <<"_id">> => <<"1">>,
+ <<"source">> => <<"http://127.0.0.1:1007">>,
+ <<"target">> => <<"https://127.0.0.1:1008">>,
+ <<"doc_ids">> => 42
+ }
+ ],
+ lists:foreach(fun(Doc) ->
+ ?assertThrow({forbidden, _}, create_doc(RepDb, Doc))
+ end, Docs).
+
+
+duplicate_persistent_replication({Source, Target, RepDb}) ->
+ DocId1 = <<"rdoc5">>,
+ RDoc1 = rep_doc(Source, Target, DocId1, #{<<"continuous">> => true}),
+ create_doc(RepDb, RDoc1),
+ wait_scheduler_docs_state(RepDb, DocId1, <<"running">>),
+
+ DocId2 = <<"rdoc6">>,
+ RDoc2 = rep_doc(Source, Target, DocId2, #{<<"continuous">> => true}),
+ create_doc(RepDb, RDoc2),
+ wait_scheduler_docs_state(RepDb, DocId2, <<"failed">>),
+
+ delete_doc(RepDb, DocId1),
+ delete_doc(RepDb, DocId2),
+
+ wait_scheduler_docs_not_found(RepDb, DocId1),
+ wait_scheduler_docs_not_found(RepDb, DocId2).
+
+
+duplicate_transient_replication({Source, Target, RepDb}) ->
+ {ok, _Pid, RepId} = couch_replicator_test_helper:replicate_continuous(
+ Source, Target),
+
+ DocId = <<"rdoc7">>,
+ RDoc = rep_doc(Source, Target, DocId, #{<<"continuous">> => true}),
+ create_doc(RepDb, RDoc),
+ wait_scheduler_docs_state(RepDb, DocId, <<"crashing">>),
+
+ couch_replicator_test_helper:cancel(RepId),
+ wait_reschedule_docs_state(RepDb, DocId, <<"running">>),
+
+ delete_doc(RepDb, DocId),
+ wait_scheduler_docs_not_found(RepDb, DocId).
+
+
+scheduler_jobs(Id) ->
+ SUrl = couch_replicator_test_helper:server_url(),
+ Url = lists:flatten(io_lib:format("~s/_scheduler/jobs/~s", [SUrl, Id])),
+ {ok, Code, _, Body} = test_request:get(Url, []),
+ {Code, jiffy:decode(Body, [return_maps])}.
+
+
+scheduler_docs(DbName, DocId) ->
+ SUrl = couch_replicator_test_helper:server_url(),
+ Fmt = "~s/_scheduler/docs/~s/~s",
+ Url = lists:flatten(io_lib:format(Fmt, [SUrl, DbName, DocId])),
+ {ok, Code, _, Body} = test_request:get(Url, []),
+ {Code, jiffy:decode(Body, [return_maps])}.
+
+
+rep_doc(Source, Target, DocId) ->
+ rep_doc(Source, Target, DocId, #{}).
+
+
+rep_doc(Source, Target, DocId, #{} = Extra) ->
+ maps:merge(#{
+ <<"_id">> => DocId,
+ <<"source">> => couch_replicator_test_helper:db_url(Source),
+ <<"target">> => couch_replicator_test_helper:db_url(Target)
+ }, Extra).
+
+
+create_doc(DbName, Doc) ->
+ couch_replicator_test_helper:create_docs(DbName, [Doc]).
+
+
+delete_doc(DbName, DocId) ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ {ok, Doc} = fabric2_db:open_doc(Db, DocId),
+ Doc1 = Doc#doc{deleted = true},
+ {ok, _} = fabric2_db:update_doc(Db, Doc1, []).
+
+
+read_doc(DbName, DocId) ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ {ok, Doc} = fabric2_db:open_doc(Db, DocId, [ejson_body]),
+ Body = Doc#doc.body,
+ couch_util:json_decode(couch_util:json_encode(Body), [return_maps]).
+
+
+wait_scheduler_docs_state(DbName, DocId, State) ->
+ test_util:wait(fun() ->
+ case scheduler_docs(DbName, DocId) of
+ {200, #{<<"state">> := State} = Res} -> Res;
+ {_, _} -> wait
+ end
+ end, 10000, 250).
+
+
+wait_scheduler_docs_not_found(DbName, DocId) ->
+ test_util:wait(fun() ->
+ case scheduler_docs(DbName, DocId) of
+ {404, _} -> ok;
+ {_, _} -> wait
+ end
+ end, 10000, 250).
+
+
+wait_reschedule_docs_state(DbName, DocId, State) ->
+ test_util:wait(fun() ->
+ couch_replicator_job_server:reschedule(),
+ case scheduler_docs(DbName, DocId) of
+ {200, #{<<"state">> := State} = Res} -> Res;
+ {_, _} -> wait
+ end
+ end, 10000, 500).
+
+
+wait_doc_state(DbName, DocId, State) ->
+ test_util:wait(fun() ->
+ case read_doc(DbName, DocId) of
+ #{<<"_replication_state">> := State} -> ok;
+ #{} -> wait
+ end
+ end, 10000, 250).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_filtered_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_filtered_tests.erl
index 7ac9a4d..4d72c84 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_filtered_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_filtered_tests.erl
@@ -12,17 +12,20 @@
-module(couch_replicator_filtered_tests).
+
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
--define(DDOC, {[
- {<<"_id">>, <<"_design/filter_ddoc">>},
- {<<"filters">>, {[
- {<<"testfilter">>, <<"
+-define(DDOC_ID, <<"_design/filter_ddoc">>).
+-define(DDOC, #{
+ <<"_id">> => ?DDOC_ID,
+ <<"filters">> => #{
+ <<"testfilter">> => <<"
function(doc, req){if (doc.class == 'mammal') return true;}
- ">>},
- {<<"queryfilter">>, <<"
+ ">>,
+ <<"queryfilter">> => <<"
function(doc, req) {
if (doc.class && req.query.starts) {
return doc.class.indexOf(req.query.starts) === 0;
@@ -31,99 +34,87 @@
return false;
}
}
- ">>}
- ]}},
- {<<"views">>, {[
- {<<"mammals">>, {[
- {<<"map">>, <<"
+ ">>
+ },
+ <<"views">> => #{
+ <<"mammals">> => #{
+ <<"map">> => <<"
function(doc) {
if (doc.class == 'mammal') {
emit(doc._id, null);
}
}
- ">>}
- ]}}
- ]}}
-]}).
-
-setup(_) ->
- Ctx = test_util:start_couch([couch_replicator]),
- Source = create_db(),
- create_docs(Source),
- Target = create_db(),
- {Ctx, {Source, Target}}.
+ ">>
+ }
+ }
+}).
-teardown(_, {Ctx, {Source, Target}}) ->
- delete_db(Source),
- delete_db(Target),
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
filtered_replication_test_() ->
- Pairs = [{remote, remote}],
{
- "Filtered replication tests",
+ "Replications with filters tests",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_succeed/2} || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(filtered_replication_test),
+ ?TDEF_FE(query_filtered_replication_test),
+ ?TDEF_FE(view_filtered_replication_test),
+ ?TDEF_FE(replication_id_changes_if_filter_changes, 15)
+ ]
+ }
}
}.
-query_filtered_replication_test_() ->
- Pairs = [{remote, remote}],
- {
- "Filtered with query replication tests",
- {
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_succeed_with_query/2} || Pair <- Pairs]
- }
- }.
-view_filtered_replication_test_() ->
- Pairs = [{remote, remote}],
- {
- "Filtered with a view replication tests",
- {
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_succeed_with_view/2} || Pair <- Pairs]
- }
- }.
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ create_docs(Source),
+ Target = couch_replicator_test_helper:create_db(),
+ config:set("replicator", "stats_update_interval_sec", "0", false),
+ config:set("replicator", "interval_sec", "1", false),
+ {Source, Target}.
+
+
+teardown({Source, Target}) ->
+ config:delete("replicator", "stats_update_interval_sec", false),
+ config:delete("replicator", "checkpoint_interval", false),
+ config:delete("replicator", "interval_sec", false),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
-should_succeed({From, To}, {_Ctx, {Source, Target}}) ->
+
+filtered_replication_test({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"filter">>, <<"filter_ddoc/testfilter">>}
]},
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- %% FilteredFun is an Erlang version of following JS function
- %% function(doc, req){if (doc.class == 'mammal') return true;}
+ {ok, _} = couch_replicator_test_helper:replicate(RepObject),
FilterFun = fun(_DocId, {Props}) ->
couch_util:get_value(<<"class">>, Props) == <<"mammal">>
end,
{ok, TargetDbInfo, AllReplies} = compare_dbs(Source, Target, FilterFun),
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])), [
- {"Target DB has proper number of docs",
- ?_assertEqual(1, proplists:get_value(doc_count, TargetDbInfo))},
- {"Target DB doesn't have deleted docs",
- ?_assertEqual(0, proplists:get_value(doc_del_count, TargetDbInfo))},
- {"All the docs filtered as expected",
- ?_assert(lists:all(fun(Valid) -> Valid end, AllReplies))}
- ]}.
-
-should_succeed_with_query({From, To}, {_Ctx, {Source, Target}}) ->
+ ?assertEqual(1, proplists:get_value(doc_count, TargetDbInfo)),
+ ?assertEqual(0, proplists:get_value(doc_del_count, TargetDbInfo)),
+ ?assert(lists:all(fun(Valid) -> Valid end, AllReplies)).
+
+
+query_filtered_replication_test({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"filter">>, <<"filter_ddoc/queryfilter">>},
{<<"query_params">>, {[
{<<"starts">>, <<"a">>}
]}}
]},
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, _} = couch_replicator_test_helper:replicate(RepObject),
FilterFun = fun(_DocId, {Props}) ->
case couch_util:get_value(<<"class">>, Props) of
<<"a", _/binary>> -> true;
@@ -131,109 +122,144 @@ should_succeed_with_query({From, To}, {_Ctx, {Source, Target}}) ->
end
end,
{ok, TargetDbInfo, AllReplies} = compare_dbs(Source, Target, FilterFun),
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])), [
- {"Target DB has proper number of docs",
- ?_assertEqual(2, proplists:get_value(doc_count, TargetDbInfo))},
- {"Target DB doesn't have deleted docs",
- ?_assertEqual(0, proplists:get_value(doc_del_count, TargetDbInfo))},
- {"All the docs filtered as expected",
- ?_assert(lists:all(fun(Valid) -> Valid end, AllReplies))}
- ]}.
-
-should_succeed_with_view({From, To}, {_Ctx, {Source, Target}}) ->
+ ?assertEqual(2, proplists:get_value(doc_count, TargetDbInfo)),
+ ?assertEqual(0, proplists:get_value(doc_del_count, TargetDbInfo)),
+ ?assert(lists:all(fun(Valid) -> Valid end, AllReplies)).
+
+
+view_filtered_replication_test({Source, Target}) ->
RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)},
+ {<<"source">>, Source},
+ {<<"target">>, Target},
{<<"filter">>, <<"_view">>},
{<<"query_params">>, {[
{<<"view">>, <<"filter_ddoc/mammals">>}
]}}
]},
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+ {ok, _} = couch_replicator_test_helper:replicate(RepObject),
FilterFun = fun(_DocId, {Props}) ->
couch_util:get_value(<<"class">>, Props) == <<"mammal">>
end,
{ok, TargetDbInfo, AllReplies} = compare_dbs(Source, Target, FilterFun),
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])), [
- {"Target DB has proper number of docs",
- ?_assertEqual(1, proplists:get_value(doc_count, TargetDbInfo))},
- {"Target DB doesn't have deleted docs",
- ?_assertEqual(0, proplists:get_value(doc_del_count, TargetDbInfo))},
- {"All the docs filtered as expected",
- ?_assert(lists:all(fun(Valid) -> Valid end, AllReplies))}
- ]}.
+ ?assertEqual(1, proplists:get_value(doc_count, TargetDbInfo)),
+ ?assertEqual(0, proplists:get_value(doc_del_count, TargetDbInfo)),
+ ?assert(lists:all(fun(Valid) -> Valid end, AllReplies)).
+
+
+replication_id_changes_if_filter_changes({Source, Target}) ->
+ config:set("replicator", "checkpoint_interval", "500", false),
+ Rep = {[
+ {<<"source">>, Source},
+ {<<"target">>, Target},
+ {<<"filter">>, <<"filter_ddoc/testfilter">>},
+ {<<"continuous">>, true}
+ ]},
+ {ok, _, RepId1} = couch_replicator_test_helper:replicate_continuous(Rep),
+
+ wait_scheduler_docs_written(1),
+
+ ?assertMatch([#{<<"id">> := RepId1}],
+ couch_replicator_test_helper:scheduler_jobs()),
+
+ FilterFun1 = fun(_, {Props}) ->
+ couch_util:get_value(<<"class">>, Props) == <<"mammal">>
+ end,
+ {ok, TargetDbInfo1, AllReplies1} = compare_dbs(Source, Target, FilterFun1),
+ ?assertEqual(1, proplists:get_value(doc_count, TargetDbInfo1)),
+ ?assert(lists:all(fun(Valid) -> Valid end, AllReplies1)),
+
+ {ok, SourceDb} = fabric2_db:open(Source, [?ADMIN_CTX]),
+ {ok, DDoc1} = fabric2_db:open_doc(SourceDb, ?DDOC_ID),
+ Flt = <<"function(doc, req) {if (doc.class == 'reptiles') return true};">>,
+ DDoc2 = DDoc1#doc{body = {[
+ {<<"filters">>, {[
+ {<<"testfilter">>, Flt}
+ ]}}
+ ]}},
+ {ok, {_, _}} = fabric2_db:update_doc(SourceDb, DDoc2),
+ Info = wait_scheduler_repid_change(RepId1),
+
+ RepId2 = maps:get(<<"id">>, Info),
+ ?assert(RepId1 =/= RepId2),
+
+ wait_scheduler_docs_written(1),
+
+ FilterFun2 = fun(_, {Props}) ->
+ Class = couch_util:get_value(<<"class">>, Props),
+ Class == <<"mammal">> orelse Class == <<"reptiles">>
+ end,
+ {ok, TargetDbInfo2, AllReplies2} = compare_dbs(Source, Target, FilterFun2),
+ ?assertEqual(2, proplists:get_value(doc_count, TargetDbInfo2)),
+ ?assert(lists:all(fun(Valid) -> Valid end, AllReplies2)),
+
+ couch_replicator_test_helper:cancel(RepId2).
+
compare_dbs(Source, Target, FilterFun) ->
- {ok, SourceDb} = couch_db:open_int(Source, []),
- {ok, TargetDb} = couch_db:open_int(Target, []),
- {ok, TargetDbInfo} = couch_db:get_db_info(TargetDb),
- Fun = fun(FullDocInfo, Acc) ->
- {ok, DocId, SourceDoc} = read_doc(SourceDb, FullDocInfo),
- TargetReply = read_doc(TargetDb, DocId),
- case FilterFun(DocId, SourceDoc) of
- true ->
- ValidReply = {ok, DocId, SourceDoc} == TargetReply,
- {ok, [ValidReply|Acc]};
- false ->
- ValidReply = {not_found, missing} == TargetReply,
- {ok, [ValidReply|Acc]}
+ {ok, TargetDb} = fabric2_db:open(Target, [?ADMIN_CTX]),
+ {ok, TargetDbInfo} = fabric2_db:get_db_info(TargetDb),
+ Fun = fun(SrcDoc, TgtDoc, Acc) ->
+ case FilterFun(SrcDoc#doc.id, SrcDoc#doc.body) of
+ true -> [SrcDoc == TgtDoc | Acc];
+ false -> [not_found == TgtDoc | Acc]
end
end,
- {ok, AllReplies} = couch_db:fold_docs(SourceDb, Fun, [], []),
- ok = couch_db:close(SourceDb),
- ok = couch_db:close(TargetDb),
- {ok, TargetDbInfo, AllReplies}.
-
-read_doc(Db, DocIdOrInfo) ->
- case couch_db:open_doc(Db, DocIdOrInfo) of
- {ok, Doc} ->
- {Props} = couch_doc:to_json_obj(Doc, [attachments]),
- DocId = couch_util:get_value(<<"_id">>, Props),
- {ok, DocId, {Props}};
- Error ->
- Error
- end.
-
-create_db() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
+ Res = couch_replicator_test_helper:compare_fold(Source, Target, Fun, []),
+ {ok, TargetDbInfo, Res}.
+
create_docs(DbName) ->
- {ok, Db} = couch_db:open(DbName, [?ADMIN_CTX]),
- DDoc = couch_doc:from_json_obj(?DDOC),
- Doc1 = couch_doc:from_json_obj({[
- {<<"_id">>, <<"doc1">>},
- {<<"class">>, <<"mammal">>},
- {<<"value">>, 1}
-
- ]}),
- Doc2 = couch_doc:from_json_obj({[
- {<<"_id">>, <<"doc2">>},
- {<<"class">>, <<"amphibians">>},
- {<<"value">>, 2}
-
- ]}),
- Doc3 = couch_doc:from_json_obj({[
- {<<"_id">>, <<"doc3">>},
- {<<"class">>, <<"reptiles">>},
- {<<"value">>, 3}
-
- ]}),
- Doc4 = couch_doc:from_json_obj({[
- {<<"_id">>, <<"doc4">>},
- {<<"class">>, <<"arthropods">>},
- {<<"value">>, 2}
-
- ]}),
- {ok, _} = couch_db:update_docs(Db, [DDoc, Doc1, Doc2, Doc3, Doc4]),
- couch_db:close(Db).
-
-delete_db(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]).
-
-db_url(remote, DbName) ->
- Addr = config:get("httpd", "bind_address", "127.0.0.1"),
- Port = mochiweb_socket_server:get(couch_httpd, port),
- ?l2b(io_lib:format("http://~s:~b/~s", [Addr, Port, DbName])).
+ couch_replicator_test_helper:create_docs(DbName, [
+ ?DDOC,
+ #{
+ <<"_id">> => <<"doc1">>,
+ <<"class">> => <<"mammal">>,
+ <<"value">> => 1
+ },
+ #{
+ <<"_id">> => <<"doc2">>,
+ <<"class">> => <<"amphibians">>,
+ <<"value">> => 2
+ },
+ #{
+ <<"_id">> => <<"doc3">>,
+ <<"class">> => <<"reptiles">>,
+ <<"value">> => 3
+ },
+ #{
+ <<"_id">> => <<"doc4">>,
+ <<"class">> => <<"arthropods">>,
+ <<"value">> => 2
+ }
+ ]).
+
+
+wait_scheduler_docs_written(DocsWritten) ->
+ test_util:wait(fun() ->
+ case couch_replicator_test_helper:scheduler_jobs() of
+ [] ->
+ wait;
+ [#{<<"info">> := null}] ->
+ wait;
+ [#{<<"info">> := Info}] ->
+ case Info of
+ #{<<"docs_written">> := DocsWritten} -> Info;
+ #{} -> wait
+ end
+ end
+ end, 10000, 250).
+
+
+wait_scheduler_repid_change(OldRepId) ->
+ test_util:wait(fun() ->
+ case couch_replicator_test_helper:scheduler_jobs() of
+ [] ->
+ wait;
+ [#{<<"id">> := OldRepId}] ->
+ wait;
+ [#{<<"id">> := null}] ->
+ wait;
+ [#{<<"id">> := NewId} = Info] when is_binary(NewId) ->
+ Info
+ end
+ end, 10000, 250).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_httpc_pool_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_httpc_pool_tests.erl
index c4ad4e9..6c61446 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_httpc_pool_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_httpc_pool_tests.erl
@@ -12,17 +12,13 @@
-module(couch_replicator_httpc_pool_tests).
+
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-
--define(TIMEOUT, 1000).
+-include_lib("fabric/test/fabric2_test.hrl").
-setup() ->
- spawn_pool().
-
-teardown(Pool) ->
- stop_pool(Pool).
+-define(TIMEOUT, 1000).
httpc_pool_test_() ->
@@ -30,75 +26,81 @@ httpc_pool_test_() ->
"httpc pool tests",
{
setup,
- fun() -> test_util:start_couch([couch_replicator]) end, fun test_util:stop_couch/1,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
{
foreach,
- fun setup/0, fun teardown/1,
+ fun setup/0,
+ fun teardown/1,
[
- fun should_block_new_clients_when_full/1,
- fun should_replace_worker_on_death/1
+ ?TDEF_FE(should_block_new_clients_when_full),
+ ?TDEF_FE(should_replace_worker_on_death)
]
}
}
}.
+setup() ->
+ spawn_pool().
+
+
+teardown(Pool) ->
+ stop_pool(Pool).
+
+
should_block_new_clients_when_full(Pool) ->
- ?_test(begin
- Client1 = spawn_client(Pool),
- Client2 = spawn_client(Pool),
- Client3 = spawn_client(Pool),
+ Client1 = spawn_client(Pool),
+ Client2 = spawn_client(Pool),
+ Client3 = spawn_client(Pool),
+
+ ?assertEqual(ok, ping_client(Client1)),
+ ?assertEqual(ok, ping_client(Client2)),
+ ?assertEqual(ok, ping_client(Client3)),
- ?assertEqual(ok, ping_client(Client1)),
- ?assertEqual(ok, ping_client(Client2)),
- ?assertEqual(ok, ping_client(Client3)),
+ Worker1 = get_client_worker(Client1, "1"),
+ Worker2 = get_client_worker(Client2, "2"),
+ Worker3 = get_client_worker(Client3, "3"),
- Worker1 = get_client_worker(Client1, "1"),
- Worker2 = get_client_worker(Client2, "2"),
- Worker3 = get_client_worker(Client3, "3"),
+ ?assert(is_process_alive(Worker1)),
+ ?assert(is_process_alive(Worker2)),
+ ?assert(is_process_alive(Worker3)),
- ?assert(is_process_alive(Worker1)),
- ?assert(is_process_alive(Worker2)),
- ?assert(is_process_alive(Worker3)),
+ ?assertNotEqual(Worker1, Worker2),
+ ?assertNotEqual(Worker2, Worker3),
+ ?assertNotEqual(Worker3, Worker1),
- ?assertNotEqual(Worker1, Worker2),
- ?assertNotEqual(Worker2, Worker3),
- ?assertNotEqual(Worker3, Worker1),
+ Client4 = spawn_client(Pool),
+ ?assertEqual(timeout, ping_client(Client4)),
- Client4 = spawn_client(Pool),
- ?assertEqual(timeout, ping_client(Client4)),
+ ?assertEqual(ok, stop_client(Client1)),
+ ?assertEqual(ok, ping_client(Client4)),
- ?assertEqual(ok, stop_client(Client1)),
- ?assertEqual(ok, ping_client(Client4)),
+ Worker4 = get_client_worker(Client4, "4"),
+ ?assertEqual(Worker1, Worker4),
- Worker4 = get_client_worker(Client4, "4"),
- ?assertEqual(Worker1, Worker4),
+ lists:foreach(fun(C) ->
+ ?assertEqual(ok, stop_client(C))
+ end, [Client2, Client3, Client4]).
- lists:foreach(
- fun(C) ->
- ?assertEqual(ok, stop_client(C))
- end, [Client2, Client3, Client4])
- end).
should_replace_worker_on_death(Pool) ->
- ?_test(begin
- Client1 = spawn_client(Pool),
- ?assertEqual(ok, ping_client(Client1)),
- Worker1 = get_client_worker(Client1, "1"),
- ?assert(is_process_alive(Worker1)),
+ Client1 = spawn_client(Pool),
+ ?assertEqual(ok, ping_client(Client1)),
+ Worker1 = get_client_worker(Client1, "1"),
+ ?assert(is_process_alive(Worker1)),
- ?assertEqual(ok, kill_client_worker(Client1)),
- ?assertNot(is_process_alive(Worker1)),
- ?assertEqual(ok, stop_client(Client1)),
+ ?assertEqual(ok, kill_client_worker(Client1)),
+ ?assertNot(is_process_alive(Worker1)),
+ ?assertEqual(ok, stop_client(Client1)),
- Client2 = spawn_client(Pool),
- ?assertEqual(ok, ping_client(Client2)),
- Worker2 = get_client_worker(Client2, "2"),
- ?assert(is_process_alive(Worker2)),
+ Client2 = spawn_client(Pool),
+ ?assertEqual(ok, ping_client(Client2)),
+ Worker2 = get_client_worker(Client2, "2"),
+ ?assert(is_process_alive(Worker2)),
- ?assertNotEqual(Worker1, Worker2),
- ?assertEqual(ok, stop_client(Client2))
- end).
+ ?assertNotEqual(Worker1, Worker2),
+ ?assertEqual(ok, stop_client(Client2)).
spawn_client(Pool) ->
@@ -110,6 +112,7 @@ spawn_client(Pool) ->
end),
{Pid, Ref}.
+
ping_client({Pid, Ref}) ->
Pid ! ping,
receive
@@ -119,18 +122,18 @@ ping_client({Pid, Ref}) ->
timeout
end.
+
get_client_worker({Pid, Ref}, ClientName) ->
Pid ! get_worker,
receive
{worker, Ref, Worker} ->
Worker
after ?TIMEOUT ->
- erlang:error(
- {assertion_failed,
- [{module, ?MODULE}, {line, ?LINE},
- {reason, "Timeout getting client " ++ ClientName ++ " worker"}]})
+ erlang:error({assertion_failed, [{module, ?MODULE}, {line, ?LINE},
+ {reason, "Timeout getting client " ++ ClientName ++ " worker"}]})
end.
+
stop_client({Pid, Ref}) ->
Pid ! stop,
receive
@@ -140,6 +143,7 @@ stop_client({Pid, Ref}) ->
timeout
end.
+
kill_client_worker({Pid, Ref}) ->
Pid ! get_worker,
receive
@@ -150,6 +154,7 @@ kill_client_worker({Pid, Ref}) ->
timeout
end.
+
loop(Parent, Ref, Worker, Pool) ->
receive
ping ->
@@ -163,12 +168,14 @@ loop(Parent, Ref, Worker, Pool) ->
Parent ! {stop, Ref}
end.
+
spawn_pool() ->
- Host = config:get("httpd", "bind_address", "127.0.0.1"),
- Port = config:get("httpd", "port", "5984"),
+ Host = config:get("chttpd", "bind_address", "127.0.0.1"),
+ Port = config:get("chttpd", "port", "5984"),
{ok, Pool} = couch_replicator_httpc_pool:start_link(
"http://" ++ Host ++ ":" ++ Port, [{max_connections, 3}]),
Pool.
+
stop_pool(Pool) ->
ok = couch_replicator_httpc_pool:stop(Pool).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_id_too_long_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_id_too_long_tests.erl
index a4696c4..3a0e6f7 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_id_too_long_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_id_too_long_tests.erl
@@ -15,76 +15,57 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
-
-
-setup(_) ->
- Ctx = test_util:start_couch([couch_replicator]),
- Source = create_db(),
- create_doc(Source),
- Target = create_db(),
- {Ctx, {Source, Target}}.
-
-
-teardown(_, {Ctx, {Source, Target}}) ->
- delete_db(Source),
- delete_db(Target),
- config:set("replicator", "max_document_id_length", "infinity"),
- ok = test_util:stop_couch(Ctx).
+-include_lib("fabric/test/fabric2_test.hrl").
id_too_long_replication_test_() ->
- Pairs = [{remote, remote}],
{
"Doc id too long tests",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_succeed/2} || Pair <- Pairs] ++
- [{Pair, fun should_fail/2} || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_succeed),
+ ?TDEF_FE(should_fail)
+
+ ]
+ }
}
}.
-should_succeed({From, To}, {_Ctx, {Source, Target}}) ->
- RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)}
- ]},
- config:set("replicator", "max_document_id_length", "5"),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- ?_assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
-
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ create_doc(Source),
+ Target = couch_replicator_test_helper:create_db(),
+ {Source, Target}.
-should_fail({From, To}, {_Ctx, {Source, Target}}) ->
- RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)}
- ]},
- config:set("replicator", "max_document_id_length", "4"),
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- ?_assertError({badmatch, {not_found, missing}},
- couch_replicator_test_helper:compare_dbs(Source, Target)).
+teardown({Source, Target}) ->
+ config:delete("replicator", "max_document_id_length", false),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
-create_db() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
+should_succeed({Source, Target}) ->
+ config:set("replicator", "max_document_id_length", "5", false),
+ {ok, _} = couch_replicator_test_helper:replicate(Source, Target),
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
-create_doc(DbName) ->
- {ok, Db} = couch_db:open(DbName, [?ADMIN_CTX]),
- Doc = couch_doc:from_json_obj({[{<<"_id">>, <<"12345">>}]}),
- {ok, _} = couch_db:update_doc(Db, Doc, []),
- couch_db:close(Db).
+should_fail({Source, Target}) ->
+ config:set("replicator", "max_document_id_length", "4", false),
+ {ok, _} = couch_replicator_test_helper:replicate(Source, Target),
+ ExceptIds = [<<"12345">>],
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target,
+ ExceptIds)).
-delete_db(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]).
-
-db_url(remote, DbName) ->
- Addr = config:get("httpd", "bind_address", "127.0.0.1"),
- Port = mochiweb_socket_server:get(couch_httpd, port),
- ?l2b(io_lib:format("http://~s:~b/~s", [Addr, Port, DbName])).
+create_doc(DbName) ->
+ Docs = [#{<<"_id">> => <<"12345">>}],
+ couch_replicator_test_helper:create_docs(DbName, Docs).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_job_server_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_job_server_tests.erl
new file mode 100644
index 0000000..698a844
--- /dev/null
+++ b/src/couch_replicator/test/eunit/couch_replicator_job_server_tests.erl
@@ -0,0 +1,437 @@
+% 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_job_server_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
+
+
+-define(SHUTDOWN_TIMEOUT, 1000).
+-define(JOB_SERVER, couch_replicator_job_server).
+
+
+job_server_test_() ->
+ {
+ "Test job server",
+ {
+ setup,
+ fun setup_all/0,
+ fun teardown_all/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_start_up),
+ ?TDEF_FE(reschedule_resets_timer),
+ ?TDEF_FE(reschedule_reads_config),
+ ?TDEF_FE(acceptors_spawned_if_pending),
+ ?TDEF_FE(acceptors_not_spawned_if_no_pending),
+ ?TDEF_FE(acceptors_not_spawned_if_no_max_churn),
+ ?TDEF_FE(acceptors_not_spawned_if_no_churn_budget),
+ ?TDEF_FE(acceptors_spawned_on_acceptor_exit),
+ ?TDEF_FE(acceptor_turns_into_worker),
+ ?TDEF_FE(acceptors_spawned_on_worker_exit),
+ ?TDEF_FE(excess_acceptors_spawned),
+ ?TDEF_FE(excess_workers_trimmed_on_reschedule),
+ ?TDEF_FE(recent_workers_are_not_stopped)
+ ]
+ }
+ }
+ }.
+
+
+setup_all() ->
+ Ctx = test_util:start_couch(),
+ meck:new(couch_replicator_job_server, [passthrough]),
+ mock_pending(0),
+ meck:expect(couch_replicator_jobs, set_timeout, 0, ok),
+ meck:expect(couch_replicator_jobs, fold_jobs, 3, ok),
+ meck:expect(couch_replicator_job, start_link, fun() ->
+ {ok, spawn_link(fun() -> start_job() end)}
+ end),
+ Ctx.
+
+
+teardown_all(Ctx) ->
+ meck:unload(),
+ config_delete("interval_sec"),
+ config_delete("max_acceptors"),
+ config_delete("max_jobs"),
+ config_delete("max_churn"),
+ config_delete("min_run_time_sec"),
+ config_delete("transient_job_max_age_sec"),
+ test_util:stop_couch(Ctx).
+
+
+setup() ->
+ config_set("interval_sec", "99999"),
+ config_set("max_acceptors", "0"),
+ config_set("max_jobs", "0"),
+ config_set("max_churn", "1"),
+ config_set("min_run_time_sec", "0"),
+ config_set("transient_job_max_age_sec", "99999"),
+
+ mock_pending(0),
+
+ {ok, SPid} = ?JOB_SERVER:start_link(?SHUTDOWN_TIMEOUT),
+ SPid.
+
+
+teardown(SPid) when is_pid(SPid) ->
+ unlink(SPid),
+ Ref = monitor(process, SPid),
+ exit(SPid, kill),
+ receive {'DOWN', Ref, _, _, _} -> ok end,
+
+ meck:reset(couch_replicator_jobs),
+ meck:reset(couch_replicator_job),
+ meck:reset(couch_replicator_job_server),
+
+ config_delete("interval_sec"),
+ config_delete("max_acceptors"),
+ config_delete("max_jobs"),
+ config_delete("max_churn"),
+ config_delete("min_run_time_sec"),
+ config_delete("transient_job_max_age_sec").
+
+
+should_start_up(SPid) ->
+ ?assert(is_process_alive(SPid)),
+ ?assertEqual(SPid, whereis(?JOB_SERVER)),
+ State = sys:get_state(?JOB_SERVER),
+ #{
+ acceptors := #{},
+ workers := #{},
+ churn := 0,
+ config := Config,
+ timer := Timer,
+ timeout := ?SHUTDOWN_TIMEOUT
+ } = State,
+
+ % Make sure it read the config
+ ?assertMatch(#{
+ max_acceptors := 0,
+ interval_sec := 99999,
+ max_jobs := 0,
+ max_churn := 1,
+ min_run_time_sec := 0,
+ transient_job_max_age_sec := 99999
+ }, Config),
+
+ % Timer was set up
+ ?assert(is_reference(Timer)),
+ ?assert(is_integer(erlang:read_timer(Timer))).
+
+
+reschedule_resets_timer(_) ->
+ #{timer := OldTimer} = sys:get_state(?JOB_SERVER),
+
+ ?assertEqual(ok, ?JOB_SERVER:reschedule()),
+
+ #{timer := Timer} = sys:get_state(?JOB_SERVER),
+ ?assert(is_reference(Timer)),
+ ?assert(Timer =/= OldTimer).
+
+
+reschedule_reads_config(_) ->
+ config_set("interval_sec", "99998"),
+
+ ?JOB_SERVER:reschedule(),
+
+ #{config := Config} = sys:get_state(?JOB_SERVER),
+ ?assertMatch(#{interval_sec := 99998}, Config).
+
+
+acceptors_spawned_if_pending(_) ->
+ config_set("max_acceptors", "1"),
+ mock_pending(1),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertMatch([Pid] when is_pid(Pid), acceptors()).
+
+
+acceptors_not_spawned_if_no_pending(_) ->
+ config_set("max_acceptors", "1"),
+ mock_pending(0),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual([], acceptors()).
+
+
+acceptors_not_spawned_if_no_max_churn(_) ->
+ config_set("max_churn", "0"),
+ config_set("max_acceptors", "1"),
+ mock_pending(1),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual([], acceptors()).
+
+
+acceptors_not_spawned_if_no_churn_budget(_) ->
+ config_set("max_churn", "1"),
+ config_set("max_acceptors", "1"),
+ mock_pending(0),
+
+ % To read the config
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual([], acceptors()),
+
+ mock_pending(1),
+
+ % Exhaust churn budget
+ sys:replace_state(couch_replicator_job_server, fun(#{} = St) ->
+ St#{churn := 1}
+ end),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual([], acceptors()).
+
+
+acceptors_spawned_on_acceptor_exit(_) ->
+ config_set("max_acceptors", "3"),
+ config_set("max_jobs", "4"),
+ mock_pending(1),
+
+ ?JOB_SERVER:reschedule(),
+
+ [A1] = acceptors(),
+
+ exit(A1, kill),
+ meck:wait(?JOB_SERVER, handle_info, [{'EXIT', A1, killed}, '_'], 2000),
+
+ ?assertEqual(3, length(acceptors())).
+
+
+acceptor_turns_into_worker(_) ->
+ config_set("max_acceptors", "3"),
+ config_set("max_jobs", "4"),
+ mock_pending(1),
+
+ ?JOB_SERVER:reschedule(),
+
+ [A1] = acceptors(),
+ accept_job(A1, true),
+ ?assertEqual(3, length(acceptors())),
+ #{workers := Workers} = sys:get_state(?JOB_SERVER),
+ ?assertMatch([{A1, {true, _}}], maps:to_list(Workers)).
+
+
+acceptors_spawned_on_worker_exit(_) ->
+ config_set("max_acceptors", "1"),
+ config_set("max_jobs", "1"),
+ mock_pending(1),
+
+ ?JOB_SERVER:reschedule(),
+
+ [A1] = acceptors(),
+ accept_job(A1, true),
+
+ % Since max_jobs = 1 no more acceptors are spawned
+ ?assertEqual(0, length(acceptors())),
+
+ % Same acceptor process is now a worker
+ ?assertEqual([A1], workers()),
+
+ exit(A1, shutdown),
+ meck:wait(?JOB_SERVER, handle_info, [{'EXIT', A1, shutdown}, '_'], 2000),
+
+ % New acceptor process started
+ ?assertEqual(1, length(acceptors())),
+ ?assertEqual(0, length(workers())).
+
+
+excess_acceptors_spawned(_) ->
+ config_set("max_acceptors", "2"),
+ config_set("max_churn", "3"),
+ config_set("max_jobs", "4"),
+ mock_pending(100),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual(3, length(acceptors())),
+
+ accept_all(),
+
+ ?assertEqual(3, length(workers())),
+ ?assertEqual(1, length(acceptors())),
+ % Check that the churn budget was consumed
+ ?assertMatch(#{churn := 3}, sys:get_state(?JOB_SERVER)),
+
+ accept_all(),
+
+ % No more acceptors spawned after reaching max_jobs
+ ?assertEqual(0, length(acceptors())),
+ ?assertEqual(4, length(workers())),
+
+ ?JOB_SERVER:reschedule(),
+
+ % Since all churn budget was consumed, no new acceptors should have beens
+ % spawned this cycle but churn budget should have been reset
+ ?assertEqual(0, length(acceptors())),
+ ?assertEqual(4, length(workers())),
+ ?assertMatch(#{churn := 0}, sys:get_state(?JOB_SERVER)),
+
+ ?JOB_SERVER:reschedule(),
+
+ % Should have spawned 3 excess acceptors
+ ?assertEqual(3, length(acceptors())),
+ ?assertEqual(4, length(workers())),
+
+ accept_all(),
+
+ % Running with an excess number of workers
+ ?assertEqual(0, length(acceptors())),
+ ?assertEqual(7, length(workers())).
+
+
+excess_workers_trimmed_on_reschedule(_) ->
+ config_set("max_acceptors", "2"),
+ config_set("max_churn", "3"),
+ config_set("max_jobs", "4"),
+ mock_pending(100),
+
+ ?JOB_SERVER:reschedule(),
+
+ [A1, A2, A3] = acceptors(),
+ accept_job(A1, true),
+ accept_job(A2, false),
+ accept_job(A3, false),
+ [A4] = acceptors(),
+ accept_job(A4, true),
+
+ ?JOB_SERVER:reschedule(),
+
+ % First reschedule was to reset the churn budget, this next one is to spawn
+ % an excess number of acceptors.
+ ?JOB_SERVER:reschedule(),
+
+ [A5, A6, A7] = acceptors(),
+ accept_job(A5, true),
+ accept_job(A6, false),
+ accept_job(A7, false),
+
+ ?assertEqual(7, length(workers())),
+
+ % Running with an excess number of workers. These should be trimmed on the
+ % during the next cycle
+ ?JOB_SERVER:reschedule(),
+
+ Workers = workers(),
+ ?assertEqual(4, length(Workers)),
+ ?assertEqual(0, length(acceptors())),
+
+ % Check that A1 and A4 were skipped since they are not continuous
+ ?assertEqual(Workers, Workers -- [A2, A3, A6]).
+
+
+recent_workers_are_not_stopped(_) ->
+ config_set("max_acceptors", "2"),
+ config_set("max_churn", "3"),
+ config_set("max_jobs", "4"),
+ mock_pending(100),
+
+ ?JOB_SERVER:reschedule(),
+
+ [A1, A2, A3] = acceptors(),
+ accept_job(A1, true),
+ accept_job(A2, false),
+ accept_job(A3, false),
+ [A4] = acceptors(),
+ accept_job(A4, true),
+
+ ?JOB_SERVER:reschedule(),
+
+ % First reschedule was to reset the churn budget, this next one is to spawn
+ % an excess number of acceptors.
+ ?JOB_SERVER:reschedule(),
+
+ [A5, A6, A7] = acceptors(),
+ accept_job(A5, true),
+ accept_job(A6, false),
+ accept_job(A7, false),
+
+ ?assertEqual(7, length(workers())),
+
+ % Running with an excess number of workers. But they won't be stopped on
+ % reschedule if they ran for a period less than min_run_time_sec during the
+ % next cycle
+ config_set("min_run_time_sec", "9999"),
+
+ % don't want to start new acceptors anymore
+ mock_pending(0),
+ config_set("max_acceptors", "0"),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual(7, length(workers())),
+ ?assertEqual(0, length(acceptors())),
+
+ config_set("min_run_time_sec", "0"),
+
+ ?JOB_SERVER:reschedule(),
+
+ ?assertEqual(4, length(workers())),
+ ?assertEqual(0, length(acceptors())).
+
+
+config_set(K, V) ->
+ config:set("replicator", K, V, _Persist = false).
+
+
+config_delete(K) ->
+ config:delete("replicator", K, _Persist = false).
+
+
+mock_pending(N) ->
+ meck:expect(couch_replicator_jobs, pending_count, 2, N).
+
+
+acceptors() ->
+ #{acceptors := Acceptors} = sys:get_state(?JOB_SERVER),
+ maps:keys(Acceptors).
+
+
+workers() ->
+ #{workers := Workers} = sys:get_state(?JOB_SERVER),
+ maps:keys(Workers).
+
+
+accept_job(APid, Normal) ->
+ APid ! {accept_job, Normal, self()},
+ receive
+ {job_accepted, APid} -> ok
+ after
+ 5000 ->
+ error(test_job_accept_timeout)
+ end.
+
+
+accept_all() ->
+ [accept_job(APid, true) || APid <- acceptors()].
+
+
+start_job() ->
+ receive
+ {accept_job, Normal, From} ->
+ ok = ?JOB_SERVER:accepted(self(), Normal),
+ From ! {job_accepted, self()},
+ start_job();
+ {exit_job, ExitSig} ->
+ exit(ExitSig)
+ end.
diff --git a/src/couch_replicator/test/eunit/couch_replicator_large_atts_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_large_atts_tests.erl
index 27c89a0..fcbdf22 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_large_atts_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_large_atts_tests.erl
@@ -14,12 +14,8 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
--import(couch_replicator_test_helper, [
- db_url/1,
- replicate/2,
- compare_dbs/2
-]).
-define(ATT_SIZE_1, 2 * 1024 * 1024).
-define(ATT_SIZE_2, round(6.6 * 1024 * 1024)).
@@ -27,90 +23,65 @@
-define(TIMEOUT_EUNIT, 120).
-setup() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
-
-setup(remote) ->
- {remote, setup()};
-setup({A, B}) ->
- Ctx = test_util:start_couch([couch_replicator]),
- config:set("attachments", "compressible_types", "text/*", false),
- Source = setup(A),
- Target = setup(B),
- {Ctx, {Source, Target}}.
-
-teardown({remote, DbName}) ->
- teardown(DbName);
-teardown(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]),
- ok.
-
-teardown(_, {Ctx, {Source, Target}}) ->
- teardown(Source),
- teardown(Target),
-
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
-
large_atts_test_() ->
- Pairs = [{remote, remote}],
{
- "Replicate docs with large attachments",
+ "Large attachment replication test",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_populate_replicate_compact/2}
- || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_replicate_attachments, 120)
+ ]
+ }
}
}.
-should_populate_replicate_compact({From, To}, {_Ctx, {Source, Target}}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [should_populate_source(Source),
- should_replicate(Source, Target),
- should_compare_databases(Source, Target)]}}.
+setup() ->
+ AttCfg = config:get("attachments", "compressible_types"),
+ config:set("attachments", "compressible_types", "text/*", false),
+ Source = couch_replicator_test_helper:create_db(),
+ ok = populate_db(Source, ?DOCS_COUNT),
+ Target = couch_replicator_test_helper:create_db(),
+ {AttCfg, Source, Target}.
+
-should_populate_source({remote, Source}) ->
- should_populate_source(Source);
-should_populate_source(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(populate_db(Source, ?DOCS_COUNT))}.
+teardown({AttCfg, Source, Target}) ->
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target),
+ case AttCfg of
+ undefined ->
+ config:delete("attachments", "compressible_types", false);
+ _ ->
+ config:set("attachments", "compressible_types", AttCfg)
+ end.
-should_replicate({remote, Source}, Target) ->
- should_replicate(db_url(Source), Target);
-should_replicate(Source, {remote, Target}) ->
- should_replicate(Source, db_url(Target));
-should_replicate(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(replicate(Source, Target))}.
-should_compare_databases({remote, Source}, Target) ->
- should_compare_databases(Source, Target);
-should_compare_databases(Source, {remote, Target}) ->
- should_compare_databases(Source, Target);
-should_compare_databases(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(compare_dbs(Source, Target))}.
+should_replicate_attachments({_AttCfg, Source, Target}) ->
+ ?assertMatch({ok, _},
+ couch_replicator_test_helper:replicate(Source, Target)),
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
populate_db(DbName, DocCount) ->
- {ok, Db} = couch_db:open_int(DbName, []),
- Docs = lists:foldl(
- fun(DocIdCounter, Acc) ->
- Doc = #doc{
- id = iolist_to_binary(["doc", integer_to_list(DocIdCounter)]),
- body = {[]},
- atts = [
- att(<<"att1">>, ?ATT_SIZE_1, <<"text/plain">>),
- att(<<"att2">>, ?ATT_SIZE_2, <<"app/binary">>)
- ]
- },
- [Doc | Acc]
- end,
- [], lists:seq(1, DocCount)),
- {ok, _} = couch_db:update_docs(Db, Docs, []),
- couch_db:close(Db).
+ Docs = lists:foldl(fun(DocIdCounter, Acc) ->
+ Doc = #doc{
+ id = iolist_to_binary(["doc", integer_to_list(DocIdCounter)]),
+ body = {[]},
+ atts = [
+ att(<<"att1">>, ?ATT_SIZE_1, <<"text/plain">>),
+ att(<<"att2">>, ?ATT_SIZE_2, <<"app/binary">>)
+ ]
+ },
+ [Doc | Acc]
+ end, [], lists:seq(1, DocCount)),
+ couch_replicator_test_helper:create_docs(DbName, Docs).
+
att(Name, Size, Type) ->
couch_att:new([
diff --git a/src/couch_replicator/test/eunit/couch_replicator_many_leaves_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_many_leaves_tests.erl
index c7933b4..3dbfa6a 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_many_leaves_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_many_leaves_tests.erl
@@ -14,11 +14,8 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
--import(couch_replicator_test_helper, [
- db_url/1,
- replicate/2
-]).
-define(DOCS_CONFLICTS, [
{<<"doc1">>, 10},
@@ -28,178 +25,150 @@
{<<"doc3">>, 210}
]).
-define(NUM_ATTS, 2).
--define(TIMEOUT_EUNIT, 60).
-define(i2l(I), integer_to_list(I)).
-define(io2b(Io), iolist_to_binary(Io)).
-setup() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
-
-
-setup(remote) ->
- {remote, setup()};
-setup({A, B}) ->
- Ctx = test_util:start_couch([couch_replicator]),
- Source = setup(A),
- Target = setup(B),
- {Ctx, {Source, Target}}.
-
-teardown({remote, DbName}) ->
- teardown(DbName);
-teardown(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]),
- ok.
-
-teardown(_, {Ctx, {Source, Target}}) ->
- teardown(Source),
- teardown(Target),
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
docs_with_many_leaves_test_() ->
- Pairs = [{remote, remote}],
{
"Replicate documents with many leaves",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_populate_replicate_compact/2}
- || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_replicate_doc_with_many_leaves, 180)
+ ]
+ }
}
}.
-should_populate_replicate_compact({From, To}, {_Ctx, {Source, Target}}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [
- should_populate_source(Source),
- should_replicate(Source, Target),
- should_verify_target(Source, Target),
- should_add_attachments_to_source(Source),
- should_replicate(Source, Target),
- should_verify_target(Source, Target)
- ]}}.
-
-should_populate_source({remote, Source}) ->
- should_populate_source(Source);
-should_populate_source(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(populate_db(Source))}.
-
-should_replicate({remote, Source}, Target) ->
- should_replicate(db_url(Source), Target);
-should_replicate(Source, {remote, Target}) ->
- should_replicate(Source, db_url(Target));
-should_replicate(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(replicate(Source, Target))}.
-
-should_verify_target({remote, Source}, Target) ->
- should_verify_target(Source, Target);
-should_verify_target(Source, {remote, Target}) ->
- should_verify_target(Source, Target);
-should_verify_target(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(begin
- {ok, SourceDb} = couch_db:open_int(Source, []),
- {ok, TargetDb} = couch_db:open_int(Target, []),
- verify_target(SourceDb, TargetDb, ?DOCS_CONFLICTS),
- ok = couch_db:close(SourceDb),
- ok = couch_db:close(TargetDb)
- end)}.
-
-should_add_attachments_to_source({remote, Source}) ->
- should_add_attachments_to_source(Source);
-should_add_attachments_to_source(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(begin
- {ok, SourceDb} = couch_db:open_int(Source, [?ADMIN_CTX]),
- add_attachments(SourceDb, ?NUM_ATTS, ?DOCS_CONFLICTS),
- ok = couch_db:close(SourceDb)
- end)}.
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ populate_db(Source),
+ Target = couch_replicator_test_helper:create_db(),
+ {Source, Target}.
+
+
+teardown({Source, Target}) ->
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
+
+
+should_replicate_doc_with_many_leaves({Source, Target}) ->
+ replicate(Source, Target),
+ {ok, SourceDb} = fabric2_db:open(Source, [?ADMIN_CTX]),
+ {ok, TargetDb} = fabric2_db:open(Target, [?ADMIN_CTX]),
+ verify_target(SourceDb, TargetDb, ?DOCS_CONFLICTS),
+ add_attachments(SourceDb, ?NUM_ATTS, ?DOCS_CONFLICTS),
+ replicate(Source, Target),
+ verify_target(SourceDb, TargetDb, ?DOCS_CONFLICTS).
+
populate_db(DbName) ->
- {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
- lists:foreach(
- fun({DocId, NumConflicts}) ->
- Value = <<"0">>,
- Doc = #doc{
- id = DocId,
- body = {[ {<<"value">>, Value} ]}
- },
- {ok, _} = couch_db:update_doc(Db, Doc, [?ADMIN_CTX]),
- {ok, _} = add_doc_siblings(Db, DocId, NumConflicts)
- end, ?DOCS_CONFLICTS),
- couch_db:close(Db).
-
-add_doc_siblings(Db, DocId, NumLeaves) when NumLeaves > 0 ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ lists:foreach(fun({DocId, NumConflicts}) ->
+ Doc = #doc{
+ id = DocId,
+ body = {[{<<"value">>, <<"0">>}]}
+ },
+ {ok, _} = fabric2_db:update_doc(Db, Doc),
+ {ok, _} = add_doc_siblings(Db, DocId, NumConflicts)
+ end, ?DOCS_CONFLICTS).
+
+
+add_doc_siblings(#{} = Db, DocId, NumLeaves) when NumLeaves > 0 ->
add_doc_siblings(Db, DocId, NumLeaves, [], []).
-add_doc_siblings(Db, _DocId, 0, AccDocs, AccRevs) ->
- {ok, []} = couch_db:update_docs(Db, AccDocs, [], replicated_changes),
+
+add_doc_siblings(#{} = Db, _DocId, 0, AccDocs, AccRevs) ->
+ {ok, []} = fabric2_db:update_docs(Db, AccDocs, [replicated_changes]),
{ok, AccRevs};
-add_doc_siblings(Db, DocId, NumLeaves, AccDocs, AccRevs) ->
+add_doc_siblings(#{} = Db, DocId, NumLeaves, AccDocs, AccRevs) ->
Value = ?l2b(?i2l(NumLeaves)),
Rev = couch_hash:md5_hash(Value),
Doc = #doc{
id = DocId,
revs = {1, [Rev]},
- body = {[ {<<"value">>, Value} ]}
+ body = {[{<<"value">>, Value}]}
},
add_doc_siblings(Db, DocId, NumLeaves - 1,
- [Doc | AccDocs], [{1, Rev} | AccRevs]).
+ [Doc | AccDocs], [{1, Rev} | AccRevs]).
+
verify_target(_SourceDb, _TargetDb, []) ->
ok;
-verify_target(SourceDb, TargetDb, [{DocId, NumConflicts} | Rest]) ->
- {ok, SourceLookups} = couch_db:open_doc_revs(
- SourceDb,
- DocId,
- all,
- [conflicts, deleted_conflicts]),
- {ok, TargetLookups} = couch_db:open_doc_revs(
- TargetDb,
- DocId,
- all,
- [conflicts, deleted_conflicts]),
+
+verify_target(#{} = SourceDb, #{} = TargetDb,
+ [{DocId, NumConflicts} | Rest]) ->
+ Opts = [conflicts, deleted_conflicts],
+ {ok, SourceLookups} = open_doc_revs(SourceDb, DocId, Opts),
+ {ok, TargetLookups} = open_doc_revs(TargetDb, DocId, Opts),
SourceDocs = [Doc || {ok, Doc} <- SourceLookups],
TargetDocs = [Doc || {ok, Doc} <- TargetLookups],
Total = NumConflicts + 1,
?assertEqual(Total, length(TargetDocs)),
- lists:foreach(
- fun({SourceDoc, TargetDoc}) ->
- SourceJson = couch_doc:to_json_obj(SourceDoc, [attachments]),
- TargetJson = couch_doc:to_json_obj(TargetDoc, [attachments]),
- ?assertEqual(SourceJson, TargetJson)
- end,
- lists:zip(SourceDocs, TargetDocs)),
+ lists:foreach(fun({SourceDoc, TargetDoc}) ->
+ ?assertEqual(json_doc(SourceDoc), json_doc(TargetDoc))
+ end, lists:zip(SourceDocs, TargetDocs)),
verify_target(SourceDb, TargetDb, Rest).
-add_attachments(_SourceDb, _NumAtts, []) ->
+
+add_attachments(_SourceDb, _NumAtts, []) ->
ok;
-add_attachments(SourceDb, NumAtts, [{DocId, NumConflicts} | Rest]) ->
- {ok, SourceLookups} = couch_db:open_doc_revs(SourceDb, DocId, all, []),
+
+add_attachments(#{} = SourceDb, NumAtts,
+ [{DocId, NumConflicts} | Rest]) ->
+ {ok, SourceLookups} = open_doc_revs(SourceDb, DocId, []),
SourceDocs = [Doc || {ok, Doc} <- SourceLookups],
Total = NumConflicts + 1,
?assertEqual(Total, length(SourceDocs)),
- NewDocs = lists:foldl(
- fun(#doc{atts = Atts, revs = {Pos, [Rev | _]}} = Doc, Acc) ->
+ NewDocs = lists:foldl(fun
+ (#doc{atts = Atts, revs = {Pos, [Rev | _]}} = Doc, Acc) ->
NewAtts = lists:foldl(fun(I, AttAcc) ->
- AttData = crypto:strong_rand_bytes(100),
- NewAtt = couch_att:new([
- {name, ?io2b(["att_", ?i2l(I), "_",
- couch_doc:rev_to_str({Pos, Rev})])},
- {type, <<"application/foobar">>},
- {att_len, byte_size(AttData)},
- {data, AttData}
- ]),
- [NewAtt | AttAcc]
+ [att(I, {Pos, Rev}, 100) | AttAcc]
end, [], lists:seq(1, NumAtts)),
[Doc#doc{atts = Atts ++ NewAtts} | Acc]
- end,
- [], SourceDocs),
- {ok, UpdateResults} = couch_db:update_docs(SourceDb, NewDocs, []),
- NewRevs = [R || {ok, R} <- UpdateResults],
- ?assertEqual(length(NewDocs), length(NewRevs)),
+ end, [], SourceDocs),
+ lists:foreach(fun(#doc{} = Doc) ->
+ ?assertMatch({ok, _}, fabric2_db:update_doc(SourceDb, Doc))
+ end, NewDocs),
add_attachments(SourceDb, NumAtts, Rest).
+
+att(I, PosRev, Size) ->
+ Name = ?io2b(["att_", ?i2l(I), "_", couch_doc:rev_to_str(PosRev)]),
+ AttData = crypto:strong_rand_bytes(Size),
+ couch_att:new([
+ {name, Name},
+ {type, <<"application/foobar">>},
+ {att_len, byte_size(AttData)},
+ {data, AttData}
+ ]).
+
+
+open_doc_revs(#{} = Db, DocId, Opts) ->
+ fabric2_db:open_doc_revs(Db, DocId, all, Opts).
+
+
+json_doc(#doc{} = Doc) ->
+ couch_doc:to_json_obj(Doc, [attachments]).
+
+
+replicate(Source, Target) ->
+ % Serialize the concurrent updates of the same document in order
+ % to prevent having to set higher timeouts due to FDB conflicts
+ RepObject = #{
+ <<"source">> => Source,
+ <<"target">> => Target,
+ <<"worker_processes">> => 1,
+ <<"http_connections">> => 1
+ },
+ ?assertMatch({ok, _},
+ couch_replicator_test_helper:replicate(RepObject)).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_missing_stubs_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_missing_stubs_tests.erl
index ff08b5e..e672c76 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_missing_stubs_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_missing_stubs_tests.erl
@@ -14,103 +14,59 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
--import(couch_replicator_test_helper, [
- db_url/1,
- replicate/2,
- compare_dbs/2
-]).
-define(REVS_LIMIT, 3).
--define(TIMEOUT_EUNIT, 30).
-setup() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
-
-setup(remote) ->
- {remote, setup()};
-setup({A, B}) ->
- Ctx = test_util:start_couch([couch_replicator]),
- Source = setup(A),
- Target = setup(B),
- {Ctx, {Source, Target}}.
-
-teardown({remote, DbName}) ->
- teardown(DbName);
-teardown(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]),
- ok.
-
-teardown(_, {Ctx, {Source, Target}}) ->
- teardown(Source),
- teardown(Target),
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
missing_stubs_test_() ->
- Pairs = [{remote, remote}],
{
"Replicate docs with missing stubs (COUCHDB-1365)",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_replicate_docs_with_missed_att_stubs/2}
- || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_replicate_docs_with_missed_att_stubs, 60)
+ ]
+ }
}
}.
-should_replicate_docs_with_missed_att_stubs({From, To}, {_Ctx, {Source, Target}}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [
- should_populate_source(Source),
- should_set_target_revs_limit(Target, ?REVS_LIMIT),
- should_replicate(Source, Target),
- should_compare_databases(Source, Target),
- should_update_source_docs(Source, ?REVS_LIMIT * 2),
- should_replicate(Source, Target),
- should_compare_databases(Source, Target)
- ]}}.
-
-should_populate_source({remote, Source}) ->
- should_populate_source(Source);
-should_populate_source(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(populate_db(Source))}.
-
-should_replicate({remote, Source}, Target) ->
- should_replicate(db_url(Source), Target);
-should_replicate(Source, {remote, Target}) ->
- should_replicate(Source, db_url(Target));
-should_replicate(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(replicate(Source, Target))}.
-
-should_set_target_revs_limit({remote, Target}, RevsLimit) ->
- should_set_target_revs_limit(Target, RevsLimit);
-should_set_target_revs_limit(Target, RevsLimit) ->
- ?_test(begin
- {ok, Db} = couch_db:open_int(Target, [?ADMIN_CTX]),
- ?assertEqual(ok, couch_db:set_revs_limit(Db, RevsLimit)),
- ok = couch_db:close(Db)
- end).
-
-should_compare_databases({remote, Source}, Target) ->
- should_compare_databases(Source, Target);
-should_compare_databases(Source, {remote, Target}) ->
- should_compare_databases(Source, Target);
-should_compare_databases(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(compare_dbs(Source, Target))}.
-
-should_update_source_docs({remote, Source}, Times) ->
- should_update_source_docs(Source, Times);
-should_update_source_docs(Source, Times) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(update_db_docs(Source, Times))}.
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ populate_db(Source),
+ Target = couch_replicator_test_helper:create_db(),
+ {Source, Target}.
+
+
+teardown({Source, Target}) ->
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
+
+
+should_replicate_docs_with_missed_att_stubs({Source, Target}) ->
+ {ok, TargetDb} = fabric2_db:open(Target, [?ADMIN_CTX]),
+ ?assertEqual(ok, fabric2_db:set_revs_limit(TargetDb, ?REVS_LIMIT)),
+
+ ?assertMatch({ok, _},
+ couch_replicator_test_helper:replicate(Source, Target)),
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)),
+
+ ok = update_db_docs(Source, ?REVS_LIMIT * 2),
+
+ ?assertMatch({ok, _},
+ couch_replicator_test_helper:replicate(Source, Target)),
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
populate_db(DbName) ->
- {ok, Db} = couch_db:open_int(DbName, []),
AttData = crypto:strong_rand_bytes(6000),
Doc = #doc{
id = <<"doc1">>,
@@ -120,35 +76,40 @@ populate_db(DbName) ->
{type, <<"application/foobar">>},
{att_len, byte_size(AttData)},
{data, AttData}
- ])
+ ])
]
},
- {ok, _} = couch_db:update_doc(Db, Doc, []),
- couch_db:close(Db).
+ couch_replicator_test_helper:create_docs(DbName, [Doc]).
+
update_db_docs(DbName, Times) ->
- {ok, Db} = couch_db:open_int(DbName, []),
- {ok, _} = couch_db:fold_docs(
- Db,
- fun(FDI, Acc) -> db_fold_fun(FDI, Acc) end,
- {DbName, Times},
- []),
- ok = couch_db:close(Db).
-
-db_fold_fun(FullDocInfo, {DbName, Times}) ->
- {ok, Db} = couch_db:open_int(DbName, []),
- {ok, Doc} = couch_db:open_doc(Db, FullDocInfo),
- lists:foldl(
- fun(_, {Pos, RevId}) ->
- {ok, Db2} = couch_db:reopen(Db),
- NewDocVersion = Doc#doc{
- revs = {Pos, [RevId]},
- body = {[{<<"value">>, base64:encode(crypto:strong_rand_bytes(100))}]}
- },
- {ok, NewRev} = couch_db:update_doc(Db2, NewDocVersion, []),
- NewRev
- end,
- {element(1, Doc#doc.revs), hd(element(2, Doc#doc.revs))},
- lists:seq(1, Times)),
- ok = couch_db:close(Db),
- {ok, {DbName, Times}}.
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ FoldFun = fun
+ ({meta, _Meta}, Acc) ->
+ {ok, Acc};
+ (complete, Acc) ->
+ {ok, Acc};
+ ({row, Row}, Acc) ->
+ {_, DocId} = lists:keyfind(id, 1, Row),
+ ok = update_doc(DbName, DocId, Times),
+ {ok, Acc}
+ end,
+ Opts = [{restart_tx, true}],
+ {ok, _} = fabric2_db:fold_docs(Db, FoldFun, ok, Opts),
+ ok.
+
+
+update_doc(_DbName, _DocId, 0) ->
+ ok;
+
+update_doc(DbName, DocId, Times) ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ {ok, Doc} = fabric2_db:open_doc(Db, DocId, []),
+ #doc{revs = {Pos, [Rev | _]}} = Doc,
+ Val = base64:encode(crypto:strong_rand_bytes(100)),
+ Doc1 = Doc#doc{
+ revs = {Pos, [Rev]},
+ body = {[{<<"value">>, Val}]}
+ },
+ {ok, _} = fabric2_db:update_doc(Db, Doc1),
+ update_doc(DbName, DocId, Times - 1).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_proxy_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_proxy_tests.erl
index da46b8a..f5e745d 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_proxy_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_proxy_tests.erl
@@ -14,15 +14,7 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
--include_lib("couch_replicator/include/couch_replicator_api_wrap.hrl").
-
-
-setup() ->
- ok.
-
-
-teardown(_) ->
- ok.
+-include_lib("fabric/test/fabric2_test.hrl").
replicator_proxy_test_() ->
@@ -30,87 +22,78 @@ replicator_proxy_test_() ->
"replicator proxy tests",
{
setup,
- fun() -> test_util:start_couch([couch_replicator]) end, fun test_util:stop_couch/1,
- {
- foreach,
- fun setup/0, fun teardown/1,
- [
- fun parse_rep_doc_without_proxy/1,
- fun parse_rep_doc_with_proxy/1,
- fun parse_rep_source_target_proxy/1,
- fun mutually_exclusive_proxy_and_source_proxy/1,
- fun mutually_exclusive_proxy_and_target_proxy/1
- ]
- }
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ with([
+ ?TDEF(parse_rep_doc_without_proxy),
+ ?TDEF(parse_rep_doc_with_proxy),
+ ?TDEF(parse_rep_source_target_proxy),
+ ?TDEF(mutually_exclusive_proxy_and_source_proxy),
+ ?TDEF(mutually_exclusive_proxy_and_target_proxy)
+ ])
}
}.
parse_rep_doc_without_proxy(_) ->
- ?_test(begin
- NoProxyDoc = {[
- {<<"source">>, <<"http://unproxied.com">>},
- {<<"target">>, <<"http://otherunproxied.com">>}
- ]},
- Rep = couch_replicator_docs:parse_rep_doc(NoProxyDoc),
- ?assertEqual((Rep#rep.source)#httpdb.proxy_url, undefined),
- ?assertEqual((Rep#rep.target)#httpdb.proxy_url, undefined)
- end).
+ NoProxyDoc = {[
+ {<<"source">>, <<"http://unproxied.com">>},
+ {<<"target">>, <<"http://otherunproxied.com">>}
+ ]},
+ Rep = couch_replicator_parse:parse_rep_doc(NoProxyDoc),
+ Src = maps:get(?SOURCE, Rep),
+ Tgt = maps:get(?TARGET, Rep),
+ ?assertEqual(null, maps:get(<<"proxy_url">>, Src)),
+ ?assertEqual(null, maps:get(<<"proxy_url">>, Tgt)).
parse_rep_doc_with_proxy(_) ->
- ?_test(begin
- ProxyURL = <<"http://myproxy.com">>,
- ProxyDoc = {[
- {<<"source">>, <<"http://unproxied.com">>},
- {<<"target">>, <<"http://otherunproxied.com">>},
- {<<"proxy">>, ProxyURL}
- ]},
- Rep = couch_replicator_docs:parse_rep_doc(ProxyDoc),
- ?assertEqual((Rep#rep.source)#httpdb.proxy_url, binary_to_list(ProxyURL)),
- ?assertEqual((Rep#rep.target)#httpdb.proxy_url, binary_to_list(ProxyURL))
- end).
+ ProxyURL = <<"http://myproxy.com">>,
+ ProxyDoc = {[
+ {<<"source">>, <<"http://unproxied.com">>},
+ {<<"target">>, <<"http://otherunproxied.com">>},
+ {<<"proxy">>, ProxyURL}
+ ]},
+ Rep = couch_replicator_parse:parse_rep_doc(ProxyDoc),
+ Src = maps:get(?SOURCE, Rep),
+ Tgt = maps:get(?TARGET, Rep),
+ ?assertEqual(ProxyURL, maps:get(<<"proxy_url">>, Src)),
+ ?assertEqual(ProxyURL, maps:get(<<"proxy_url">>, Tgt)).
parse_rep_source_target_proxy(_) ->
- ?_test(begin
- SrcProxyURL = <<"http://mysrcproxy.com">>,
- TgtProxyURL = <<"http://mytgtproxy.com:9999">>,
- ProxyDoc = {[
- {<<"source">>, <<"http://unproxied.com">>},
- {<<"target">>, <<"http://otherunproxied.com">>},
- {<<"source_proxy">>, SrcProxyURL},
- {<<"target_proxy">>, TgtProxyURL}
- ]},
- Rep = couch_replicator_docs:parse_rep_doc(ProxyDoc),
- ?assertEqual((Rep#rep.source)#httpdb.proxy_url,
- binary_to_list(SrcProxyURL)),
- ?assertEqual((Rep#rep.target)#httpdb.proxy_url,
- binary_to_list(TgtProxyURL))
- end).
+ SrcProxyURL = <<"http://mysrcproxy.com">>,
+ TgtProxyURL = <<"http://mytgtproxy.com:9999">>,
+ ProxyDoc = {[
+ {<<"source">>, <<"http://unproxied.com">>},
+ {<<"target">>, <<"http://otherunproxied.com">>},
+ {<<"source_proxy">>, SrcProxyURL},
+ {<<"target_proxy">>, TgtProxyURL}
+ ]},
+ Rep = couch_replicator_parse:parse_rep_doc(ProxyDoc),
+ Src = maps:get(?SOURCE, Rep),
+ Tgt = maps:get(?TARGET, Rep),
+ ?assertEqual(SrcProxyURL, maps:get(<<"proxy_url">>, Src)),
+ ?assertEqual(TgtProxyURL, maps:get(<<"proxy_url">>, Tgt)).
mutually_exclusive_proxy_and_source_proxy(_) ->
- ?_test(begin
- ProxyDoc = {[
- {<<"source">>, <<"http://unproxied.com">>},
- {<<"target">>, <<"http://otherunproxied.com">>},
- {<<"proxy">>, <<"oldstyleproxy.local">>},
- {<<"source_proxy">>, <<"sourceproxy.local">>}
- ]},
- ?assertThrow({bad_rep_doc, _},
- couch_replicator_docs:parse_rep_doc(ProxyDoc))
- end).
+ ProxyDoc = {[
+ {<<"source">>, <<"http://unproxied.com">>},
+ {<<"target">>, <<"http://otherunproxied.com">>},
+ {<<"proxy">>, <<"oldstyleproxy.local">>},
+ {<<"source_proxy">>, <<"sourceproxy.local">>}
+ ]},
+ ?assertThrow({bad_rep_doc, _},
+ couch_replicator_parse:parse_rep_doc(ProxyDoc)).
mutually_exclusive_proxy_and_target_proxy(_) ->
- ?_test(begin
- ProxyDoc = {[
- {<<"source">>, <<"http://unproxied.com">>},
- {<<"target">>, <<"http://otherunproxied.com">>},
- {<<"proxy">>, <<"oldstyleproxy.local">>},
- {<<"target_proxy">>, <<"targetproxy.local">>}
- ]},
- ?assertThrow({bad_rep_doc, _},
- couch_replicator_docs:parse_rep_doc(ProxyDoc))
- end).
+ ProxyDoc = {[
+ {<<"source">>, <<"http://unproxied.com">>},
+ {<<"target">>, <<"http://otherunproxied.com">>},
+ {<<"proxy">>, <<"oldstyleproxy.local">>},
+ {<<"target_proxy">>, <<"targetproxy.local">>}
+ ]},
+ ?assertThrow({bad_rep_doc, _},
+ couch_replicator_parse:parse_rep_doc(ProxyDoc)).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_rate_limiter_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_rate_limiter_tests.erl
index 034550a..fb98920 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_rate_limiter_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_rate_limiter_tests.erl
@@ -1,6 +1,7 @@
-module(couch_replicator_rate_limiter_tests).
-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
rate_limiter_test_() ->
@@ -9,64 +10,52 @@ rate_limiter_test_() ->
fun setup/0,
fun teardown/1,
[
- t_new_key(),
- t_1_failure(),
- t_2_failures_back_to_back(),
- t_2_failures(),
- t_success_threshold(),
- t_1_failure_2_successes()
+ ?TDEF_FE(t_new_key),
+ ?TDEF_FE(t_1_failure),
+ ?TDEF_FE(t_2_failures_back_to_back),
+ ?TDEF_FE(t_2_failures),
+ ?TDEF_FE(t_success_threshold),
+ ?TDEF_FE(t_1_failure_2_successes)
]
}.
-t_new_key() ->
- ?_test(begin
- ?assertEqual(0, couch_replicator_rate_limiter:interval({"foo", get}))
- end).
+t_new_key(_) ->
+ ?assertEqual(0, couch_replicator_rate_limiter:interval({"foo", get})).
-t_1_failure() ->
- ?_test(begin
- ?assertEqual(24, couch_replicator_rate_limiter:failure({"foo", get}))
- end).
+t_1_failure(_) ->
+ ?assertEqual(24, couch_replicator_rate_limiter:failure({"foo", get})).
-t_2_failures() ->
- ?_test(begin
- couch_replicator_rate_limiter:failure({"foo", get}),
- low_pass_filter_delay(),
- Interval = couch_replicator_rate_limiter:failure({"foo", get}),
- ?assertEqual(29, Interval)
- end).
+t_2_failures(_) ->
+ couch_replicator_rate_limiter:failure({"foo", get}),
+ low_pass_filter_delay(),
+ Interval = couch_replicator_rate_limiter:failure({"foo", get}),
+ ?assertEqual(29, Interval).
-t_2_failures_back_to_back() ->
- ?_test(begin
- couch_replicator_rate_limiter:failure({"foo", get}),
- Interval = couch_replicator_rate_limiter:failure({"foo", get}),
- ?assertEqual(24, Interval)
- end).
+t_2_failures_back_to_back(_) ->
+ couch_replicator_rate_limiter:failure({"foo", get}),
+ Interval = couch_replicator_rate_limiter:failure({"foo", get}),
+ ?assertEqual(24, Interval).
-t_success_threshold() ->
- ?_test(begin
- Interval = couch_replicator_rate_limiter:success({"foo", get}),
- ?assertEqual(0, Interval),
- Interval = couch_replicator_rate_limiter:success({"foo", get}),
- ?assertEqual(0, Interval)
- end).
+t_success_threshold(_) ->
+ Interval = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(0, Interval),
+ Interval = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(0, Interval).
-t_1_failure_2_successes() ->
- ?_test(begin
- couch_replicator_rate_limiter:failure({"foo", get}),
- low_pass_filter_delay(),
- Succ1 = couch_replicator_rate_limiter:success({"foo", get}),
- ?assertEqual(20, Succ1),
- low_pass_filter_delay(),
- Succ2 = couch_replicator_rate_limiter:success({"foo", get}),
- ?assertEqual(0, Succ2)
- end).
+t_1_failure_2_successes(_) ->
+ couch_replicator_rate_limiter:failure({"foo", get}),
+ low_pass_filter_delay(),
+ Succ1 = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(20, Succ1),
+ low_pass_filter_delay(),
+ Succ2 = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(0, Succ2).
low_pass_filter_delay() ->
diff --git a/src/couch_replicator/test/eunit/couch_replicator_retain_stats_between_job_runs.erl b/src/couch_replicator/test/eunit/couch_replicator_retain_stats_between_job_runs.erl
index 037f371..4b7c37d 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_retain_stats_between_job_runs.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_retain_stats_between_job_runs.erl
@@ -15,139 +15,72 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
-
--define(DELAY, 500).
--define(TIMEOUT, 60000).
-
-
-setup_all() ->
- test_util:start_couch([couch_replicator, chttpd, mem3, fabric]).
-
-
-teardown_all(Ctx) ->
- ok = test_util:stop_couch(Ctx).
-
-
-setup() ->
- Source = setup_db(),
- Target = setup_db(),
- {Source, Target}.
+-include_lib("fabric/test/fabric2_test.hrl").
-teardown({Source, Target}) ->
- teardown_db(Source),
- teardown_db(Target),
- ok.
+-define(DELAY, 500).
stats_retained_test_() ->
{
setup,
- fun setup_all/0,
- fun teardown_all/1,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
{
foreach,
fun setup/0,
fun teardown/1,
[
- fun t_stats_retained_by_scheduler/1,
- fun t_stats_retained_on_job_removal/1
+ ?TDEF_FE(t_stats_retained_on_job_removal, 60)
]
}
}.
-t_stats_retained_by_scheduler({Source, Target}) ->
- ?_test(begin
- {ok, _} = add_vdu(Target),
- populate_db_reject_even_docs(Source, 1, 10),
- {ok, RepPid, RepId} = replicate(Source, Target),
- wait_target_in_sync(6, Target),
-
- check_active_tasks(10, 5, 5),
- check_scheduler_jobs(10, 5, 5),
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ Target = couch_replicator_test_helper:create_db(),
+ config:set("replicator", "stats_update_interval_sec", "0", false),
+ config:set("replicator", "checkpoint_interval", "1000", false),
+ {Source, Target}.
- stop_job(RepPid),
- check_scheduler_jobs(10, 5, 5),
- start_job(),
- check_active_tasks(10, 5, 5),
- check_scheduler_jobs(10, 5, 5),
- couch_replicator_scheduler:remove_job(RepId)
- end).
+teardown({Source, Target}) ->
+ config:delete("replicator", "stats_update_interval_sec", false),
+ config:delete("replicator", "checkpoint_interval", false),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
t_stats_retained_on_job_removal({Source, Target}) ->
- ?_test(begin
- {ok, _} = add_vdu(Target),
- populate_db_reject_even_docs(Source, 1, 10),
- {ok, _, RepId} = replicate(Source, Target),
- wait_target_in_sync(6, Target), % 5 + 1 vdu
-
- check_active_tasks(10, 5, 5),
- check_scheduler_jobs(10, 5, 5),
+ {ok, _} = add_vdu(Target),
+ populate_db_reject_even_docs(Source, 1, 10),
+ {ok, Pid1, RepId} = replicate(Source, Target),
+ wait_target_in_sync(6, Target), % 5 + 1 vdu
- couch_replicator_scheduler:remove_job(RepId),
+ check_scheduler_jobs(10, 5, 5),
- populate_db_reject_even_docs(Source, 11, 20),
- {ok, _, RepId} = replicate(Source, Target),
- wait_target_in_sync(11, Target), % 6 + 5
+ cancel(RepId, Pid1),
- check_scheduler_jobs(20, 10, 10),
- check_active_tasks(20, 10, 10),
+ populate_db_reject_even_docs(Source, 11, 20),
+ {ok, Pid2, RepId} = replicate(Source, Target),
+ wait_target_in_sync(11, Target), % 6 + 5
- couch_replicator_scheduler:remove_job(RepId),
+ check_scheduler_jobs(20, 10, 10),
- populate_db_reject_even_docs(Source, 21, 30),
- {ok, _, RepId} = replicate(Source, Target),
- wait_target_in_sync(16, Target), % 11 + 5
+ cancel(RepId, Pid2),
- check_scheduler_jobs(30, 15, 15),
- check_active_tasks(30, 15, 15),
-
- couch_replicator_scheduler:remove_job(RepId)
- end).
+ populate_db_reject_even_docs(Source, 21, 30),
+ {ok, Pid3, RepId} = replicate(Source, Target),
+ wait_target_in_sync(16, Target), % 11 + 5
+ check_scheduler_jobs(30, 15, 15),
-setup_db() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
-
-
-teardown_db(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]),
- ok.
-
-
-stop_job(RepPid) ->
- Ref = erlang:monitor(process, RepPid),
- gen_server:cast(couch_replicator_scheduler, {set_max_jobs, 0}),
- couch_replicator_scheduler:reschedule(),
- receive
- {'DOWN', Ref, _, _, _} -> ok
- after ?TIMEOUT ->
- erlang:error(timeout)
- end.
-
-
-start_job() ->
- gen_server:cast(couch_replicator_scheduler, {set_max_jobs, 500}),
- couch_replicator_scheduler:reschedule().
-
-
-check_active_tasks(DocsRead, DocsWritten, DocsFailed) ->
- RepTask = wait_for_task_status(),
- ?assertNotEqual(timeout, RepTask),
- ?assertEqual(DocsRead, couch_util:get_value(docs_read, RepTask)),
- ?assertEqual(DocsWritten, couch_util:get_value(docs_written, RepTask)),
- ?assertEqual(DocsFailed, couch_util:get_value(doc_write_failures,
- RepTask)).
+ cancel(RepId, Pid3).
check_scheduler_jobs(DocsRead, DocsWritten, DocFailed) ->
- Info = wait_scheduler_info(),
+ Info = wait_scheduler_info(DocsRead),
?assert(maps:is_key(<<"changes_pending">>, Info)),
?assert(maps:is_key(<<"doc_write_failures">>, Info)),
?assert(maps:is_key(<<"docs_read">>, Info)),
@@ -161,27 +94,18 @@ check_scheduler_jobs(DocsRead, DocsWritten, DocFailed) ->
?assertMatch(#{<<"doc_write_failures">> := DocFailed}, Info).
-replication_tasks() ->
- lists:filter(fun(P) ->
- couch_util:get_value(type, P) =:= replication
- end, couch_task_status:all()).
-
-
-wait_for_task_status() ->
+wait_scheduler_info(DocsRead) ->
test_util:wait(fun() ->
- case replication_tasks() of
- [] -> wait;
- [RepTask] -> RepTask
- end
- end).
-
-
-wait_scheduler_info() ->
- test_util:wait(fun() ->
- case scheduler_jobs() of
- [] -> wait;
- [#{<<"info">> := null}] -> wait;
- [#{<<"info">> := Info}] -> Info
+ case couch_replicator_test_helper:scheduler_jobs() of
+ [] ->
+ wait;
+ [#{<<"info">> := null}] ->
+ wait;
+ [#{<<"info">> := Info}] ->
+ case Info of
+ #{<<"docs_read">> := DocsRead} -> Info;
+ #{} -> wait
+ end
end
end).
@@ -197,16 +121,12 @@ populate_db_reject_even_docs(DbName, Start, End) ->
populate_db(DbName, Start, End, BodyFun) when is_function(BodyFun, 1) ->
- {ok, Db} = couch_db:open_int(DbName, []),
- Docs = lists:foldl(
- fun(DocIdCounter, Acc) ->
- Id = integer_to_binary(DocIdCounter),
- Doc = #doc{id = Id, body = BodyFun(DocIdCounter)},
- [Doc | Acc]
- end,
- [], lists:seq(Start, End)),
- {ok, _} = couch_db:update_docs(Db, Docs, []),
- ok = couch_db:close(Db).
+ Docs = lists:foldl(fun(DocIdCounter, Acc) ->
+ Id = integer_to_binary(DocIdCounter),
+ Doc = #doc{id = Id, body = BodyFun(DocIdCounter)},
+ [Doc | Acc]
+ end, [], lists:seq(Start, End)),
+ couch_replicator_test_helper:create_docs(DbName, Docs).
wait_target_in_sync(DocCount, Target) when is_integer(DocCount) ->
@@ -215,14 +135,13 @@ wait_target_in_sync(DocCount, Target) when is_integer(DocCount) ->
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"}
+ {module, ?MODULE}, {line, ?LINE},
+ {reason, "Could not get source and target databases in sync"}
]});
wait_target_in_sync_loop(DocCount, TargetName, RetriesLeft) ->
- {ok, Target} = couch_db:open_int(TargetName, []),
- {ok, TargetInfo} = couch_db:get_db_info(Target),
- ok = couch_db:close(Target),
+ {ok, Db} = fabric2_db:open(TargetName, [?ADMIN_CTX]),
+ {ok, TargetInfo} = fabric2_db:get_db_info(Db),
TargetDocCount = couch_util:get_value(doc_count, TargetInfo),
case TargetDocCount == DocCount of
true ->
@@ -234,27 +153,11 @@ wait_target_in_sync_loop(DocCount, TargetName, RetriesLeft) ->
replicate(Source, Target) ->
- SrcUrl = couch_replicator_test_helper:db_url(Source),
- TgtUrl = couch_replicator_test_helper:db_url(Target),
- RepObject = {[
- {<<"source">>, SrcUrl},
- {<<"target">>, TgtUrl},
- {<<"continuous">>, true}
- ]},
- {ok, Rep} = couch_replicator_utils:parse_rep_doc(RepObject, ?ADMIN_USER),
- ok = couch_replicator_scheduler:add_job(Rep),
- couch_replicator_scheduler:reschedule(),
- Pid = couch_replicator_test_helper:get_pid(Rep#rep.id),
- {ok, Pid, Rep#rep.id}.
-
-
-scheduler_jobs() ->
- Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
- Port = mochiweb_socket_server:get(chttpd, port),
- Url = lists:flatten(io_lib:format("http://~s:~b/_scheduler/jobs", [Addr, Port])),
- {ok, 200, _, Body} = test_request:get(Url, []),
- Json = jiffy:decode(Body, [return_maps]),
- maps:get(<<"jobs">>, Json).
+ couch_replicator_test_helper:replicate_continuous(Source, Target).
+
+
+cancel(RepId, Pid) ->
+ couch_replicator_test_helper:cancel(RepId, Pid).
vdu() ->
@@ -274,9 +177,5 @@ add_vdu(DbName) ->
{<<"validate_doc_update">>, vdu()}
],
Doc = couch_doc:from_json_obj({DocProps}, []),
- {ok, Db} = couch_db:open_int(DbName, [?ADMIN_CTX]),
- try
- {ok, _Rev} = couch_db:update_doc(Db, Doc, [])
- after
- couch_db:close(Db)
- end.
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ {ok, _} = fabric2_db:update_doc(Db, Doc, []).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_selector_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_selector_tests.erl
index 5026c14..5dfe4ba 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_selector_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_selector_tests.erl
@@ -15,103 +15,69 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_replicator/src/couch_replicator.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
-setup(_) ->
- Ctx = test_util:start_couch([couch_replicator]),
- Source = create_db(),
- create_docs(Source),
- Target = create_db(),
- {Ctx, {Source, Target}}.
-
-teardown(_, {Ctx, {Source, Target}}) ->
- delete_db(Source),
- delete_db(Target),
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
-
selector_replication_test_() ->
- Pairs = [{remote, remote}],
{
"Selector filtered replication tests",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_succeed/2} || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_replicate_with_selector)
+ ]
+ }
}
}.
-should_succeed({From, To}, {_Ctx, {Source, Target}}) ->
- RepObject = {[
- {<<"source">>, db_url(From, Source)},
- {<<"target">>, db_url(To, Target)},
- {<<"selector">>, {[{<<"_id">>, <<"doc2">>}]}}
- ]},
- {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
- %% FilteredFun is an Erlang version of following mango selector
- FilterFun = fun(_DocId, {Props}) ->
- couch_util:get_value(<<"_id">>, Props) == <<"doc2">>
- end,
- {ok, TargetDbInfo, AllReplies} = compare_dbs(Source, Target, FilterFun),
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])), [
- {"Target DB has proper number of docs",
- ?_assertEqual(1, proplists:get_value(doc_count, TargetDbInfo))},
- {"All the docs selected as expected",
- ?_assert(lists:all(fun(Valid) -> Valid end, AllReplies))}
- ]}.
-compare_dbs(Source, Target, FilterFun) ->
- {ok, SourceDb} = couch_db:open_int(Source, []),
- {ok, TargetDb} = couch_db:open_int(Target, []),
- {ok, TargetDbInfo} = couch_db:get_db_info(TargetDb),
- Fun = fun(FullDocInfo, Acc) ->
- {ok, DocId, SourceDoc} = read_doc(SourceDb, FullDocInfo),
- TargetReply = read_doc(TargetDb, DocId),
- case FilterFun(DocId, SourceDoc) of
- true ->
- ValidReply = {ok, DocId, SourceDoc} == TargetReply,
- {ok, [ValidReply|Acc]};
- false ->
- ValidReply = {not_found, missing} == TargetReply,
- {ok, [ValidReply|Acc]}
- end
- end,
- {ok, AllReplies} = couch_db:fold_docs(SourceDb, Fun, [], []),
- ok = couch_db:close(SourceDb),
- ok = couch_db:close(TargetDb),
- {ok, TargetDbInfo, AllReplies}.
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ create_docs(Source),
+ Target = couch_replicator_test_helper:create_db(),
+ {Source, Target}.
-read_doc(Db, DocIdOrInfo) ->
- case couch_db:open_doc(Db, DocIdOrInfo) of
- {ok, Doc} ->
- {Props} = couch_doc:to_json_obj(Doc, [attachments]),
- DocId = couch_util:get_value(<<"_id">>, Props),
- {ok, DocId, {Props}};
- Error ->
- Error
- end.
-create_db() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
+teardown({Source, Target}) ->
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
-create_docs(DbName) ->
- {ok, Db} = couch_db:open(DbName, [?ADMIN_CTX]),
- Doc1 = couch_doc:from_json_obj({[
- {<<"_id">>, <<"doc1">>}
- ]}),
- Doc2 = couch_doc:from_json_obj({[
- {<<"_id">>, <<"doc2">>}
- ]}),
- {ok, _} = couch_db:update_docs(Db, [Doc1, Doc2]),
- couch_db:close(Db).
-delete_db(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]).
+should_replicate_with_selector({Source, Target}) ->
+ RepObject = #{
+ <<"source">> => Source,
+ <<"target">> => Target,
+ <<"selector">> => #{
+ <<"_id">> => <<"doc2">>
+ }
+ },
+ ?assertMatch({ok, _}, couch_replicator_test_helper:replicate(RepObject)),
+ {ok, TargetDbInfo, AllReplies} = compare_dbs(Source, Target),
+ ?assertEqual(1, proplists:get_value(doc_count, TargetDbInfo)),
+ ?assert(lists:all(fun(Valid) -> Valid end, AllReplies)).
+
-db_url(remote, DbName) ->
- Addr = config:get("httpd", "bind_address", "127.0.0.1"),
- Port = mochiweb_socket_server:get(couch_httpd, port),
- ?l2b(io_lib:format("http://~s:~b/~s", [Addr, Port, DbName])).
+compare_dbs(Source, Target) ->
+ {ok, TargetDb} = fabric2_db:open(Target, []),
+ {ok, TargetDbInfo} = fabric2_db:get_db_info(TargetDb),
+ Fun = fun(SrcDoc, TgtDoc, Acc) ->
+ case SrcDoc#doc.id == <<"doc2">> of
+ true -> [SrcDoc#doc.body == TgtDoc#doc.body | Acc];
+ false -> [not_found == TgtDoc | Acc]
+ end
+ end,
+ Res = couch_replicator_test_helper:compare_fold(Source, Target, Fun, []),
+ {ok, TargetDbInfo, Res}.
+
+
+create_docs(DbName) ->
+ couch_replicator_test_helper:create_docs(DbName, [
+ #{<<"_id">> => <<"doc1">>},
+ #{<<"_id">> => <<"doc2">>}
+ ]).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_small_max_request_size_target.erl b/src/couch_replicator/test/eunit/couch_replicator_small_max_request_size_target.erl
index 8aebbe1..b113c53 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_small_max_request_size_target.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_small_max_request_size_target.erl
@@ -1,139 +1,70 @@
-module(couch_replicator_small_max_request_size_target).
+
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
-
--import(couch_replicator_test_helper, [
- db_url/1,
- replicate/1,
- compare_dbs/3
-]).
-
--define(TIMEOUT_EUNIT, 360).
-
-
-setup() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
-
-
-setup(remote) ->
- {remote, setup()};
-
-setup({A, B}) ->
- Ctx = test_util:start_couch([couch_replicator]),
- config:set("httpd", "max_http_request_size", "10000", false),
- Source = setup(A),
- Target = setup(B),
- {Ctx, {Source, Target}}.
-
-
-teardown({remote, DbName}) ->
- teardown(DbName);
-teardown(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]),
- ok.
-
-teardown(_, {Ctx, {Source, Target}}) ->
- teardown(Source),
- teardown(Target),
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
+-include_lib("fabric/test/fabric2_test.hrl").
reduce_max_request_size_test_() ->
- Pairs = [{remote, remote}],
{
"Replicate docs when target has a small max_http_request_size",
{
- foreachx,
- fun setup/1, fun teardown/2,
- [{Pair, fun should_replicate_all_docs/2}
- || Pair <- Pairs]
- ++ [{Pair, fun should_replicate_one/2}
- || Pair <- Pairs]
- % Disabled. See issue 574. Sometimes PUTs with a doc and
- % attachment which exceed maximum request size are simply
- % closed instead of returning a 413 request. That makes these
- % tests flaky.
- ++ [{Pair, fun should_replicate_one_with_attachment/2}
- || Pair <- Pairs]
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(should_replicate_all_docs, 120),
+ ?TDEF_FE(should_replicate_one, 120),
+ ?TDEF_FE(should_replicate_one_with_attachment, 120)
+ ]
+ }
}
}.
-% Test documents which are below max_http_request_size but when batched, batch size
-% will be greater than max_http_request_size. Replicator could automatically split
-% the batch into smaller batches and POST those separately.
-should_replicate_all_docs({From, To}, {_Ctx, {Source, Target}}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [should_populate_source(Source),
- should_replicate(Source, Target),
- should_compare_databases(Source, Target, [])]}}.
-
-
-% If a document is too large to post as a single request, that document is
-% skipped but replication overall will make progress and not crash.
-should_replicate_one({From, To}, {_Ctx, {Source, Target}}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [should_populate_source_one_large_one_small(Source),
- should_replicate(Source, Target),
- should_compare_databases(Source, Target, [<<"doc0">>])]}}.
-
-
-% If a document has an attachment > 64 * 1024 bytes, replicator will switch to
-% POST-ing individual documents directly and skip bulk_docs. Test that case
-% separately
-% See note in main test function why this was disabled.
-should_replicate_one_with_attachment({From, To}, {_Ctx, {Source, Target}}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [should_populate_source_one_large_attachment(Source),
- should_populate_source(Source),
- should_replicate(Source, Target),
- should_compare_databases(Source, Target, [<<"doc0">>])]}}.
-
-
-should_populate_source({remote, Source}) ->
- should_populate_source(Source);
-
-should_populate_source(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(add_docs(Source, 5, 3000, 0))}.
-
-
-should_populate_source_one_large_one_small({remote, Source}) ->
- should_populate_source_one_large_one_small(Source);
-
-should_populate_source_one_large_one_small(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(one_large_one_small(Source, 12000, 3000))}.
-
-
-should_populate_source_one_large_attachment({remote, Source}) ->
- should_populate_source_one_large_attachment(Source);
-
-should_populate_source_one_large_attachment(Source) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(one_large_attachment(Source, 70000, 70000))}.
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ Target = couch_replicator_test_helper:create_db(),
+ config:set("httpd", "max_http_request_size", "10000", false),
+ {Source, Target}.
-should_replicate({remote, Source}, Target) ->
- should_replicate(db_url(Source), Target);
+teardown({Source, Target}) ->
+ config:delete("httpd", "max_http_request_size", false),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
-should_replicate(Source, {remote, Target}) ->
- should_replicate(Source, db_url(Target));
-should_replicate(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(replicate(Source, Target))}.
+% Test documents which are below max_http_request_size but when batched, batch
+% size will be greater than max_http_request_size. Replicator could
+% automatically split the batch into smaller batches and POST those separately.
+should_replicate_all_docs({Source, Target}) ->
+ ?assertEqual(ok, add_docs(Source, 5, 3000, 0)),
+ replicate(Source, Target),
+ compare_dbs(Source, Target, []).
-should_compare_databases({remote, Source}, Target, ExceptIds) ->
- should_compare_databases(Source, Target, ExceptIds);
+% If a document is too large to post as a single request, that document is
+% skipped but replication overall will make progress and not crash.
+should_replicate_one({Source, Target}) ->
+ ?assertEqual(ok, one_large_one_small(Source, 12000, 3000)),
+ replicate(Source, Target),
+ compare_dbs(Source, Target, [<<"doc0">>]).
-should_compare_databases(Source, {remote, Target}, ExceptIds) ->
- should_compare_databases(Source, Target, ExceptIds);
-should_compare_databases(Source, Target, ExceptIds) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(compare_dbs(Source, Target, ExceptIds))}.
+% If a document has an attachment > 64 * 1024 bytes, replicator will switch to
+% POST-ing individual documents directly and skip bulk_docs. Test that case
+% separately See note in main test function why this was disabled.
+should_replicate_one_with_attachment({Source, Target}) ->
+ ?assertEqual(ok, one_large_attachment(Source, 70000, 70000)),
+ ?assertEqual(ok, add_docs(Source, 5, 3000, 0)),
+ replicate(Source, Target),
+ compare_dbs(Source, Target, [<<"doc0">>]).
binary_chunk(Size) when is_integer(Size), Size > 0 ->
@@ -150,19 +81,21 @@ add_docs(DbName, DocCount, DocSize, AttSize) ->
one_large_one_small(DbName, Large, Small) ->
add_doc(DbName, <<"doc0">>, Large, 0),
- add_doc(DbName, <<"doc1">>, Small, 0).
+ add_doc(DbName, <<"doc1">>, Small, 0),
+ ok.
one_large_attachment(DbName, Size, AttSize) ->
- add_doc(DbName, <<"doc0">>, Size, AttSize).
+ add_doc(DbName, <<"doc0">>, Size, AttSize),
+ ok.
add_doc(DbName, DocId, Size, AttSize) when is_binary(DocId) ->
- {ok, Db} = couch_db:open_int(DbName, []),
- Doc0 = #doc{id = DocId, body = {[{<<"x">>, binary_chunk(Size)}]}},
- Doc = Doc0#doc{atts = atts(AttSize)},
- {ok, _} = couch_db:update_doc(Db, Doc, []),
- couch_db:close(Db).
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ Doc0 = #doc{id = DocId, body = {[{<<"x">>, binary_chunk(Size)}]}},
+ Doc = Doc0#doc{atts = atts(AttSize)},
+ {ok, _} = fabric2_db:update_doc(Db, Doc, []),
+ ok.
atts(0) ->
@@ -178,8 +111,13 @@ atts(Size) ->
replicate(Source, Target) ->
- replicate({[
- {<<"source">>, Source},
- {<<"target">>, Target},
- {<<"worker_processes">>, "1"} % This make batch_size predictable
- ]}).
+ ?assertMatch({ok, _}, couch_replicator_test_helper:replicate(#{
+ <<"source">> => Source,
+ <<"target">> => Target,
+ <<"worker_processes">> => 1 % This make batch_size predictable
+ })).
+
+
+compare_dbs(Source, Target, ExceptIds) ->
+ ?assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target,
+ ExceptIds)).
diff --git a/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl b/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl
index fd04091..2ac447e 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_test_helper.erl
@@ -1,51 +1,166 @@
+% 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_test_helper).
--include_lib("couch/include/couch_eunit.hrl").
--include_lib("couch/include/couch_db.hrl").
--include_lib("couch_replicator/src/couch_replicator.hrl").
-export([
+ start_couch/0,
+ stop_couch/1,
+
+ create_db/0,
+ create_db/1,
+ delete_db/1,
+
+ server_url/0,
+ db_url/1,
+
+ create_docs/2,
+
compare_dbs/2,
compare_dbs/3,
- db_url/1,
- replicate/1,
+ compare_fold/4,
+
+ compare_docs/2,
+
get_pid/1,
- replicate/2
+
+ replicate/1,
+ replicate/2,
+ replicate_continuous/1,
+ replicate_continuous/2,
+
+ cancel/1,
+ cancel/2,
+
+ scheduler_jobs/0
]).
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_replicator/src/couch_replicator.hrl").
+
+
+-define(USERNAME, "rep_eunit_admin").
+-define(PASSWORD, "rep_eunit_password").
+
+
+start_couch() ->
+ Ctx = test_util:start_couch([fabric, chttpd, couch_replicator]),
+ Hashed = couch_passwords:hash_admin_password(?PASSWORD),
+ ok = config:set("admins", ?USERNAME, ?b2l(Hashed), _Persist = false),
+ Ctx.
+
+
+stop_couch(Ctx) ->
+ config:delete("admins", ?USERNAME, _Persist = false),
+ test_util:stop_couch(Ctx).
+
+
+create_db() ->
+ {ok, Db} = fabric2_db:create(?tempdb(), [?ADMIN_CTX]),
+ fabric2_db:name(Db).
+
+
+create_db(DbName) when is_binary(DbName) ->
+ {ok, Db} = fabric2_db:create(DbName, [?ADMIN_CTX]),
+ fabric2_db:name(Db).
+
+
+delete_db(DbName) ->
+ try
+ ok = fabric2_db:delete(DbName, [?ADMIN_CTX])
+ catch
+ error:database_does_not_exist ->
+ ok
+ end.
+
+
+server_url() ->
+ Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
+ Port = mochiweb_socket_server:get(chttpd, port),
+ Fmt = "http://~s:~s@~s:~b",
+ ?l2b(io_lib:format(Fmt, [?USERNAME, ?PASSWORD, Addr, Port])).
+
+
+db_url(DbName) ->
+ ?l2b(io_lib:format("~s/~s", [server_url(), DbName])).
+
+
+create_docs(DbName, Docs) when is_binary(DbName), is_list(Docs) ->
+ {ok, Db} = fabric2_db:open(DbName, [?ADMIN_CTX]),
+ Docs1 = lists:map(fun(Doc) ->
+ case Doc of
+ #{} ->
+ Doc1 = couch_util:json_decode(couch_util:json_encode(Doc)),
+ couch_doc:from_json_obj(Doc1);
+ #doc{} ->
+ Doc
+ end
+ end, Docs),
+ {ok, ResList} = fabric2_db:update_docs(Db, Docs1),
+ lists:foreach(fun(Res) ->
+ ?assertMatch({ok, {_, Rev}} when is_binary(Rev), Res)
+ end, ResList).
+
+
compare_dbs(Source, Target) ->
- compare_dbs(Source, Target, []).
-
-
-compare_dbs(Source, Target, ExceptIds) ->
- {ok, SourceDb} = couch_db:open_int(Source, []),
- {ok, TargetDb} = couch_db:open_int(Target, []),
-
- Fun = fun(FullDocInfo, Acc) ->
- {ok, DocSource} = couch_db:open_doc(SourceDb, FullDocInfo),
- Id = DocSource#doc.id,
- case lists:member(Id, ExceptIds) of
- true ->
- ?assertEqual(not_found, couch_db:get_doc_info(TargetDb, Id));
- false ->
- {ok, TDoc} = couch_db:open_doc(TargetDb, Id),
- compare_docs(DocSource, TDoc)
+ Fun = fun(SrcDoc, TgtDoc, ok) -> compare_docs(SrcDoc, TgtDoc) end,
+ compare_fold(Source, Target, Fun, ok).
+
+
+compare_dbs(Source, Target, ExceptIds) when is_binary(Source),
+ is_binary(Target), is_list(ExceptIds) ->
+ Fun = fun(SrcDoc, TgtDoc, ok) ->
+ case lists:member(SrcDoc#doc.id, ExceptIds) of
+ true -> ?assertEqual(not_found, TgtDoc);
+ false -> compare_docs(SrcDoc, TgtDoc)
end,
- {ok, Acc}
+ ok
end,
+ compare_fold(Source, Target, Fun, ok).
+
- {ok, _} = couch_db:fold_docs(SourceDb, Fun, [], []),
- ok = couch_db:close(SourceDb),
- ok = couch_db:close(TargetDb).
+compare_fold(Source, Target, Fun, Acc0) when
+ is_binary(Source), is_binary(Target), is_function(Fun, 3) ->
+ {ok, SourceDb} = fabric2_db:open(Source, [?ADMIN_CTX]),
+ {ok, TargetDb} = fabric2_db:open(Target, [?ADMIN_CTX]),
+ fabric2_fdb:transactional(SourceDb, fun(TxSourceDb) ->
+ FoldFun = fun
+ ({meta, _Meta}, Acc) ->
+ {ok, Acc};
+ (complete, Acc) ->
+ {ok, Acc};
+ ({row, Row}, Acc) ->
+ {_, Id} = lists:keyfind(id, 1, Row),
+ SrcDoc = open_doc(TxSourceDb, Id),
+ TgtDoc = open_doc(TargetDb, Id),
+ {ok, Fun(SrcDoc, TgtDoc, Acc)}
+ end,
+ Opts = [{restart_tx, true}],
+ {ok, AccF} = fabric2_db:fold_docs(TxSourceDb, FoldFun, Acc0, Opts),
+ AccF
+ end).
-compare_docs(Doc1, Doc2) ->
+compare_docs(#doc{} = Doc1, Doc2) when
+ is_record(Doc2, doc) orelse Doc2 =:= not_found ->
+ ?assert(Doc2 =/= not_found),
?assertEqual(Doc1#doc.body, Doc2#doc.body),
#doc{atts = Atts1} = Doc1,
#doc{atts = Atts2} = Doc2,
?assertEqual(lists:sort([couch_att:fetch(name, Att) || Att <- Atts1]),
- lists:sort([couch_att:fetch(name, Att) || Att <- Atts2])),
+ lists:sort([couch_att:fetch(name, Att) || Att <- Atts2])),
FunCompareAtts = fun(Att) ->
AttName = couch_att:fetch(name, Att),
{ok, AttTarget} = find_att(Atts2, AttName),
@@ -68,19 +183,109 @@ compare_docs(Doc1, Doc2) ->
?assert(is_integer(couch_att:fetch(disk_len, AttTarget))),
?assert(is_integer(couch_att:fetch(att_len, AttTarget))),
?assertEqual(couch_att:fetch(disk_len, Att),
- couch_att:fetch(disk_len, AttTarget)),
+ couch_att:fetch(disk_len, AttTarget)),
?assertEqual(couch_att:fetch(att_len, Att),
- couch_att:fetch(att_len, AttTarget)),
+ couch_att:fetch(att_len, AttTarget)),
?assertEqual(couch_att:fetch(type, Att),
- couch_att:fetch(type, AttTarget)),
+ couch_att:fetch(type, AttTarget)),
?assertEqual(couch_att:fetch(md5, Att),
- couch_att:fetch(md5, AttTarget))
+ couch_att:fetch(md5, AttTarget))
end,
lists:foreach(FunCompareAtts, Atts1).
+get_pid(RepId) ->
+ JobId = case couch_replicator_jobs:get_job_id(undefined, RepId) of
+ {ok, JobId0} -> JobId0;
+ {error, not_found} -> RepId
+ end,
+ {ok, #{<<"state">> := <<"running">>, <<"pid">> := Pid0}} =
+ couch_replicator_jobs:get_job_data(undefined, JobId),
+ Pid = list_to_pid(binary_to_list(Pid0)),
+ ?assert(is_pid(Pid)),
+ ?assert(is_process_alive(Pid)),
+ Pid.
+
+
+replicate({[_ | _]} = EJson) ->
+ Str = couch_util:json_encode(EJson),
+ replicate(couch_util:json_decode(Str, [return_maps]));
+
+replicate(#{} = Rep0) ->
+ Rep = maybe_db_urls(Rep0),
+ {ok, Id, _} = couch_replicator_parse:parse_transient_rep(Rep, null),
+ ok = cancel(Id),
+ try
+ couch_replicator:replicate(Rep, ?ADMIN_USER)
+ after
+ ok = cancel(Id)
+ end.
+
+
+replicate(Source, Target) ->
+ replicate(#{
+ <<"source">> => Source,
+ <<"target">> => Target
+ }).
+
+
+replicate_continuous({[_ | _]} = EJson) ->
+ Str = couch_util:json_encode(EJson),
+ replicate_continuous(couch_util:json_decode(Str, [return_maps]));
+
+replicate_continuous(#{<<"continuous">> := true} = Rep0) ->
+ Rep = maybe_db_urls(Rep0),
+ {ok, {continuous, RepId}} = couch_replicator:replicate(Rep, ?ADMIN_USER),
+ {ok, get_pid(RepId), RepId}.
+
+
+replicate_continuous(Source, Target) ->
+ replicate_continuous(#{
+ <<"source">> => Source,
+ <<"target">> => Target,
+ <<"continuous">> => true
+ }).
+
+
+cancel(Id) when is_binary(Id) ->
+ CancelRep = #{<<"cancel">> => true, <<"id">> => Id},
+ case couch_replicator:replicate(CancelRep, ?ADMIN_USER) of
+ {ok, {cancelled, <<_/binary>>}} -> ok;
+ {error, not_found} -> ok
+ end.
+
+
+cancel(Id, Pid) when is_pid(Pid), is_binary(Id) ->
+ Ref = monitor(process, Pid),
+ try
+ cancel(Id)
+ after
+ receive
+ {'DOWN', Ref, _, _, _} -> ok
+ after 60000 ->
+ error(replicator_pid_death_timeout)
+ end
+ end.
+
+
+scheduler_jobs() ->
+ ServerUrl = couch_replicator_test_helper:server_url(),
+ Url = lists:flatten(io_lib:format("~s/_scheduler/jobs", [ServerUrl])),
+ {ok, 200, _, Body} = test_request:get(Url, []),
+ Json = jiffy:decode(Body, [return_maps]),
+ maps:get(<<"jobs">>, Json).
+
+
+open_doc(Db, DocId) ->
+ case fabric2_db:open_doc(Db, DocId, []) of
+ {ok, #doc{deleted = false} = Doc} -> Doc;
+ {not_found, missing} -> not_found
+ end.
+
+
find_att([], _Name) ->
nil;
+
find_att([Att | Rest], Name) ->
case couch_att:fetch(name, Att) of
Name ->
@@ -91,45 +296,29 @@ find_att([Att | Rest], Name) ->
att_md5(Att) ->
- Md50 = couch_att:foldl(
- Att,
- fun(Chunk, Acc) -> couch_hash:md5_hash_update(Acc, Chunk) end,
- couch_hash:md5_hash_init()),
+ Md50 = couch_att:foldl(Att, fun(Chunk, Acc) ->
+ couch_hash:md5_hash_update(Acc, Chunk)
+ end, couch_hash:md5_hash_init()),
couch_hash:md5_hash_final(Md50).
+
att_decoded_md5(Att) ->
- Md50 = couch_att:foldl_decode(
- Att,
- fun(Chunk, Acc) -> couch_hash:md5_hash_update(Acc, Chunk) end,
- couch_hash:md5_hash_init()),
+ Md50 = couch_att:foldl_decode(Att, fun(Chunk, Acc) ->
+ couch_hash:md5_hash_update(Acc, Chunk)
+ end, couch_hash:md5_hash_init()),
couch_hash:md5_hash_final(Md50).
-db_url(DbName) ->
- iolist_to_binary([
- "http://", config:get("httpd", "bind_address", "127.0.0.1"),
- ":", integer_to_list(mochiweb_socket_server:get(couch_httpd, port)),
- "/", DbName
- ]).
-
-get_pid(RepId) ->
- Pid = global:whereis_name({couch_replicator_scheduler_job,RepId}),
- ?assert(is_pid(Pid)),
- Pid.
-replicate(Source, Target) ->
- replicate({[
- {<<"source">>, Source},
- {<<"target">>, Target}
- ]}).
-
-replicate({[_ | _]} = RepObject) ->
- {ok, Rep} = couch_replicator_utils:parse_rep_doc(RepObject, ?ADMIN_USER),
- ok = couch_replicator_scheduler:add_job(Rep),
- couch_replicator_scheduler:reschedule(),
- Pid = get_pid(Rep#rep.id),
- MonRef = erlang:monitor(process, Pid),
- receive
- {'DOWN', MonRef, process, Pid, _} ->
- ok
+maybe_db_urls(#{} = Rep) ->
+ #{<<"source">> := Src, <<"target">> := Tgt} = Rep,
+ Src1 = case Src of
+ <<"http://", _/binary>> -> Src;
+ <<"https://", _/binary>> -> Src;
+ <<_/binary>> -> db_url(Src)
+ end,
+ Tgt1 = case Tgt of
+ <<"http://", _/binary>> -> Tgt;
+ <<"https://", _/binary>> -> Tgt;
+ <<_/binary>> -> db_url(Tgt)
end,
- ok = couch_replicator_scheduler:remove_job(Rep#rep.id).
+ Rep#{<<"source">> := Src1, <<"target">> := Tgt1}.
diff --git a/src/couch_replicator/test/eunit/couch_replicator_transient_jobs_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_transient_jobs_tests.erl
new file mode 100644
index 0000000..25fc6a3
--- /dev/null
+++ b/src/couch_replicator/test/eunit/couch_replicator_transient_jobs_tests.erl
@@ -0,0 +1,106 @@
+% 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_transient_jobs_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_replicator/src/couch_replicator.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
+
+
+transient_jobs_test_() ->
+ {
+ "Transient jobs tests",
+ {
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(transient_job_is_removed, 10),
+ ?TDEF_FE(posting_same_job_is_a_noop, 10)
+ ]
+ }
+ }
+ }.
+
+
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ couch_replicator_test_helper:create_docs(Source, [
+ #{<<"_id">> => <<"doc1">>}
+ ]),
+ Target = couch_replicator_test_helper:create_db(),
+ config:set("replicator", "stats_update_interval_sec", "0", false),
+ config:set("replicator", "transient_job_max_age_sec", "9999", false),
+ {Source, Target}.
+
+
+teardown({Source, Target}) ->
+ config:delete("replicator", "stats_update_interval_sec", false),
+ config:delete("replicator", "transient_job_max_age_sec", false),
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
+
+
+transient_job_is_removed({Source, Target}) ->
+ {ok, #{}} = replicate(Source, Target),
+ JobId = get_rep_id(Source, Target),
+
+ couch_replicator_job_server:reschedule(),
+
+ % Still there after clean up attempt ran
+ ?assertMatch({200, #{}}, scheduler_jobs(JobId)),
+
+ config:set("replicator", "transient_job_max_age_sec", "0", false),
+ couch_replicator_job_server:reschedule(),
+
+ % Should be gone now
+ ?assertMatch({404, #{}}, scheduler_jobs(JobId)).
+
+
+posting_same_job_is_a_noop({Source, Target}) ->
+ {ok, Pid1, RepId1} = replicate_continuous(Source, Target),
+ {ok, Pid2, RepId2} = replicate_continuous(Source, Target),
+ ?assertEqual(RepId1, RepId2),
+ ?assertEqual(Pid1, Pid2),
+ couch_replicator_test_helper:cancel(RepId1).
+
+
+get_rep_id(Source, Target) ->
+ {ok, Id, _} = couch_replicator_parse:parse_transient_rep(#{
+ <<"source">> => couch_replicator_test_helper:db_url(Source),
+ <<"target">> => couch_replicator_test_helper:db_url(Target)
+ }, null),
+ Id.
+
+
+replicate(Source, Target) ->
+ couch_replicator:replicate(#{
+ <<"source">> => couch_replicator_test_helper:db_url(Source),
+ <<"target">> => couch_replicator_test_helper:db_url(Target)
+ }, ?ADMIN_USER).
+
+
+replicate_continuous(Source, Target) ->
+ couch_replicator_test_helper:replicate_continuous(Source, Target).
+
+
+scheduler_jobs(Id) ->
+ SUrl = couch_replicator_test_helper:server_url(),
+ Url = lists:flatten(io_lib:format("~s/_scheduler/jobs/~s", [SUrl, Id])),
+ {ok, Code, _, Body} = test_request:get(Url, []),
+ {Code, jiffy:decode(Body, [return_maps])}.
diff --git a/src/couch_replicator/test/eunit/couch_replicator_use_checkpoints_tests.erl b/src/couch_replicator/test/eunit/couch_replicator_use_checkpoints_tests.erl
index 8e4a21d..4371eff 100644
--- a/src/couch_replicator/test/eunit/couch_replicator_use_checkpoints_tests.erl
+++ b/src/couch_replicator/test/eunit/couch_replicator_use_checkpoints_tests.erl
@@ -14,165 +14,82 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
+-include_lib("fabric/test/fabric2_test.hrl").
--import(couch_replicator_test_helper, [
- db_url/1,
- replicate/1
-]).
-define(DOCS_COUNT, 100).
--define(TIMEOUT_EUNIT, 30).
-define(i2l(I), integer_to_list(I)).
-define(io2b(Io), iolist_to_binary(Io)).
-start(false) ->
- fun
- ({finished, _, {CheckpointHistory}}) ->
- ?assertEqual([{<<"use_checkpoints">>,false}], CheckpointHistory);
- (_) ->
- ok
- end;
-start(true) ->
- fun
- ({finished, _, {CheckpointHistory}}) ->
- ?assertNotEqual(false, lists:keyfind(<<"session_id">>,
- 1, CheckpointHistory));
- (_) ->
- ok
- end.
-
-stop(_, _) ->
- ok.
-
-setup() ->
- DbName = ?tempdb(),
- {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
- ok = couch_db:close(Db),
- DbName.
-
-setup(remote) ->
- {remote, setup()};
-setup({_, Fun, {A, B}}) ->
- Ctx = test_util:start_couch([couch_replicator]),
- {ok, Listener} = couch_replicator_notifier:start_link(Fun),
- Source = setup(A),
- Target = setup(B),
- {Ctx, {Source, Target, Listener}}.
-
-teardown({remote, DbName}) ->
- teardown(DbName);
-teardown(DbName) ->
- ok = couch_server:delete(DbName, [?ADMIN_CTX]),
- ok.
-
-teardown(_, {Ctx, {Source, Target, Listener}}) ->
- teardown(Source),
- teardown(Target),
-
- couch_replicator_notifier:stop(Listener),
- ok = application:stop(couch_replicator),
- ok = test_util:stop_couch(Ctx).
-
use_checkpoints_test_() ->
{
- "Replication use_checkpoints feature tests",
+ setup,
+ fun couch_replicator_test_helper:start_couch/0,
+ fun couch_replicator_test_helper:stop_couch/1,
{
- foreachx,
- fun start/1, fun stop/2,
- [{UseCheckpoints, fun use_checkpoints_tests/2}
- || UseCheckpoints <- [false, true]]
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_replicate_with_checkpoints, 15),
+ ?TDEF_FE(t_replicate_without_checkpoints, 15)
+ ]
}
}.
-use_checkpoints_tests(UseCheckpoints, Fun) ->
- Pairs = [{remote, remote}],
- {
- "use_checkpoints: " ++ atom_to_list(UseCheckpoints),
- {
- foreachx,
- fun setup/1, fun teardown/2,
- [{{UseCheckpoints, Fun, Pair}, fun should_test_checkpoints/2}
- || Pair <- Pairs]
- }
- }.
-should_test_checkpoints({UseCheckpoints, _, {From, To}}, {_Ctx, {Source, Target, _}}) ->
- should_test_checkpoints(UseCheckpoints, {From, To}, {Source, Target}).
-should_test_checkpoints(UseCheckpoints, {From, To}, {Source, Target}) ->
- {lists:flatten(io_lib:format("~p -> ~p", [From, To])),
- {inorder, [
- should_populate_source(Source, ?DOCS_COUNT),
- should_replicate(Source, Target, UseCheckpoints),
- should_compare_databases(Source, Target)
- ]}}.
-
-should_populate_source({remote, Source}, DocCount) ->
- should_populate_source(Source, DocCount);
-should_populate_source(Source, DocCount) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(populate_db(Source, DocCount))}.
-
-should_replicate({remote, Source}, Target, UseCheckpoints) ->
- should_replicate(db_url(Source), Target, UseCheckpoints);
-should_replicate(Source, {remote, Target}, UseCheckpoints) ->
- should_replicate(Source, db_url(Target), UseCheckpoints);
-should_replicate(Source, Target, UseCheckpoints) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(replicate(Source, Target, UseCheckpoints))}.
-
-should_compare_databases({remote, Source}, Target) ->
- should_compare_databases(Source, Target);
-should_compare_databases(Source, {remote, Target}) ->
- should_compare_databases(Source, Target);
-should_compare_databases(Source, Target) ->
- {timeout, ?TIMEOUT_EUNIT, ?_test(compare_dbs(Source, Target))}.
+setup() ->
+ Source = couch_replicator_test_helper:create_db(),
+ Target = couch_replicator_test_helper:create_db(),
+ {Source, Target}.
-populate_db(DbName, DocCount) ->
- {ok, Db} = couch_db:open_int(DbName, []),
- Docs = lists:foldl(
- fun(DocIdCounter, Acc) ->
- Id = ?io2b(["doc", ?i2l(DocIdCounter)]),
- Value = ?io2b(["val", ?i2l(DocIdCounter)]),
- Doc = #doc{
- id = Id,
- body = {[ {<<"value">>, Value} ]}
- },
- [Doc | Acc]
- end,
- [], lists:seq(1, DocCount)),
- {ok, _} = couch_db:update_docs(Db, Docs, []),
- ok = couch_db:close(Db).
-
-compare_dbs(Source, Target) ->
- {ok, SourceDb} = couch_db:open_int(Source, []),
- {ok, TargetDb} = couch_db:open_int(Target, []),
- Fun = fun(FullDocInfo, Acc) ->
- {ok, Doc} = couch_db:open_doc(SourceDb, FullDocInfo),
- {Props} = DocJson = couch_doc:to_json_obj(Doc, [attachments]),
- DocId = couch_util:get_value(<<"_id">>, Props),
- DocTarget = case couch_db:open_doc(TargetDb, DocId) of
- {ok, DocT} ->
- DocT;
- Error ->
- erlang:error(
- {assertion_failed,
- [{module, ?MODULE}, {line, ?LINE},
- {reason, lists:concat(["Error opening document '",
- ?b2l(DocId), "' from target: ",
- couch_util:to_list(Error)])}]})
- end,
- DocTargetJson = couch_doc:to_json_obj(DocTarget, [attachments]),
- ?assertEqual(DocJson, DocTargetJson),
- {ok, Acc}
- end,
- {ok, _} = couch_db:fold_docs(SourceDb, Fun, [], []),
- ok = couch_db:close(SourceDb),
- ok = couch_db:close(TargetDb).
-
-replicate(Source, Target, UseCheckpoints) ->
- replicate({[
- {<<"source">>, Source},
- {<<"target">>, Target},
- {<<"use_checkpoints">>, UseCheckpoints}
- ]}).
+teardown({Source, Target}) ->
+ couch_replicator_test_helper:delete_db(Source),
+ couch_replicator_test_helper:delete_db(Target).
+
+t_replicate_with_checkpoints({Source, Target}) ->
+ populate_db(Source, ?DOCS_COUNT),
+ Res = couch_replicator_test_helper:replicate(#{
+ <<"source">> => Source,
+ <<"target">> => Target,
+ <<"use_checkpoints">> => true
+ }),
+ ?assertMatch({ok, _}, Res),
+
+ {ok, History} = Res,
+ ?assertMatch(#{<<"history">> := _, <<"session_id">> := _}, History),
+
+ Checkpoints = maps:get(<<"history">>, History),
+ SessionId = maps:get(<<"session_id">>, History),
+ ?assert(is_binary(SessionId)),
+ ?assert(is_list(Checkpoints)),
+ ?assert(length(Checkpoints) >= 1),
+
+ couch_replicator_test_helper:compare_dbs(Source, Target).
+
+
+t_replicate_without_checkpoints({Source, Target}) ->
+ populate_db(Source, ?DOCS_COUNT),
+ Res = couch_replicator_test_helper:replicate(#{
+ <<"source">> => Source,
+ <<"target">> => Target,
+ <<"use_checkpoints">> => false
+ }),
+ ?assertEqual({ok, #{<<"use_checkpoints">> => false}}, Res),
+ couch_replicator_test_helper:compare_dbs(Source, Target).
+
+
+populate_db(DbName, DocCount) ->
+ Docs = lists:foldl(fun(DocIdCounter, Acc) ->
+ Id = ?io2b(["doc", ?i2l(DocIdCounter)]),
+ Value = ?io2b(["val", ?i2l(DocIdCounter)]),
+ Doc = #doc{
+ id = Id,
+ body = {[{<<"value">>, Value}]}
+ },
+ [Doc | Acc]
+ end, [], lists:seq(1, DocCount)),
+ couch_replicator_test_helper:create_docs(DbName, Docs).
diff --git a/test/elixir/test/replication_test.exs b/test/elixir/test/replication_test.exs
index 8b657d9..9af5ef8 100644
--- a/test/elixir/test/replication_test.exs
+++ b/test/elixir/test/replication_test.exs
@@ -14,7 +14,10 @@ defmodule ReplicationTest do
# This should probably go into `make elixir` like what
# happens for JavaScript tests.
- @moduletag config: [{"replicator", "startup_jitter", "0"}]
+ @moduletag config: [
+ {"replicator", "startup_jitter", "0"},
+ {"replicator", "stats_update_interval_sec", "0"}
+ ]
test "source database not found with host" do
name = random_db_name()