Commit 5917ccda authored by Dominik Charousset's avatar Dominik Charousset

Check for multiple finalizer calls in unit tests

parent 84285fb1
......@@ -65,9 +65,10 @@ std::function<bool(const buf&)> is_done(scheduled_actor* self) {
};
}
template <class T>
std::function<void(T&, const error&)> fin(scheduled_actor* self) {
template <class T, class Self>
std::function<void(T&, const error&)> fin(Self* self) {
return [=](T&, const error& err) {
self->state.fin_called += 1;
if (err == none) {
CAF_MESSAGE(self->name() << " is done");
} else {
......@@ -76,20 +77,30 @@ std::function<void(T&, const error&)> fin(scheduled_actor* self) {
};
}
TESTEE_STATE(infinite_source) {
int fin_called = 0;
};
TESTEE(infinite_source) {
return {[=](string& fname) -> result<stream<int>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
self, [](int& x) { x = 0; },
[](int& x, downstream<int>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) { return false; }, fin<int>(self));
}};
return {
[=](string& fname) -> result<stream<int>> {
CAF_CHECK_EQUAL(fname, "numbers.txt");
CAF_CHECK_EQUAL(self->mailbox().empty(), true);
return attach_stream_source(
self, [](int& x) { x = 0; },
[](int& x, downstream<int>& out, size_t num) {
for (size_t i = 0; i < num; ++i)
out.push(x++);
},
[](const int&) { return false; }, fin<int>(self));
},
};
}
TESTEE_STATE(file_reader) {
int fin_called = 0;
};
VARARGS_TESTEE(file_reader, size_t buf_size) {
return {
[=](string& fname) -> result<stream<int>> {
......@@ -109,6 +120,7 @@ VARARGS_TESTEE(file_reader, size_t buf_size) {
TESTEE_STATE(sum_up) {
int x = 0;
int fin_called = 0;
};
TESTEE(sum_up) {
......@@ -129,6 +141,7 @@ TESTEE(sum_up) {
TESTEE_STATE(delayed_sum_up) {
int x = 0;
int fin_called = 0;
};
TESTEE(delayed_sum_up) {
......@@ -153,6 +166,10 @@ TESTEE(delayed_sum_up) {
};
}
TESTEE_STATE(broken_sink) {
int fin_called = 0;
};
TESTEE(broken_sink) {
CAF_IGNORE_UNUSED(self);
return {
......@@ -162,6 +179,10 @@ TESTEE(broken_sink) {
};
}
TESTEE_STATE(filter) {
int fin_called = 0;
};
TESTEE(filter) {
CAF_IGNORE_UNUSED(self);
return {
......@@ -185,6 +206,10 @@ TESTEE(filter) {
};
}
TESTEE_STATE(doubler) {
int fin_called = 0;
};
TESTEE(doubler) {
CAF_IGNORE_UNUSED(self);
return {
......@@ -241,6 +266,9 @@ CAF_TEST(depth_2_pipeline_50_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_setup2_50_items) {
......@@ -261,6 +289,9 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(delayed_depth_2_pipeline_50_items) {
......@@ -287,6 +318,9 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<delayed_sum_up_actor>(snk).state.x, 1275);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<delayed_sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_500_items) {
......@@ -314,6 +348,9 @@ CAF_TEST(depth_2_pipeline_500_items) {
CAF_MESSAGE("expect close message from src and then result from snk");
expect((downstream_msg::close), from(src).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125250);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_error_during_handshake) {
......@@ -326,6 +363,9 @@ CAF_TEST(depth_2_pipeline_error_during_handshake) {
expect((open_stream_msg), from(self).to(snk));
expect((upstream_msg::forced_drop), from(_).to(src));
expect((error), from(snk).to(self).with(sec::stream_init_failed));
run();
CAF_MESSAGE("verify that the file reader called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_2_pipeline_error_at_source) {
......@@ -342,6 +382,8 @@ CAF_TEST(depth_2_pipeline_error_at_source) {
hard_kill(src);
expect((downstream_msg::batch), from(src).to(snk));
expect((downstream_msg::forced_close), from(_).to(snk));
CAF_MESSAGE("verify that the sink called its finalizer once");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_2_pipelin_error_at_sink) {
......@@ -357,6 +399,8 @@ CAF_TEST(depth_2_pipelin_error_at_sink) {
hard_kill(snk);
expect((upstream_msg::ack_open), from(snk).to(src));
expect((upstream_msg::forced_drop), from(_).to(src));
CAF_MESSAGE("verify that the source called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
}
CAF_TEST(depth_3_pipeline_50_items) {
......@@ -391,6 +435,10 @@ CAF_TEST(depth_3_pipeline_50_items) {
expect((upstream_msg::ack_batch), from(snk).to(stg));
expect((downstream_msg::close), from(stg).to(snk));
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<filter_actor>(stg).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_4_pipeline_500_items) {
......@@ -413,6 +461,11 @@ CAF_TEST(depth_4_pipeline_500_items) {
run();
CAF_MESSAGE("check sink result");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<filter_actor>(stg1).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<doubler_actor>(stg2).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_3_pipeline_graceful_shutdown) {
......@@ -433,6 +486,10 @@ CAF_TEST(depth_3_pipeline_graceful_shutdown) {
run();
CAF_MESSAGE("check sink result");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 625);
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<file_reader_actor>(src).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<filter_actor>(stg).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST(depth_3_pipeline_infinite_source) {
......@@ -450,6 +507,9 @@ CAF_TEST(depth_3_pipeline_infinite_source) {
CAF_MESSAGE("send exit to the source and expect the stream to terminate");
anon_send_exit(src, exit_reason::user_shutdown);
run();
CAF_MESSAGE("verify that each actor called its finalizer once");
CAF_CHECK_EQUAL(deref<filter_actor>(stg).state.fin_called, 1);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.fin_called, 1);
}
CAF_TEST_FIXTURE_SCOPE_END()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment