3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-03-10 16:06:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3224 from sigiesec/migrate-further-tests

Browse Source 
Migrate further tests to unity
This commit is contained in:
Luca Boccassi 2018-08-17 11:14:59 +01:00 committed by GitHub
commit 4ecfb81c7c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 532 additions and 557 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Makefile.am
View File

@ -517,7 +517,8 @@ tests_test_last_endpoint_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_last_endpoint_CPPFLAGS = ${UNITY_CPPFLAGS} tests_test_last_endpoint_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_term_endpoint_SOURCES = tests/test_term_endpoint.cpp tests_test_term_endpoint_SOURCES = tests/test_term_endpoint.cpp
tests_test_term_endpoint_LDADD = src/libzmq.la tests_test_term_endpoint_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_term_endpoint_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_srcfd_SOURCES = tests/test_srcfd.cpp tests_test_srcfd_SOURCES = tests/test_srcfd.cpp
tests_test_srcfd_LDADD = src/libzmq.la tests_test_srcfd_LDADD = src/libzmq.la
@ -591,7 +592,8 @@ tests_test_spec_dealer_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_spec_dealer_CPPFLAGS = ${UNITY_CPPFLAGS} tests_test_spec_dealer_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_spec_router_SOURCES = tests/test_spec_router.cpp tests_test_spec_router_SOURCES = tests/test_spec_router.cpp
tests_test_spec_router_LDADD = src/libzmq.la tests_test_spec_router_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_spec_router_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_spec_pushpull_SOURCES = tests/test_spec_pushpull.cpp tests_test_spec_pushpull_SOURCES = tests/test_spec_pushpull.cpp
tests_test_spec_pushpull_LDADD = src/libzmq.la tests_test_spec_pushpull_LDADD = src/libzmq.la
@ -632,7 +634,8 @@ tests_test_diffserv_SOURCES = tests/test_diffserv.cpp
tests_test_diffserv_LDADD = src/libzmq.la tests_test_diffserv_LDADD = src/libzmq.la
tests_test_connect_rid_SOURCES = tests/test_connect_rid.cpp tests_test_connect_rid_SOURCES = tests/test_connect_rid.cpp
tests_test_connect_rid_LDADD = src/libzmq.la tests_test_connect_rid_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_connect_rid_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_bind_src_address_SOURCES = tests/test_bind_src_address.cpp tests_test_bind_src_address_SOURCES = tests/test_bind_src_address.cpp
tests_test_bind_src_address_LDADD = src/libzmq.la ${UNITY_LIBS} tests_test_bind_src_address_LDADD = src/libzmq.la ${UNITY_LIBS}
@ -666,7 +669,8 @@ tests_test_sockopt_hwm_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_sockopt_hwm_CPPFLAGS = ${UNITY_CPPFLAGS} tests_test_sockopt_hwm_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_setsockopt_SOURCES = tests/test_setsockopt.cpp tests_test_setsockopt_SOURCES = tests/test_setsockopt.cpp
tests_test_setsockopt_LDADD = src/libzmq.la tests_test_setsockopt_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_setsockopt_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_heartbeats_SOURCES = tests/test_heartbeats.cpp tests_test_heartbeats_SOURCES = tests/test_heartbeats.cpp
tests_test_heartbeats_LDADD = src/libzmq.la ${UNITY_LIBS} tests_test_heartbeats_LDADD = src/libzmq.la ${UNITY_LIBS}
@ -775,7 +779,8 @@ tests_test_timeo_SOURCES = tests/test_timeo.cpp
tests_test_timeo_LDADD = src/libzmq.la tests_test_timeo_LDADD = src/libzmq.la
tests_test_filter_ipc_SOURCES = tests/test_filter_ipc.cpp tests_test_filter_ipc_SOURCES = tests/test_filter_ipc.cpp
tests_test_filter_ipc_LDADD = src/libzmq.la tests_test_filter_ipc_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_filter_ipc_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_use_fd_ipc_SOURCES = \ tests_test_use_fd_ipc_SOURCES = \
tests/test_use_fd_ipc.cpp \ tests/test_use_fd_ipc.cpp \

318
tests/test_connect_rid.cpp
View File

@ -28,275 +28,231 @@
*/ */
#include "testutil.hpp" #include "testutil.hpp"
#include "testutil_unity.hpp"
void setUp ()
{
setup_test_context ();
}
void tearDown ()
{
teardown_test_context ();
}
const char *rconn1routing_id = "conn1";
const char *x_routing_id = "X";
const char *y_routing_id = "Y";
const char *z_routing_id = "Z";
void test_stream_2_stream () void test_stream_2_stream ()
{ {
void *rbind, *rconn1;
int ret;
char buff[256]; char buff[256];
char msg[] = "hi 1"; const char msg[] = "hi 1";
const char *bindip = "tcp://127.0.0.1:*"; const int disabled = 0;
int disabled = 0; const int zero = 0;
int zero = 0;
size_t len = MAX_SOCKET_STRING;
char my_endpoint[MAX_SOCKET_STRING]; char my_endpoint[MAX_SOCKET_STRING];
void *ctx = zmq_ctx_new ();
// Set up listener STREAM. // Set up listener STREAM.
rbind = zmq_socket (ctx, ZMQ_STREAM); void *rbind = test_context_socket (ZMQ_STREAM);
assert (rbind); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (rbind, ZMQ_STREAM_NOTIFY, &disabled, sizeof (disabled)));
zmq_setsockopt (rbind, ZMQ_STREAM_NOTIFY, &disabled, sizeof (disabled));
assert (ret == 0); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof zero));
assert (0 == ret); bind_loopback_ipv4 (rbind, my_endpoint, sizeof my_endpoint);
ret = zmq_bind (rbind, bindip);
assert (0 == ret);
ret = zmq_getsockopt (rbind, ZMQ_LAST_ENDPOINT, my_endpoint, &len);
assert (0 == ret);
// Set up connection stream. // Set up connection stream.
rconn1 = zmq_socket (ctx, ZMQ_STREAM); void *rconn1 = test_context_socket (ZMQ_STREAM);
assert (rconn1); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof zero));
assert (0 == ret);
// Do the connection. // Do the connection.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID,
assert (0 == ret); rconn1routing_id,
ret = zmq_connect (rconn1, my_endpoint); strlen (rconn1routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, my_endpoint));
/* Uncomment to test assert on duplicate routing id. /* Uncomment to test assert on duplicate routing id.
// Test duplicate connect attempt. // Test duplicate connect attempt.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, rconn1routing_id, strlen(rconn1routing_id)));
assert (0 == ret); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, bindip));
ret = zmq_connect (rconn1, bindip);
assert (0 == ret);
*/ */
// Send data to the bound stream. // Send data to the bound stream.
ret = zmq_send (rconn1, "conn1", 6, ZMQ_SNDMORE); send_string_expect_success (rconn1, rconn1routing_id, ZMQ_SNDMORE);
assert (6 == ret); send_string_expect_success (rconn1, msg, 0);
ret = zmq_send (rconn1, msg, 5, 0);
assert (5 == ret);
// Accept data on the bound stream. // Accept data on the bound stream.
ret = zmq_recv (rbind, buff, 256, 0); TEST_ASSERT_GREATER_THAN (
assert (ret); 0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (rbind, buff, 256, 0)));
assert (0 == buff[0]); TEST_ASSERT_EQUAL (0, buff[0]); // an auto-generated routing id
ret = zmq_recv (rbind, buff + 128, 128, 0); recv_string_expect_success (rbind, msg, 0);
assert (5 == ret);
assert ('h' == buff[128]);
// Handle close of the socket. // Handle close of the socket.
ret = zmq_unbind (rbind, my_endpoint); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (rbind, my_endpoint));
assert (0 == ret); test_context_socket_close (rbind);
ret = zmq_close (rbind); test_context_socket_close (rconn1);
assert (0 == ret);
ret = zmq_close (rconn1);
assert (0 == ret);
zmq_ctx_destroy (ctx);
} }
void test_router_2_router (bool named_) void test_router_2_router (bool named_)
{ {
void *rbind, *rconn1;
int ret;
char buff[256]; char buff[256];
char msg[] = "hi 1"; const char msg[] = "hi 1";
const char *bindip = "tcp://127.0.0.1:*"; const int zero = 0;
int zero = 0;
size_t len = MAX_SOCKET_STRING;
char my_endpoint[MAX_SOCKET_STRING]; char my_endpoint[MAX_SOCKET_STRING];
void *ctx = zmq_ctx_new ();
// Create bind socket. // Create bind socket.
rbind = zmq_socket (ctx, ZMQ_ROUTER); void *rbind = test_context_socket (ZMQ_ROUTER);
assert (rbind); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof (zero)));
assert (0 == ret); bind_loopback_ipv4 (rbind, my_endpoint, sizeof my_endpoint);
ret = zmq_bind (rbind, bindip);
assert (0 == ret);
ret = zmq_getsockopt (rbind, ZMQ_LAST_ENDPOINT, my_endpoint, &len);
assert (0 == ret);
// Create connection socket. // Create connection socket.
rconn1 = zmq_socket (ctx, ZMQ_ROUTER); void *rconn1 = test_context_socket (ZMQ_ROUTER);
assert (rconn1); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof (zero)));
assert (0 == ret);
// If we're in named mode, set some identities. // If we're in named mode, set some identities.
if (named_) { if (named_) {
ret = zmq_setsockopt (rbind, ZMQ_ROUTING_ID, "X", 1); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (rconn1, ZMQ_ROUTING_ID, "Y", 1); zmq_setsockopt (rbind, ZMQ_ROUTING_ID, x_routing_id, 1));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (rconn1, ZMQ_ROUTING_ID, y_routing_id, 1));
} }
// Make call to connect using a connect_routing_id. // Make call to connect using a connect_routing_id.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID,
assert (0 == ret); rconn1routing_id,
ret = zmq_connect (rconn1, my_endpoint); strlen (rconn1routing_id)));
assert (0 == ret); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, my_endpoint));
/* Uncomment to test assert on duplicate routing id /* Uncomment to test assert on duplicate routing id
// Test duplicate connect attempt. // Test duplicate connect attempt.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, rconn1routing_id, strlen (rconn1routing_id)));
assert (0 == ret); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, bindip));
ret = zmq_connect (rconn1, bindip);
assert (0 == ret);
*/ */
// Send some data. // Send some data.
ret = zmq_send (rconn1, "conn1", 6, ZMQ_SNDMORE);
assert (6 == ret); send_string_expect_success (rconn1, rconn1routing_id, ZMQ_SNDMORE);
ret = zmq_send (rconn1, msg, 5, 0); send_string_expect_success (rconn1, msg, 0);
assert (5 == ret);
// Receive the name. // Receive the name.
ret = zmq_recv (rbind, buff, 256, 0); const int routing_id_len = zmq_recv (rbind, buff, 256, 0);
if (named_) if (named_) {
assert (ret && 'Y' == buff[0]); TEST_ASSERT_EQUAL_INT (strlen (y_routing_id), routing_id_len);
else TEST_ASSERT_EQUAL_STRING_LEN (y_routing_id, buff, routing_id_len);
assert (ret && 0 == buff[0]); } else {
TEST_ASSERT_TRUE (routing_id_len && 0 == buff[0]);
}
// Receive the data. // Receive the data.
ret = zmq_recv (rbind, buff + 128, 128, 0); recv_string_expect_success (rbind, msg, 0);
assert (5 == ret && 'h' == buff[128]);
// Send some data back. // Send some data back.
if (named_) { const int ret = zmq_send (rbind, buff, routing_id_len, ZMQ_SNDMORE);
ret = zmq_send (rbind, buff, 1, ZMQ_SNDMORE); TEST_ASSERT_EQUAL_INT (routing_id_len, ret);
assert (1 == ret); send_string_expect_success (rbind, "ok", 0);
} else {
ret = zmq_send (rbind, buff, 5, ZMQ_SNDMORE);
assert (5 == ret);
}
ret = zmq_send_const (rbind, "ok", 3, 0);
assert (3 == ret);
// If bound socket identity naming a problem, we'll likely see something funky here. // If bound socket identity naming a problem, we'll likely see something funky here.
ret = zmq_recv (rconn1, buff, 256, 0); recv_string_expect_success (rconn1, rconn1routing_id, 0);
assert ('c' == buff[0] && 6 == ret); recv_string_expect_success (rconn1, "ok", 0);
ret = zmq_recv (rconn1, buff + 128, 128, 0);
assert (3 == ret && 'o' == buff[128]);
ret = zmq_unbind (rbind, my_endpoint); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (rbind, my_endpoint));
assert (0 == ret); test_context_socket_close (rbind);
ret = zmq_close (rbind); test_context_socket_close (rconn1);
assert (0 == ret);
ret = zmq_close (rconn1);
assert (0 == ret);
zmq_ctx_destroy (ctx);
} }
void test_router_2_router_while_receiving () void test_router_2_router_while_receiving ()
{ {
void *xbind, *zbind, *yconn;
int ret;
char buff[256]; char buff[256];
char msg[] = "hi 1"; const char msg[] = "hi 1";
const char *wildcard_bind = "tcp://127.0.0.1:*"; const int zero = 0;
int zero = 0;
size_t len = MAX_SOCKET_STRING;
char x_endpoint[MAX_SOCKET_STRING]; char x_endpoint[MAX_SOCKET_STRING];
char z_endpoint[MAX_SOCKET_STRING]; char z_endpoint[MAX_SOCKET_STRING];
void *ctx = zmq_ctx_new ();
// Create xbind socket. // Create xbind socket.
xbind = zmq_socket (ctx, ZMQ_ROUTER); void *xbind = test_context_socket (ZMQ_ROUTER);
assert (xbind); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (xbind, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (xbind, ZMQ_LINGER, &zero, sizeof (zero)));
assert (0 == ret); bind_loopback_ipv4 (xbind, x_endpoint, sizeof x_endpoint);
ret = zmq_bind (xbind, wildcard_bind);
assert (0 == ret);
ret = zmq_getsockopt (xbind, ZMQ_LAST_ENDPOINT, x_endpoint, &len);
assert (0 == ret);
// Create zbind socket. // Create zbind socket.
zbind = zmq_socket (ctx, ZMQ_ROUTER); void *zbind = test_context_socket (ZMQ_ROUTER);
assert (zbind); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (zbind, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (zbind, ZMQ_LINGER, &zero, sizeof (zero)));
assert (0 == ret); bind_loopback_ipv4 (zbind, z_endpoint, sizeof z_endpoint);
ret = zmq_bind (zbind, wildcard_bind);
assert (0 == ret);
ret = zmq_getsockopt (zbind, ZMQ_LAST_ENDPOINT, z_endpoint, &len);
assert (0 == ret);
// Create connection socket. // Create connection socket.
yconn = zmq_socket (ctx, ZMQ_ROUTER); void *yconn = test_context_socket (ZMQ_ROUTER);
assert (yconn); TEST_ASSERT_SUCCESS_ERRNO (
ret = zmq_setsockopt (yconn, ZMQ_LINGER, &zero, sizeof (zero)); zmq_setsockopt (yconn, ZMQ_LINGER, &zero, sizeof (zero)));
assert (0 == ret);
// set identites for each socket // set identities for each socket
ret = zmq_setsockopt (xbind, ZMQ_ROUTING_ID, "X", 2); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
ret = zmq_setsockopt (yconn, ZMQ_ROUTING_ID, "Y", 2); xbind, ZMQ_ROUTING_ID, x_routing_id, strlen (x_routing_id)));
ret = zmq_setsockopt (zbind, ZMQ_ROUTING_ID, "Z", 2); TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (yconn, ZMQ_ROUTING_ID, y_routing_id, 2));
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
zbind, ZMQ_ROUTING_ID, z_routing_id, strlen (z_routing_id)));
// Connect Y to X using a routing id // Connect Y to X using a routing id
ret = zmq_setsockopt (yconn, ZMQ_CONNECT_ROUTING_ID, "X", 2); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
assert (0 == ret); yconn, ZMQ_CONNECT_ROUTING_ID, x_routing_id, strlen (x_routing_id)));
ret = zmq_connect (yconn, x_endpoint); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (yconn, x_endpoint));
assert (0 == ret);
// Send some data from Y to X. // Send some data from Y to X.
ret = zmq_send (yconn, "X", 2, ZMQ_SNDMORE); send_string_expect_success (yconn, x_routing_id, ZMQ_SNDMORE);
assert (2 == ret); send_string_expect_success (yconn, msg, 0);
ret = zmq_send (yconn, msg, 5, 0);
assert (5 == ret);
// wait for the Y->X message to be received // wait for the Y->X message to be received
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
// Now X tries to connect to Z and send a message // Now X tries to connect to Z and send a message
ret = zmq_setsockopt (xbind, ZMQ_CONNECT_ROUTING_ID, "Z", 2); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
assert (0 == ret); xbind, ZMQ_CONNECT_ROUTING_ID, z_routing_id, strlen (z_routing_id)));
ret = zmq_connect (xbind, z_endpoint); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (xbind, z_endpoint));
assert (0 == ret);
// Try to send some data from X to Z. // Try to send some data from X to Z.
ret = zmq_send (xbind, "Z", 2, ZMQ_SNDMORE); send_string_expect_success (xbind, z_routing_id, ZMQ_SNDMORE);
assert (2 == ret); send_string_expect_success (xbind, msg, 0);
ret = zmq_send (xbind, msg, 5, 0);
assert (5 == ret);
// wait for the X->Z message to be received (so that our non-blocking check will actually // wait for the X->Z message to be received (so that our non-blocking check will actually
// fail if the message is routed to Y) // fail if the message is routed to Y)
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
// nothing should have been received on the Y socket // nothing should have been received on the Y socket
ret = zmq_recv (yconn, buff, 256, ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (EAGAIN,
assert (ret == -1); zmq_recv (yconn, buff, 256, ZMQ_DONTWAIT));
assert (zmq_errno () == EAGAIN);
// the message should have been received on the Z socket // the message should have been received on the Z socket
ret = zmq_recv (zbind, buff, 256, 0); recv_string_expect_success (zbind, x_routing_id, 0);
assert (ret && 'X' == buff[0]); recv_string_expect_success (zbind, msg, 0);
ret = zmq_recv (zbind, buff + 128, 128, 0);
assert (5 == ret && 'h' == buff[128]);
ret = zmq_unbind (xbind, x_endpoint); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (xbind, x_endpoint));
assert (0 == ret); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (zbind, z_endpoint));
ret = zmq_unbind (zbind, z_endpoint);
assert (0 == ret);
ret = zmq_close (yconn);
assert (0 == ret);
ret = zmq_close (xbind);
assert (0 == ret);
ret = zmq_close (zbind);
assert (0 == ret);
zmq_ctx_destroy (ctx); test_context_socket_close (yconn);
test_context_socket_close (xbind);
test_context_socket_close (zbind);
} }
int main (void) void test_router_2_router_unnamed ()
{
test_router_2_router (false);
}
void test_router_2_router_named ()
{
test_router_2_router (true);
}
int main ()
{ {
setup_test_environment (); setup_test_environment ();
test_stream_2_stream (); UNITY_BEGIN ();
test_router_2_router (false); RUN_TEST (test_stream_2_stream);
test_router_2_router (true); RUN_TEST (test_router_2_router_unnamed);
test_router_2_router_while_receiving (); RUN_TEST (test_router_2_router_named);
RUN_TEST (test_router_2_router_while_receiving);
return 0; return UNITY_END ();
} }

188
tests/test_filter_ipc.cpp
View File

@ -28,6 +28,17 @@
*/ */
#include "testutil.hpp" #include "testutil.hpp"
#include "testutil_unity.hpp"
void setUp ()
{
setup_test_context ();
}
void tearDown ()
{
teardown_test_context ();
}
static void bounce_fail (void *server_, void *client_) static void bounce_fail (void *server_, void *client_)
{ {
@ -35,131 +46,178 @@ static void bounce_fail (void *server_, void *client_)
char buffer[32]; char buffer[32];
// Send message from client to server // Send message from client to server
int rc = zmq_send (client_, content, 32, ZMQ_SNDMORE); send_string_expect_success (client_, content, ZMQ_SNDMORE);
assert (rc == 32); send_string_expect_success (client_, content, 0);
rc = zmq_send (client_, content, 32, 0);
assert (rc == 32);
// Receive message at server side (should not succeed) // Receive message at server side (should not succeed)
int timeout = 250; int timeout = SETTLE_TIME;
rc = zmq_setsockopt (server_, ZMQ_RCVTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (server_, ZMQ_RCVTIMEO, &timeout, sizeof (int)));
rc = zmq_recv (server_, buffer, 32, 0); TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_recv (server_, buffer, 32, 0));
assert (rc == -1);
assert (zmq_errno () == EAGAIN);
// Send message from server to client to test other direction // Send message from server to client to test other direction
rc = zmq_setsockopt (server_, ZMQ_SNDTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (server_, ZMQ_SNDTIMEO, &timeout, sizeof (int)));
rc = zmq_send (server_, content, 32, ZMQ_SNDMORE); TEST_ASSERT_FAILURE_ERRNO (EAGAIN,
assert (rc == -1); zmq_send (server_, content, 32, ZMQ_SNDMORE));
assert (zmq_errno () == EAGAIN);
} }
template <class T> template <class T>
static void static void
run_test (int opt_, T optval_, int expected_error_, int bounce_test_) run_test (int opt_, T optval_, int expected_error_, int bounce_test_)
{ {
int rc; void *sb = test_context_socket (ZMQ_DEALER);
void *ctx = zmq_ctx_new ();
assert (ctx);
void *sb = zmq_socket (ctx, ZMQ_DEALER);
assert (sb);
if (opt_) { if (opt_) {
rc = zmq_setsockopt (sb, opt_, &optval_, sizeof (optval_)); const int rc = zmq_setsockopt (sb, opt_, &optval_, sizeof (optval_));
if (expected_error_) { if (expected_error_) {
assert (rc == -1); TEST_ASSERT_FAILURE_ERRNO (expected_error_, rc);
assert (zmq_errno () == expected_error_); } else {
} else TEST_ASSERT_SUCCESS_ERRNO (rc);
assert (rc == 0); }
} }
void *sc = zmq_socket (ctx, ZMQ_DEALER); void *sc = test_context_socket (ZMQ_DEALER);
assert (sc);
// If a test fails, don't hang for too long // If a test fails, don't hang for too long
int timeout = 2500; int timeout = 2500;
rc = zmq_setsockopt (sb, ZMQ_RCVTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (sb, ZMQ_RCVTIMEO, &timeout, sizeof (int)));
rc = zmq_setsockopt (sb, ZMQ_SNDTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (sb, ZMQ_SNDTIMEO, &timeout, sizeof (int)));
rc = zmq_setsockopt (sc, ZMQ_RCVTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (sc, ZMQ_RCVTIMEO, &timeout, sizeof (int)));
rc = zmq_setsockopt (sc, ZMQ_SNDTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (sc, ZMQ_SNDTIMEO, &timeout, sizeof (int)));
int interval = -1; int interval = -1;
rc = zmq_setsockopt (sc, ZMQ_RECONNECT_IVL, &interval, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (sc, ZMQ_RECONNECT_IVL, &interval, sizeof (int)));
if (bounce_test_) { if (bounce_test_) {
const char *endpoint = "ipc://test_filter_ipc.sock"; const char *endpoint = "ipc://test_filter_ipc.sock";
int rc = zmq_bind (sb, endpoint); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (sb, endpoint));
assert (rc == 0); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (sc, endpoint));
rc = zmq_connect (sc, endpoint);
assert (rc == 0);
if (bounce_test_ > 0) if (bounce_test_ > 0)
bounce (sb, sc); bounce (sb, sc);
else else
bounce_fail (sb, sc); bounce_fail (sb, sc);
} }
close_zero_linger (sc);
close_zero_linger (sb);
rc = zmq_ctx_term (ctx); // TODO only use zero linger when bounce_test_ < 0?
assert (rc == 0); test_context_socket_close_zero_linger (sc);
test_context_socket_close_zero_linger (sb);
} }
int main (void)
{
#if !defined(ZMQ_HAVE_WINDOWS)
setup_test_environment ();
// No filters
run_test<int> (0, 0, 0, 1);
#if defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED #if defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED
// Get the group and supplimental groups of the process owner gid_t group, supgroup, notgroup;
void init_groups ()
{
// Get the group and supplemental groups of the process owner
gid_t groups[100]; gid_t groups[100];
int ngroups = getgroups (100, groups); int ngroups = getgroups (100, groups);
assert (ngroups != -1); assert (ngroups != -1);
gid_t group = getgid (), supgroup = group, notgroup = group + 1; group = getgid ();
supgroup = group;
notgroup = group + 1;
for (int i = 0; i < ngroups; i++) { for (int i = 0; i < ngroups; i++) {
if (supgroup == group && group != groups[i]) if (supgroup == group && group != groups[i])
supgroup = groups[i]; supgroup = groups[i];
if (notgroup <= groups[i]) if (notgroup <= groups[i])
notgroup = groups[i] + 1; notgroup = groups[i] + 1;
} }
}
#endif
// Test filter with UID of process owner void test_no_filters ()
{
run_test<int> (0, 0, 0, 1);
}
#if defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED
void test_filter_with_process_owner_uid ()
{
run_test<uid_t> (ZMQ_IPC_FILTER_UID, getuid (), 0, 1); run_test<uid_t> (ZMQ_IPC_FILTER_UID, getuid (), 0, 1);
// Test filter with UID of another (possibly non-existent) user }
void test_filter_with_possibly_nonexistent_uid ()
{
run_test<uid_t> (ZMQ_IPC_FILTER_UID, getuid () + 1, 0, -1); run_test<uid_t> (ZMQ_IPC_FILTER_UID, getuid () + 1, 0, -1);
// Test filter with GID of process owner }
void test_filter_with_process_owner_gid ()
{
run_test<gid_t> (ZMQ_IPC_FILTER_GID, group, 0, 1); run_test<gid_t> (ZMQ_IPC_FILTER_GID, group, 0, 1);
// Test filter with supplimental group of process owner }
void test_filter_with_supplemental_process_owner_gid ()
{
run_test<gid_t> (ZMQ_IPC_FILTER_GID, supgroup, 0, 1); run_test<gid_t> (ZMQ_IPC_FILTER_GID, supgroup, 0, 1);
// Test filter with GID of another (possibly non-existent) group }
void test_filter_with_possibly_nonexistent_gid ()
{
run_test<gid_t> (ZMQ_IPC_FILTER_GID, notgroup, 0, -1); run_test<gid_t> (ZMQ_IPC_FILTER_GID, notgroup, 0, -1);
}
#if defined ZMQ_HAVE_SO_PEERCRED #if defined ZMQ_HAVE_SO_PEERCRED
// Test filter with PID of current process void test_filter_with_current_process_pid ()
{
run_test<pid_t> (ZMQ_IPC_FILTER_PID, getpid (), 0, 1); run_test<pid_t> (ZMQ_IPC_FILTER_PID, getpid (), 0, 1);
// Test filter with PID of another (possibly non-existent) process }
void test_filter_with_possibly_nonexistent_pid ()
{
run_test<pid_t> (ZMQ_IPC_FILTER_PID, getpid () + 1, 0, -1); run_test<pid_t> (ZMQ_IPC_FILTER_PID, getpid () + 1, 0, -1);
}
#else #else
void test_filter_with_pid_fails ()
{
// Setup of PID filter should fail with operation not supported error // Setup of PID filter should fail with operation not supported error
// TODO EINVAL is not ENOTSUP (!)
run_test<pid_t> (ZMQ_IPC_FILTER_PID, getpid (), EINVAL, 0); run_test<pid_t> (ZMQ_IPC_FILTER_PID, getpid (), EINVAL, 0);
}
#endif #endif
#else #else
void test_filter_with_zero_uid_fails ()
{
run_test<uid_t> (ZMQ_IPC_FILTER_UID, 0, EINVAL, 0); run_test<uid_t> (ZMQ_IPC_FILTER_UID, 0, EINVAL, 0);
}
void test_filter_with_zero_gid_fails ()
{
run_test<gid_t> (ZMQ_IPC_FILTER_GID, 0, EINVAL, 0); run_test<gid_t> (ZMQ_IPC_FILTER_GID, 0, EINVAL, 0);
}
void test_filter_with_zero_pid_fails ()
{
run_test<pid_t> (ZMQ_IPC_FILTER_PID, 0, EINVAL, 0); run_test<pid_t> (ZMQ_IPC_FILTER_PID, 0, EINVAL, 0);
}
#endif // defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED #endif // defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED
int main (void)
{
#if !defined(ZMQ_HAVE_WINDOWS)
setup_test_environment ();
#if defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED
init_groups ();
#endif #endif
UNITY_BEGIN ();
RUN_TEST (test_no_filters);
#if defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED
RUN_TEST (test_filter_with_process_owner_uid);
RUN_TEST (test_filter_with_possibly_nonexistent_uid);
RUN_TEST (test_filter_with_process_owner_gid);
RUN_TEST (test_filter_with_supplemental_process_owner_gid);
RUN_TEST (test_filter_with_possibly_nonexistent_gid);
#if defined ZMQ_HAVE_SO_PEERCRED
RUN_TEST (test_filter_with_current_process_pid);
RUN_TEST (test_filter_with_possibly_nonexistent_pid);
#else
RUN_TEST (test_filter_with_pid_fails ());
#endif
#else
RUN_TEST (test_filter_with_zero_uid_fails);
RUN_TEST (test_filter_with_zero_gid_fails);
RUN_TEST (test_filter_with_zero_pid_fails);
#endif // defined ZMQ_HAVE_SO_PEERCRED || defined ZMQ_HAVE_LOCAL_PEERCRED
return UNITY_END ();
#else
return 0; return 0;
#endif
} }

147
tests/test_setsockopt.cpp
View File

@ -28,126 +28,129 @@
*/ */
#include "testutil.hpp" #include "testutil.hpp"
#include "testutil_unity.hpp"
void test_setsockopt_tcp_recv_buffer (void) void setUp ()
{ {
int rc; setup_test_context ();
void *ctx = zmq_ctx_new ();
void *socket = zmq_socket (ctx, ZMQ_PUSH);
int val = 0;
size_t placeholder = sizeof (val);
rc = zmq_getsockopt (socket, ZMQ_RCVBUF, &val, &placeholder);
assert (rc == 0);
assert (val == -1);
val = 16384;
rc = zmq_setsockopt (socket, ZMQ_RCVBUF, &val, sizeof (val));
assert (rc == 0);
assert (val == 16384);
rc = zmq_getsockopt (socket, ZMQ_RCVBUF, &val, &placeholder);
assert (rc == 0);
assert (val == 16384);
zmq_close (socket);
zmq_ctx_term (ctx);
} }
void test_setsockopt_tcp_send_buffer (void) void tearDown ()
{ {
int rc; teardown_test_context ();
void *ctx = zmq_ctx_new (); }
void *socket = zmq_socket (ctx, ZMQ_PUSH);
void test_setsockopt_tcp_recv_buffer ()
{
void *socket = test_context_socket (ZMQ_PUSH);
int val = 0; int val = 0;
size_t placeholder = sizeof (val); size_t placeholder = sizeof (val);
rc = zmq_getsockopt (socket, ZMQ_SNDBUF, &val, &placeholder); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (socket, ZMQ_RCVBUF, &val, &placeholder));
assert (val == -1); TEST_ASSERT_EQUAL_INT (-1, val);
val = 16384; val = 16384;
rc = zmq_setsockopt (socket, ZMQ_SNDBUF, &val, sizeof (val)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (socket, ZMQ_RCVBUF, &val, sizeof (val)));
assert (val == 16384); TEST_ASSERT_EQUAL_INT (16384, val);
rc = zmq_getsockopt (socket, ZMQ_SNDBUF, &val, &placeholder); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (socket, ZMQ_RCVBUF, &val, &placeholder));
assert (val == 16384); TEST_ASSERT_EQUAL_INT (16384, val);
zmq_close (socket); test_context_socket_close (socket);
zmq_ctx_term (ctx); }
void test_setsockopt_tcp_send_buffer ()
{
void *socket = test_context_socket (ZMQ_PUSH);
int val = 0;
size_t placeholder = sizeof (val);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_getsockopt (socket, ZMQ_SNDBUF, &val, &placeholder));
TEST_ASSERT_EQUAL_INT (-1, val);
val = 16384;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (socket, ZMQ_SNDBUF, &val, sizeof (val)));
TEST_ASSERT_EQUAL_INT (16384, val);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_getsockopt (socket, ZMQ_SNDBUF, &val, &placeholder));
TEST_ASSERT_EQUAL_INT (16384, val);
test_context_socket_close (socket);
} }
void test_setsockopt_use_fd () void test_setsockopt_use_fd ()
{ {
int rc; void *socket = test_context_socket (ZMQ_PUSH);
void *ctx = zmq_ctx_new ();
void *socket = zmq_socket (ctx, ZMQ_PUSH);
int val = 0; int val = 0;
size_t placeholder = sizeof (val); size_t placeholder = sizeof (val);
rc = zmq_getsockopt (socket, ZMQ_USE_FD, &val, &placeholder); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (socket, ZMQ_USE_FD, &val, &placeholder));
assert (val == -1); TEST_ASSERT_EQUAL_INT (-1, val);
val = 3; val = 3;
rc = zmq_setsockopt (socket, ZMQ_USE_FD, &val, sizeof (val)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (socket, ZMQ_USE_FD, &val, sizeof (val)));
assert (val == 3); TEST_ASSERT_EQUAL_INT (3, val);
rc = zmq_getsockopt (socket, ZMQ_USE_FD, &val, &placeholder); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (socket, ZMQ_USE_FD, &val, &placeholder));
assert (val == 3); TEST_ASSERT_EQUAL_INT (3, val);
zmq_close (socket); test_context_socket_close (socket);
zmq_ctx_term (ctx);
} }
#define BOUNDDEVBUFSZ 16 #define BOUNDDEVBUFSZ 16
void test_setsockopt_bindtodevice () void test_setsockopt_bindtodevice ()
{ {
void *ctx = zmq_ctx_new (); void *socket = test_context_socket (ZMQ_PUSH);
void *socket = zmq_socket (ctx, ZMQ_PUSH);
#ifdef ZMQ_BINDTODEVICE #ifdef ZMQ_BINDTODEVICE
int rc;
char devname[BOUNDDEVBUFSZ]; char devname[BOUNDDEVBUFSZ];
size_t buflen = BOUNDDEVBUFSZ; size_t buflen = BOUNDDEVBUFSZ;
rc = zmq_getsockopt (socket, ZMQ_BINDTODEVICE, devname, &buflen); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (socket, ZMQ_BINDTODEVICE, devname, &buflen));
assert (devname[0] == '\0'); TEST_ASSERT_EQUAL_INT8 ('\0', devname[0]);
assert (buflen == 1); TEST_ASSERT_EQUAL_UINT (1, buflen);
sprintf (devname, "testdev"); sprintf (devname, "testdev");
buflen = strlen (devname); buflen = strlen (devname);
rc = zmq_setsockopt (socket, ZMQ_BINDTODEVICE, devname, buflen); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (socket, ZMQ_BINDTODEVICE, devname, buflen));
buflen = BOUNDDEVBUFSZ; buflen = BOUNDDEVBUFSZ;
memset (devname, 0, buflen); memset (devname, 0, buflen);
rc = zmq_getsockopt (socket, ZMQ_BINDTODEVICE, devname, &buflen); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (socket, ZMQ_BINDTODEVICE, devname, &buflen));
assert (!strncmp ("testdev", devname, buflen)); TEST_ASSERT_EQUAL_STRING_LEN ("testdev", devname, buflen);
#endif #endif
zmq_close (socket); test_context_socket_close (socket);
zmq_ctx_term (ctx);
} }
int main (void) int main ()
{ {
test_setsockopt_tcp_recv_buffer (); setup_test_environment ();
test_setsockopt_tcp_send_buffer ();
test_setsockopt_use_fd (); UNITY_BEGIN ();
test_setsockopt_bindtodevice (); RUN_TEST (test_setsockopt_tcp_recv_buffer);
RUN_TEST (test_setsockopt_tcp_send_buffer);
RUN_TEST (test_setsockopt_use_fd);
RUN_TEST (test_setsockopt_bindtodevice);
return UNITY_END ();
} }

178
tests/test_spec_router.cpp
View File

@ -28,50 +28,59 @@
*/ */
#include "testutil.hpp" #include "testutil.hpp"
#include "testutil_unity.hpp"
const char *bind_address = 0; #include <unity.h>
char connect_address[MAX_SOCKET_STRING];
void test_fair_queue_in (void *ctx_) void setUp ()
{ {
void *receiver = zmq_socket (ctx_, ZMQ_ROUTER); setup_test_context ();
assert (receiver); }
void tearDown ()
{
teardown_test_context ();
}
// SHALL receive incoming messages from its peers using a fair-queuing
// strategy.
void test_fair_queue_in (const char *bind_address)
{
char connect_address[MAX_SOCKET_STRING];
void *receiver = test_context_socket (ZMQ_ROUTER);
int timeout = 250; int timeout = 250;
int rc = zmq_setsockopt (receiver, ZMQ_RCVTIMEO, &timeout, sizeof (int)); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (receiver, ZMQ_RCVTIMEO, &timeout, sizeof (int)));
rc = zmq_bind (receiver, bind_address); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (receiver, bind_address));
assert (rc == 0);
size_t len = MAX_SOCKET_STRING; size_t len = MAX_SOCKET_STRING;
rc = zmq_getsockopt (receiver, ZMQ_LAST_ENDPOINT, connect_address, &len); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (receiver, ZMQ_LAST_ENDPOINT, connect_address, &len));
const unsigned char services = 5; const unsigned char services = 5;
void *senders[services]; void *senders[services];
for (unsigned char peer = 0; peer < services; ++peer) { for (unsigned char peer = 0; peer < services; ++peer) {
senders[peer] = zmq_socket (ctx_, ZMQ_DEALER); senders[peer] = test_context_socket (ZMQ_DEALER);
assert (senders[peer]);
rc = TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (senders[peer], ZMQ_RCVTIMEO, &timeout, sizeof (int)); zmq_setsockopt (senders[peer], ZMQ_RCVTIMEO, &timeout, sizeof (int)));
assert (rc == 0);
char *str = strdup ("A"); char *str = strdup ("A");
str[0] += peer; str[0] += peer;
rc = zmq_setsockopt (senders[peer], ZMQ_ROUTING_ID, str, 2); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_setsockopt (senders[peer], ZMQ_ROUTING_ID, str, 2));
free (str); free (str);
rc = zmq_connect (senders[peer], connect_address); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_connect (senders[peer], connect_address));
} }
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
zmq_msg_t msg; zmq_msg_t msg;
rc = zmq_msg_init (&msg); TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
assert (rc == 0);
s_send_seq (senders[0], "M", SEQ_END); s_send_seq (senders[0], "M", SEQ_END);
s_recv_seq (receiver, "A", "M", SEQ_END); s_recv_seq (receiver, "A", "M", SEQ_END);
@ -87,56 +96,53 @@ void test_fair_queue_in (void *ctx_)
sum += 'A' + peer; sum += 'A' + peer;
} }
assert (sum == services * 'A' + services * (services - 1) / 2); TEST_ASSERT_EQUAL_INT (services * 'A' + services * (services - 1) / 2, sum);
// handle N requests // handle N requests
for (unsigned char peer = 0; peer < services; ++peer) { for (unsigned char peer = 0; peer < services; ++peer) {
rc = zmq_msg_recv (&msg, receiver, 0); TEST_ASSERT_EQUAL_INT (
assert (rc == 2); 2, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&msg, receiver, 0)));
const char *id = (const char *) zmq_msg_data (&msg); const char *id = static_cast<const char *> (zmq_msg_data (&msg));
sum -= id[0]; sum -= id[0];
s_recv_seq (receiver, "M", SEQ_END); s_recv_seq (receiver, "M", SEQ_END);
} }
assert (sum == 0); TEST_ASSERT_EQUAL_INT (0, sum);
rc = zmq_msg_close (&msg); TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));
assert (rc == 0);
close_zero_linger (receiver); test_context_socket_close_zero_linger (receiver);
for (size_t peer = 0; peer < services; ++peer) for (size_t peer = 0; peer < services; ++peer)
close_zero_linger (senders[peer]); test_context_socket_close_zero_linger (senders[peer]);
// Wait for disconnects. // Wait for disconnects.
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
} }
void test_destroy_queue_on_disconnect (void *ctx_) // SHALL create a double queue when a peer connects to it. If this peer
// disconnects, the ROUTER socket SHALL destroy its double queue and SHALL
// discard any messages it contains.
void test_destroy_queue_on_disconnect (const char *bind_address)
{ {
void *a = zmq_socket (ctx_, ZMQ_ROUTER); void *a = test_context_socket (ZMQ_ROUTER);
assert (a);
int enabled = 1; int enabled = 1;
int rc = TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (a, ZMQ_ROUTER_MANDATORY, &enabled, sizeof (enabled)); zmq_setsockopt (a, ZMQ_ROUTER_MANDATORY, &enabled, sizeof (enabled)));
assert (rc == 0);
rc = zmq_bind (a, bind_address); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (a, bind_address));
assert (rc == 0);
size_t len = MAX_SOCKET_STRING; size_t len = MAX_SOCKET_STRING;
rc = zmq_getsockopt (a, ZMQ_LAST_ENDPOINT, connect_address, &len); char connect_address[MAX_SOCKET_STRING];
assert (rc == 0); TEST_ASSERT_SUCCESS_ERRNO (
zmq_getsockopt (a, ZMQ_LAST_ENDPOINT, connect_address, &len));
void *b = zmq_socket (ctx_, ZMQ_DEALER); void *b = test_context_socket (ZMQ_DEALER);
assert (b);
rc = zmq_setsockopt (b, ZMQ_ROUTING_ID, "B", 2); TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (b, ZMQ_ROUTING_ID, "B", 2));
assert (rc == 0);
rc = zmq_connect (b, connect_address); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (b, connect_address));
assert (rc == 0);
// Wait for connection. // Wait for connection.
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
@ -145,75 +151,57 @@ void test_destroy_queue_on_disconnect (void *ctx_)
s_send_seq (a, "B", "ABC", SEQ_END); s_send_seq (a, "B", "ABC", SEQ_END);
s_send_seq (b, "DEF", SEQ_END); s_send_seq (b, "DEF", SEQ_END);
rc = zmq_disconnect (b, connect_address); TEST_ASSERT_SUCCESS_ERRNO (zmq_disconnect (b, connect_address));
assert (rc == 0);
// Disconnect may take time and need command processing. // Disconnect may take time and need command processing.
zmq_pollitem_t poller[2] = {{a, 0, 0, 0}, {b, 0, 0, 0}}; zmq_pollitem_t poller[2] = {{a, 0, 0, 0}, {b, 0, 0, 0}};
rc = zmq_poll (poller, 2, 100); TEST_ASSERT_SUCCESS_ERRNO (zmq_poll (poller, 2, 100));
assert (rc == 0); TEST_ASSERT_SUCCESS_ERRNO (zmq_poll (poller, 2, 100));
rc = zmq_poll (poller, 2, 100);
assert (rc == 0);
// No messages should be available, sending should fail. // No messages should be available, sending should fail.
zmq_msg_t msg; zmq_msg_t msg;
zmq_msg_init (&msg); zmq_msg_init (&msg);
rc = zmq_send (a, "B", 2, ZMQ_SNDMORE | ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (
assert (rc == -1); EHOSTUNREACH, zmq_send (a, "B", 2, ZMQ_SNDMORE | ZMQ_DONTWAIT));
assert (errno == EHOSTUNREACH);
rc = zmq_msg_recv (&msg, a, ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_msg_recv (&msg, a, ZMQ_DONTWAIT));
assert (rc == -1);
assert (errno == EAGAIN);
// After a reconnect of B, the messages should still be gone // After a reconnect of B, the messages should still be gone
rc = zmq_connect (b, connect_address); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (b, connect_address));
assert (rc == 0);
rc = zmq_msg_recv (&msg, a, ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_msg_recv (&msg, a, ZMQ_DONTWAIT));
assert (rc == -1);
assert (errno == EAGAIN);
rc = zmq_msg_recv (&msg, b, ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_msg_recv (&msg, b, ZMQ_DONTWAIT));
assert (rc == -1);
assert (errno == EAGAIN);
rc = zmq_msg_close (&msg); TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_close (&msg));
assert (rc == 0);
close_zero_linger (a); test_context_socket_close_zero_linger (a);
close_zero_linger (b); test_context_socket_close_zero_linger (b);
// Wait for disconnects. // Wait for disconnects.
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
} }
#define TEST_SUITE(name, bind_address) \
int main (void) void test_fair_queue_in_##name () { test_fair_queue_in (bind_address); } \
{ void test_destroy_queue_on_disconnect_##name () \
setup_test_environment (); { \
void *ctx = zmq_ctx_new (); test_destroy_queue_on_disconnect (bind_address); \
assert (ctx);
const char *binds[] = {"inproc://a", "tcp://127.0.0.1:*"};
for (int transport = 0; transport < 2; ++transport) {
bind_address = binds[transport];
// SHALL receive incoming messages from its peers using a fair-queuing
// strategy.
test_fair_queue_in (ctx);
// SHALL create a double queue when a peer connects to it. If this peer
// disconnects, the ROUTER socket SHALL destroy its double queue and SHALL
// discard any messages it contains.
// *** Test disabled until libzmq does this properly ***
// test_destroy_queue_on_disconnect (ctx);
} }
int rc = zmq_ctx_term (ctx); TEST_SUITE (inproc, "inproc://a")
assert (rc == 0); TEST_SUITE (tcp, "tcp://127.0.0.1:*")
return 0; int main ()
{
setup_test_environment ();
UNITY_BEGIN ();
RUN_TEST (test_fair_queue_in_tcp);
RUN_TEST (test_fair_queue_in_inproc);
// TODO commented out until libzmq implements this properly
// RUN_TEST (test_destroy_queue_on_disconnect_tcp);
// RUN_TEST (test_destroy_queue_on_disconnect_inproc);
return UNITY_END ();
} }

239
tests/test_term_endpoint.cpp
View File

@ -27,200 +27,165 @@
along with this program. If not, see <http://www.gnu.org/licenses/>. along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include <stdio.h>
#include "testutil.hpp" #include "testutil.hpp"
#include "testutil_unity.hpp"
/* Use the worst case filename size for the buffer (+1 for trailing NUL) */ void setUp ()
{
setup_test_context ();
}
void tearDown ()
{
teardown_test_context ();
}
/* Use the worst case filename size for the buffer (+1 for trailing NUL), this
* is larger than MAX_SOCKET_STRING, which is not large enough for IPC */
#define BUF_SIZE (FILENAME_MAX + 1) #define BUF_SIZE (FILENAME_MAX + 1)
int main (void) const char *ep_wc_tcp = "tcp://127.0.0.1:*";
{
setup_test_environment ();
int rc;
char buf[BUF_SIZE];
size_t buf_size;
const char *ep_wc_tcp = "tcp://127.0.0.1:*";
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
const char *ep_wc_ipc = "ipc://*"; const char *ep_wc_ipc = "ipc://*";
#endif #endif
#if defined ZMQ_HAVE_VMCI #if defined ZMQ_HAVE_VMCI
const char *ep_wc_vmci = "vmci://*:*"; const char *ep_wc_vmci = "vmci://*:*";
#endif #endif
void test_send_after_unbind_fails ()
{
char my_endpoint[BUF_SIZE];
// Create infrastructure. // Create infrastructure.
void *ctx = zmq_ctx_new (); void *push = test_context_socket (ZMQ_PUSH);
assert (ctx); bind_loopback_ipv4 (push, my_endpoint, BUF_SIZE);
void *push = zmq_socket (ctx, ZMQ_PUSH);
assert (push); void *pull = test_context_socket (ZMQ_PULL);
rc = zmq_bind (push, ep_wc_tcp); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (pull, my_endpoint));
assert (rc == 0);
buf_size = sizeof (buf);
rc = zmq_getsockopt (push, ZMQ_LAST_ENDPOINT, buf, &buf_size);
assert (rc == 0);
void *pull = zmq_socket (ctx, ZMQ_PULL);
assert (pull);
rc = zmq_connect (pull, buf);
assert (rc == 0);
// Pass one message through to ensure the connection is established // Pass one message through to ensure the connection is established
rc = zmq_send (push, "ABC", 3, 0); send_string_expect_success (push, "ABC", 0);
assert (rc == 3); recv_string_expect_success (pull, "ABC", 0);
rc = zmq_recv (pull, buf, sizeof (buf), 0);
assert (rc == 3);
// Unbind the listening endpoint // Unbind the listening endpoint
buf_size = sizeof (buf); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (push, my_endpoint));
rc = zmq_getsockopt (push, ZMQ_LAST_ENDPOINT, buf, &buf_size);
assert (rc == 0);
rc = zmq_unbind (push, buf);
assert (rc == 0);
// Allow unbind to settle // Allow unbind to settle
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
// Check that sending would block (there's no outbound connection) // Check that sending would block (there's no outbound connection)
rc = zmq_send (push, "ABC", 3, ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_send (push, "ABC", 3, ZMQ_DONTWAIT));
assert (rc == -1 && zmq_errno () == EAGAIN);
// Clean up // Clean up
rc = zmq_close (pull); test_context_socket_close (pull);
assert (rc == 0); test_context_socket_close (push);
rc = zmq_close (push); }
assert (rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
void test_send_after_disconnect_fails ()
{
// Create infrastructure // Create infrastructure
ctx = zmq_ctx_new (); void *pull = test_context_socket (ZMQ_PULL);
assert (ctx); char my_endpoint[BUF_SIZE];
pull = zmq_socket (ctx, ZMQ_PULL); bind_loopback_ipv4 (pull, my_endpoint, BUF_SIZE);
assert (pull);
rc = zmq_bind (pull, ep_wc_tcp); void *push = test_context_socket (ZMQ_PUSH);
assert (rc == 0); TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (push, my_endpoint));
buf_size = sizeof (buf);
rc = zmq_getsockopt (pull, ZMQ_LAST_ENDPOINT, buf, &buf_size);
assert (rc == 0);
push = zmq_socket (ctx, ZMQ_PUSH);
assert (push);
rc = zmq_connect (push, buf);
assert (rc == 0);
// Pass one message through to ensure the connection is established. // Pass one message through to ensure the connection is established.
rc = zmq_send (push, "ABC", 3, 0); send_string_expect_success (push, "ABC", 0);
assert (rc == 3); recv_string_expect_success (pull, "ABC", 0);
rc = zmq_recv (pull, buf, sizeof (buf), 0);
assert (rc == 3);
// Disconnect the bound endpoint // Disconnect the bound endpoint
buf_size = sizeof (buf); TEST_ASSERT_SUCCESS_ERRNO (zmq_disconnect (push, my_endpoint));
rc = zmq_getsockopt (pull, ZMQ_LAST_ENDPOINT, buf, &buf_size);
assert (rc == 0);
rc = zmq_disconnect (push, buf);
assert (rc == 0);
// Allow disconnect to settle // Allow disconnect to settle
msleep (SETTLE_TIME); msleep (SETTLE_TIME);
// Check that sending would block (there's no inbound connections). // Check that sending would block (there's no inbound connections).
rc = zmq_send (push, "ABC", 3, ZMQ_DONTWAIT); TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_send (push, "ABC", 3, ZMQ_DONTWAIT));
assert (rc == -1 && zmq_errno () == EAGAIN);
// Clean up. // Clean up
rc = zmq_close (pull); test_context_socket_close (pull);
assert (rc == 0); test_context_socket_close (push);
rc = zmq_close (push); }
assert (rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
void test_unbind_via_last_endpoint ()
{
// Create infrastructure (wild-card binding) // Create infrastructure (wild-card binding)
ctx = zmq_ctx_new (); void *push = test_context_socket (ZMQ_PUSH);
assert (ctx); char my_endpoint[BUF_SIZE];
push = zmq_socket (ctx, ZMQ_PUSH); bind_loopback_ipv4 (push, my_endpoint, BUF_SIZE);
assert (push);
rc = zmq_bind (push, ep_wc_tcp); void *pull = test_context_socket (ZMQ_PULL);
assert (rc == 0);
pull = zmq_socket (ctx, ZMQ_PULL);
assert (pull);
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
rc = zmq_bind (pull, ep_wc_ipc); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pull, ep_wc_ipc));
assert (rc == 0);
#endif #endif
#if defined ZMQ_HAVE_VMCI #if defined ZMQ_HAVE_VMCI
void *req = zmq_socket (ctx, ZMQ_REQ); void *req = test_context_socket (ZMQ_REQ);
assert (req); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (req, ep_wc_vmci));
rc = zmq_bind (req, ep_wc_vmci);
assert (rc == 0);
#endif #endif
// Unbind sockets binded by wild-card address // Unbind sockets binded by wild-card address
buf_size = sizeof (buf); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (push, my_endpoint));
rc = zmq_getsockopt (push, ZMQ_LAST_ENDPOINT, buf, &buf_size);
assert (rc == 0); size_t buf_size = 0;
rc = zmq_unbind (push, buf); (void) buf_size;
assert (rc == 0);
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
buf_size = sizeof (buf); buf_size = sizeof (my_endpoint);
rc = zmq_getsockopt (pull, ZMQ_LAST_ENDPOINT, buf, &buf_size); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (pull, ZMQ_LAST_ENDPOINT, my_endpoint, &buf_size));
rc = zmq_unbind (pull, buf); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (pull, my_endpoint));
assert (rc == 0);
#endif #endif
#if defined ZMQ_HAVE_VMCI #if defined ZMQ_HAVE_VMCI
buf_size = sizeof (buf); buf_size = sizeof (my_endpoint);
rc = zmq_getsockopt (req, ZMQ_LAST_ENDPOINT, buf, &buf_size); TEST_ASSERT_SUCCESS_ERRNO (
assert (rc == 0); zmq_getsockopt (req, ZMQ_LAST_ENDPOINT, my_endpoint, &buf_size));
rc = zmq_unbind (req, buf); TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (req, my_endpoint));
assert (rc == 0);
#endif #endif
// Clean up. // Clean up
rc = zmq_close (pull); test_context_socket_close (pull);
assert (rc == 0); test_context_socket_close (push);
rc = zmq_close (push); }
assert (rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
void test_wildcard_unbind_fails ()
{
// Create infrastructure (wild-card binding) // Create infrastructure (wild-card binding)
ctx = zmq_ctx_new (); void *push = test_context_socket (ZMQ_PUSH);
assert (ctx); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (push, ep_wc_tcp));
push = zmq_socket (ctx, ZMQ_PUSH); void *pull = test_context_socket (ZMQ_PULL);
assert (push);
rc = zmq_bind (push, ep_wc_tcp);
assert (rc == 0);
pull = zmq_socket (ctx, ZMQ_PULL);
assert (pull);
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
rc = zmq_bind (pull, ep_wc_ipc); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (pull, ep_wc_ipc));
assert (rc == 0);
#endif #endif
#if defined ZMQ_HAVE_VMCI #if defined ZMQ_HAVE_VMCI
req = zmq_socket (ctx, ZMQ_REQ); void *req = test_context_socket (ZMQ_REQ);
assert (req); TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (req, ep_wc_vmci));
rc = zmq_bind (req, ep_wc_vmci);
assert (rc == 0);
#endif #endif
// Sockets binded by wild-card address can't be unbinded by wild-card address // Sockets binded by wild-card address can't be unbinded by wild-card address
rc = zmq_unbind (push, ep_wc_tcp); TEST_ASSERT_FAILURE_ERRNO (ENOENT, zmq_unbind (push, ep_wc_tcp));
assert (rc == -1 && zmq_errno () == ENOENT);
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS #if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
rc = zmq_unbind (pull, ep_wc_ipc); TEST_ASSERT_FAILURE_ERRNO (ENOENT, zmq_unbind (pull, ep_wc_ipc));
assert (rc == -1 && zmq_errno () == ENOENT);
#endif #endif
#if defined ZMQ_HAVE_VMCI #if defined ZMQ_HAVE_VMCI
rc = zmq_unbind (req, ep_wc_vmci); TEST_ASSERT_FAILURE_ERRNO (ENOENT, zmq_unbind (req, ep_wc_vmci));
assert (rc == -1 && zmq_errno () == ENOENT);
#endif #endif
// Clean up. // Clean up
rc = zmq_close (pull); test_context_socket_close (pull);
assert (rc == 0); test_context_socket_close (push);
rc = zmq_close (push); }
assert (rc == 0);
rc = zmq_ctx_term (ctx); int main ()
assert (rc == 0); {
setup_test_environment ();
return 0;
UNITY_BEGIN ();
RUN_TEST (test_send_after_unbind_fails);
RUN_TEST (test_send_after_disconnect_fails);
RUN_TEST (test_unbind_via_last_endpoint);
RUN_TEST (test_wildcard_unbind_fails);
return UNITY_END ();
} }

4
tests/testutil_unity.hpp
View File

@ -90,8 +90,8 @@ int test_assert_success_message_raw_errno_helper (int rc_,
#define TEST_ASSERT_FAILURE_ERRNO(error_code, expr) \ #define TEST_ASSERT_FAILURE_ERRNO(error_code, expr) \
{ \ { \
int rc = (expr); \ int _rc = (expr); \
TEST_ASSERT_EQUAL_INT (-1, rc); \ TEST_ASSERT_EQUAL_INT (-1, _rc); \
TEST_ASSERT_EQUAL_INT (error_code, errno); \ TEST_ASSERT_EQUAL_INT (error_code, errno); \
} }