test_tport.c

Sofia SIP is an open-source SIP User-Agent library, compliant with the IETF RFC3261 specification.

    *rname = *myname;

    /* Check that we cannot bind to an already used socket */

    memset(su, 0, sulen = sizeof(su->su_sin));
    s = su_socket(su->su_family = AF_INET, SOCK_STREAM, 0); TEST_1(s != -1);
    TEST_1(bind(s, &su->su_sa, sulen) != -1);
    TEST_1(listen(s, 5) != -1);
    TEST_1(getsockname(s, &su->su_sa, &sulen) != -1);

    sprintf(port, "%u", ntohs(su->su_port));

    rname->tpn_port = port;
    rname->tpn_ident = "failure";
    
    before = count_tports(tt->tt_srv_tports);

    /* Bind server transports to an reserved port - this should fail */
    TEST(tport_tbind(tt->tt_srv_tports, rname, transports, 
		     TPTAG_SERVER(1),
		     TAG_END()), 
	 -1);

    after = count_tports(tt->tt_srv_tports);

    /* Check that no new primary transports has been added by failed call */
    TEST(before, after);

    /* Add new transports to an ephemeral port with new identity */

    for (tp = tport_primaries(tt->tt_srv_tports); tp; tp = tport_next(tp))
      TEST_S(tport_name(tp)->tpn_ident, "server");

    rname->tpn_port = "*";
    rname->tpn_ident = "server2";

    /* Bind server transports to another port */
    TEST(tport_tbind(tt->tt_srv_tports, rname, transports, 
		     TPTAG_SERVER(1),
		     TAG_END()), 
	 0);

    /* Check that new transports are after old ones. */
    for (i = 0, tp = tport_primaries(tt->tt_srv_tports);
	 i < before;
	 i++, tp = tport_next(tp))
      TEST_S(tport_name(tp)->tpn_ident, "server");

    for (; tp; tp = tport_next(tp))
      TEST_S(tport_name(tp)->tpn_ident, "server2");
  }

#if HAVE_TLS
  {
    tp_name_t tlsname[1] = {{ "tls", "*", "*", "*", NULL }};
    char const * transports[] = { "tls", NULL };

    char const *srcdir = getenv("srcdir");

    if (srcdir == NULL)
      srcdir = ".";

    tlsname->tpn_host = myname->tpn_host;
    tlsname->tpn_ident = "server";

    /* Bind client transports */
    TEST(tport_tbind(tt->tt_tports, tlsname, transports,
		     TPTAG_SERVER(0), 
		     TPTAG_CERTIFICATE(srcdir),
		     TAG_END()), 
	 0);

    /* Bind tls server transport */
    TEST(tport_tbind(tt->tt_srv_tports, tlsname, transports, 
		     TPTAG_SERVER(1),
		     TPTAG_CERTIFICATE(srcdir),
		     TAG_END()), 
	 0);
  }
#endif

  for (tp = tport_primaries(tt->tt_srv_tports); tp; tp = tport_next(tp)) {
    TEST_1(tpn = tport_name(tp));

    if (tt->tt_flags & tst_verbatim) {
      char const *host = tpn->tpn_host != tpn->tpn_canon ? tpn->tpn_host : "";
      printf("bound transport to %s/%s:%s%s%s%s%s\n",
	     tpn->tpn_proto, tpn->tpn_canon, tpn->tpn_port, 
	     host[0] ? ";maddr=" : "", host,
	     tpn->tpn_comp ? ";comp=" : "", 
	     tpn->tpn_comp ? tpn->tpn_comp : "");
    }

    /* Ignore server2 tports for now */
    if (strcmp(tpn->tpn_ident, "server"))
      continue;

    if (strcmp(tpn->tpn_proto, "udp") == 0) {
      *tt->tt_udp_name = *tpn;
      tt->tt_udp_name->tpn_comp = NULL;
      tt->tt_udp_name->tpn_ident = NULL;
      *tt->tt_udp_comp = *tpn;
      tt->tt_udp_comp->tpn_ident = NULL;
    }
    else if (strcmp(tpn->tpn_proto, "tcp") == 0) {
      *tt->tt_tcp_name = *tpn;
      tt->tt_tcp_name->tpn_comp = NULL;
      tt->tt_tcp_name->tpn_ident = NULL;
      *tt->tt_tcp_comp = *tpn;
      tt->tt_tcp_comp->tpn_ident = NULL;

      if (tt->tt_tcp_addr == NULL) {
	tt->tt_tcp_addr = tport_get_address(tp); 
	tt->tt_tcp = tp;
      }
    } 
    else if (strcmp(tpn->tpn_proto, "sctp") == 0) {
      *tt->tt_sctp_name = *tpn;
      tt->tt_sctp_name->tpn_ident = NULL;
    }
    else if (strcmp(tpn->tpn_proto, "tls") == 0) {
      *tt->tt_tls_name = *tpn;
      tt->tt_tls_name->tpn_ident = NULL;
    }
  }

  END();
}

char const payload[] = 
"Some data\n"
"More data\n";

#include <time.h>

int 
tport_test_run(tp_test_t *tt, unsigned timeout)
{
  time_t now = time(NULL);

  tt->tt_status = 0;

  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;
  tt->tt_rtport = NULL;

  while (!tt->tt_status) {
    if (tt->tt_flags & tst_verbatim) {
      fputs(".", stdout); fflush(stdout);
    }
    su_root_step(tt->tt_root, 500L);

    if (!getenv("TPORT_TEST_DEBUG") && 
	time(NULL) > (time_t)(now + timeout))
      return 0;
  }

  return tt->tt_status;
}

static int udp_test(tp_test_t *tt)
{
  tport_t *tp;
  msg_t *msg;
  msg_test_t *tst;
  su_home_t *home;
  msg_request_t *rq;
  msg_unknown_t *u;
  msg_content_length_t *l;
  msg_content_md5_t *md5;
  msg_separator_t *sep;
  msg_payload_t *pl;

  BEGIN();

  TEST_1(msg = msg_create(tt->tt_mclass, 0));
  TEST_1(tst = msg_test_public(msg));
  TEST_1(home = msg_home(msg));

  TEST_1(rq = msg_request_make(home, "DO im:foo@faa " TPORT_TEST_VERSION));
  TEST(msg_header_insert(msg, (void *)tst, (msg_header_t *)rq), 0);

  TEST_1(u = msg_unknown_make(home, "Foo: faa"));
  TEST(msg_header_insert(msg, (void *)tst, (msg_header_t *)u), 0);

  TEST_1(pl = msg_payload_make(home, payload));
  TEST(msg_header_insert(msg, (void *)tst, (msg_header_t *)pl), 0);

  TEST_1(l = msg_content_length_format(home, MOD_ZU, (size_t)pl->pl_len));
  TEST(msg_header_insert(msg, (void *)tst, (msg_header_t *)l), 0);

  TEST_1(md5 = msg_content_md5_make(home, "R6nitdrtJFpxYzrPaSXfrA=="));
  TEST(msg_header_insert(msg, (void *)tst, (msg_header_t *)md5), 0);

  TEST_1(sep = msg_separator_create(home));
  TEST(msg_header_insert(msg, (void *)tst, (msg_header_t *)sep), 0);

  TEST(msg_serialize(msg, (void *)tst), 0);

  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_udp_name, TAG_END()));

  TEST_S(tport_name(tp)->tpn_ident, "client");

  TEST(tport_test_run(tt, 5), 1);

  msg_destroy(msg);

#if 0
  tp_name_t tpn[1] = {{ NULL }};

  TEST_1(msg = tt->tt_rmsg); tt->tt_rmsg = NULL;

  TEST_1(home = msg_home(msg));

  TEST_1(tport_convert_addr(home, tpn, "udp", NULL, msg_addr(msg)) == 0);

  tpn->tpn_comp = tport_name(tt->tt_rtport)->tpn_comp;

  /* reply */
  TEST_1(tport_tsend(tt->tt_rtport, msg, tpn, TAG_END()) != NULL);

  msg_destroy(msg);

  TEST(tport_test_run(tt, 5), 1);

  msg_destroy(msg);
#endif

  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  END();
}

int pending_server_close, pending_client_close;

void server_closed_callback(tp_stack_t *tt, tp_client_t *client,
			    tport_t *tp, msg_t *msg, int error)
{
  assert(msg == NULL);
  assert(client == NULL);
  if (msg == NULL) {
    tport_release(tp, pending_server_close, NULL, NULL, client, 0);
    pending_server_close = 0;
  }
}

void client_closed_callback(tp_stack_t *tt, tp_client_t *client,
			    tport_t *tp, msg_t *msg, int error)
{
  assert(msg == NULL);
  assert(client == NULL);
  if (msg == NULL) {
    tport_release(tp, pending_client_close, NULL, NULL, client, 0);
    pending_client_close = 0;
  }
}

static int tcp_test(tp_test_t *tt)
{
  BEGIN();

  msg_t *msg = NULL;
  int i, N;
  tport_t *tp, *tp0;
  char ident[16];
  su_time_t started;

  /* Send a single message */
  TEST_1(!new_test_msg(tt, &msg, "tcp-first", 1, 1024));
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  tp0 = tport_incref(tp);
  msg_destroy(msg);

  tport_set_params(tp,
       	    TPTAG_KEEPALIVE(100), 
       	    TPTAG_PINGPONG(500),
       	    TPTAG_IDLE(500),
       	    TAG_END());

  TEST(tport_test_run(tt, 5), 1);
  TEST_1(!check_msg(tt, tt->tt_rmsg, "tcp-first"));
  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  /* Ask for notification upon close */
  pending_client_close = tport_pend(tp0, NULL, client_closed_callback, NULL);
  TEST_1(pending_client_close > 0);
  tp = tt->tt_rtport;
  pending_server_close = tport_pend(tp, NULL, server_closed_callback, NULL);
  TEST_1(pending_server_close > 0);

  N = 0; tt->tt_received = 0;

#ifndef WIN32			/* Windows seems to be buffering too much */

  /* Create a large message, just to force queueing in sending end */
  TEST(new_test_msg(tt, &msg, "tcp-0", 1, 16 * 64 * 1024), 0);
  test_create_md5(tt, msg);
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  N++;
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_P(tport_incref(tp), tp0); tport_decref(&tp);
  msg_destroy(msg);

  /* Fill up the queue */
  for (i = 1; i < TPORT_QUEUESIZE; i++) {
    snprintf(ident, sizeof ident, "tcp-%u", i);

    TEST(new_test_msg(tt, &msg, ident, 1, 64 * 1024), 0);
    TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
    N++;
    TEST_S(tport_name(tp)->tpn_ident, "client");
    TEST_P(tport_incref(tp), tp0); tport_decref(&tp);
    msg_destroy(msg);
  }

  /* This overflows the queue */
  TEST(new_test_msg(tt, &msg, "tcp-overflow", 1, 1024), 0);
  TEST_1(!tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  msg_destroy(msg);

  TEST(tport_test_run(tt, 60), 1);
  TEST_1(!check_msg(tt, tt->tt_rmsg, "tcp-0"));
  test_check_md5(tt, tt->tt_rmsg);
  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  if (tt->tt_received < TPORT_QUEUESIZE) { /* We have not received it all */
    snprintf(ident, sizeof ident, "tcp-%u", tt->tt_received);
    TEST(tport_test_run(tt, 5), 1);
    TEST_1(!check_msg(tt, tt->tt_rmsg, ident));
    msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;
  }
#else
 (void)i; (void)ident;
#endif

  /* This uses a new connection */
  TEST_1(!new_test_msg(tt, &msg, "tcp-no-reuse", 1, 1024));
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, 
       		   TPTAG_REUSE(0), TAG_END()));
  N++;
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_1(tport_incref(tp) != tp0); tport_decref(&tp);
  msg_destroy(msg);

  /* This uses the old connection */
  TEST_1(!new_test_msg(tt, &msg, "tcp-reuse", 1, 1024));
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, 
       		   TPTAG_REUSE(1), TAG_END()));
  N++;
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_1(tport_incref(tp) == tp0); tport_decref(&tp);
  msg_destroy(msg);

  /* Receive every message from queue */
  while (tt->tt_received < N) {
    TEST(tport_test_run(tt, 5), 1);
    /* Validate message */
    TEST_1(!check_msg(tt, tt->tt_rmsg, NULL));
    msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;
  }

  /* Try to send a single message */
  TEST_1(!new_test_msg(tt, &msg, "tcp-last", 1, 1024));
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_P(tport_incref(tp), tp0); tport_decref(&tp);
  msg_destroy(msg);

  TEST(tport_test_run(tt, 5), 1);

  TEST_1(!check_msg(tt, tt->tt_rmsg, "tcp-last"));
  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  TEST_1(pending_server_close && pending_client_close);
  SU_DEBUG_3(("tport_test(%p): waiting for PONG timeout\n", (void *)tp0));

  /* Wait until notifications - 
     client closes when no pong is received and notifys pending, 
     then server closes and notifys pending */
  while (pending_server_close || pending_client_close)
    su_root_step(tt->tt_root, 50);

  tport_decref(&tp0);

  /* Again a single message */
  TEST_1(!new_test_msg(tt, &msg, "tcp-pingpong", 1, 512));
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  tp0 = tport_incref(tp);
  msg_destroy(msg);

  tport_set_params(tp0,
       	    TPTAG_KEEPALIVE(250), 
       	    TPTAG_PINGPONG(200),
       	    TAG_END());

  TEST(tport_test_run(tt, 5), 1);
  TEST_1(!check_msg(tt, tt->tt_rmsg, "tcp-pingpong"));
  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  /* Ask for notifications upon close */
  pending_client_close = tport_pend(tp0, NULL, client_closed_callback, NULL);
  TEST_1(pending_client_close > 0);

  tp = tt->tt_rtport;
  pending_server_close = tport_pend(tp, NULL, server_closed_callback, NULL);
  TEST_1(pending_server_close > 0);

  /* Now server responds with pong ... */
  TEST(tport_set_params(tp, TPTAG_PONG2PING(1), TAG_END()), 1);

  started = su_now();

  while (pending_server_close && pending_client_close) {
    su_root_step(tt->tt_root, 50);
    if (su_duration(su_now(), started) > 1000)
      break;
  }

  /* ... and we are still pending after a second */
  TEST_1(pending_client_close && pending_server_close);
  TEST_1(su_duration(su_now(), started) > 1000);

  tport_shutdown(tp0, 2);
  tport_unref(tp0);

  while (pending_server_close || pending_client_close)
    su_root_step(tt->tt_root, 50);

  END();
}

static int test_incomplete(tp_test_t *tt)
{
  BEGIN();
  
  su_addrinfo_t const *ai = tt->tt_tcp_addr;
  su_socket_t s;
  int connected;

  TEST_1(ai != NULL);

  TEST(tport_set_params(tt->tt_tcp, TPTAG_TIMEOUT(500), TAG_END()), 1);

  s = su_socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
  TEST_1(s != SOCKET_ERROR);
  
  su_setblocking(s, 1);
  connected = connect(s, ai->ai_addr, (socklen_t)ai->ai_addrlen);

  su_root_step(tt->tt_root, 50);
  
  TEST(su_send(s, "F", 1, 0), 1);
  su_root_step(tt->tt_root, 50);
  TEST(su_send(s, "O", 1, 0), 1);
  su_root_step(tt->tt_root, 50);
  TEST(su_send(s, "O", 1, 0), 1);
  su_root_step(tt->tt_root, 50);
  TEST(su_send(s, " ", 1, 0), 1);
  su_root_step(tt->tt_root, 50);
  
  tt->tt_received = 0;
  TEST(tport_test_run(tt, 5), -1);
  TEST(tt->tt_received, 1);
  TEST_P(tt->tt_rmsg, NULL);
  
  su_close(s);

  END();
}

static int reuse_test(tp_test_t *tt)
{
  msg_t *msg = NULL;
  int i, reuse = -1;
  tport_t *tp, *tp0, *tp1;
  tp_name_t tpn[1];

  BEGIN();

  /* Flush existing connections */
  *tpn = *tt->tt_tcp_name;
  tpn->tpn_port = "*";
  TEST_1(tp = tport_by_name(tt->tt_tports, tpn));
  TEST_1(tport_is_primary(tp));
  TEST(tport_flush(tp), 0);
  
  for (i = 0; i < 10; i++)
    su_root_step(tt->tt_root, 10L);

  TEST(tport_set_params(tp, TPTAG_REUSE(0), TAG_END()), 1);

  /* Send two messages */
  TEST(new_test_msg(tt, &msg, "reuse-1", 1, 1024), 0);
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_1(tp0 = tport_incref(tp));
  TEST(tport_get_params(tp, TPTAG_REUSE_REF(reuse), TAG_END()), 1);
  TEST(reuse, 0);
  msg_destroy(msg), msg = NULL;

  TEST(new_test_msg(tt, &msg, "reuse-2", 1, 1024), 0);
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_1(tp1 = tport_incref(tp)); TEST_1(tp0 != tp1);
  TEST(tport_get_params(tp, TPTAG_REUSE_REF(reuse), TAG_END()), 1);
  TEST(reuse, 0);
  msg_destroy(msg), msg = NULL;

  /* Receive every message from queue */
  for (tt->tt_received = 0;
       tt->tt_received < 2;) {
    TEST(tport_test_run(tt, 5), 1);
    msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;
  }

  /* Enable reuse on single connection */
  TEST(tport_set_params(tp1, TPTAG_REUSE(1), TAG_END()), 1);
  TEST(new_test_msg(tt, &msg, "reuse-3", 1, 1024), 0);
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, 
			  TPTAG_REUSE(1),
			  TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_1(tp1 == tp);
  TEST(tport_get_params(tp, TPTAG_REUSE_REF(reuse), TAG_END()), 1);
  TEST(reuse, 1);
  msg_destroy(msg), msg = NULL;

  TEST(tport_test_run(tt, 5), 1);
  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  TEST_1(tp = tport_by_name(tt->tt_tports, tpn));
  TEST_1(tport_is_primary(tp));
  TEST(tport_set_params(tp, TPTAG_REUSE(1), TAG_END()), 1);

  /* Send a single message with different connection */
  TEST_1(!new_test_msg(tt, &msg, "fresh-1", 1, 1024));
  TEST_1(tp = tport_tsend(tt->tt_tports, msg, tt->tt_tcp_name, 
			  TPTAG_FRESH(1),
			  TPTAG_REUSE(1),
			  TAG_END()));
  TEST_S(tport_name(tp)->tpn_ident, "client");
  TEST_1(tport_incref(tp) != tp1);  tport_decref(&tp);
  msg_destroy(msg);

  TEST(tport_test_run(tt, 5), 1);

  TEST_1(!check_msg(tt, tt->tt_rmsg, "fresh-1"));
  msg_destroy(tt->tt_rmsg), tt->tt_rmsg = NULL;

  TEST_1(tport_shutdown(tp0, 2) >= 0);
  TEST_1(tport_shutdown(tp1, 2) >= 0);
  TEST_1(tport_shutdown(tp0, 1) >= 0);

  TEST(tport_shutdown(NULL, 0), -1);

  tport_decref(&tp0);
  tport_decref(&tp1);

  END();
}

static int sctp_test(tp_test_t *tt)
{
  BEGIN();