talk/base/testutils.h

/*
 * libjingle
 * Copyright 2004--2011, Google Inc.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *  1. Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *  2. Redistributions in binary form must reproduce the above copyright notice,
 *     this list of conditions and the following disclaimer in the documentation
 *     and/or other materials provided with the distribution.
 *  3. The name of the author may not be used to endorse or promote products
 *     derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
 * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
 * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#ifndef TALK_BASE_TESTUTILS_H__
#define TALK_BASE_TESTUTILS_H__

// Utilities for testing talk_base infrastructure in unittests

#include <map>
#include <vector>
#include "talk/base/asyncsocket.h"
#include "talk/base/common.h"
#include "talk/base/gunit.h"
#include "talk/base/nethelpers.h"
#include "talk/base/stream.h"
#include "talk/base/stringencode.h"
#include "talk/base/stringutils.h"
#include "talk/base/thread.h"

namespace testing {

using namespace talk_base;

///////////////////////////////////////////////////////////////////////////////
// StreamSink - Monitor asynchronously signalled events from StreamInterface
// or AsyncSocket (which should probably be a StreamInterface.
///////////////////////////////////////////////////////////////////////////////

// Note: Any event that is an error is treaded as SSE_ERROR instead of that
// event.

enum StreamSinkEvent {
  SSE_OPEN  = SE_OPEN,
  SSE_READ  = SE_READ,
  SSE_WRITE = SE_WRITE,
  SSE_CLOSE = SE_CLOSE,
  SSE_ERROR = 16
};

class StreamSink : public sigslot::has_slots<> {
 public:
  void Monitor(StreamInterface* stream) {
   stream->SignalEvent.connect(this, &StreamSink::OnEvent);
   events_.erase(stream);
  }
  void Unmonitor(StreamInterface* stream) {
   stream->SignalEvent.disconnect(this);
   // In case you forgot to unmonitor a previous object with this address
   events_.erase(stream);
  }
  bool Check(StreamInterface* stream, StreamSinkEvent event, bool reset = true) {
    return DoCheck(stream, event, reset);
  }
  int Events(StreamInterface* stream, bool reset = true) {
    return DoEvents(stream, reset);
  }

  void Monitor(AsyncSocket* socket) {
   socket->SignalConnectEvent.connect(this, &StreamSink::OnConnectEvent);
   socket->SignalReadEvent.connect(this, &StreamSink::OnReadEvent);
   socket->SignalWriteEvent.connect(this, &StreamSink::OnWriteEvent);
   socket->SignalCloseEvent.connect(this, &StreamSink::OnCloseEvent);
   // In case you forgot to unmonitor a previous object with this address
   events_.erase(socket);
  }
  void Unmonitor(AsyncSocket* socket) {
   socket->SignalConnectEvent.disconnect(this);
   socket->SignalReadEvent.disconnect(this);
   socket->SignalWriteEvent.disconnect(this);
   socket->SignalCloseEvent.disconnect(this);
   events_.erase(socket);
  }
  bool Check(AsyncSocket* socket, StreamSinkEvent event, bool reset = true) {
    return DoCheck(socket, event, reset);
  }
  int Events(AsyncSocket* socket, bool reset = true) {
    return DoEvents(socket, reset);
  }

 private:
  typedef std::map<void*,int> EventMap;

  void OnEvent(StreamInterface* stream, int events, int error) {
    if (error) {
      events = SSE_ERROR;
    }
    AddEvents(stream, events);
  }
  void OnConnectEvent(AsyncSocket* socket) {
    AddEvents(socket, SSE_OPEN);
  }
  void OnReadEvent(AsyncSocket* socket) {
    AddEvents(socket, SSE_READ);
  }
  void OnWriteEvent(AsyncSocket* socket) {
    AddEvents(socket, SSE_WRITE);
  }
  void OnCloseEvent(AsyncSocket* socket, int error) {
    AddEvents(socket, (0 == error) ? SSE_CLOSE : SSE_ERROR);
  }

  void AddEvents(void* obj, int events) {
    EventMap::iterator it = events_.find(obj);
    if (events_.end() == it) {
      events_.insert(EventMap::value_type(obj, events));
    } else {
      it->second |= events;
    }
  }
  bool DoCheck(void* obj, StreamSinkEvent event, bool reset) {
    EventMap::iterator it = events_.find(obj);
    if ((events_.end() == it) || (0 == (it->second & event))) {
      return false;
    }
    if (reset) {
      it->second &= ~event;
    }
    return true;
  }
  int DoEvents(void* obj, bool reset) {
    EventMap::iterator it = events_.find(obj);
    if (events_.end() == it)
      return 0;
    int events = it->second;
    if (reset) {
      it->second = 0;
    }
    return events;
  }

  EventMap events_;
};

///////////////////////////////////////////////////////////////////////////////
// StreamSource - Implements stream interface and simulates asynchronous
// events on the stream, without a network.  Also buffers written data.
///////////////////////////////////////////////////////////////////////////////

class StreamSource : public StreamInterface {
public:
  StreamSource() {
    Clear();
  }

  void Clear() {
    readable_data_.clear();
    written_data_.clear();
    state_ = SS_CLOSED;
    read_block_ = 0;
    write_block_ = SIZE_UNKNOWN;
  }
  void QueueString(const char* data) {
    QueueData(data, strlen(data));
  }
  void QueueStringF(const char* format, ...) {
    va_list args;
    va_start(args, format);
    char buffer[1024];
    size_t len = vsprintfn(buffer, sizeof(buffer), format, args);
    ASSERT(len < sizeof(buffer) - 1);
    va_end(args);
    QueueData(buffer, len);
  }
  void QueueData(const char* data, size_t len) {
    readable_data_.insert(readable_data_.end(), data, data + len);
    if ((SS_OPEN == state_) && (readable_data_.size() == len)) {
      SignalEvent(this, SE_READ, 0);
    }
  }
  std::string ReadData() {
    std::string data;
    // avoid accessing written_data_[0] if it is undefined
    if (written_data_.size() > 0) {
      data.insert(0, &written_data_[0], written_data_.size());
    }
    written_data_.clear();
    return data;
  }
  void SetState(StreamState state) {
    int events = 0;
    if ((SS_OPENING == state_) && (SS_OPEN == state)) {
      events |= SE_OPEN;
      if (!readable_data_.empty()) {
        events |= SE_READ;
      }
    } else if ((SS_CLOSED != state_) && (SS_CLOSED == state)) {
      events |= SE_CLOSE;
    }
    state_ = state;
    if (events) {
      SignalEvent(this, events, 0);
    }
  }
  // Will cause Read to block when there are pos bytes in the read queue.
  void SetReadBlock(size_t pos) { read_block_ = pos; }
  // Will cause Write to block when there are pos bytes in the write queue.
  void SetWriteBlock(size_t pos) { write_block_ = pos; }

  virtual StreamState GetState() const { return state_; }
  virtual StreamResult Read(void* buffer, size_t buffer_len,
                            size_t* read, int* error) {
    if (SS_CLOSED == state_) {
      if (error) *error = -1;
      return SR_ERROR;
    }
    if ((SS_OPENING == state_) || (readable_data_.size() <= read_block_)) {
      return SR_BLOCK;
    }
    size_t count = _min(buffer_len, readable_data_.size() - read_block_);
    memcpy(buffer, &readable_data_[0], count);
    size_t new_size = readable_data_.size() - count;
    // Avoid undefined access beyond the last element of the vector.
    // This only happens when new_size is 0.
    if (count < readable_data_.size()) {
      memmove(&readable_data_[0], &readable_data_[count], new_size);
    }
    readable_data_.resize(new_size);
    if (read) *read = count;
    return SR_SUCCESS;
  }
  virtual StreamResult Write(const void* data, size_t data_len,
                             size_t* written, int* error) {
    if (SS_CLOSED == state_) {
      if (error) *error = -1;
      return SR_ERROR;
    }
    if (SS_OPENING == state_) {
      return SR_BLOCK;
    }
    if (SIZE_UNKNOWN != write_block_) {
      if (written_data_.size() >= write_block_) {
        return SR_BLOCK;
      }
      if (data_len > (write_block_ - written_data_.size())) {
        data_len = write_block_ - written_data_.size();
      }
    }
    if (written) *written = data_len;
    const char* cdata = static_cast<const char*>(data);
    written_data_.insert(written_data_.end(), cdata, cdata + data_len);
    return SR_SUCCESS;
  }
  virtual void Close() { state_ = SS_CLOSED; }

private:
  typedef std::vector<char> Buffer;
  Buffer readable_data_, written_data_;
  StreamState state_;
  size_t read_block_, write_block_;
};

///////////////////////////////////////////////////////////////////////////////
// SocketTestClient
// Creates a simulated client for testing.  Works on real and virtual networks.
///////////////////////////////////////////////////////////////////////////////

class SocketTestClient : public sigslot::has_slots<> {
public:
  SocketTestClient() {
    Init(NULL, AF_INET);
  }
  SocketTestClient(AsyncSocket* socket) {
    Init(socket, socket->GetLocalAddress().family());
  }
  SocketTestClient(const SocketAddress& address) {
    Init(NULL, address.family());
    socket_->Connect(address);
  }

  AsyncSocket* socket() { return socket_.get(); }

  void QueueString(const char* data) {
    QueueData(data, strlen(data));
  }
  void QueueStringF(const char* format, ...) {
    va_list args;
    va_start(args, format);
    char buffer[1024];
    size_t len = vsprintfn(buffer, sizeof(buffer), format, args);
    ASSERT(len < sizeof(buffer) - 1);
    va_end(args);
    QueueData(buffer, len);
  }
  void QueueData(const char* data, size_t len) {
    send_buffer_.insert(send_buffer_.end(), data, data + len);
    if (Socket::CS_CONNECTED == socket_->GetState()) {
      Flush();
    }
  }
  std::string ReadData() {
    std::string data(&recv_buffer_[0], recv_buffer_.size());
    recv_buffer_.clear();
    return data;
  }

  bool IsConnected() const {
    return (Socket::CS_CONNECTED == socket_->GetState());
  }
  bool IsClosed() const {
    return (Socket::CS_CLOSED == socket_->GetState());
  }

private:
  typedef std::vector<char> Buffer;

  void Init(AsyncSocket* socket, int family) {
    if (!socket) {
      socket = Thread::Current()->socketserver()
          ->CreateAsyncSocket(family, SOCK_STREAM);
    }
    socket_.reset(socket);
    socket_->SignalConnectEvent.connect(this,
      &SocketTestClient::OnConnectEvent);
    socket_->SignalReadEvent.connect(this, &SocketTestClient::OnReadEvent);
    socket_->SignalWriteEvent.connect(this, &SocketTestClient::OnWriteEvent);
    socket_->SignalCloseEvent.connect(this, &SocketTestClient::OnCloseEvent);
  }

  void Flush() {
    size_t sent = 0;
    while (sent < send_buffer_.size()) {
      int result = socket_->Send(&send_buffer_[sent],
                                 send_buffer_.size() - sent);
      if (result > 0) {
        sent += result;
      } else {
        break;
      }
    }
    size_t new_size = send_buffer_.size() - sent;
    memmove(&send_buffer_[0], &send_buffer_[sent], new_size);
    send_buffer_.resize(new_size);
  }

  void OnConnectEvent(AsyncSocket* socket) {
    if (!send_buffer_.empty()) {
      Flush();
    }
  }
  void OnReadEvent(AsyncSocket* socket) {
    char data[64 * 1024];
    int result = socket_->Recv(data, ARRAY_SIZE(data));
    if (result > 0) {
      recv_buffer_.insert(recv_buffer_.end(), data, data + result);
    }
  }
  void OnWriteEvent(AsyncSocket* socket) {
    if (!send_buffer_.empty()) {
      Flush();
    }
  }
  void OnCloseEvent(AsyncSocket* socket, int error) {
  }

  scoped_ptr<AsyncSocket> socket_;
  Buffer send_buffer_, recv_buffer_;
};

///////////////////////////////////////////////////////////////////////////////
// SocketTestServer
// Creates a simulated server for testing.  Works on real and virtual networks.
///////////////////////////////////////////////////////////////////////////////

class SocketTestServer : public sigslot::has_slots<> {
 public:
  SocketTestServer(const SocketAddress& address)
      : socket_(Thread::Current()->socketserver()
                ->CreateAsyncSocket(address.family(), SOCK_STREAM))
  {
    socket_->SignalReadEvent.connect(this, &SocketTestServer::OnReadEvent);
    socket_->Bind(address);
    socket_->Listen(5);
  }
  virtual ~SocketTestServer() {
    clear();
  }

  size_t size() const { return clients_.size(); }
  SocketTestClient* client(size_t index) const { return clients_[index]; }
  SocketTestClient* operator[](size_t index) const { return client(index); }

  void clear() {
    for (size_t i=0; i<clients_.size(); ++i) {
      delete clients_[i];
    }
    clients_.clear();
  }

 private:
  void OnReadEvent(AsyncSocket* socket) {
    AsyncSocket* accepted =
      static_cast<AsyncSocket*>(socket_->Accept(NULL));
    if (!accepted)
      return;
    clients_.push_back(new SocketTestClient(accepted));
  }

  scoped_ptr<AsyncSocket> socket_;
  std::vector<SocketTestClient*> clients_;
};

///////////////////////////////////////////////////////////////////////////////
// Generic Utilities
///////////////////////////////////////////////////////////////////////////////

inline bool ReadFile(const char* filename, std::string* contents) {
  FILE* fp = fopen(filename, "rb");
  if (!fp)
    return false;
  char buffer[1024*64];
  size_t read;
  contents->clear();
  while ((read = fread(buffer, 1, sizeof(buffer), fp))) {
    contents->append(buffer, read);
  }
  bool success = (0 != feof(fp));
  fclose(fp);
  return success;
}

///////////////////////////////////////////////////////////////////////////////
// Unittest predicates which are similar to STREQ, but for raw memory
///////////////////////////////////////////////////////////////////////////////

inline AssertionResult CmpHelperMemEq(const char* expected_expression,
                                      const char* expected_length_expression,
                                      const char* actual_expression,
                                      const char* actual_length_expression,
                                      const void* expected,
                                      size_t expected_length,
                                      const void* actual,
                                      size_t actual_length)
{
  if ((expected_length == actual_length)
      && (0 == memcmp(expected, actual, expected_length))) {
    return AssertionSuccess();
  }

  Message msg;
  msg << "Value of: " << actual_expression
      << " [" << actual_length_expression << "]";
  if (true) {  //!actual_value.Equals(actual_expression)) {
    size_t buffer_size = actual_length * 2 + 1;
    char* buffer = STACK_ARRAY(char, buffer_size);
    hex_encode(buffer, buffer_size,
               reinterpret_cast<const char*>(actual), actual_length);
    msg << "\n  Actual: " << buffer << " [" << actual_length << "]";
  }

  msg << "\nExpected: " << expected_expression
      << " [" << expected_length_expression << "]";
  if (true) {  //!expected_value.Equals(expected_expression)) {
    size_t buffer_size = expected_length * 2 + 1;
    char* buffer = STACK_ARRAY(char, buffer_size);
    hex_encode(buffer, buffer_size,
               reinterpret_cast<const char*>(expected), expected_length);
    msg << "\nWhich is: " << buffer << " [" << expected_length << "]";
  }

  return AssertionFailure(msg);
}

inline AssertionResult CmpHelperFileEq(const char* expected_expression,
                                       const char* expected_length_expression,
                                       const char* actual_filename,
                                       const void* expected,
                                       size_t expected_length,
                                       const char* filename)
{
  std::string contents;
  if (!ReadFile(filename, &contents)) {
    Message msg;
    msg << "File '" << filename << "' could not be read.";
    return AssertionFailure(msg);
  }
  return CmpHelperMemEq(expected_expression, expected_length_expression,
                        actual_filename, "",
                        expected, expected_length,
                        contents.c_str(), contents.size());
}

#define EXPECT_MEMEQ(expected, expected_length, actual, actual_length) \
  EXPECT_PRED_FORMAT4(::testing::CmpHelperMemEq, expected, expected_length, \
                      actual, actual_length)

#define ASSERT_MEMEQ(expected, expected_length, actual, actual_length) \
  ASSERT_PRED_FORMAT4(::testing::CmpHelperMemEq, expected, expected_length, \
                      actual, actual_length)

#define EXPECT_FILEEQ(expected, expected_length, filename) \
  EXPECT_PRED_FORMAT3(::testing::CmpHelperFileEq, expected, expected_length, \
                      filename)

#define ASSERT_FILEEQ(expected, expected_length, filename) \
  ASSERT_PRED_FORMAT3(::testing::CmpHelperFileEq, expected, expected_length, \
                      filename)

///////////////////////////////////////////////////////////////////////////////
// Helpers for initializing constant memory with integers in a particular byte
// order
///////////////////////////////////////////////////////////////////////////////

#define BYTE_CAST(x) static_cast<uint8>((x) & 0xFF)

// Declare a N-bit integer as a little-endian sequence of bytes
#define LE16(x) BYTE_CAST(((uint16)x) >>  0), BYTE_CAST(((uint16)x) >>  8)

#define LE32(x) BYTE_CAST(((uint32)x) >>  0), BYTE_CAST(((uint32)x) >>  8), \
                BYTE_CAST(((uint32)x) >> 16), BYTE_CAST(((uint32)x) >> 24)

#define LE64(x) BYTE_CAST(((uint64)x) >>  0), BYTE_CAST(((uint64)x) >>  8), \
                BYTE_CAST(((uint64)x) >> 16), BYTE_CAST(((uint64)x) >> 24), \
                BYTE_CAST(((uint64)x) >> 32), BYTE_CAST(((uint64)x) >> 40), \
                BYTE_CAST(((uint64)x) >> 48), BYTE_CAST(((uint64)x) >> 56)

// Declare a N-bit integer as a big-endian (Internet) sequence of bytes
#define BE16(x) BYTE_CAST(((uint16)x) >>  8), BYTE_CAST(((uint16)x) >>  0)

#define BE32(x) BYTE_CAST(((uint32)x) >> 24), BYTE_CAST(((uint32)x) >> 16), \
                BYTE_CAST(((uint32)x) >>  8), BYTE_CAST(((uint32)x) >>  0)

#define BE64(x) BYTE_CAST(((uint64)x) >> 56), BYTE_CAST(((uint64)x) >> 48), \
                BYTE_CAST(((uint64)x) >> 40), BYTE_CAST(((uint64)x) >> 32), \
                BYTE_CAST(((uint64)x) >> 24), BYTE_CAST(((uint64)x) >> 16), \
                BYTE_CAST(((uint64)x) >>  8), BYTE_CAST(((uint64)x) >>  0)

// Declare a N-bit integer as a this-endian (local machine) sequence of bytes
#ifndef BIG_ENDIAN
#define BIG_ENDIAN 1
#endif  // BIG_ENDIAN

#if BIG_ENDIAN
#define TE16 BE16
#define TE32 BE32
#define TE64 BE64
#else  // !BIG_ENDIAN
#define TE16 LE16
#define TE32 LE32
#define TE64 LE64
#endif  // !BIG_ENDIAN

///////////////////////////////////////////////////////////////////////////////

}  // namespace testing

#endif  // TALK_BASE_TESTUTILS_H__