Skip to content

L
libzmq
Commit 30b9e5a1 authored by Simon Giesecke's avatar Simon Giesecke
Browse files
Options

Problem: test_connect_rid not yet using unity 

Solution: migrate to unity
parent 0fc2f0f0
Hide whitespace changes
Inline Side-by-side
Showing with 142 additions and 185 deletions
Makefile.am
View file @ 30b9e5a1
......@@ -633,7 +633,8 @@ tests_test_diffserv_SOURCES = tests/test_diffserv.cpp
tests_test_diffserv_LDADD = src/libzmq.la
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_LDADD = src/libzmq.la ${UNITY_LIBS}
......
tests/test_connect_rid.cpp
View file @ 30b9e5a1
......@@ -28,275 +28,231 @@
*/
#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 *rbind, *rconn1;
int ret;
char buff[256];
char msg[] = "hi 1";
const char *bindip = "tcp://127.0.0.1:*";
int disabled = 0;
int zero = 0;
size_t len = MAX_SOCKET_STRING;
const char msg[] = "hi 1";
const int disabled = 0;
const int zero = 0;
char my_endpoint[MAX_SOCKET_STRING];
void *ctx = zmq_ctx_new ();
// Set up listener STREAM.
rbind = zmq_socket (ctx, ZMQ_STREAM);
assert (rbind);
ret =
zmq_setsockopt (rbind, ZMQ_STREAM_NOTIFY, &disabled, sizeof (disabled));
assert (ret == 0);
ret = zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
ret = zmq_bind (rbind, bindip);
assert (0 == ret);
ret = zmq_getsockopt (rbind, ZMQ_LAST_ENDPOINT, my_endpoint, &len);
assert (0 == ret);
void *rbind = test_context_socket (ZMQ_STREAM);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (rbind, ZMQ_STREAM_NOTIFY, &disabled, sizeof (disabled)));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof zero));
bind_loopback_ipv4 (rbind, my_endpoint, sizeof my_endpoint);
// Set up connection stream.
rconn1 = zmq_socket (ctx, ZMQ_STREAM);
assert (rconn1);
ret = zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
void *rconn1 = test_context_socket (ZMQ_STREAM);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof zero));
// Do the connection.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6);
assert (0 == ret);
ret = zmq_connect (rconn1, my_endpoint);
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID,
rconn1routing_id,
strlen (rconn1routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, my_endpoint));
/* Uncomment to test assert on duplicate routing id.
// Test duplicate connect attempt.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6);
assert (0 == ret);
ret = zmq_connect (rconn1, bindip);
assert (0 == ret);
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, rconn1routing_id, strlen(rconn1routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, bindip));
*/
// Send data to the bound stream.
ret = zmq_send (rconn1, "conn1", 6, ZMQ_SNDMORE);
assert (6 == ret);
ret = zmq_send (rconn1, msg, 5, 0);
assert (5 == ret);
send_string_expect_success (rconn1, rconn1routing_id, ZMQ_SNDMORE);
send_string_expect_success (rconn1, msg, 0);
// Accept data on the bound stream.
ret = zmq_recv (rbind, buff, 256, 0);
assert (ret);
assert (0 == buff[0]);
ret = zmq_recv (rbind, buff + 128, 128, 0);
assert (5 == ret);
assert ('h' == buff[128]);
TEST_ASSERT_GREATER_THAN (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (rbind, buff, 256, 0)));
TEST_ASSERT_EQUAL (0, buff[0]); // an auto-generated routing id
recv_string_expect_success (rbind, msg, 0);
// Handle close of the socket.
ret = zmq_unbind (rbind, my_endpoint);
assert (0 == ret);
ret = zmq_close (rbind);
assert (0 == ret);
ret = zmq_close (rconn1);
assert (0 == ret);
zmq_ctx_destroy (ctx);
TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (rbind, my_endpoint));
test_context_socket_close (rbind);
test_context_socket_close (rconn1);
}
void test_router_2_router (bool named_)
{
void *rbind, *rconn1;
int ret;
char buff[256];
char msg[] = "hi 1";
const char *bindip = "tcp://127.0.0.1:*";
int zero = 0;
size_t len = MAX_SOCKET_STRING;
const char msg[] = "hi 1";
const int zero = 0;
char my_endpoint[MAX_SOCKET_STRING];
void *ctx = zmq_ctx_new ();
// Create bind socket.
rbind = zmq_socket (ctx, ZMQ_ROUTER);
assert (rbind);
ret = zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
ret = zmq_bind (rbind, bindip);
assert (0 == ret);
ret = zmq_getsockopt (rbind, ZMQ_LAST_ENDPOINT, my_endpoint, &len);
assert (0 == ret);
void *rbind = test_context_socket (ZMQ_ROUTER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (rbind, ZMQ_LINGER, &zero, sizeof (zero)));
bind_loopback_ipv4 (rbind, my_endpoint, sizeof my_endpoint);
// Create connection socket.
rconn1 = zmq_socket (ctx, ZMQ_ROUTER);
assert (rconn1);
ret = zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
void *rconn1 = test_context_socket (ZMQ_ROUTER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (rconn1, ZMQ_LINGER, &zero, sizeof (zero)));
// If we're in named mode, set some identities.
if (named_) {
ret = zmq_setsockopt (rbind, ZMQ_ROUTING_ID, "X", 1);
ret = zmq_setsockopt (rconn1, ZMQ_ROUTING_ID, "Y", 1);
TEST_ASSERT_SUCCESS_ERRNO (
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.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6);
assert (0 == ret);
ret = zmq_connect (rconn1, my_endpoint);
assert (0 == ret);
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID,
rconn1routing_id,
strlen (rconn1routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, my_endpoint));
/* Uncomment to test assert on duplicate routing id
// Test duplicate connect attempt.
ret = zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, "conn1", 6);
assert (0 == ret);
ret = zmq_connect (rconn1, bindip);
assert (0 == ret);
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (rconn1, ZMQ_CONNECT_ROUTING_ID, rconn1routing_id, strlen (rconn1routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (rconn1, bindip));
*/
// Send some data.
ret = zmq_send (rconn1, "conn1", 6, ZMQ_SNDMORE);
assert (6 == ret);
ret = zmq_send (rconn1, msg, 5, 0);
assert (5 == ret);
send_string_expect_success (rconn1, rconn1routing_id, ZMQ_SNDMORE);
send_string_expect_success (rconn1, msg, 0);
// Receive the name.
ret = zmq_recv (rbind, buff, 256, 0);
if (named_)
assert (ret && 'Y' == buff[0]);
else
assert (ret && 0 == buff[0]);
const int routing_id_len = zmq_recv (rbind, buff, 256, 0);
if (named_) {
TEST_ASSERT_EQUAL_INT (strlen (y_routing_id), routing_id_len);
TEST_ASSERT_EQUAL_STRING_LEN (y_routing_id, buff, routing_id_len);
} else {
TEST_ASSERT_TRUE (routing_id_len && 0 == buff[0]);
}
// Receive the data.
ret = zmq_recv (rbind, buff + 128, 128, 0);
assert (5 == ret && 'h' == buff[128]);
recv_string_expect_success (rbind, msg, 0);
// Send some data back.
if (named_) {
ret = zmq_send (rbind, buff, 1, ZMQ_SNDMORE);
assert (1 == ret);
} else {
ret = zmq_send (rbind, buff, 5, ZMQ_SNDMORE);
assert (5 == ret);
}
ret = zmq_send_const (rbind, "ok", 3, 0);
assert (3 == ret);
const int ret = zmq_send (rbind, buff, routing_id_len, ZMQ_SNDMORE);
TEST_ASSERT_EQUAL_INT (routing_id_len, ret);
send_string_expect_success (rbind, "ok", 0);
// If bound socket identity naming a problem, we'll likely see something funky here.
ret = zmq_recv (rconn1, buff, 256, 0);
assert ('c' == buff[0] && 6 == ret);
ret = zmq_recv (rconn1, buff + 128, 128, 0);
assert (3 == ret && 'o' == buff[128]);
ret = zmq_unbind (rbind, my_endpoint);
assert (0 == ret);
ret = zmq_close (rbind);
assert (0 == ret);
ret = zmq_close (rconn1);
assert (0 == ret);
zmq_ctx_destroy (ctx);
recv_string_expect_success (rconn1, rconn1routing_id, 0);
recv_string_expect_success (rconn1, "ok", 0);
TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (rbind, my_endpoint));
test_context_socket_close (rbind);
test_context_socket_close (rconn1);
}
void test_router_2_router_while_receiving ()
{
void *xbind, *zbind, *yconn;
int ret;
char buff[256];
char msg[] = "hi 1";
const char *wildcard_bind = "tcp://127.0.0.1:*";
int zero = 0;
size_t len = MAX_SOCKET_STRING;
const char msg[] = "hi 1";
const int zero = 0;
char x_endpoint[MAX_SOCKET_STRING];
char z_endpoint[MAX_SOCKET_STRING];
void *ctx = zmq_ctx_new ();
// Create xbind socket.
xbind = zmq_socket (ctx, ZMQ_ROUTER);
assert (xbind);
ret = zmq_setsockopt (xbind, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
ret = zmq_bind (xbind, wildcard_bind);
assert (0 == ret);
ret = zmq_getsockopt (xbind, ZMQ_LAST_ENDPOINT, x_endpoint, &len);
assert (0 == ret);
void *xbind = test_context_socket (ZMQ_ROUTER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (xbind, ZMQ_LINGER, &zero, sizeof (zero)));
bind_loopback_ipv4 (xbind, x_endpoint, sizeof x_endpoint);
// Create zbind socket.
zbind = zmq_socket (ctx, ZMQ_ROUTER);
assert (zbind);
ret = zmq_setsockopt (zbind, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
ret = zmq_bind (zbind, wildcard_bind);
assert (0 == ret);
ret = zmq_getsockopt (zbind, ZMQ_LAST_ENDPOINT, z_endpoint, &len);
assert (0 == ret);
void *zbind = test_context_socket (ZMQ_ROUTER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (zbind, ZMQ_LINGER, &zero, sizeof (zero)));
bind_loopback_ipv4 (zbind, z_endpoint, sizeof z_endpoint);
// Create connection socket.
yconn = zmq_socket (ctx, ZMQ_ROUTER);
assert (yconn);
ret = zmq_setsockopt (yconn, ZMQ_LINGER, &zero, sizeof (zero));
assert (0 == ret);
// set identites for each socket
ret = zmq_setsockopt (xbind, ZMQ_ROUTING_ID, "X", 2);
ret = zmq_setsockopt (yconn, ZMQ_ROUTING_ID, "Y", 2);
ret = zmq_setsockopt (zbind, ZMQ_ROUTING_ID, "Z", 2);
void *yconn = test_context_socket (ZMQ_ROUTER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (yconn, ZMQ_LINGER, &zero, sizeof (zero)));
// set identities for each socket
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
xbind, ZMQ_ROUTING_ID, x_routing_id, strlen (x_routing_id)));
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
ret = zmq_setsockopt (yconn, ZMQ_CONNECT_ROUTING_ID, "X", 2);
assert (0 == ret);
ret = zmq_connect (yconn, x_endpoint);
assert (0 == ret);
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
yconn, ZMQ_CONNECT_ROUTING_ID, x_routing_id, strlen (x_routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (yconn, x_endpoint));
// Send some data from Y to X.
ret = zmq_send (yconn, "X", 2, ZMQ_SNDMORE);
assert (2 == ret);
ret = zmq_send (yconn, msg, 5, 0);
assert (5 == ret);
send_string_expect_success (yconn, x_routing_id, ZMQ_SNDMORE);
send_string_expect_success (yconn, msg, 0);
// wait for the Y->X message to be received
msleep (SETTLE_TIME);
// Now X tries to connect to Z and send a message
ret = zmq_setsockopt (xbind, ZMQ_CONNECT_ROUTING_ID, "Z", 2);
assert (0 == ret);
ret = zmq_connect (xbind, z_endpoint);
assert (0 == ret);
TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
xbind, ZMQ_CONNECT_ROUTING_ID, z_routing_id, strlen (z_routing_id)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (xbind, z_endpoint));
// Try to send some data from X to Z.
ret = zmq_send (xbind, "Z", 2, ZMQ_SNDMORE);
assert (2 == ret);
ret = zmq_send (xbind, msg, 5, 0);
assert (5 == ret);
send_string_expect_success (xbind, z_routing_id, ZMQ_SNDMORE);
send_string_expect_success (xbind, msg, 0);
// 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)
msleep (SETTLE_TIME);
// nothing should have been received on the Y socket
ret = zmq_recv (yconn, buff, 256, ZMQ_DONTWAIT);
assert (ret == -1);
assert (zmq_errno () == EAGAIN);
TEST_ASSERT_FAILURE_ERRNO (EAGAIN,
zmq_recv (yconn, buff, 256, ZMQ_DONTWAIT));
// the message should have been received on the Z socket
ret = zmq_recv (zbind, buff, 256, 0);
assert (ret && 'X' == buff[0]);
ret = zmq_recv (zbind, buff + 128, 128, 0);
assert (5 == ret && 'h' == buff[128]);
ret = zmq_unbind (xbind, x_endpoint);
assert (0 == ret);
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);
recv_string_expect_success (zbind, x_routing_id, 0);
recv_string_expect_success (zbind, msg, 0);
TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (xbind, x_endpoint));
TEST_ASSERT_SUCCESS_ERRNO (zmq_unbind (zbind, z_endpoint));
test_context_socket_close (yconn);
test_context_socket_close (xbind);
test_context_socket_close (zbind);
}
int main (void)
void test_router_2_router_unnamed ()
{
setup_test_environment ();
test_stream_2_stream ();
test_router_2_router (false);
}
void test_router_2_router_named ()
{
test_router_2_router (true);
test_router_2_router_while_receiving ();
}
int main ()
{
setup_test_environment ();
return 0;
UNITY_BEGIN ();
RUN_TEST (test_stream_2_stream);
RUN_TEST (test_router_2_router_unnamed);
RUN_TEST (test_router_2_router_named);
RUN_TEST (test_router_2_router_while_receiving);
return UNITY_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