net/quic/bidirectional_stream_quic_impl_unittest.cc

// Copyright 2016 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 "net/quic/bidirectional_stream_quic_impl.h"

#include <stdint.h>
#include <vector>

#include "base/callback_helpers.h"
#include "base/memory/scoped_ptr.h"
#include "base/message_loop/message_loop.h"
#include "base/run_loop.h"
#include "base/strings/string_number_conversions.h"
#include "net/base/net_errors.h"
#include "net/http/bidirectional_stream_request_info.h"
#include "net/http/transport_security_state.h"
#include "net/quic/crypto/crypto_protocol.h"
#include "net/quic/crypto/quic_decrypter.h"
#include "net/quic/crypto/quic_encrypter.h"
#include "net/quic/crypto/quic_server_info.h"
#include "net/quic/quic_chromium_client_session.h"
#include "net/quic/quic_chromium_client_stream.h"
#include "net/quic/quic_chromium_connection_helper.h"
#include "net/quic/quic_chromium_packet_reader.h"
#include "net/quic/quic_chromium_packet_writer.h"
#include "net/quic/quic_connection.h"
#include "net/quic/quic_http_utils.h"
#include "net/quic/spdy_utils.h"
#include "net/quic/test_tools/crypto_test_utils.h"
#include "net/quic/test_tools/mock_clock.h"
#include "net/quic/test_tools/mock_crypto_client_stream_factory.h"
#include "net/quic/test_tools/mock_random.h"
#include "net/quic/test_tools/quic_connection_peer.h"
#include "net/quic/test_tools/quic_test_packet_maker.h"
#include "net/quic/test_tools/quic_test_utils.h"
#include "net/quic/test_tools/test_task_runner.h"
#include "net/socket/socket_test_util.h"
#include "testing/gmock/include/gmock/gmock.h"
#include "testing/gtest/include/gtest/gtest.h"

namespace net {

namespace test {

namespace {

const char kUploadData[] = "Really nifty data!";
const char kDefaultServerHostName[] = "www.google.com";
const uint16_t kDefaultServerPort = 80;
// Size of the buffer to be allocated for each read.
const size_t kReadBufferSize = 4096;

class TestDelegateBase : public BidirectionalStreamJob::Delegate {
 public:
  TestDelegateBase(IOBuffer* read_buf, int read_buf_len)
      : TestDelegateBase(read_buf,
                         read_buf_len,
                         make_scoped_ptr(new base::Timer(false, false))) {}

  TestDelegateBase(IOBuffer* read_buf,
                   int read_buf_len,
                   scoped_ptr<base::Timer> timer)
      : read_buf_(read_buf),
        read_buf_len_(read_buf_len),
        timer_(std::move(timer)),
        loop_(nullptr),
        error_(OK),
        on_data_read_count_(0),
        on_data_sent_count_(0),
        not_expect_callback_(false) {
    loop_.reset(new base::RunLoop);
  }

  ~TestDelegateBase() override {}

  void OnHeadersSent() override {
    CHECK(!not_expect_callback_);
    loop_->Quit();
  }

  void OnHeadersReceived(const SpdyHeaderBlock& response_headers) override {
    CHECK(!not_expect_callback_);

    response_headers_ = response_headers;
    loop_->Quit();
  }

  void OnDataRead(int bytes_read) override {
    CHECK(!not_expect_callback_);
    CHECK(!callback_.is_null());

    ++on_data_read_count_;
    CHECK_GE(bytes_read, OK);
    data_received_.append(read_buf_->data(), bytes_read);
    base::ResetAndReturn(&callback_).Run(bytes_read);
  }

  void OnDataSent() override {
    CHECK(!not_expect_callback_);

    ++on_data_sent_count_;
    loop_->Quit();
  }

  void OnTrailersReceived(const SpdyHeaderBlock& trailers) override {
    CHECK(!not_expect_callback_);

    trailers_ = trailers;
    loop_->Quit();
  }

  void OnFailed(int error) override {
    CHECK(!not_expect_callback_);
    CHECK_EQ(OK, error_);
    CHECK_NE(OK, error);

    error_ = error;
    loop_->Quit();
  }

  void Start(const BidirectionalStreamRequestInfo* request_info,
             const BoundNetLog& net_log,
             const base::WeakPtr<QuicChromiumClientSession> session) {
    stream_job_.reset(new BidirectionalStreamQuicImpl(session));
    stream_job_->Start(request_info, net_log, this, nullptr);
  }

  void SendData(IOBuffer* data, int length, bool end_of_stream) {
    not_expect_callback_ = true;
    stream_job_->SendData(data, length, end_of_stream);
    not_expect_callback_ = false;
  }

  // Waits until next Delegate callback.
  void WaitUntilNextCallback() {
    loop_->Run();
    loop_.reset(new base::RunLoop);
  }

  // Calls ReadData on the |stream_| and updates |data_received_|.
  int ReadData(const CompletionCallback& callback) {
    not_expect_callback_ = true;
    int rv = stream_job_->ReadData(read_buf_.get(), read_buf_len_);
    not_expect_callback_ = false;
    if (rv > 0)
      data_received_.append(read_buf_->data(), rv);
    if (rv == ERR_IO_PENDING)
      callback_ = callback;
    return rv;
  }

  // Cancels |stream_|.
  void CancelStream() { stream_job_->Cancel(); }

  NextProto GetProtocol() const { return stream_job_->GetProtocol(); }

  int64_t GetTotalReceivedBytes() const {
    return stream_job_->GetTotalReceivedBytes();
  }

  int64_t GetTotalSentBytes() const { return stream_job_->GetTotalSentBytes(); }

  // Const getters for internal states.
  const std::string& data_received() const { return data_received_; }
  int error() const { return error_; }
  const SpdyHeaderBlock& response_headers() const { return response_headers_; }
  const SpdyHeaderBlock& trailers() const { return trailers_; }
  int on_data_read_count() const { return on_data_read_count_; }
  int on_data_sent_count() const { return on_data_sent_count_; }

 protected:
  // Quits |loop_|.
  void QuitLoop() { loop_->Quit(); }

  // Deletes |stream_|.
  void DeleteStream() { stream_job_.reset(); }

 private:
  scoped_ptr<BidirectionalStreamQuicImpl> stream_job_;
  scoped_refptr<IOBuffer> read_buf_;
  int read_buf_len_;
  scoped_ptr<base::Timer> timer_;
  std::string data_received_;
  scoped_ptr<base::RunLoop> loop_;
  SpdyHeaderBlock response_headers_;
  SpdyHeaderBlock trailers_;
  int error_;
  int on_data_read_count_;
  int on_data_sent_count_;
  // This is to ensure that delegate callback is not invoked synchronously when
  // calling into |stream_|.
  bool not_expect_callback_;
  CompletionCallback callback_;

  DISALLOW_COPY_AND_ASSIGN(TestDelegateBase);
};

// A delegate that deletes the stream in a particular callback.
class DeleteStreamDelegate : public TestDelegateBase {
 public:
  // Specifies in which callback the stream can be deleted.
  enum Phase {
    ON_HEADERS_RECEIVED,
    ON_DATA_READ,
    ON_TRAILERS_RECEIVED,
    ON_FAILED,
  };

  DeleteStreamDelegate(IOBuffer* buf, int buf_len, Phase phase, bool do_cancel)
      : TestDelegateBase(buf, buf_len), phase_(phase), do_cancel_(do_cancel) {}
  ~DeleteStreamDelegate() override {}

  void OnHeadersReceived(const SpdyHeaderBlock& response_headers) override {
    if (phase_ == ON_HEADERS_RECEIVED) {
      DeleteStream();
    }
    TestDelegateBase::OnHeadersReceived(response_headers);
  }

  void OnDataSent() override { NOTREACHED(); }

  void OnDataRead(int bytes_read) override {
    DCHECK_NE(ON_HEADERS_RECEIVED, phase_);
    if (phase_ == ON_DATA_READ)
      DeleteStream();
    TestDelegateBase::OnDataRead(bytes_read);
  }

  void OnTrailersReceived(const SpdyHeaderBlock& trailers) override {
    DCHECK_NE(ON_HEADERS_RECEIVED, phase_);
    DCHECK_NE(ON_DATA_READ, phase_);
    if (phase_ == ON_TRAILERS_RECEIVED)
      DeleteStream();
    TestDelegateBase::OnTrailersReceived(trailers);
  }

  void OnFailed(int error) override {
    DCHECK_EQ(ON_FAILED, phase_);
    DeleteStream();
    TestDelegateBase::OnFailed(error);
  }

 private:
  // Indicates in which callback the delegate should cancel or delete the
  // stream.
  Phase phase_;
  // Indicates whether to cancel or delete the stream.
  bool do_cancel_;

  DISALLOW_COPY_AND_ASSIGN(DeleteStreamDelegate);
};

}  // namespace

class BidirectionalStreamQuicImplTest
    : public ::testing::TestWithParam<QuicVersion> {
 protected:
  static const bool kFin = true;
  static const bool kIncludeVersion = true;
  static const bool kIncludeCongestionFeedback = true;

  // Holds a packet to be written to the wire, and the IO mode that should
  // be used by the mock socket when performing the write.
  struct PacketToWrite {
    PacketToWrite(IoMode mode, QuicEncryptedPacket* packet)
        : mode(mode), packet(packet) {}
    PacketToWrite(IoMode mode, int rv) : mode(mode), packet(nullptr), rv(rv) {}
    IoMode mode;
    QuicEncryptedPacket* packet;
    int rv;
  };

  BidirectionalStreamQuicImplTest()
      : net_log_(BoundNetLog()),
        crypto_config_(CryptoTestUtils::ProofVerifierForTesting()),
        read_buffer_(new IOBufferWithSize(4096)),
        connection_id_(2),
        stream_id_(kClientDataStreamId1),
        maker_(GetParam(), connection_id_, &clock_, kDefaultServerHostName),
        random_generator_(0) {
    IPAddressNumber ip;
    CHECK(ParseIPLiteralToNumber("192.0.2.33", &ip));
    peer_addr_ = IPEndPoint(ip, 443);
    self_addr_ = IPEndPoint(ip, 8435);
    clock_.AdvanceTime(QuicTime::Delta::FromMilliseconds(20));
  }

  ~BidirectionalStreamQuicImplTest() {
    session_->CloseSessionOnError(ERR_ABORTED, QUIC_INTERNAL_ERROR);
    for (size_t i = 0; i < writes_.size(); i++) {
      delete writes_[i].packet;
    }
  }

  void TearDown() override {
    EXPECT_TRUE(socket_data_->AllReadDataConsumed());
    EXPECT_TRUE(socket_data_->AllWriteDataConsumed());
  }

  // Adds a packet to the list of expected writes.
  void AddWrite(scoped_ptr<QuicEncryptedPacket> packet) {
    writes_.push_back(PacketToWrite(SYNCHRONOUS, packet.release()));
  }

  void ProcessPacket(scoped_ptr<QuicEncryptedPacket> packet) {
    connection_->ProcessUdpPacket(self_addr_, peer_addr_, *packet);
  }

  // Configures the test fixture to use the list of expected writes.
  void Initialize() {
    mock_writes_.reset(new MockWrite[writes_.size()]);
    for (size_t i = 0; i < writes_.size(); i++) {
      if (writes_[i].packet == nullptr) {
        mock_writes_[i] = MockWrite(writes_[i].mode, writes_[i].rv, i);
      } else {
        mock_writes_[i] = MockWrite(writes_[i].mode, writes_[i].packet->data(),
                                    writes_[i].packet->length());
      }
    };

    socket_data_.reset(new StaticSocketDataProvider(
        nullptr, 0, mock_writes_.get(), writes_.size()));

    MockUDPClientSocket* socket =
        new MockUDPClientSocket(socket_data_.get(), net_log_.net_log());
    socket->Connect(peer_addr_);
    runner_ = new TestTaskRunner(&clock_);
    helper_.reset(new QuicChromiumConnectionHelper(runner_.get(), &clock_,
                                                   &random_generator_));
    connection_ = new QuicConnection(
        connection_id_, peer_addr_, helper_.get(),
        new QuicChromiumPacketWriter(socket), true /* owns_writer */,
        Perspective::IS_CLIENT, SupportedVersions(GetParam()));

    session_.reset(new QuicChromiumClientSession(
        connection_, scoped_ptr<DatagramClientSocket>(socket),
        /*stream_factory=*/nullptr, &crypto_client_stream_factory_, &clock_,
        &transport_security_state_, make_scoped_ptr((QuicServerInfo*)nullptr),
        QuicServerId(kDefaultServerHostName, kDefaultServerPort,
                     PRIVACY_MODE_DISABLED),
        kQuicYieldAfterPacketsRead,
        QuicTime::Delta::FromMilliseconds(kQuicYieldAfterDurationMilliseconds),
        /*cert_verify_flags=*/0, DefaultQuicConfig(), &crypto_config_,
        "CONNECTION_UNKNOWN", base::TimeTicks::Now(), &push_promise_index_,
        base::ThreadTaskRunnerHandle::Get().get(),
        /*socket_performance_watcher=*/nullptr, nullptr));
    session_->Initialize();
    session_->GetCryptoStream()->CryptoConnect();
    EXPECT_TRUE(session_->IsCryptoHandshakeConfirmed());
  }

  void SetRequest(const std::string& method,
                  const std::string& path,
                  RequestPriority priority) {
    request_headers_ = maker_.GetRequestHeaders(method, "http", path);
  }

  SpdyHeaderBlock ConstructResponseHeaders(const std::string& response_code) {
    return maker_.GetResponseHeaders(response_code);
  }

  scoped_ptr<QuicEncryptedPacket> ConstructDataPacket(
      QuicPacketNumber packet_number,
      bool should_include_version,
      bool fin,
      QuicStreamOffset offset,
      base::StringPiece data) {
    scoped_ptr<QuicEncryptedPacket> packet(maker_.MakeDataPacket(
        packet_number, stream_id_, should_include_version, fin, offset, data));
    DVLOG(2) << "packet(" << packet_number << "): " << std::endl
             << QuicUtils::StringToHexASCIIDump(packet->AsStringPiece());
    return packet;
  }

  scoped_ptr<QuicEncryptedPacket> ConstructRequestHeadersPacket(
      QuicPacketNumber packet_number,
      bool fin,
      RequestPriority request_priority,
      size_t* spdy_headers_frame_length) {
    SpdyPriority priority =
        ConvertRequestPriorityToQuicPriority(request_priority);
    return maker_.MakeRequestHeadersPacket(
        packet_number, stream_id_, kIncludeVersion, fin, priority,
        request_headers_, spdy_headers_frame_length);
  }

  scoped_ptr<QuicEncryptedPacket> ConstructResponseHeadersPacket(
      QuicPacketNumber packet_number,
      bool fin,
      const SpdyHeaderBlock& response_headers,
      size_t* spdy_headers_frame_length,
      QuicStreamOffset* offset) {
    return maker_.MakeResponseHeadersPacket(
        packet_number, stream_id_, !kIncludeVersion, fin, response_headers,
        spdy_headers_frame_length, offset);
  }

  scoped_ptr<QuicEncryptedPacket> ConstructResponseTrailersPacket(
      QuicPacketNumber packet_number,
      bool fin,
      const SpdyHeaderBlock& trailers,
      size_t* spdy_headers_frame_length,
      QuicStreamOffset* offset) {
    return maker_.MakeResponseHeadersPacket(packet_number, stream_id_,
                                            !kIncludeVersion, fin, trailers,
                                            spdy_headers_frame_length, offset);
  }

  scoped_ptr<QuicEncryptedPacket> ConstructRstStreamPacket(
      QuicPacketNumber packet_number) {
    return ConstructRstStreamCancelledPacket(packet_number, 0);
  }

  scoped_ptr<QuicEncryptedPacket> ConstructRstStreamCancelledPacket(
      QuicPacketNumber packet_number,
      size_t bytes_written) {
    scoped_ptr<QuicEncryptedPacket> packet(
        maker_.MakeRstPacket(packet_number, !kIncludeVersion, stream_id_,
                             QUIC_STREAM_CANCELLED, bytes_written));
    DVLOG(2) << "packet(" << packet_number << "): " << std::endl
             << QuicUtils::StringToHexASCIIDump(packet->AsStringPiece());
    return packet;
  }

  scoped_ptr<QuicEncryptedPacket> ConstructAckAndRstStreamPacket(
      QuicPacketNumber packet_number,
      QuicPacketNumber largest_received,
      QuicPacketNumber ack_least_unacked,
      QuicPacketNumber stop_least_unacked) {
    return maker_.MakeAckAndRstPacket(
        packet_number, !kIncludeVersion, stream_id_, QUIC_STREAM_CANCELLED,
        largest_received, ack_least_unacked, stop_least_unacked,
        !kIncludeCongestionFeedback);
  }

  scoped_ptr<QuicEncryptedPacket> ConstructAckAndDataPacket(
      QuicPacketNumber packet_number,
      bool should_include_version,
      QuicPacketNumber largest_received,
      QuicPacketNumber least_unacked,
      bool fin,
      QuicStreamOffset offset,
      base::StringPiece data) {
    scoped_ptr<QuicEncryptedPacket> packet(maker_.MakeAckAndDataPacket(
        packet_number, should_include_version, stream_id_, largest_received,
        least_unacked, fin, offset, data));
    DVLOG(2) << "packet(" << packet_number << "): " << std::endl
             << QuicUtils::StringToHexASCIIDump(packet->AsStringPiece());
    return packet;
  }

  scoped_ptr<QuicEncryptedPacket> ConstructAckPacket(
      QuicPacketNumber packet_number,
      QuicPacketNumber largest_received,
      QuicPacketNumber least_unacked) {
    return maker_.MakeAckPacket(packet_number, largest_received, least_unacked,
                                !kIncludeCongestionFeedback);
  }

  const BoundNetLog& net_log() const { return net_log_; }

  QuicChromiumClientSession* session() const { return session_.get(); }

 private:
  BoundNetLog net_log_;
  scoped_refptr<TestTaskRunner> runner_;
  scoped_ptr<MockWrite[]> mock_writes_;
  MockClock clock_;
  QuicConnection* connection_;
  scoped_ptr<QuicChromiumConnectionHelper> helper_;
  TransportSecurityState transport_security_state_;
  scoped_ptr<QuicChromiumClientSession> session_;
  QuicCryptoClientConfig crypto_config_;
  HttpRequestHeaders headers_;
  HttpResponseInfo response_;
  scoped_refptr<IOBufferWithSize> read_buffer_;
  SpdyHeaderBlock request_headers_;
  const QuicConnectionId connection_id_;
  const QuicStreamId stream_id_;
  QuicTestPacketMaker maker_;
  IPEndPoint self_addr_;
  IPEndPoint peer_addr_;
  MockRandom random_generator_;
  MockCryptoClientStreamFactory crypto_client_stream_factory_;
  scoped_ptr<StaticSocketDataProvider> socket_data_;
  std::vector<PacketToWrite> writes_;
  QuicClientPushPromiseIndex push_promise_index_;
};

INSTANTIATE_TEST_CASE_P(Version,
                        BidirectionalStreamQuicImplTest,
                        ::testing::ValuesIn(QuicSupportedVersions()));

TEST_P(BidirectionalStreamQuicImplTest, GetRequest) {
  SetRequest("GET", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));

  AddWrite(ConstructAckPacket(2, 3, 1));
  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "GET";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = true;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");

  size_t spdy_response_headers_frame_length;
  QuicStreamOffset offset = 0;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      &offset));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);
  const char kResponseBody[] = "Hello world!";
  // Server sends data.
  ProcessPacket(
      ConstructDataPacket(3, !kIncludeVersion, !kFin, 0, kResponseBody));
  EXPECT_EQ(12, cb.WaitForResult());

  EXPECT_EQ(std::string(kResponseBody), delegate->data_received());
  TestCompletionCallback cb2;
  EXPECT_EQ(ERR_IO_PENDING, delegate->ReadData(cb2.callback()));

  SpdyHeaderBlock trailers;
  size_t spdy_trailers_frame_length;
  trailers["foo"] = "bar";
  trailers[kFinalOffsetHeaderKey] = base::IntToString(strlen(kResponseBody));
  // Server sends trailers.
  ProcessPacket(ConstructResponseTrailersPacket(
      4, kFin, trailers, &spdy_trailers_frame_length, &offset));

  delegate->WaitUntilNextCallback();  // OnTrailersReceived
  EXPECT_EQ(OK, cb2.WaitForResult());
  trailers.erase(kFinalOffsetHeaderKey);
  EXPECT_EQ(trailers, delegate->trailers());

  EXPECT_EQ(OK, delegate->ReadData(cb2.callback()));
  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(2, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
  EXPECT_EQ(kProtoQUIC1SPDY3, delegate->GetProtocol());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(
      static_cast<int64_t>(spdy_response_headers_frame_length +
                           strlen(kResponseBody) + spdy_trailers_frame_length),
      delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, PostRequest) {
  SetRequest("POST", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, !kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructDataPacket(2, kIncludeVersion, kFin, 0, kUploadData));
  AddWrite(ConstructAckPacket(3, 3, 1));

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "POST";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = false;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Send a DATA frame.
  scoped_refptr<StringIOBuffer> buf(new StringIOBuffer(kUploadData));

  delegate->SendData(buf.get(), buf->size(), true);
  delegate->WaitUntilNextCallback();  // OnDataSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");
  size_t spdy_response_headers_frame_length;
  QuicStreamOffset offset = 0;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      &offset));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);
  const char kResponseBody[] = "Hello world!";
  // Server sends data.
  ProcessPacket(
      ConstructDataPacket(3, !kIncludeVersion, !kFin, 0, kResponseBody));

  EXPECT_EQ(static_cast<int>(strlen(kResponseBody)), cb.WaitForResult());

  size_t spdy_trailers_frame_length;
  SpdyHeaderBlock trailers;
  trailers["foo"] = "bar";
  trailers[kFinalOffsetHeaderKey] = base::IntToString(strlen(kResponseBody));
  // Server sends trailers.
  ProcessPacket(ConstructResponseTrailersPacket(
      4, kFin, trailers, &spdy_trailers_frame_length, &offset));

  delegate->WaitUntilNextCallback();  // OnTrailersReceived
  trailers.erase(kFinalOffsetHeaderKey);
  EXPECT_EQ(trailers, delegate->trailers());
  EXPECT_EQ(OK, delegate->ReadData(cb.callback()));

  EXPECT_EQ(1, delegate->on_data_read_count());
  EXPECT_EQ(1, delegate->on_data_sent_count());
  EXPECT_EQ(kProtoQUIC1SPDY3, delegate->GetProtocol());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length +
                                 strlen(kUploadData)),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(
      static_cast<int64_t>(spdy_response_headers_frame_length +
                           strlen(kResponseBody) + spdy_trailers_frame_length),
      delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, InterleaveReadDataAndSendData) {
  SetRequest("POST", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, !kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructAckAndDataPacket(2, !kIncludeVersion, 2, 1, !kFin, 0,
                                     kUploadData));
  AddWrite(ConstructAckAndDataPacket(3, !kIncludeVersion, 3, 3, kFin,
                                     strlen(kUploadData), kUploadData));
  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "POST";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = false;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");
  size_t spdy_response_headers_frame_length;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length, 0));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  // Client sends a data packet.
  scoped_refptr<StringIOBuffer> buf(new StringIOBuffer(kUploadData));

  delegate->SendData(buf.get(), buf->size(), false);
  delegate->WaitUntilNextCallback();  // OnDataSent

  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  const char kResponseBody[] = "Hello world!";

  // Server sends a data packet.
  ProcessPacket(ConstructAckAndDataPacket(3, !kIncludeVersion, 2, 1, !kFin, 0,
                                          kResponseBody));

  EXPECT_EQ(static_cast<int64_t>(strlen(kResponseBody)), cb.WaitForResult());
  EXPECT_EQ(std::string(kResponseBody), delegate->data_received());

  // Client sends a data packet.
  delegate->SendData(buf.get(), buf->size(), true);
  delegate->WaitUntilNextCallback();  // OnDataSent

  TestCompletionCallback cb2;
  rv = delegate->ReadData(cb2.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  ProcessPacket(ConstructAckAndDataPacket(
      4, !kIncludeVersion, 3, 1, kFin, strlen(kResponseBody), kResponseBody));

  EXPECT_EQ(static_cast<int64_t>(strlen(kResponseBody)), cb2.WaitForResult());

  std::string expected_body(kResponseBody);
  expected_body.append(kResponseBody);
  EXPECT_EQ(expected_body, delegate->data_received());

  EXPECT_EQ(OK, delegate->ReadData(cb.callback()));
  EXPECT_EQ(2, delegate->on_data_read_count());
  EXPECT_EQ(2, delegate->on_data_sent_count());
  EXPECT_EQ(kProtoQUIC1SPDY3, delegate->GetProtocol());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length +
                                 2 * strlen(kUploadData)),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(static_cast<int64_t>(spdy_response_headers_frame_length +
                                 2 * strlen(kResponseBody)),
            delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, ServerSendsRstAfterHeaders) {
  SetRequest("GET", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "GET";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = true;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server sends a Rst.
  ProcessPacket(ConstructRstStreamPacket(1));

  delegate->WaitUntilNextCallback();  // OnFailed
  TestCompletionCallback cb;
  EXPECT_EQ(ERR_QUIC_PROTOCOL_ERROR, delegate->ReadData(cb.callback()));

  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(ERR_QUIC_PROTOCOL_ERROR, delegate->error());
  EXPECT_EQ(0, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(0, delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, ServerSendsRstAfterReadData) {
  SetRequest("GET", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  // Why does QUIC ack Rst? Is this expected?
  AddWrite(ConstructAckPacket(2, 3, 1));

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "GET";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = true;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");

  size_t spdy_response_headers_frame_length;
  QuicStreamOffset offset = 0;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      &offset));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);

  // Server sends a Rst.
  ProcessPacket(ConstructRstStreamPacket(3));

  delegate->WaitUntilNextCallback();  // OnFailed

  EXPECT_EQ(ERR_QUIC_PROTOCOL_ERROR, delegate->ReadData(cb.callback()));
  EXPECT_EQ(ERR_QUIC_PROTOCOL_ERROR, delegate->error());
  EXPECT_EQ(0, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(static_cast<int64_t>(spdy_response_headers_frame_length),
            delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, CancelStreamAfterSendData) {
  SetRequest("POST", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, !kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructAckAndDataPacket(2, !kIncludeVersion, 2, 1, !kFin, 0,
                                     kUploadData));
  AddWrite(ConstructRstStreamCancelledPacket(3, strlen(kUploadData)));

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "POST";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = false;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");
  size_t spdy_response_headers_frame_length;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length, 0));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  // Send a DATA frame.
  scoped_refptr<StringIOBuffer> buf(new StringIOBuffer(kUploadData));

  delegate->SendData(buf.get(), buf->size(), false);
  delegate->WaitUntilNextCallback();  // OnDataSent

  delegate->CancelStream();
  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(0, delegate->on_data_read_count());
  EXPECT_EQ(1, delegate->on_data_sent_count());
  EXPECT_EQ(kProtoQUIC1SPDY3, delegate->GetProtocol());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length +
                                 strlen(kUploadData)),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(static_cast<int64_t>(spdy_response_headers_frame_length),
            delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, SessionClosedBeforeReadData) {
  SetRequest("GET", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "GET";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = true;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");

  size_t spdy_response_headers_frame_length;
  QuicStreamOffset offset = 0;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      &offset));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  session()->connection()->CloseConnection(QUIC_NO_ERROR,
                                           ConnectionCloseSource::FROM_PEER);
  delegate->WaitUntilNextCallback();  // OnFailed

  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(ERR_UNEXPECTED, delegate->ReadData(cb.callback()));
  EXPECT_EQ(ERR_UNEXPECTED, delegate->error());
  EXPECT_EQ(0, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
  EXPECT_EQ(kProtoQUIC1SPDY3, delegate->GetProtocol());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(static_cast<int64_t>(spdy_response_headers_frame_length),
            delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, CancelStreamAfterReadData) {
  SetRequest("POST", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, !kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructAckAndRstStreamPacket(2, 2, 1, 1));

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "POST";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = false;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<TestDelegateBase> delegate(
      new TestDelegateBase(read_buffer.get(), kReadBufferSize));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");
  size_t spdy_response_headers_frame_length;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length, 0));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  // Cancel the stream after ReadData returns ERR_IO_PENDING.
  TestCompletionCallback cb;
  EXPECT_EQ(ERR_IO_PENDING, delegate->ReadData(cb.callback()));
  delegate->CancelStream();

  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(0, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
  EXPECT_EQ(kProtoQUIC1SPDY3, delegate->GetProtocol());
  EXPECT_EQ(static_cast<int64_t>(spdy_request_headers_frame_length),
            delegate->GetTotalSentBytes());
  EXPECT_EQ(static_cast<int64_t>(spdy_response_headers_frame_length),
            delegate->GetTotalReceivedBytes());
}

TEST_P(BidirectionalStreamQuicImplTest, DeleteStreamDuringOnHeadersReceived) {
  SetRequest("POST", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, !kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructAckAndRstStreamPacket(2, 2, 1, 1));

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "POST";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = false;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<DeleteStreamDelegate> delegate(new DeleteStreamDelegate(
      read_buffer.get(), kReadBufferSize,
      DeleteStreamDelegate::ON_HEADERS_RECEIVED, true));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");

  size_t spdy_response_headers_frame_length;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      nullptr));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived
  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(0, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
}

TEST_P(BidirectionalStreamQuicImplTest, DeleteStreamDuringOnDataRead) {
  SetRequest("POST", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, !kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructAckPacket(2, 3, 1));
  AddWrite(ConstructRstStreamPacket(3));

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "POST";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = false;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<DeleteStreamDelegate> delegate(
      new DeleteStreamDelegate(read_buffer.get(), kReadBufferSize,
                               DeleteStreamDelegate::ON_DATA_READ, true));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");

  size_t spdy_response_headers_frame_length;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      nullptr));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived

  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  const char kResponseBody[] = "Hello world!";
  // Server sends data.
  ProcessPacket(
      ConstructDataPacket(3, !kIncludeVersion, !kFin, 0, kResponseBody));
  EXPECT_EQ(static_cast<int64_t>(strlen(kResponseBody)), cb.WaitForResult());

  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(1, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
}

TEST_P(BidirectionalStreamQuicImplTest, DeleteStreamDuringOnTrailersReceived) {
  SetRequest("GET", "/", DEFAULT_PRIORITY);
  size_t spdy_request_headers_frame_length;
  AddWrite(ConstructRequestHeadersPacket(1, kFin, DEFAULT_PRIORITY,
                                         &spdy_request_headers_frame_length));
  AddWrite(ConstructAckPacket(2, 3, 1));  // Ack the data packet

  Initialize();

  BidirectionalStreamRequestInfo request;
  request.method = "GET";
  request.url = GURL("http://www.google.com/");
  request.end_stream_on_headers = true;
  request.priority = DEFAULT_PRIORITY;

  scoped_refptr<IOBuffer> read_buffer(new IOBuffer(kReadBufferSize));
  scoped_ptr<DeleteStreamDelegate> delegate(new DeleteStreamDelegate(
      read_buffer.get(), kReadBufferSize,
      DeleteStreamDelegate::ON_TRAILERS_RECEIVED, true));
  delegate->Start(&request, net_log(), session()->GetWeakPtr());
  delegate->WaitUntilNextCallback();  // OnHeadersSent

  // Server acks the request.
  ProcessPacket(ConstructAckPacket(1, 0, 0));

  // Server sends the response headers.
  SpdyHeaderBlock response_headers = ConstructResponseHeaders("200");

  QuicStreamOffset offset = 0;
  size_t spdy_response_headers_frame_length;
  ProcessPacket(ConstructResponseHeadersPacket(
      2, !kFin, response_headers, &spdy_response_headers_frame_length,
      &offset));

  delegate->WaitUntilNextCallback();  // OnHeadersReceived

  EXPECT_EQ("200", delegate->response_headers().find(":status")->second);

  TestCompletionCallback cb;
  int rv = delegate->ReadData(cb.callback());
  EXPECT_EQ(ERR_IO_PENDING, rv);
  const char kResponseBody[] = "Hello world!";
  // Server sends data.
  ProcessPacket(
      ConstructDataPacket(3, !kIncludeVersion, !kFin, 0, kResponseBody));

  EXPECT_EQ(static_cast<int64_t>(strlen(kResponseBody)), cb.WaitForResult());
  EXPECT_EQ(std::string(kResponseBody), delegate->data_received());

  size_t spdy_trailers_frame_length;
  SpdyHeaderBlock trailers;
  trailers["foo"] = "bar";
  trailers[kFinalOffsetHeaderKey] = base::IntToString(strlen(kResponseBody));
  // Server sends trailers.
  ProcessPacket(ConstructResponseTrailersPacket(
      4, kFin, trailers, &spdy_trailers_frame_length, &offset));

  delegate->WaitUntilNextCallback();  // OnTrailersReceived
  trailers.erase(kFinalOffsetHeaderKey);
  EXPECT_EQ(trailers, delegate->trailers());

  base::MessageLoop::current()->RunUntilIdle();

  EXPECT_EQ(1, delegate->on_data_read_count());
  EXPECT_EQ(0, delegate->on_data_sent_count());
}

}  // namespace test

}  // namespace net