A change for the setPause() api in chrome.sockets.tcp:

extensions/browser/api/sockets_tcp/sockets_tcp_apitest.cc

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+#include <sched.h>
+#include <unistd.h>
+
 #include "base/memory/ref_counted.h"
+#include "base/run_loop.h"
 #include "base/strings/stringprintf.h"
+#include "base/synchronization/lock.h"
 #include "extensions/browser/api/dns/host_resolver_wrapper.h"
 #include "extensions/browser/api/dns/mock_host_resolver_creator.h"
 #include "extensions/browser/api/sockets_tcp/sockets_tcp_api.h"
 #include "extensions/browser/api_test_utils.h"
+#include "extensions/browser/event_router.h"
+#include "extensions/browser/event_router_factory.h"
+#include "extensions/browser/extension_function_dispatcher.h"
+#include "extensions/browser/extension_prefs.h"
+#include "extensions/common/api/sockets/sockets_manifest_data.h"
 #include "extensions/common/extension.h"
+#include "extensions/common/extension_builder.h"
+#include "extensions/common/manifest_constants.h"
 #include "extensions/common/test_util.h"
+#include "extensions/common/value_builder.h"
 #include "extensions/shell/test/shell_apitest.h"
 #include "extensions/shell/test/shell_test.h"
 #include "extensions/test/extension_test_message_listener.h"
 #include "extensions/test/result_catcher.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
 #include "net/dns/mock_host_resolver.h"
+#include "net/socket/tcp_server_socket.h"
 #include "net/test/spawned_test_server/spawned_test_server.h"
+#include "testing/gmock/include/gmock/gmock.h"
 
 namespace extensions {
+using ::testing::_;
 
 const std::string kHostname = "127.0.0.1";
 
+// Runs a single-connection TCP echo server on a separate thread.
+class LocalServer : public base::RefCountedThreadSafe<LocalServer> {
+ public:
+  LocalServer()
+      : echo_socket_(nullptr, net::NetLog::Source()),
+        thread_("localserver"),
+        active_(false),
+        issued_write_count_(0),
+        finished_write_count_(0),
+        port_(-1),
+        sock_(nullptr),
+        state_cvar_(&state_lock_) {}
+
+  void Start() {
+    // Should run in an IO thread (or at least one with a MessageLoopForIO)
+    base::Thread::Options thread_opts(base::MessageLoop::TYPE_IO, 0);
+    thread_.StartWithOptions(thread_opts);
+    thread_.WaitUntilThreadStarted();
+    thread_.message_loop()->PostTask(
+        FROM_HERE, base::Bind(&LocalServer::IOStart, base::Unretained(this)));
+  }
+
+  void Close() {
+    thread_.message_loop()->PostTask(FROM_HERE,
+                                     base::Bind(&LocalServer::DoClose, this));
+  }
+
+  // This is -1 until IOStart() has run on this LocalServer's IO thread
+  int port() {
+    int ret = -1;
+    do {
+      {
+        base::AutoLock l(state_lock_);
+        ret = port_;
+      }
+      if (port_ < 0) {
+        LOG(INFO) << "port(): waiting for thread to finish doing whatever it's "
+                     "doing.";
+        usleep(25000);
+      }
+    } while (ret < 0);
+    return ret;
+  }
+
+  // Whether Accept() completed, making sock_ valid.
+  bool accepted() {
+    base::AutoLock l(state_lock_);
+    return active_;
+  }
+
+  // Returns the total number of issued writes. If this call returns N, then
+  // you can verify that the underlying write finished with
+  // wait_for_n_finished_writes(N).
+  int Write(const char* message) { return Write(std::string(message)); }
+
+  int Write(const std::string& message) {
+    thread_.message_loop()->PostTask(
+        FROM_HERE,
+        base::Bind(&LocalServer::DoWrite, base::Unretained(this), message));
+    return ++issued_write_count_;
+  }
+
+  int issued_write_count() {
+    base::AutoLock l(state_lock_);
+    return issued_write_count_;
+  }
+
+  int finished_write_count() {
+    base::AutoLock l(state_lock_);
+    return finished_write_count_;
+  }
+
+  void WaitForNFinishedWrites(int nr) {
+    base::AutoLock l(state_lock_);
+    while (finished_write_count_ < nr) {
+      state_cvar_.Wait();
+    }
+  }
+
+ private:
+  net::TCPServerSocket echo_socket_;
+  base::Thread thread_;  // Our IO thread.
+  bool active_;
+  int issued_write_count_;    // How many calls to Write() have been made?
+  int finished_write_count_;  // How many calls to FinishWrite() have been made?
+  int port_;
+  scoped_ptr<net::StreamSocket> sock_;  // not for use on the primary thread!
+  base::Lock state_lock_;
+  base::ConditionVariable state_cvar_;
+
+  friend class base::RefCountedThreadSafe<LocalServer>;
+  virtual ~LocalServer() {
+    bool active;
+    {
+      base::AutoLock l(state_lock_);
+      active = active_;
+    }
+    DCHECK(!active);
+  }
+
+  void FinishWrite(int result) {
+    base::AutoLock l(state_lock_);
+    ASSERT_GT(result, 0);
+    LOG(INFO) << "FinishWrite(result=" << result << ")";
+    finished_write_count_++;
+    state_cvar_.Broadcast();
+  }
+
+  void DoWrite(const std::string& message) {
+    {
+      base::AutoLock l(state_lock_);
+      DCHECK(active_);
+      issued_write_count_++;
+    }
+    scoped_refptr<net::StringIOBuffer> buf(new net::StringIOBuffer(message));
+    LOG(INFO) << "Write(" << message << ")";
+    int result = sock_->Write(
+        buf.get(), buf->size(),
+        base::Bind(&LocalServer::FinishWrite, base::Unretained(this)));
+    if (result != net::ERR_IO_PENDING) {
+      FinishWrite(result);
+    }
+  }
+
+  void IOStart() {
+    base::AutoLock l(state_lock_);
+    net::IPAddressNumber localhost;
+    net::IPEndPoint local;
+    net::ParseIPLiteralToNumber("127.0.0.1", &localhost);
+    // setup echo_socket_.
+    echo_socket_.Listen(net::IPEndPoint(localhost, 0), 5);
+    echo_socket_.GetLocalAddress(&local);
+    port_ = local.port();
+    echo_socket_.Accept(
+        &sock_, base::Bind(&LocalServer::AcceptedSock, base::Unretained(this)));
+  }
+
+  void AcceptedSock(int result) {
+    base::AutoLock l(state_lock_);
+    ASSERT_EQ(net::OK, result);
+    active_ = true;
+  }
+
+  void DoClose() {
+    base::AutoLock l(state_lock_);
+    // Repeatedly calling this is harmless.
+    sock_->Disconnect();
+    active_ = false;
+  }
+};
+
+class MockEventRouter : public EventRouter {
+ public:
+  MockEventRouter(content::BrowserContext* ctx, ExtensionPrefs* prefs)
+      : EventRouter(ctx, prefs), do_dispatch_(true), call_count_(0) {}
+
+  MOCK_METHOD3(DoDispatchEventToExtension,
+               void(const std::string&, const std::string&, Event*));
+
+  void set_dispatch(bool v) { do_dispatch_ = v; }
+  int call_count() { return call_count_; }
+
+  // Trampoline method to work around GMOCK/scoped_ptr problems.
+  void DispatchEventToExtension(const std::string& str,
+                                scoped_ptr<Event> evt) override {
+    LOG(INFO) << "MockEventRouter: Got event: " << evt->event_name;
+    DCHECK_CURRENTLY_ON(content::BrowserThread::UI);
+    call_count_++;
+    DoDispatchEventToExtension(str, evt->event_name, evt.get());
+    if (do_dispatch_) {
+      EventRouter::DispatchEventToExtension(str,
+                                            scoped_ptr<Event>(evt.release()));
+    }
+  }
+
+ private:
+  bool do_dispatch_;
+  int call_count_;
+};
+
 class SocketsTcpApiTest : public ShellApiTest {
  public:
   SocketsTcpApiTest()
       : resolver_event_(true, false),
         resolver_creator_(new MockHostResolverCreator()) {}
 
+  void EnableMockRouter() {
+    // Setup a Mock EventRouter to catch output from the
+    // TCPSocketEventDispatcher
+    EventRouterFactory::GetInstance()->SetTestingFactoryAndUse(
+        browser_context(), &BuildMockRouter);
+  }
+
   void SetUpOnMainThread() override {
     ShellApiTest::SetUpOnMainThread();
 
     HostResolverWrapper::GetInstance()->SetHostResolverForTesting(
         resolver_creator_->CreateMockHostResolver());
   }
 
   void TearDownOnMainThread() override {
+    LOG(INFO) << "TearDown starting";
     HostResolverWrapper::GetInstance()->SetHostResolverForTesting(NULL);
     resolver_creator_->DeleteMockHostResolver();
 
+    LOG(INFO) << "super TearDown starting";
     ShellApiTest::TearDownOnMainThread();
+    LOG(INFO) << "TearDown done";
+  }
+
+ protected:
+  // Makes the extension if we haven't built it yet.
+  scoped_refptr<Extension> extension() {
+    if (!extension_.get()) {
+      DictionaryBuilder app;
+      app.Set("background",
+              DictionaryBuilder().Set("scripts",
+                                      ListBuilder().Append("background.J's")));
+      manifest_builder_.Set("name", "Test")
+          .Set("version", "1.0")
+          .Set(manifest_keys::kApp, app);
+      extension_ = ExtensionBuilder().SetManifest(manifest_builder_).Build();
+    }
+    return extension_;
+  }
+
+  static scoped_ptr<KeyedService> BuildMockRouter(
+      content::BrowserContext* ctx) {
+    return scoped_ptr<KeyedService>(
+        new MockEventRouter(ctx, ExtensionPrefs::Get(ctx)));
+  }
+
+  MockEventRouter& mock_router() {
+    return *static_cast<MockEventRouter*>(EventRouter::Get(browser_context()));
+  }
+
+  template <class F>
+  scoped_ptr<base::Value> CallFunction(std::string args) {
+    scoped_refptr<F> function = new F();
+    function->set_extension(extension());
+    return scoped_ptr<base::Value>(
+        api_test_utils::RunFunctionAndReturnSingleResult(function.get(), args,
+                                                         browser_context()));
+  }
+
+  // Adds parts to the currently-building manifest.  Call |extension()| to
+  // create the extension with the manifest.
+  void SetManifestPermissions(const std::string& perms) {
+    base::string16 error;
+    scoped_ptr<base::DictionaryValue> permissionsDict =
+        api_test_utils::ParseDictionary(perms);
+    // Validate the permissions.
+    DCHECK(extension_.get() == nullptr);
+    if (SocketsManifestData::FromValue(*permissionsDict.get(), &error).get()) {
+      DictionaryBuilder dbuilder(*permissionsDict.get());
+      manifest_builder_.Set(manifest_keys::kSockets, dbuilder);
+    } else {
+      LOG(ERROR) << "Failed to setup socket manifest permissions: " << error
+                 << " for value: " << perms;
+    }
   }
 
  private:
   base::WaitableEvent resolver_event_;
+  DictionaryBuilder manifest_builder_;
+  scoped_refptr<Extension> extension_;
 
   // The MockHostResolver asserts that it's used on the same thread on which
   // it's created, which is actually a stronger rule than its real counterpart.
   // But that's fine; it's good practice.
   scoped_refptr<MockHostResolverCreator> resolver_creator_;
 };
 
-IN_PROC_BROWSER_TEST_F(SocketsTcpApiTest, SocketsTcpCreateGood) {
-  scoped_refptr<api::SocketsTcpCreateFunction> socket_create_function(
-      new api::SocketsTcpCreateFunction());
-  scoped_refptr<Extension> empty_extension = test_util::CreateEmptyExtension();
-
-  socket_create_function->set_extension(empty_extension.get());
-  socket_create_function->set_has_callback(true);
-
+IN_PROC_BROWSER_TEST_F(SocketsTcpApiTest, SocketsTcpCreateIsGood) {
   scoped_ptr<base::Value> result(
-      api_test_utils::RunFunctionAndReturnSingleResult(
-          socket_create_function.get(), "[]", browser_context()));
+      CallFunction<api::SocketsTcpCreateFunction>("[]"));
 
   ASSERT_EQ(base::Value::TYPE_DICTIONARY, result->GetType());
   base::DictionaryValue* value =
       static_cast<base::DictionaryValue*>(result.get());
   int socketId = -1;
   EXPECT_TRUE(value->GetInteger("socketId", &socketId));
-  ASSERT_TRUE(socketId > 0);
+  ASSERT_GT(socketId, 0);
 }
 
 IN_PROC_BROWSER_TEST_F(SocketsTcpApiTest, SocketTcpExtension) {
   scoped_ptr<net::SpawnedTestServer> test_server(new net::SpawnedTestServer(
       net::SpawnedTestServer::TYPE_TCP_ECHO, net::SpawnedTestServer::kLocalhost,
       base::FilePath(FILE_PATH_LITERAL("net/data"))));
   EXPECT_TRUE(test_server->Start());
 
   net::HostPortPair host_port_pair = test_server->host_port_pair();
   int port = host_port_pair.port();
-  ASSERT_TRUE(port > 0);
+  ASSERT_GT(port, 0);
 
   // Test that connect() is properly resolving hostnames.
   host_port_pair.set_host("lOcAlHoSt");
 
   ResultCatcher catcher;
   catcher.RestrictToBrowserContext(browser_context());
 
   ExtensionTestMessageListener listener("info_please", true);
 
   ASSERT_TRUE(LoadApp("sockets_tcp/api"));
@@ skipping 21 matching lines @@
   ExtensionTestMessageListener listener("info_please", true);
 
   ASSERT_TRUE(LoadApp("sockets_tcp/api"));
   EXPECT_TRUE(listener.WaitUntilSatisfied());
   listener.Reply(base::StringPrintf(
       "https:%s:%d", https_host_port_pair.host().c_str(), https_port));
 
   EXPECT_TRUE(catcher.GetNextResult()) << catcher.message();
 }
 
+IN_PROC_BROWSER_TEST_F(SocketsTcpApiTest, SocketsTcpPauseIsGood) {
+  EnableMockRouter();
+
+  scoped_refptr<LocalServer> server(new LocalServer);
+  server->Start();
+  int port = server->port();
+
+  LOG(INFO) << "Port " << port;
+  SetManifestPermissions(base::StringPrintf(
+      "{ \"tcp\": { \"connect\": \"localhost:%d\" } }", port));
+
+  scoped_ptr<base::Value> create(
+      CallFunction<api::SocketsTcpCreateFunction>("[{\"persistent\":true}]"));
+  int sockfd;
+  base::DictionaryValue* create_val =
+      static_cast<base::DictionaryValue*>(create.get());
+  EXPECT_TRUE(create_val->GetInteger("socketId", &sockfd));
+  ASSERT_GT(sockfd, 0);
+
+  scoped_ptr<base::Value> conn_result =
+      CallFunction<api::SocketsTcpConnectFunction>(
+          base::StringPrintf("[%d, \"localhost\", %d]", sockfd, port));
+  LOG(INFO) << "Finished running connect()";
+
+  // Require that we receive two OnReceive events, with a "MARK" event in the
+  // middle.  So we can make sure that we don't receive an OnReceive while
+  // paused.
+  {
+    ::testing::InSequence seq;
+    EXPECT_CALL(mock_router(),
+                DoDispatchEventToExtension(_, "sockets.tcp.onReceive", _))
+        .Times(1);
+    EXPECT_CALL(mock_router(), DoDispatchEventToExtension(_, "MARK", _))
+        .Times(1);
+    EXPECT_CALL(mock_router(),
+                DoDispatchEventToExtension(_, "sockets.tcp.onReceive", _))
+        .Times(1)
+        .RetiresOnSaturation();
+  }
+
+  // Write the first message, and wait for it to have been issued.
+  server->Write("first message\n");
+  server->WaitForNFinishedWrites(1);
+  // Make sure that the onReceive comes before the MARK gets sent over.
+  while (mock_router().call_count() < 1) {
+    LOG(INFO) << "RunLoop call {";
+    base::RunLoop().RunUntilIdle();
+    LOG(INFO) << "} RunLoop call finished.";
+  }
+
+  CallFunction<api::SocketsTcpSetPausedFunction>(
+      base::StringPrintf("[%d, true]", sockfd));
+
+  server->Write("second message\n");
+  server->WaitForNFinishedWrites(2);
+  // This will result in a message before the Write() above, as it's paused.
+  mock_router().DoDispatchEventToExtension("", "MARK", nullptr);
+
+  // Let the Write() get delivered.
+  CallFunction<api::SocketsTcpSetPausedFunction>(
+      base::StringPrintf("[%d, false]", sockfd));
+
+  while (mock_router().call_count() < 2) {
+    // Let event queues drain.
+    base::RunLoop().RunUntilIdle();
+  }
+
+  server->Close();
+}
+
 }  // namespace extensions