test for refetching history after failure restart

test/dps_channels_manager_tests.erl

@@ -6,12 +6,15 @@
 
 
 manager_test_() ->
-  TestFunctions = [F || {F,0} <- ?MODULE:module_info(exports),
+  TestFunctions = [{F,Arity} || {F,Arity} <- ?MODULE:module_info(exports),
                             lists:prefix("test_", atom_to_list(F))],
   {foreach,
     fun setup/0,
     fun teardown/1,
-    [{atom_to_list(F), fun ?MODULE:F/0} || F <- TestFunctions]
+    [case Fun of
+      {F,0} -> {atom_to_list(F), fun ?MODULE:F/0};
+      {F,1} -> {with, [fun ?MODULE:F/1]}
+    end || Fun <- TestFunctions]
   }.
 
 -record(env, {
@@ -32,17 +35,15 @@ setup() ->
   Slaves = lists:map(fun(I) ->
     Name = list_to_atom("dps_test" ++ integer_to_list(I)),
     {ok, Slave} = slave:start_link(Host, Name, "-setcookie mytestcookie -pa ebin "),
-    Slave
+    {Slave, Host, Name}
   end, lists:seq(1,?NODE_COUNT)),
 
   application:start(dps),
-  {ok, [AppDesc]} = file:consult("../ebin/dps.app"),
-  rpc:multicall(application, load, [AppDesc]),
   gen_event:delete_handler(error_logger, error_logger_tty_h, []),
   #env{modules = Modules, slaves = Slaves}.
 
 teardown(#env{modules = Modules, slaves = Slaves}) ->
-  [slave:stop(Slave) || Slave <- Slaves],
+  [slave:stop(Slave) || {Slave,_,_} <- Slaves],
   meck:unload(Modules),
   application:stop(dps),
   ?assertEqual([], nodes()),
@@ -92,9 +93,7 @@ test_publish_on_channel() ->
 
 
 test_remote_channels_start() ->
-  {Replies, BadNodes} = rpc:multicall(nodes(), application, start, [dps]),
-  ?assertEqual(length(Replies), ?NODE_COUNT),
-  ?assertEqual([], BadNodes),
+  assert_remote_start(),
 
   ?assertEqual(undefined, dps_channels_manager:find(test_channel)),
   assert_create(),
@@ -108,7 +107,7 @@ test_remote_channels_start() ->
 
 test_remote_slaves_take_our_channels_on_start() ->
   assert_create(),
-  ?assertMatch({_,[]}, rpc:multicall(nodes(), application, start, [dps])),
+  assert_remote_start(),
 
   {Replies, BadNodes2} = rpc:multicall(nodes(), dps_channels_manager, find, [test_channel]),
   RemoteChannels = [Pid || Pid <- Replies, is_pid(Pid)],
@@ -122,7 +121,7 @@ test_remote_channels_take_our_history_on_start() ->
   assert_create(),
   [dps_channel:publish(test_channel, N) || N <- lists:seq(1,100)],
   ?assertMatch({ok, TS, Messages} when is_number(TS) andalso length(Messages) == 100, dps_channel:messages(test_channel, 0)),
-  ?assertMatch({_,[]}, rpc:multicall(nodes(), application, start, [dps])),
+  assert_remote_start(),
 
   {Replies, BadNodes2} = rpc:multicall(nodes(), dps_channel, messages, [test_channel, 0]),
   ?assertEqual([], BadNodes2),
@@ -134,10 +133,41 @@ test_remote_channels_take_our_history_on_start() ->
 assert_create() ->
   ?assertMatch(Pid when is_pid(Pid), dps_channels_manager:create(test_channel)).
 
+assert_remote_start() ->
+  {ok, [AppDesc]} = file:consult("../ebin/dps.app"),
+  rpc:multicall(nodes(), application, load, [AppDesc]),
+  ?assertMatch({_,[]}, rpc:multicall(nodes(), application, start, [dps])).
+
+
+
 test_replication_publish_when_node_is_not_initialized() ->
   assert_create(),
   [?assertMatch(TS when is_number(TS), dps_channel:publish(test_channel, N)) || N <- lists:seq(1,100)],
   ok.  
 
 
+
+test_node_refill_history_after_restart(#env{slaves = Slaves}) ->
+  ?assertMatch(Pid when is_pid(Pid), dps_channels_manager:create(chan1)),
+  ?assertMatch(Pid when is_pid(Pid), dps_channels_manager:create(chan2)),
+
+  assert_remote_start(),
+
+  [dps_channel:publish(chan1, N) || N <- lists:seq(1,200)],
+  [dps_channel:publish(chan2, N) || N <- lists:seq(1,200)],
+
+  {Slave, Host, Name} = hd(Slaves),
+
+  ?assertMatch({ok, _, Messages} when length(Messages) == 100, rpc:call(Slave, dps_channel, messages, [chan1, 0])),
+
+  slave:stop(Slave),
+  {ok, Slave} = slave:start_link(Host, Name, "-setcookie mytestcookie -pa ebin "),
+  {ok, [AppDesc]} = file:consult("../ebin/dps.app"),
+  rpc:call(Slave, application, load, [AppDesc]),
+  ?assertEqual(ok, rpc:call(Slave, application, start, [dps])),
+  ?assertMatch({ok, _, Messages} when length(Messages) == 100, rpc:call(Slave, dps_channel, messages, [chan1, 0])),
+
+  ok.
+
+
 -endif.