content/browser/service_worker/service_worker_data_pipe_reader_unittest.cc - chromium/src - Git at Google

 // Copyright 2017 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 "content/browser/service_worker/service_worker_data_pipe_reader.h"

 #include "base/run_loop.h"
 #include "content/browser/service_worker/embedded_worker_test_helper.h"
 #include "content/browser/service_worker/service_worker_context_core.h"
 #include "content/browser/service_worker/service_worker_context_wrapper.h"
 #include "content/browser/service_worker/service_worker_registration.h"
 #include "content/browser/service_worker/service_worker_url_request_job.h"
 #include "content/browser/service_worker/service_worker_version.h"
 #include "content/public/common/resource_request_body.h"
 #include "content/public/test/test_browser_thread_bundle.h"
 #include "net/base/io_buffer.h"
 #include "testing/gtest/include/gtest/gtest.h"

 namespace content {

 namespace {

 const char kTestData[] = "Here is sample text for the blob.";

 }  // namespace

 class MockServiceWorkerURLRequestJob : public ServiceWorkerURLRequestJob {
  public:
   explicit MockServiceWorkerURLRequestJob(
       ServiceWorkerURLRequestJob::Delegate* delegate)
       : ServiceWorkerURLRequestJob(nullptr,
                                    nullptr,
                                    "",
                                    nullptr,
                                    nullptr,
                                    FETCH_REQUEST_MODE_NO_CORS,
                                    FETCH_CREDENTIALS_MODE_OMIT,
                                    FetchRedirectMode::FOLLOW_MODE,
                                    std::string() /* integrity */,
                                    RESOURCE_TYPE_MAIN_FRAME,
                                    REQUEST_CONTEXT_TYPE_HYPERLINK,
                                    REQUEST_CONTEXT_FRAME_TYPE_TOP_LEVEL,
                                    scoped_refptr<ResourceRequestBody>(),
                                    ServiceWorkerFetchType::FETCH,
                                    base::Optional<base::TimeDelta>(),
                                    delegate),
         is_response_started_(false) {}

   void OnResponseStarted() override { is_response_started_ = true; }

   void OnReadRawDataComplete(int bytes_read) override {
     async_read_bytes_.push_back(bytes_read);
   }

   void RecordResult(ServiceWorkerMetrics::URLRequestJobResult result) override {
     results_.push_back(result);
   }

   bool is_response_started() { return is_response_started_; }
   const std::vector<int>& async_read_bytes() { return async_read_bytes_; }
   const std::vector<ServiceWorkerMetrics::URLRequestJobResult>& results() {
     return results_;
   }

  private:
   bool is_response_started_;
   std::vector<int> async_read_bytes_;
   std::vector<ServiceWorkerMetrics::URLRequestJobResult> results_;
 };

 class ServiceWorkerDataPipeReaderTest
     : public testing::Test,
       public ServiceWorkerURLRequestJob::Delegate {
  public:
   ServiceWorkerDataPipeReaderTest()
       : thread_bundle_(TestBrowserThreadBundle::IO_MAINLOOP) {}

   void SetUp() override {
     helper_ = base::MakeUnique<EmbeddedWorkerTestHelper>(base::FilePath());
     mock_url_request_job_ =
         base::MakeUnique<MockServiceWorkerURLRequestJob>(this);
     ServiceWorkerRegistrationOptions options(GURL("https://example.com/"));
     registration_ = new ServiceWorkerRegistration(
         options, 1L, helper_->context()->AsWeakPtr());
     version_ = new ServiceWorkerVersion(
         registration_.get(), GURL("https://example.com/service_worker.js"), 1L,
         helper_->context()->AsWeakPtr());
     std::vector<ServiceWorkerDatabase::ResourceRecord> records;
     records.push_back(
         ServiceWorkerDatabase::ResourceRecord(10, version_->script_url(), 100));
     version_->script_cache_map()->SetResources(records);
     version_->set_fetch_handler_existence(
         ServiceWorkerVersion::FetchHandlerExistence::EXISTS);
   }

   std::unique_ptr<ServiceWorkerDataPipeReader> CreateTargetDataPipeReader(
       blink::mojom::ServiceWorkerStreamCallbackPtr* stream_callback,
       mojo::DataPipe* data_pipe) {
     blink::mojom::ServiceWorkerStreamHandlePtr stream_handle =
         blink::mojom::ServiceWorkerStreamHandle::New();
     stream_handle->stream = std::move(data_pipe->consumer_handle);
     stream_handle->callback_request = mojo::MakeRequest(stream_callback);
     return base::MakeUnique<ServiceWorkerDataPipeReader>(
         mock_url_request_job_.get(), version_, std::move(stream_handle));
   }

   // Implements ServiceWorkerURLRequestJob::Delegate.
   void OnPrepareToRestart() override { NOTREACHED(); }

   ServiceWorkerVersion* GetServiceWorkerVersion(
       ServiceWorkerMetrics::URLRequestJobResult*) override {
     NOTREACHED();
     return nullptr;
   }

   bool RequestStillValid(ServiceWorkerMetrics::URLRequestJobResult*) override {
     NOTREACHED();
     return false;
   }

   void TearDown() override { helper_.reset(); }

   MockServiceWorkerURLRequestJob* mock_url_request_job() {
     return mock_url_request_job_.get();
   }

  protected:
   TestBrowserThreadBundle thread_bundle_;

   std::unique_ptr<EmbeddedWorkerTestHelper> helper_;
   std::unique_ptr<MockServiceWorkerURLRequestJob> mock_url_request_job_;
   scoped_refptr<ServiceWorkerRegistration> registration_;
   scoped_refptr<ServiceWorkerVersion> version_;
 };

 class ServiceWorkerDataPipeReaderTestP
     : public ServiceWorkerDataPipeReaderTest,
       public testing::WithParamInterface<
           std::tuple<bool /* should_close_connection_first */,
                      bool /* has_body */>> {
  public:
   ServiceWorkerDataPipeReaderTestP() {}
   virtual ~ServiceWorkerDataPipeReaderTestP() {}

  protected:
   bool should_close_connection_first() const { return std::get<0>(GetParam()); }
   bool has_body() const { return std::get<1>(GetParam()); }
 };

 TEST_P(ServiceWorkerDataPipeReaderTestP, SyncRead) {
   blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
   mojo::DataPipe data_pipe;
   std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
       CreateTargetDataPipeReader(&stream_callback, &data_pipe);

   // Push enough data.
   if (has_body()) {
     std::string expected_response;
     expected_response.reserve((sizeof(kTestData) - 1) * 1024);
     for (int i = 0; i < 1024; ++i) {
       expected_response += kTestData;
       uint32_t written_bytes = sizeof(kTestData) - 1;
       MojoResult result =
           mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
                              &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
       ASSERT_EQ(MOJO_RESULT_OK, result);
       EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
     }
   }
   data_pipe.producer_handle.reset();
   stream_callback->OnCompleted();
   base::RunLoop().RunUntilIdle();

   // Nothing has started.
   EXPECT_FALSE(mock_url_request_job()->is_response_started());
   EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(0UL, mock_url_request_job()->results().size());

   // Start to read.
   data_pipe_reader->Start();
   EXPECT_TRUE(mock_url_request_job()->is_response_started());
   const int buffer_size = sizeof(kTestData);
   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(buffer_size);
   buffer->data()[buffer_size - 1] = '\0';

   // Read successfully.
   if (has_body()) {
     std::string retrieved_response;
     retrieved_response.reserve(buffer_size * 1024);
     for (int i = 0; i < 1024; ++i) {
       EXPECT_EQ(buffer_size - 1,
                 data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
       EXPECT_STREQ(kTestData, buffer->data());
       retrieved_response += buffer->data();
     }
   }

   // Finish successfully.
   EXPECT_EQ(net::OK,
             data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
   EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
   ASSERT_EQ(1UL, mock_url_request_job()->results().size());
   EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE,
             mock_url_request_job()->results()[0]);
 }

 TEST_P(ServiceWorkerDataPipeReaderTestP, SyncAbort) {
   blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
   mojo::DataPipe data_pipe;
   std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
       CreateTargetDataPipeReader(&stream_callback, &data_pipe);

   // Push enough data.
   if (has_body()) {
     std::string expected_response;
     expected_response.reserve((sizeof(kTestData) - 1) * 1024);
     for (int i = 0; i < 1024; ++i) {
       expected_response += kTestData;
       uint32_t written_bytes = sizeof(kTestData) - 1;
       MojoResult result =
           mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
                              &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
       ASSERT_EQ(MOJO_RESULT_OK, result);
       EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
     }
   }
   data_pipe.producer_handle.reset();
   stream_callback->OnAborted();
   base::RunLoop().RunUntilIdle();

   // Nothing has started.
   EXPECT_FALSE(mock_url_request_job()->is_response_started());
   EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(0UL, mock_url_request_job()->results().size());

   // Start to read.
   data_pipe_reader->Start();
   EXPECT_TRUE(mock_url_request_job()->is_response_started());
   const int buffer_size = sizeof(kTestData);
   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(buffer_size);
   buffer->data()[buffer_size - 1] = '\0';

   // Read successfully.
   if (has_body()) {
     std::string retrieved_response;
     retrieved_response.reserve(buffer_size * 1024);
     for (int i = 0; i < 1024; ++i) {
       EXPECT_EQ(buffer_size - 1,
                 data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
       EXPECT_STREQ(kTestData, buffer->data());
       retrieved_response += buffer->data();
     }
   }

   // Abort after all data has been read.
   EXPECT_EQ(net::ERR_CONNECTION_RESET,
             data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
   EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
   ASSERT_EQ(1UL, mock_url_request_job()->results().size());
   EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED,
             mock_url_request_job()->results()[0]);
 }

 TEST_P(ServiceWorkerDataPipeReaderTestP, AsyncRead) {
   blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
   mojo::DataPipe data_pipe;
   std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
       CreateTargetDataPipeReader(&stream_callback, &data_pipe);

   // Nothing has started.
   EXPECT_FALSE(mock_url_request_job()->is_response_started());
   EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(0UL, mock_url_request_job()->results().size());

   // Start to read.
   data_pipe_reader->Start();
   EXPECT_TRUE(mock_url_request_job()->is_response_started());
   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(sizeof(kTestData));
   buffer->data()[sizeof(kTestData) - 1] = '\0';
   std::string expected_response;
   std::string retrieved_response;
   expected_response.reserve((sizeof(kTestData) - 1) * 1024);
   retrieved_response.reserve((sizeof(kTestData) - 1) * 1024);

   if (has_body()) {
     for (int i = 0; i < 1024; ++i) {
       // Data is not coming. It should be pending state.
       EXPECT_EQ(net::ERR_IO_PENDING, data_pipe_reader->ReadRawData(
                                          buffer.get(), sizeof(kTestData) - 1));

       // Push a portion of data.
       uint32_t written_bytes = sizeof(kTestData) - 1;
       MojoResult result =
           mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
                              &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
       ASSERT_EQ(MOJO_RESULT_OK, result);
       EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
       expected_response += kTestData;
       base::RunLoop().RunUntilIdle();

       // Read the pushed data correctly.
       ASSERT_EQ(static_cast<size_t>(i + 1),
                 mock_url_request_job()->async_read_bytes().size());
       EXPECT_EQ(static_cast<int>(sizeof(kTestData) - 1),
                 mock_url_request_job()->async_read_bytes()[i]);
       EXPECT_STREQ(kTestData, buffer->data());
     }
   }

   // Data is not coming. It should be pending state.
   EXPECT_EQ(net::ERR_IO_PENDING,
             data_pipe_reader->ReadRawData(buffer.get(), sizeof(kTestData) - 1));

   // Finish successfully when connection is closed AND OnCompleted is delivered.
   size_t num_read = mock_url_request_job()->async_read_bytes().size();
   if (should_close_connection_first()) {
     data_pipe.producer_handle.reset();
   } else {
     stream_callback->OnCompleted();
   }
   base::RunLoop().RunUntilIdle();
   EXPECT_EQ(num_read, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(0UL, mock_url_request_job()->results().size());

   if (should_close_connection_first()) {
     stream_callback->OnCompleted();
   } else {
     data_pipe.producer_handle.reset();
   }
   base::RunLoop().RunUntilIdle();
   ASSERT_EQ(num_read + 1, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(net::OK, mock_url_request_job()->async_read_bytes().back());
   ASSERT_EQ(1UL, mock_url_request_job()->results().size());
   EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE,
             mock_url_request_job()->results()[0]);
 }

 TEST_P(ServiceWorkerDataPipeReaderTestP, AsyncAbort) {
   blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
   mojo::DataPipe data_pipe;
   std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
       CreateTargetDataPipeReader(&stream_callback, &data_pipe);

   // Nothing has started.
   EXPECT_FALSE(mock_url_request_job()->is_response_started());
   EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(0UL, mock_url_request_job()->results().size());

   // Start to read.
   data_pipe_reader->Start();
   EXPECT_TRUE(mock_url_request_job()->is_response_started());
   scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(sizeof(kTestData));
   buffer->data()[sizeof(kTestData) - 1] = '\0';
   std::string expected_response;
   std::string retrieved_response;
   expected_response.reserve((sizeof(kTestData) - 1) * 1024);
   retrieved_response.reserve((sizeof(kTestData) - 1) * 1024);

   if (has_body()) {
     for (int i = 0; i < 1024; ++i) {
       // Data is not coming. It should be pending state.
       EXPECT_EQ(net::ERR_IO_PENDING, data_pipe_reader->ReadRawData(
                                          buffer.get(), sizeof(kTestData) - 1));

       // Push a portion of data.
       uint32_t written_bytes = sizeof(kTestData) - 1;
       MojoResult result =
           mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
                              &written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
       ASSERT_EQ(MOJO_RESULT_OK, result);
       EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
       expected_response += kTestData;
       base::RunLoop().RunUntilIdle();

       // Read the pushed data correctly.
       ASSERT_EQ(static_cast<size_t>(i + 1),
                 mock_url_request_job()->async_read_bytes().size());
       EXPECT_EQ(static_cast<int>(sizeof(kTestData) - 1),
                 mock_url_request_job()->async_read_bytes()[i]);
       EXPECT_STREQ(kTestData, buffer->data());
     }
   }

   // Data is not coming. It should be pending state.
   EXPECT_EQ(net::ERR_IO_PENDING,
             data_pipe_reader->ReadRawData(buffer.get(), sizeof(kTestData) - 1));

   // Abort when connection is closed AND OnAborted is delivered.
   size_t num_read = mock_url_request_job()->async_read_bytes().size();
   if (should_close_connection_first()) {
     data_pipe.producer_handle.reset();
   } else {
     stream_callback->OnAborted();
   }
   base::RunLoop().RunUntilIdle();
   EXPECT_EQ(num_read, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(0UL, mock_url_request_job()->results().size());

   if (should_close_connection_first()) {
     stream_callback->OnAborted();
   } else {
     data_pipe.producer_handle.reset();
   }
   base::RunLoop().RunUntilIdle();
   ASSERT_EQ(num_read + 1, mock_url_request_job()->async_read_bytes().size());
   EXPECT_EQ(net::ERR_CONNECTION_RESET,
             mock_url_request_job()->async_read_bytes().back());
   ASSERT_EQ(1UL, mock_url_request_job()->results().size());
   EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED,
             mock_url_request_job()->results()[0]);
 }

 INSTANTIATE_TEST_CASE_P(ServiceWorkerDataPipeReaderTest,
                         ServiceWorkerDataPipeReaderTestP,
                         testing::Combine(testing::Bool(), testing::Bool()));

 }  // namespace content
	// Copyright 2017 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 "content/browser/service_worker/service_worker_data_pipe_reader.h"

	#include "base/run_loop.h"
	#include "content/browser/service_worker/embedded_worker_test_helper.h"
	#include "content/browser/service_worker/service_worker_context_core.h"
	#include "content/browser/service_worker/service_worker_context_wrapper.h"
	#include "content/browser/service_worker/service_worker_registration.h"
	#include "content/browser/service_worker/service_worker_url_request_job.h"
	#include "content/browser/service_worker/service_worker_version.h"
	#include "content/public/common/resource_request_body.h"
	#include "content/public/test/test_browser_thread_bundle.h"
	#include "net/base/io_buffer.h"
	#include "testing/gtest/include/gtest/gtest.h"

	namespace content {

	namespace {

	const char kTestData[] = "Here is sample text for the blob.";

	} // namespace

	class MockServiceWorkerURLRequestJob : public ServiceWorkerURLRequestJob {
	public:
	explicit MockServiceWorkerURLRequestJob(
	ServiceWorkerURLRequestJob::Delegate* delegate)
	: ServiceWorkerURLRequestJob(nullptr,
	nullptr,
	"",
	nullptr,
	nullptr,
	FETCH_REQUEST_MODE_NO_CORS,
	FETCH_CREDENTIALS_MODE_OMIT,
	FetchRedirectMode::FOLLOW_MODE,
	std::string() /* integrity */,
	RESOURCE_TYPE_MAIN_FRAME,
	REQUEST_CONTEXT_TYPE_HYPERLINK,
	REQUEST_CONTEXT_FRAME_TYPE_TOP_LEVEL,
	scoped_refptr<ResourceRequestBody>(),
	ServiceWorkerFetchType::FETCH,
	base::Optional<base::TimeDelta>(),
	delegate),
	is_response_started_(false) {}

	void OnResponseStarted() override { is_response_started_ = true; }

	void OnReadRawDataComplete(int bytes_read) override {
	async_read_bytes_.push_back(bytes_read);
	}

	void RecordResult(ServiceWorkerMetrics::URLRequestJobResult result) override {
	results_.push_back(result);
	}

	bool is_response_started() { return is_response_started_; }
	const std::vector<int>& async_read_bytes() { return async_read_bytes_; }
	const std::vector<ServiceWorkerMetrics::URLRequestJobResult>& results() {
	return results_;
	}

	private:
	bool is_response_started_;
	std::vector<int> async_read_bytes_;
	std::vector<ServiceWorkerMetrics::URLRequestJobResult> results_;
	};

	class ServiceWorkerDataPipeReaderTest
	: public testing::Test,
	public ServiceWorkerURLRequestJob::Delegate {
	public:
	ServiceWorkerDataPipeReaderTest()
	: thread_bundle_(TestBrowserThreadBundle::IO_MAINLOOP) {}

	void SetUp() override {
	helper_ = base::MakeUnique<EmbeddedWorkerTestHelper>(base::FilePath());
	mock_url_request_job_ =
	base::MakeUnique<MockServiceWorkerURLRequestJob>(this);
	ServiceWorkerRegistrationOptions options(GURL("https://example.com/"));
	registration_ = new ServiceWorkerRegistration(
	options, 1L, helper_->context()->AsWeakPtr());
	version_ = new ServiceWorkerVersion(
	registration_.get(), GURL("https://example.com/service_worker.js"), 1L,
	helper_->context()->AsWeakPtr());
	std::vector<ServiceWorkerDatabase::ResourceRecord> records;
	records.push_back(
	ServiceWorkerDatabase::ResourceRecord(10, version_->script_url(), 100));
	version_->script_cache_map()->SetResources(records);
	version_->set_fetch_handler_existence(
	ServiceWorkerVersion::FetchHandlerExistence::EXISTS);
	}

	std::unique_ptr<ServiceWorkerDataPipeReader> CreateTargetDataPipeReader(
	blink::mojom::ServiceWorkerStreamCallbackPtr* stream_callback,
	mojo::DataPipe* data_pipe) {
	blink::mojom::ServiceWorkerStreamHandlePtr stream_handle =
	blink::mojom::ServiceWorkerStreamHandle::New();
	stream_handle->stream = std::move(data_pipe->consumer_handle);
	stream_handle->callback_request = mojo::MakeRequest(stream_callback);
	return base::MakeUnique<ServiceWorkerDataPipeReader>(
	mock_url_request_job_.get(), version_, std::move(stream_handle));
	}

	// Implements ServiceWorkerURLRequestJob::Delegate.
	void OnPrepareToRestart() override { NOTREACHED(); }

	ServiceWorkerVersion* GetServiceWorkerVersion(
	ServiceWorkerMetrics::URLRequestJobResult*) override {
	NOTREACHED();
	return nullptr;
	}

	bool RequestStillValid(ServiceWorkerMetrics::URLRequestJobResult*) override {
	NOTREACHED();
	return false;
	}

	void TearDown() override { helper_.reset(); }

	MockServiceWorkerURLRequestJob* mock_url_request_job() {
	return mock_url_request_job_.get();
	}

	protected:
	TestBrowserThreadBundle thread_bundle_;

	std::unique_ptr<EmbeddedWorkerTestHelper> helper_;
	std::unique_ptr<MockServiceWorkerURLRequestJob> mock_url_request_job_;
	scoped_refptr<ServiceWorkerRegistration> registration_;
	scoped_refptr<ServiceWorkerVersion> version_;
	};

	class ServiceWorkerDataPipeReaderTestP
	: public ServiceWorkerDataPipeReaderTest,
	public testing::WithParamInterface<
	std::tuple<bool /* should_close_connection_first */,
	bool /* has_body */>> {
	public:
	ServiceWorkerDataPipeReaderTestP() {}
	virtual ~ServiceWorkerDataPipeReaderTestP() {}

	protected:
	bool should_close_connection_first() const { return std::get<0>(GetParam()); }
	bool has_body() const { return std::get<1>(GetParam()); }
	};

	TEST_P(ServiceWorkerDataPipeReaderTestP, SyncRead) {
	blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
	mojo::DataPipe data_pipe;
	std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
	CreateTargetDataPipeReader(&stream_callback, &data_pipe);

	// Push enough data.
	if (has_body()) {
	std::string expected_response;
	expected_response.reserve((sizeof(kTestData) - 1) * 1024);
	for (int i = 0; i < 1024; ++i) {
	expected_response += kTestData;
	uint32_t written_bytes = sizeof(kTestData) - 1;
	MojoResult result =
	mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
	&written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
	ASSERT_EQ(MOJO_RESULT_OK, result);
	EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
	}
	}
	data_pipe.producer_handle.reset();
	stream_callback->OnCompleted();
	base::RunLoop().RunUntilIdle();

	// Nothing has started.
	EXPECT_FALSE(mock_url_request_job()->is_response_started());
	EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(0UL, mock_url_request_job()->results().size());

	// Start to read.
	data_pipe_reader->Start();
	EXPECT_TRUE(mock_url_request_job()->is_response_started());
	const int buffer_size = sizeof(kTestData);
	scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(buffer_size);
	buffer->data()[buffer_size - 1] = '\0';

	// Read successfully.
	if (has_body()) {
	std::string retrieved_response;
	retrieved_response.reserve(buffer_size * 1024);
	for (int i = 0; i < 1024; ++i) {
	EXPECT_EQ(buffer_size - 1,
	data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
	EXPECT_STREQ(kTestData, buffer->data());
	retrieved_response += buffer->data();
	}
	}

	// Finish successfully.
	EXPECT_EQ(net::OK,
	data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
	EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
	ASSERT_EQ(1UL, mock_url_request_job()->results().size());
	EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE,
	mock_url_request_job()->results()[0]);
	}

	TEST_P(ServiceWorkerDataPipeReaderTestP, SyncAbort) {
	blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
	mojo::DataPipe data_pipe;
	std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
	CreateTargetDataPipeReader(&stream_callback, &data_pipe);

	// Push enough data.
	if (has_body()) {
	std::string expected_response;
	expected_response.reserve((sizeof(kTestData) - 1) * 1024);
	for (int i = 0; i < 1024; ++i) {
	expected_response += kTestData;
	uint32_t written_bytes = sizeof(kTestData) - 1;
	MojoResult result =
	mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
	&written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
	ASSERT_EQ(MOJO_RESULT_OK, result);
	EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
	}
	}
	data_pipe.producer_handle.reset();
	stream_callback->OnAborted();
	base::RunLoop().RunUntilIdle();

	// Nothing has started.
	EXPECT_FALSE(mock_url_request_job()->is_response_started());
	EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(0UL, mock_url_request_job()->results().size());

	// Start to read.
	data_pipe_reader->Start();
	EXPECT_TRUE(mock_url_request_job()->is_response_started());
	const int buffer_size = sizeof(kTestData);
	scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(buffer_size);
	buffer->data()[buffer_size - 1] = '\0';

	// Read successfully.
	if (has_body()) {
	std::string retrieved_response;
	retrieved_response.reserve(buffer_size * 1024);
	for (int i = 0; i < 1024; ++i) {
	EXPECT_EQ(buffer_size - 1,
	data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
	EXPECT_STREQ(kTestData, buffer->data());
	retrieved_response += buffer->data();
	}
	}

	// Abort after all data has been read.
	EXPECT_EQ(net::ERR_CONNECTION_RESET,
	data_pipe_reader->ReadRawData(buffer.get(), buffer_size - 1));
	EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
	ASSERT_EQ(1UL, mock_url_request_job()->results().size());
	EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED,
	mock_url_request_job()->results()[0]);
	}

	TEST_P(ServiceWorkerDataPipeReaderTestP, AsyncRead) {
	blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
	mojo::DataPipe data_pipe;
	std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
	CreateTargetDataPipeReader(&stream_callback, &data_pipe);

	// Nothing has started.
	EXPECT_FALSE(mock_url_request_job()->is_response_started());
	EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(0UL, mock_url_request_job()->results().size());

	// Start to read.
	data_pipe_reader->Start();
	EXPECT_TRUE(mock_url_request_job()->is_response_started());
	scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(sizeof(kTestData));
	buffer->data()[sizeof(kTestData) - 1] = '\0';
	std::string expected_response;
	std::string retrieved_response;
	expected_response.reserve((sizeof(kTestData) - 1) * 1024);
	retrieved_response.reserve((sizeof(kTestData) - 1) * 1024);

	if (has_body()) {
	for (int i = 0; i < 1024; ++i) {
	// Data is not coming. It should be pending state.
	EXPECT_EQ(net::ERR_IO_PENDING, data_pipe_reader->ReadRawData(
	buffer.get(), sizeof(kTestData) - 1));

	// Push a portion of data.
	uint32_t written_bytes = sizeof(kTestData) - 1;
	MojoResult result =
	mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
	&written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
	ASSERT_EQ(MOJO_RESULT_OK, result);
	EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
	expected_response += kTestData;
	base::RunLoop().RunUntilIdle();

	// Read the pushed data correctly.
	ASSERT_EQ(static_cast<size_t>(i + 1),
	mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(static_cast<int>(sizeof(kTestData) - 1),
	mock_url_request_job()->async_read_bytes()[i]);
	EXPECT_STREQ(kTestData, buffer->data());
	}
	}

	// Data is not coming. It should be pending state.
	EXPECT_EQ(net::ERR_IO_PENDING,
	data_pipe_reader->ReadRawData(buffer.get(), sizeof(kTestData) - 1));

	// Finish successfully when connection is closed AND OnCompleted is delivered.
	size_t num_read = mock_url_request_job()->async_read_bytes().size();
	if (should_close_connection_first()) {
	data_pipe.producer_handle.reset();
	} else {
	stream_callback->OnCompleted();
	}
	base::RunLoop().RunUntilIdle();
	EXPECT_EQ(num_read, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(0UL, mock_url_request_job()->results().size());

	if (should_close_connection_first()) {
	stream_callback->OnCompleted();
	} else {
	data_pipe.producer_handle.reset();
	}
	base::RunLoop().RunUntilIdle();
	ASSERT_EQ(num_read + 1, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(net::OK, mock_url_request_job()->async_read_bytes().back());
	ASSERT_EQ(1UL, mock_url_request_job()->results().size());
	EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_STREAM_RESPONSE,
	mock_url_request_job()->results()[0]);
	}

	TEST_P(ServiceWorkerDataPipeReaderTestP, AsyncAbort) {
	blink::mojom::ServiceWorkerStreamCallbackPtr stream_callback;
	mojo::DataPipe data_pipe;
	std::unique_ptr<ServiceWorkerDataPipeReader> data_pipe_reader =
	CreateTargetDataPipeReader(&stream_callback, &data_pipe);

	// Nothing has started.
	EXPECT_FALSE(mock_url_request_job()->is_response_started());
	EXPECT_EQ(0UL, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(0UL, mock_url_request_job()->results().size());

	// Start to read.
	data_pipe_reader->Start();
	EXPECT_TRUE(mock_url_request_job()->is_response_started());
	scoped_refptr<net::IOBuffer> buffer = new net::IOBuffer(sizeof(kTestData));
	buffer->data()[sizeof(kTestData) - 1] = '\0';
	std::string expected_response;
	std::string retrieved_response;
	expected_response.reserve((sizeof(kTestData) - 1) * 1024);
	retrieved_response.reserve((sizeof(kTestData) - 1) * 1024);

	if (has_body()) {
	for (int i = 0; i < 1024; ++i) {
	// Data is not coming. It should be pending state.
	EXPECT_EQ(net::ERR_IO_PENDING, data_pipe_reader->ReadRawData(
	buffer.get(), sizeof(kTestData) - 1));

	// Push a portion of data.
	uint32_t written_bytes = sizeof(kTestData) - 1;
	MojoResult result =
	mojo::WriteDataRaw(data_pipe.producer_handle.get(), kTestData,
	&written_bytes, MOJO_WRITE_DATA_FLAG_NONE);
	ASSERT_EQ(MOJO_RESULT_OK, result);
	EXPECT_EQ(sizeof(kTestData) - 1, written_bytes);
	expected_response += kTestData;
	base::RunLoop().RunUntilIdle();

	// Read the pushed data correctly.
	ASSERT_EQ(static_cast<size_t>(i + 1),
	mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(static_cast<int>(sizeof(kTestData) - 1),
	mock_url_request_job()->async_read_bytes()[i]);
	EXPECT_STREQ(kTestData, buffer->data());
	}
	}

	// Data is not coming. It should be pending state.
	EXPECT_EQ(net::ERR_IO_PENDING,
	data_pipe_reader->ReadRawData(buffer.get(), sizeof(kTestData) - 1));

	// Abort when connection is closed AND OnAborted is delivered.
	size_t num_read = mock_url_request_job()->async_read_bytes().size();
	if (should_close_connection_first()) {
	data_pipe.producer_handle.reset();
	} else {
	stream_callback->OnAborted();
	}
	base::RunLoop().RunUntilIdle();
	EXPECT_EQ(num_read, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(0UL, mock_url_request_job()->results().size());

	if (should_close_connection_first()) {
	stream_callback->OnAborted();
	} else {
	data_pipe.producer_handle.reset();
	}
	base::RunLoop().RunUntilIdle();
	ASSERT_EQ(num_read + 1, mock_url_request_job()->async_read_bytes().size());
	EXPECT_EQ(net::ERR_CONNECTION_RESET,
	mock_url_request_job()->async_read_bytes().back());
	ASSERT_EQ(1UL, mock_url_request_job()->results().size());
	EXPECT_EQ(ServiceWorkerMetrics::REQUEST_JOB_ERROR_STREAM_ABORTED,
	mock_url_request_job()->results()[0]);
	}

	INSTANTIATE_TEST_CASE_P(ServiceWorkerDataPipeReaderTest,
	ServiceWorkerDataPipeReaderTestP,
	testing::Combine(testing::Bool(), testing::Bool()));

	} // namespace content