1 files changed, 182 insertions, 21 deletions

diff --git a/ppapi/tests/test_transport.cc b/ppapi/tests/test_transport.cc
index 4035676..bcf9701 100644
--- a/ppapi/tests/test_transport.cc
+++ b/ppapi/tests/test_transport.cc
@@ -4,9 +4,11 @@
 
 #include "ppapi/tests/test_transport.h"
 
+#include <stdlib.h>
 #include <string.h>
 
-#include <string>
+#include <list>
+#include <map>
 
 #include "ppapi/c/dev/ppb_testing_dev.h"
 #include "ppapi/c/pp_errors.h"
@@ -19,6 +21,72 @@
 
 REGISTER_TEST_CASE(Transport);
 
+#define RUN_SUBTEST(function) { \
+    std::string result = function(); \
+    if (!result.empty()) \
+      return result; \
+  }
+
+namespace {
+
+const char kTestChannelName[] = "test";
+const int kNumPackets = 100;
+const int kSendBufferSize = 1200;
+const int kReadBufferSize = 65536;
+
+class StreamReader {
+ public:
+  explicit StreamReader(pp::Transport_Dev* transport)
+      : ALLOW_THIS_IN_INITIALIZER_LIST(callback_factory_(this)),
+        transport_(transport),
+        errors_(0) {
+    Read();
+  }
+
+  const std::list<std::vector<char> >& received() { return received_; }
+  std::list<std::string> errors() { return errors_; }
+
+ private:
+  void Read() {
+    while (true) {
+      buffer_.resize(kReadBufferSize);
+      int result = transport_->Recv(
+          &buffer_[0], buffer_.size(),
+          callback_factory_.NewCallback(&StreamReader::OnReadFinished));
+      if (result > 0)
+        DidFinishRead(result);
+      else
+        break;
+    }
+  }
+
+  void OnReadFinished(int result) {
+    DidFinishRead(result);
+    if (result > 0)
+      Read();
+  }
+
+  void DidFinishRead(int result) {
+    if (result > 0) {
+      if (result > static_cast<int>(buffer_.size())) {
+        errors_.push_back(TestCase::MakeFailureMessage(
+            __FILE__, __LINE__,
+            "Recv() returned value that is bigger than the buffer."));
+      }
+      buffer_.resize(result);
+      received_.push_back(buffer_);
+    }
+  }
+
+  pp::CompletionCallbackFactory<StreamReader> callback_factory_;
+  pp::Transport_Dev* transport_;
+  std::vector<char> buffer_;
+  std::list<std::vector<char> > received_;
+  std::list<std::string> errors_;
+};
+
+}  // namespace
+
 bool TestTransport::Init() {
   transport_interface_ = reinterpret_cast<PPB_Transport_Dev const*>(
       pp::Module::Get()->GetBrowserInterface(PPB_TRANSPORT_DEV_INTERFACE));
@@ -26,42 +94,135 @@ bool TestTransport::Init() {
 }
 
 void TestTransport::RunTest() {
-  RUN_TEST(Basics);
-  // TODO(juberti): more Transport tests here...
+  RUN_TEST(Create);
+  RUN_TEST(Connect);
+  RUN_TEST(SendData);
+  RUN_TEST(ConnectAndClose);
 }
 
-std::string TestTransport::TestBasics() {
-  pp::Transport_Dev transport1(instance_, "test", "");
-  pp::Transport_Dev transport2(instance_, "test", "");
+std::string TestTransport::InitTargets() {
+  transport1_.reset(new pp::Transport_Dev(instance_, kTestChannelName, ""));
+  transport2_.reset(new pp::Transport_Dev(instance_, kTestChannelName, ""));
+
+  ASSERT_TRUE(transport1_.get() != NULL);
+  ASSERT_TRUE(transport2_.get() != NULL);
 
+  PASS();
+}
+
+std::string TestTransport::Connect() {
   TestCompletionCallback connect_cb1(instance_->pp_instance());
   TestCompletionCallback connect_cb2(instance_->pp_instance());
-  ASSERT_EQ(transport1.Connect(connect_cb1), PP_OK_COMPLETIONPENDING);
-  ASSERT_EQ(transport2.Connect(connect_cb2), PP_OK_COMPLETIONPENDING);
+  ASSERT_EQ(transport1_->Connect(connect_cb1), PP_OK_COMPLETIONPENDING);
+  ASSERT_EQ(transport2_->Connect(connect_cb2), PP_OK_COMPLETIONPENDING);
 
   pp::Var address1;
   pp::Var address2;
   TestCompletionCallback next_address_cb1(instance_->pp_instance());
   TestCompletionCallback next_address_cb2(instance_->pp_instance());
-  ASSERT_EQ(transport1.GetNextAddress(&address1, next_address_cb1),
+  ASSERT_EQ(transport1_->GetNextAddress(&address1, next_address_cb1),
             PP_OK_COMPLETIONPENDING);
-  ASSERT_EQ(transport2.GetNextAddress(&address2, next_address_cb2),
+  ASSERT_EQ(transport2_->GetNextAddress(&address2, next_address_cb2),
             PP_OK_COMPLETIONPENDING);
-  ASSERT_EQ(next_address_cb1.WaitForResult(), 0);
-  ASSERT_EQ(next_address_cb2.WaitForResult(), 0);
-  ASSERT_EQ(transport1.GetNextAddress(&address1, next_address_cb1), PP_OK);
-  ASSERT_EQ(transport2.GetNextAddress(&address2, next_address_cb2), PP_OK);
+  ASSERT_EQ(next_address_cb1.WaitForResult(), PP_OK);
+  ASSERT_EQ(next_address_cb2.WaitForResult(), PP_OK);
+  ASSERT_EQ(transport1_->GetNextAddress(&address1, next_address_cb1), PP_OK);
+  ASSERT_EQ(transport2_->GetNextAddress(&address2, next_address_cb2), PP_OK);
+
+  ASSERT_EQ(transport1_->ReceiveRemoteAddress(address2), PP_OK);
+  ASSERT_EQ(transport2_->ReceiveRemoteAddress(address1), PP_OK);
+
+  ASSERT_EQ(connect_cb1.WaitForResult(), PP_OK);
+  ASSERT_EQ(connect_cb2.WaitForResult(), PP_OK);
+
+  ASSERT_TRUE(transport1_->IsWritable());
+  ASSERT_TRUE(transport2_->IsWritable());
+
+  PASS();
+}
+
+std::string TestTransport::Clean() {
+  transport1_.reset();
+  transport2_.reset();
+
+  PASS();
+}
+
+std::string TestTransport::TestCreate() {
+  RUN_SUBTEST(InitTargets);
+
+  Clean();
+
+  PASS();
+}
+
+std::string TestTransport::TestConnect() {
+  RUN_SUBTEST(InitTargets);
+  RUN_SUBTEST(Connect);
+
+  Clean();
+
+  PASS();
+}
+
+std::string TestTransport::TestSendData() {
+  RUN_SUBTEST(InitTargets);
+  RUN_SUBTEST(Connect);
+
+  StreamReader reader(transport1_.get());
+
+  std::map<int, std::vector<char> > sent_packets;
+  for (int i = 0; i < kNumPackets; ++i) {
+    std::vector<char> send_buffer(kSendBufferSize);
+    for (size_t j = 0; j < send_buffer.size(); ++j) {
+      send_buffer[j] = rand() % 256;
+    }
+    // Put packet index in the beginning.
+    memcpy(&send_buffer[0], &i, sizeof(i));
+
+    TestCompletionCallback send_cb(instance_->pp_instance());
+    ASSERT_EQ(
+        transport2_->Send(&send_buffer[0], send_buffer.size(), send_cb),
+        static_cast<int>(send_buffer.size()));
+    sent_packets[i] = send_buffer;
+  }
+
+  // Wait for 1 second.
+  TestCompletionCallback wait_cb(instance_->pp_instance());
+  pp::Module::Get()->core()->CallOnMainThread(1000, wait_cb);
+  ASSERT_EQ(wait_cb.WaitForResult(), PP_OK);
+
+  ASSERT_TRUE(reader.errors().size() == 0);
+  ASSERT_TRUE(reader.received().size() > 0);
+  for (std::list<std::vector<char> >::const_iterator it =
+           reader.received().begin(); it != reader.received().end(); ++it) {
+    int index;
+    memcpy(&index, &(*it)[0], sizeof(index));
+    ASSERT_TRUE(sent_packets[index] == *it);
+  }
+
+  Clean();
+
+  PASS();
+}
+
+std::string TestTransport::TestConnectAndClose() {
+  RUN_SUBTEST(InitTargets);
+  RUN_SUBTEST(Connect);
 
-  ASSERT_EQ(transport1.ReceiveRemoteAddress(address2), 0);
-  ASSERT_EQ(transport2.ReceiveRemoteAddress(address1), 0);
+  std::vector<char> recv_buffer(kReadBufferSize);
+  TestCompletionCallback recv_cb(instance_->pp_instance());
+  ASSERT_EQ(
+      transport1_->Recv(&recv_buffer[0], recv_buffer.size(), recv_cb),
+      PP_OK_COMPLETIONPENDING);
 
-  ASSERT_EQ(connect_cb1.WaitForResult(), 0);
-  ASSERT_EQ(connect_cb2.WaitForResult(), 0);
+  // Close the transport and verify that callback is aborted.
+  ASSERT_EQ(transport1_->Close(), PP_OK);
 
-  ASSERT_TRUE(transport1.IsWritable());
-  ASSERT_TRUE(transport2.IsWritable());
+  ASSERT_EQ(recv_cb.run_count(), 1);
+  ASSERT_EQ(recv_cb.result(), PP_ERROR_ABORTED);
 
-  // TODO(sergeyu): Verify that data can be sent/received.
+  Clean();
 
   PASS();
 }