Skip to content

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

Problem: test_stream not yet using unity 

Solution: migrate to unity
parent 7ad0ff5e
Hide whitespace changes
Inline Side-by-side
Showing with 134 additions and 160 deletions
Makefile.am
View file @ 23340907
......@@ -544,7 +544,8 @@ tests_test_probe_router_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_probe_router_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_stream_SOURCES = tests/test_stream.cpp
tests_test_stream_LDADD = src/libzmq.la
tests_test_stream_LDADD = src/libzmq.la ${UNITY_LIBS}
tests_test_stream_CPPFLAGS = ${UNITY_CPPFLAGS}
tests_test_stream_empty_SOURCES = tests/test_stream_empty.cpp
tests_test_stream_empty_LDADD = src/libzmq.la ${UNITY_LIBS}
......
tests/test_stream.cpp
View file @ 23340907
......@@ -28,10 +28,21 @@
*/
#include "testutil.hpp"
#include "testutil_unity.hpp"
void setUp ()
{
setup_test_context ();
}
void tearDown ()
{
teardown_test_context ();
}
// ZMTP protocol greeting structure
typedef unsigned char byte;
typedef uint8_t byte;
typedef struct
{
byte signature[10]; // 0xFF 8*0x00 0x7F
......@@ -49,113 +60,97 @@ typedef struct
static zmtp_greeting_t greeting = {
{0xFF, 0, 0, 0, 0, 0, 0, 0, 1, 0x7F}, {3, 0}, {'N', 'U', 'L', 'L'}, 0, {0}};
static void test_stream_to_dealer (void)
static void test_stream_to_dealer ()
{
int rc;
size_t len = MAX_SOCKET_STRING;
char my_endpoint[MAX_SOCKET_STRING];
// Set up our context and sockets
void *ctx = zmq_ctx_new ();
assert (ctx);
// We'll be using this socket in raw mode
void *stream = zmq_socket (ctx, ZMQ_STREAM);
assert (stream);
void *stream = test_context_socket (ZMQ_STREAM);
int zero = 0;
rc = zmq_setsockopt (stream, ZMQ_LINGER, &zero, sizeof (zero));
assert (rc == 0);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (stream, ZMQ_LINGER, &zero, sizeof (zero)));
int enabled = 1;
rc = zmq_setsockopt (stream, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled));
assert (rc == 0);
rc = zmq_bind (stream, "tcp://127.0.0.1:*");
assert (rc == 0);
rc = zmq_getsockopt (stream, ZMQ_LAST_ENDPOINT, my_endpoint, &len);
assert (rc == 0);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (stream, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled)));
bind_loopback_ipv4 (stream, my_endpoint, sizeof my_endpoint);
// We'll be using this socket as the other peer
void *dealer = zmq_socket (ctx, ZMQ_DEALER);
assert (dealer);
rc = zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero));
assert (rc == 0);
rc = zmq_connect (dealer, my_endpoint);
void *dealer = test_context_socket (ZMQ_DEALER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (dealer, ZMQ_LINGER, &zero, sizeof (zero)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (dealer, my_endpoint));
// Send a message on the dealer socket
rc = zmq_send (dealer, "Hello", 5, 0);
assert (rc == 5);
send_string_expect_success (dealer, "Hello", 0);
// Connecting sends a zero message
// First frame is routing id
zmq_msg_t routing_id;
rc = zmq_msg_init (&routing_id);
assert (rc == 0);
rc = zmq_msg_recv (&routing_id, stream, 0);
assert (rc > 0);
assert (zmq_msg_more (&routing_id));
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&routing_id));
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Verify the existence of Peer-Address metadata
char const *peer_address = zmq_msg_gets (&routing_id, "Peer-Address");
assert (peer_address != 0);
assert (streq (peer_address, "127.0.0.1"));
TEST_ASSERT_NOT_NULL (peer_address);
TEST_ASSERT_EQUAL_STRING ("127.0.0.1", peer_address);
// Second frame is zero
byte buffer[255];
rc = zmq_recv (stream, buffer, 255, 0);
assert (rc == 0);
TEST_ASSERT_EQUAL_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (stream, buffer, 255, 0)));
// Verify the existence of Peer-Address metadata
peer_address = zmq_msg_gets (&routing_id, "Peer-Address");
assert (peer_address != 0);
assert (streq (peer_address, "127.0.0.1"));
TEST_ASSERT_NOT_NULL (peer_address);
TEST_ASSERT_EQUAL_STRING ("127.0.0.1", peer_address);
// Real data follows
// First frame is routing id
rc = zmq_msg_recv (&routing_id, stream, 0);
assert (rc > 0);
assert (zmq_msg_more (&routing_id));
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Verify the existence of Peer-Address metadata
peer_address = zmq_msg_gets (&routing_id, "Peer-Address");
assert (peer_address != 0);
assert (streq (peer_address, "127.0.0.1"));
TEST_ASSERT_NOT_NULL (peer_address);
TEST_ASSERT_EQUAL_STRING ("127.0.0.1", peer_address);
// Second frame is greeting signature
rc = zmq_recv (stream, buffer, 255, 0);
assert (rc == 10);
assert (memcmp (buffer, greeting.signature, 10) == 0);
recv_array_expect_success (stream, greeting.signature, 0);
// Send our own protocol greeting
rc = zmq_msg_send (&routing_id, stream, ZMQ_SNDMORE);
assert (rc > 0);
rc = zmq_send (stream, &greeting, sizeof (greeting), 0);
assert (rc == sizeof (greeting));
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (&routing_id, stream, ZMQ_SNDMORE));
TEST_ASSERT_EQUAL_INT (
sizeof (greeting), TEST_ASSERT_SUCCESS_ERRNO (
zmq_send (stream, &greeting, sizeof (greeting), 0)));
// Now we expect the data from the DEALER socket
// We want the rest of greeting along with the Ready command
int bytes_read = 0;
while (bytes_read < 97) {
// First frame is the routing id of the connection (each time)
rc = zmq_msg_recv (&routing_id, stream, 0);
assert (rc > 0);
assert (zmq_msg_more (&routing_id));
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0)));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Second frame contains the next chunk of data
rc = zmq_recv (stream, buffer + bytes_read, 255 - bytes_read, 0);
assert (rc >= 0);
TEST_ASSERT_SUCCESS_ERRNO (
rc = zmq_recv (stream, buffer + bytes_read, 255 - bytes_read, 0));
bytes_read += rc;
}
// First two bytes are major and minor version numbers.
assert (buffer[0] == 3); // ZMTP/3.0
assert (buffer[1] == 0);
TEST_ASSERT_EQUAL_INT (3, buffer[0]); // ZMTP/3.0
TEST_ASSERT_EQUAL_INT (0, buffer[1]);
// Mechanism is "NULL"
assert (memcmp (buffer + 2, "NULL\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20)
== 0);
assert (memcmp (buffer + 54, "\4\51\5READY", 8) == 0);
assert (memcmp (buffer + 62, "\13Socket-Type\0\0\0\6DEALER", 22) == 0);
assert (memcmp (buffer + 84, "\10Identity\0\0\0\0", 13) == 0);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 2,
"NULL\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 54, "\4\51\5READY", 8);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 62, "\13Socket-Type\0\0\0\6DEALER",
22);
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 84, "\10Identity\0\0\0\0", 13);
// Announce we are ready
memcpy (buffer, "\4\51\5READY", 8);
......@@ -163,37 +158,36 @@ static void test_stream_to_dealer (void)
memcpy (buffer + 30, "\10Identity\0\0\0\0", 13);
// Send Ready command
rc = zmq_msg_send (&routing_id, stream, ZMQ_SNDMORE);
assert (rc > 0);
rc = zmq_send (stream, buffer, 43, 0);
assert (rc == 43);
TEST_ASSERT_GREATER_THAN_INT (0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (
&routing_id, stream, ZMQ_SNDMORE)));
TEST_ASSERT_EQUAL_INT (
43, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (stream, buffer, 43, 0)));
// Now we expect the data from the DEALER socket
// First frame is, again, the routing id of the connection
rc = zmq_msg_recv (&routing_id, stream, 0);
assert (rc > 0);
assert (zmq_msg_more (&routing_id));
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_recv (&routing_id, stream, 0)));
TEST_ASSERT_TRUE (zmq_msg_more (&routing_id));
// Third frame contains Hello message from DEALER
rc = zmq_recv (stream, buffer, sizeof buffer, 0);
assert (rc == 7);
TEST_ASSERT_EQUAL_INT (7, TEST_ASSERT_SUCCESS_ERRNO (
zmq_recv (stream, buffer, sizeof buffer, 0)));
// Then we have a 5-byte message "Hello"
assert (buffer[0] == 0); // Flags = 0
assert (buffer[1] == 5); // Size = 5
assert (memcmp (buffer + 2, "Hello", 5) == 0);
TEST_ASSERT_EQUAL_INT (0, buffer[0]); // Flags = 0
TEST_ASSERT_EQUAL_INT (5, buffer[1]); // Size = 5
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer + 2, "Hello", 5);
// Send "World" back to DEALER
rc = zmq_msg_send (&routing_id, stream, ZMQ_SNDMORE);
assert (rc > 0);
TEST_ASSERT_GREATER_THAN_INT (0, TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_send (
&routing_id, stream, ZMQ_SNDMORE)));
byte world[] = {0, 5, 'W', 'o', 'r', 'l', 'd'};
rc = zmq_send (stream, world, sizeof (world), 0);
assert (rc == sizeof (world));
TEST_ASSERT_EQUAL_INT (
sizeof (world),
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (stream, world, sizeof (world), 0)));
// Expect response on DEALER socket
rc = zmq_recv (dealer, buffer, 255, 0);
assert (rc == 5);
assert (memcmp (buffer, "World", 5) == 0);
recv_string_expect_success (dealer, "World", 0);
// Test large messages over STREAM socket
#define size 64000
......@@ -206,110 +200,92 @@ static void test_stream_to_dealer (void)
bytes_read = 0;
while (bytes_read < 9 + size) {
// Get routing id frame
rc = zmq_recv (stream, buffer, 256, 0);
assert (rc > 0);
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (stream, buffer, 256, 0)));
// Get next chunk
rc = zmq_recv (stream, msgin + bytes_read, 9 + size - bytes_read, 0);
assert (rc > 0);
TEST_ASSERT_GREATER_THAN_INT (
0,
TEST_ASSERT_SUCCESS_ERRNO (rc = zmq_recv (stream, msgin + bytes_read,
9 + size - bytes_read, 0)));
bytes_read += rc;
}
int byte_nbr;
for (byte_nbr = 0; byte_nbr < size; byte_nbr++) {
if (msgin[9 + byte_nbr] != 0xAB)
assert (false);
for (int byte_nbr = 0; byte_nbr < size; byte_nbr++) {
TEST_ASSERT_EQUAL_UINT8 (0xAB, msgin[9 + byte_nbr]);
}
rc = zmq_close (dealer);
assert (rc == 0);
rc = zmq_close (stream);
assert (rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
test_context_socket_close (dealer);
test_context_socket_close (stream);
}
static void test_stream_to_stream (void)
static void test_stream_to_stream ()
{
int rc;
size_t len = MAX_SOCKET_STRING;
char my_endpoint[MAX_SOCKET_STRING];
// Set-up our context and sockets
void *ctx = zmq_ctx_new ();
assert (ctx);
void *server = zmq_socket (ctx, ZMQ_STREAM);
assert (server);
void *server = test_context_socket (ZMQ_STREAM);
int enabled = 1;
rc = zmq_setsockopt (server, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled));
assert (rc == 0);
rc = zmq_bind (server, "tcp://127.0.0.1:*");
assert (rc == 0);
rc = zmq_getsockopt (server, ZMQ_LAST_ENDPOINT, my_endpoint, &len);
assert (rc == 0);
void *client = zmq_socket (ctx, ZMQ_STREAM);
assert (client);
rc = zmq_setsockopt (client, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled));
assert (rc == 0);
rc = zmq_connect (client, my_endpoint);
assert (rc == 0);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (server, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled)));
bind_loopback_ipv4 (server, my_endpoint, sizeof my_endpoint);
void *client = test_context_socket (ZMQ_STREAM);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (client, ZMQ_STREAM_NOTIFY, &enabled, sizeof (enabled)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (client, my_endpoint));
uint8_t id[256];
size_t id_size = 256;
uint8_t buffer[256];
// Connecting sends a zero message
// Server: First frame is routing id, second frame is zero
id_size = zmq_recv (server, id, 256, 0);
assert (id_size > 0);
rc = zmq_recv (server, buffer, 256, 0);
assert (rc == 0);
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, id, 256, 0)));
TEST_ASSERT_EQUAL_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, buffer, 256, 0)));
// Client: First frame is routing id, second frame is zero
id_size = zmq_recv (client, id, 256, 0);
assert (id_size > 0);
rc = zmq_recv (client, buffer, 256, 0);
assert (rc == 0);
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, id, 256, 0)));
TEST_ASSERT_EQUAL_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, buffer, 256, 0)));
// Sent HTTP request on client socket
// Get server routing id
rc = zmq_getsockopt (client, ZMQ_ROUTING_ID, id, &id_size);
assert (rc == 0);
size_t id_size = sizeof id;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_getsockopt (client, ZMQ_ROUTING_ID, id, &id_size));
// First frame is server routing id
rc = zmq_send (client, id, id_size, ZMQ_SNDMORE);
assert (rc == (int) id_size);
TEST_ASSERT_EQUAL_INT ((int) id_size, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (
client, id, id_size, ZMQ_SNDMORE)));
// Second frame is HTTP GET request
rc = zmq_send (client, "GET /\n\n", 7, 0);
assert (rc == 7);
TEST_ASSERT_EQUAL_INT (
7, TEST_ASSERT_SUCCESS_ERRNO (zmq_send (client, "GET /\n\n", 7, 0)));
// Get HTTP request; ID frame and then request
id_size = zmq_recv (server, id, 256, 0);
assert (id_size > 0);
rc = zmq_recv (server, buffer, 256, 0);
assert (rc != -1);
assert (memcmp (buffer, "GET /\n\n", 7) == 0);
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, id, 256, 0)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (server, buffer, 256, 0));
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer, "GET /\n\n", 7);
// Send reply back to client
char http_response[] = "HTTP/1.0 200 OK\r\n"
"Content-Type: text/plain\r\n"
"\r\n"
"Hello, World!";
rc = zmq_send (server, id, id_size, ZMQ_SNDMORE);
assert (rc != -1);
rc = zmq_send (server, http_response, sizeof (http_response), ZMQ_SNDMORE);
assert (rc != -1);
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (server, id, id_size, ZMQ_SNDMORE));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_send (server, http_response, sizeof (http_response), ZMQ_SNDMORE));
// Send zero to close connection to client
rc = zmq_send (server, id, id_size, ZMQ_SNDMORE);
assert (rc != -1);
rc = zmq_send (server, NULL, 0, ZMQ_SNDMORE);
assert (rc != -1);
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (server, id, id_size, ZMQ_SNDMORE));
TEST_ASSERT_SUCCESS_ERRNO (zmq_send (server, NULL, 0, ZMQ_SNDMORE));
// Get reply at client and check that it's complete
id_size = zmq_recv (client, id, 256, 0);
assert (id_size > 0);
rc = zmq_recv (client, buffer, 256, 0);
assert (rc == sizeof (http_response));
assert (memcmp (buffer, http_response, sizeof (http_response)) == 0);
TEST_ASSERT_GREATER_THAN_INT (
0, TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, id, 256, 0)));
TEST_ASSERT_EQUAL_INT (
sizeof http_response,
TEST_ASSERT_SUCCESS_ERRNO (zmq_recv (client, buffer, 256, 0)));
TEST_ASSERT_EQUAL_INT8_ARRAY (buffer, http_response,
sizeof (http_response));
// // Get disconnection notification
// FIXME: why does this block? Bug in STREAM disconnect notification?
......@@ -318,19 +294,16 @@ static void test_stream_to_stream (void)
// rc = zmq_recv (client, buffer, 256, 0);
// assert (rc == 0);
rc = zmq_close (server);
assert (rc == 0);
rc = zmq_close (client);
assert (rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);
test_context_socket_close (server);
test_context_socket_close (client);
}
int main (void)
int main ()
{
setup_test_environment ();
test_stream_to_dealer ();
test_stream_to_stream ();
UNITY_BEGIN ();
RUN_TEST (test_stream_to_dealer);
RUN_TEST (test_stream_to_stream);
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