Port couch_replicator/07-use-checkpoints.t etap test to eunit

3 files changed, 201 insertions, 274 deletions

diff --git a/src/couch_replicator/Makefile.am b/src/couch_replicator/Makefile.am
index 41b7da9d4..c8960e1d3 100644
--- a/src/couch_replicator/Makefile.am
+++ b/src/couch_replicator/Makefile.am
@@ -42,7 +42,7 @@ test_files = \
 	test/couch_replicator_many_leaves_tests.erl \
 	test/couch_replicator_missing_stubs_tests.erl \
 	test/couch_replicator_modules_load_tests.erl \
-	test/07-use-checkpoints.t
+	test/couch_replicator_use_checkpoints_tests.erl
 
 compiled_files = \
 	ebin/couch_replicator_api_wrap.beam \
diff --git a/src/couch_replicator/test/07-use-checkpoints.t b/src/couch_replicator/test/07-use-checkpoints.t
deleted file mode 100755
index a3295c7a1..000000000
--- a/src/couch_replicator/test/07-use-checkpoints.t
+++ /dev/null
@@ -1,273 +0,0 @@
-#!/usr/bin/env escript
-%% -*- erlang -*-
-% 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.
-
-% Verify that compacting databases that are being used as the source or
-% target of a replication doesn't affect the replication and that the
-% replication doesn't hold their reference counters forever.
-
--define(b2l(B), binary_to_list(B)).
-
--record(user_ctx, {
-    name = null,
-    roles = [],
-    handler
-}).
-
--record(doc, {
-    id = <<"">>,
-    revs = {0, []},
-    body = {[]},
-    atts = [],
-    deleted = false,
-    meta = []
-}).
-
--record(db, {
-    main_pid = nil,
-    update_pid = nil,
-    compactor_pid = nil,
-    instance_start_time, % number of microsecs since jan 1 1970 as a binary string
-    fd,
-    updater_fd,
-    fd_ref_counter,
-    header = nil,
-    committed_update_seq,
-    fulldocinfo_by_id_btree,
-    docinfo_by_seq_btree,
-    local_docs_btree,
-    update_seq,
-    name,
-    filepath,
-    validate_doc_funs = [],
-    security = [],
-    security_ptr = nil,
-    user_ctx = #user_ctx{},
-    waiting_delayed_commit = nil,
-    revs_limit = 1000,
-    fsync_options = [],
-    options = [],
-    compression,
-    before_doc_update,
-    after_doc_read
-}).
-
--record(rep, {
-    id,
-    source,
-    target,
-    options,
-    user_ctx,
-    doc_id
-}).
-
-
-source_db_name() -> <<"couch_test_rep_db_a">>.
-target_db_name() -> <<"couch_test_rep_db_b">>.
-
-
-main(_) ->
-    test_util:init_code_path(),
-
-    etap:plan(16),
-    case (catch test()) of
-        ok ->
-            etap:end_tests();
-        Other ->
-            etap:diag(io_lib:format("Test died abnormally: ~p", [Other])),
-            etap:bail(Other)
-    end,
-    ok.
-
-
-test() ->
-    couch_server_sup:start_link(test_util:config_files()),
-    ibrowse:start(),
-
-    % order matters
-    test_use_checkpoints(false),
-    test_use_checkpoints(true),
-
-    couch_server_sup:stop(),
-    ok.
-
-
-test_use_checkpoints(UseCheckpoints) ->
-    Pairs = [
-        {source_db_name(), target_db_name()},
-        {{remote, source_db_name()}, target_db_name()},
-        {source_db_name(), {remote, target_db_name()}},
-        {{remote, source_db_name()}, {remote, (target_db_name())}}
-    ],
-
-    ListenerFun = case UseCheckpoints of
-    false ->
-        fun({finished, _, {CheckpointHistory}}) ->
-            etap:is(CheckpointHistory,
-            [{<<"use_checkpoints">>,false}],
-            "No checkpoints found");
-        (_) ->
-            ok
-        end;
-    true ->
-        fun({finished, _, {CheckpointHistory}}) ->
-            SessionId = lists:keyfind(
-                <<"session_id">>, 1, CheckpointHistory),
-            case SessionId of
-                false ->
-                    OtpRel = erlang:system_info(otp_release),
-                    case OtpRel >= "R14B01" orelse OtpRel < "R14B03" of
-                        false ->
-                            etap:bail("Checkpoint expected, but not found");
-                        true ->
-                            etap:ok(true,
-                                " Checkpoint expected, but wan't found."
-                                " Your Erlang " ++ OtpRel ++ " version is"
-                                " affected to OTP-9167 issue which causes"
-                                " failure of this test. Try to upgrade Erlang"
-                                " and if this failure repeats file the bug.")
-                    end;
-                _ ->
-                    etap:ok(true, "There's a checkpoint")
-            end;
-        (_) ->
-            ok
-        end
-    end,
-    {ok, Listener} = couch_replicator_notifier:start_link(ListenerFun),
-
-    lists:foreach(
-        fun({Source, Target}) ->
-            {ok, SourceDb} = create_db(source_db_name()),
-            etap:diag("Populating source database"),
-            populate_db(SourceDb, 100),
-            ok = couch_db:close(SourceDb),
-
-            etap:diag("Creating target database"),
-            {ok, TargetDb} = create_db(target_db_name()),
-            ok = couch_db:close(TargetDb),
-
-            etap:diag("Setup replicator notifier listener"),
-
-            etap:diag("Triggering replication"),
-            replicate(Source, Target, UseCheckpoints),
-
-            etap:diag("Replication finished, comparing source and target databases"),
-            compare_dbs(SourceDb, TargetDb),
-
-            etap:diag("Deleting databases"),
-            delete_db(TargetDb),
-            delete_db(SourceDb),
-
-            ok = timer:sleep(1000)
-        end,
-        Pairs),
-
-    couch_replicator_notifier:stop(Listener).
-
-
-populate_db(Db, DocCount) ->
-    Docs = lists:foldl(
-        fun(DocIdCounter, Acc) ->
-            Id = iolist_to_binary(["doc", integer_to_list(DocIdCounter)]),
-            Value = iolist_to_binary(["val", integer_to_list(DocIdCounter)]),
-            Doc = #doc{
-                id = Id,
-                body = {[ {<<"value">>, Value} ]}
-            },
-            [Doc | Acc]
-        end,
-        [], lists:seq(1, DocCount)),
-    {ok, _} = couch_db:update_docs(Db, Docs, []).
-
-
-compare_dbs(#db{name = SourceName}, #db{name = TargetName}) ->
-    {ok, SourceDb} = couch_db:open_int(SourceName, []),
-    {ok, TargetDb} = couch_db:open_int(TargetName, []),
-    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 ->
-            etap:bail("Error opening document '" ++ ?b2l(DocId) ++
-                "' from target: " ++ couch_util:to_list(Error))
-        end,
-        DocTargetJson = couch_doc:to_json_obj(DocTarget, [attachments]),
-        case DocTargetJson of
-        DocJson ->
-            ok;
-        _ ->
-            etap:bail("Content from document '" ++ ?b2l(DocId) ++
-                "' differs in target database")
-        end,
-        {ok, Acc}
-    end,
-    {ok, _, _} = couch_db:enum_docs(SourceDb, Fun, [], []),
-    etap:diag("Target database has the same documents as the source database"),
-    ok = couch_db:close(SourceDb),
-    ok = couch_db:close(TargetDb).
-
-
-db_url(DbName) ->
-    iolist_to_binary([
-        "http://", couch_config:get("httpd", "bind_address", "127.0.0.1"),
-        ":", integer_to_list(mochiweb_socket_server:get(couch_httpd, port)),
-        "/", DbName
-    ]).
-
-
-create_db(DbName) ->
-    {ok, Db} = couch_db:create(
-        DbName,
-        [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}, overwrite]),
-    couch_db:close(Db),
-    {ok, Db}.
-
-
-delete_db(#db{name = DbName, main_pid = Pid}) ->
-    ok = couch_server:delete(
-        DbName, [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}]),
-    MonRef = erlang:monitor(process, Pid),
-    receive
-    {'DOWN', MonRef, process, Pid, _Reason} ->
-        ok
-    after 30000 ->
-        etap:bail("Timeout deleting database")
-    end.
-
-
-replicate({remote, Db}, Target, UseCheckpoints) ->
-    replicate(db_url(Db), Target, UseCheckpoints);
-
-replicate(Source, {remote, Db}, UseCheckpoints) ->
-    replicate(Source, db_url(Db), UseCheckpoints);
-
-replicate(Source, Target, UseCheckpoints) ->
-    RepObject = {[
-        {<<"source">>, Source},
-        {<<"target">>, Target},
-        {<<"use_checkpoints">>, UseCheckpoints}
-    ]},
-    {ok, Rep} = couch_replicator_utils:parse_rep_doc(
-        RepObject, #user_ctx{roles = [<<"_admin">>]}),
-    {ok, Pid} = couch_replicator:async_replicate(Rep),
-    MonRef = erlang:monitor(process, Pid),
-    receive
-    {'DOWN', MonRef, process, Pid, Reason} ->
-        etap:is(Reason, normal, "Replication finished successfully")
-    after 300000 ->
-        etap:bail("Timeout waiting for replication to finish")
-    end.
diff --git a/src/couch_replicator/test/couch_replicator_use_checkpoints_tests.erl b/src/couch_replicator/test/couch_replicator_use_checkpoints_tests.erl
new file mode 100644
index 000000000..5356a37c2
--- /dev/null
+++ b/src/couch_replicator/test/couch_replicator_use_checkpoints_tests.erl
@@ -0,0 +1,200 @@
+% 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_use_checkpoints_tests).
+
+-include("couch_eunit.hrl").
+-include_lib("couchdb/couch_db.hrl").
+
+-define(ADMIN_ROLE, #user_ctx{roles=[<<"_admin">>]}).
+-define(ADMIN_USER, {user_ctx, ?ADMIN_ROLE}).
+-define(DOCS_COUNT, 100).
+-define(TIMEOUT_STOP, 1000).
+-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_USER]),
+    ok = couch_db:close(Db),
+    DbName.
+
+setup(local) ->
+    setup();
+setup(remote) ->
+    {remote, setup()};
+setup({_, Fun, {A, B}}) ->
+    {ok, _} = couch_server_sup:start_link(?CONFIG_CHAIN),
+    {ok, Listener} = couch_replicator_notifier:start_link(Fun),
+    Source = setup(A),
+    Target = setup(B),
+    {Source, Target, Listener}.
+
+teardown({remote, DbName}) ->
+    teardown(DbName);
+teardown(DbName) ->
+    ok = couch_server:delete(DbName, [?ADMIN_USER]),
+    ok.
+
+teardown(_, {Source, Target, Listener}) ->
+    teardown(Source),
+    teardown(Target),
+
+    couch_replicator_notifier:stop(Listener),
+    Pid = whereis(couch_server_sup),
+    erlang:monitor(process, Pid),
+    couch_server_sup:stop(),
+    receive
+        {'DOWN', _, _, Pid, _} ->
+            ok
+    after ?TIMEOUT_STOP ->
+        throw({timeout, server_stop})
+    end.
+
+
+use_checkpoints_test_() ->
+    {
+        "Replication use_checkpoints feature tests",
+        {
+            foreachx,
+            fun start/1, fun stop/2,
+            [{UseCheckpoints, fun use_checkpoints_tests/2}
+             || UseCheckpoints <- [false, true]]
+        }
+    }.
+
+use_checkpoints_tests(UseCheckpoints, Fun) ->
+    Pairs = [{local, local}, {local, remote},
+             {remote, local}, {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}}, {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))}.
+
+
+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:enum_docs(SourceDb, Fun, [], []),
+    ok = couch_db:close(SourceDb),
+    ok = couch_db:close(TargetDb).
+
+db_url(DbName) ->
+    iolist_to_binary([
+        "http://", couch_config:get("httpd", "bind_address", "127.0.0.1"),
+        ":", integer_to_list(mochiweb_socket_server:get(couch_httpd, port)),
+        "/", DbName
+    ]).
+
+replicate(Source, Target, UseCheckpoints) ->
+    RepObject = {[
+        {<<"source">>, Source},
+        {<<"target">>, Target},
+        {<<"use_checkpoints">>, UseCheckpoints}
+    ]},
+    {ok, Rep} = couch_replicator_utils:parse_rep_doc(RepObject, ?ADMIN_ROLE),
+    {ok, Pid} = couch_replicator:async_replicate(Rep),
+    MonRef = erlang:monitor(process, Pid),
+    receive
+        {'DOWN', MonRef, process, Pid, _} ->
+            ok
+    end.