nearby/connections/implementation/mediums/webrtc/connection_flow_test.cc

// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "connections/implementation/mediums/webrtc/connection_flow.h"

#include <memory>
#include <utility>
#include <vector>

#include "gtest/gtest.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/mediums/webrtc/data_channel_listener.h"
#include "connections/implementation/mediums/webrtc/local_ice_candidate_listener.h"
#include "connections/implementation/mediums/webrtc/session_description_wrapper.h"
#include "connections/implementation/mediums/webrtc_socket.h"
#include "internal/platform/byte_array.h"
#include "internal/platform/count_down_latch.h"
#include "internal/platform/exception.h"
#include "internal/platform/future.h"
#include "internal/platform/medium_environment.h"
#include "internal/platform/webrtc.h"
#include "webrtc/api/jsep.h"
#include "webrtc/api/scoped_refptr.h"

namespace nearby {
namespace connections {
namespace mediums {
namespace {

class ConnectionFlowTest : public ::testing::Test {
 protected:
  ConnectionFlowTest() {
    MediumEnvironment::Instance().Start({.webrtc_enabled = true});
  }
  ~ConnectionFlowTest() override { MediumEnvironment::Instance().Stop(); }
};

std::unique_ptr<webrtc::IceCandidate> CopyCandidate(
    const webrtc::IceCandidate* candidate) {
  return webrtc::CreateIceCandidate(candidate->sdp_mid(),
                                    candidate->sdp_mline_index(),
                                    candidate->candidate());
}

// TODO(bfranz) - Add test that deterministically sends answerer_ice_candidates
// before answer is sent.
TEST_F(ConnectionFlowTest, SuccessfulOfferAnswerFlow) {
  WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;

  Future<ByteArray> message_received_future;

  Future<WebRtcSocketWrapper> offerer_socket_future, answerer_socket_future;

  std::unique_ptr<ConnectionFlow> offerer, answerer;

  // Send Ice Candidates immediately when you retrieve them
  offerer = ConnectionFlow::Create(
      {.local_ice_candidate_found_cb =
           [&answerer](const webrtc::IceCandidate* candidate) {
             std::vector<std::unique_ptr<webrtc::IceCandidate>> vec;
             vec.push_back(CopyCandidate(candidate));
             // The callback might be alive while the objects in test are
             // destroyed.
             if (answerer)
               answerer->OnRemoteIceCandidatesReceived(std::move(vec));
           }},
      {.data_channel_open_cb =
           [&offerer_socket_future](WebRtcSocketWrapper socket) {
             offerer_socket_future.Set(std::move(socket));
           }},
      {.adapter_type_changed_cb =
           [](webrtc::AdapterType adapter_type) {
             // Do nothing
           }},
      webrtc_medium_offerer);
  ASSERT_NE(offerer, nullptr);
  answerer = ConnectionFlow::Create(
      {.local_ice_candidate_found_cb =
           [&offerer](const webrtc::IceCandidate* candidate) {
             std::vector<std::unique_ptr<webrtc::IceCandidate>> vec;
             vec.push_back(CopyCandidate(candidate));
             // The callback might be alive while the objects in test are
             // destroyed.
             if (offerer)
               offerer->OnRemoteIceCandidatesReceived(std::move(vec));
           }},
      {.data_channel_open_cb =
           [&answerer_socket_future](WebRtcSocketWrapper socket) {
             answerer_socket_future.Set(std::move(socket));
           }},
      {.adapter_type_changed_cb =
           [](webrtc::AdapterType adapter_type) {
             // Do nothing
           }},
      webrtc_medium_answerer);
  ASSERT_NE(answerer, nullptr);

  // Create and send offer
  SessionDescriptionWrapper offer = offerer->CreateOffer();
  ASSERT_TRUE(offer.IsValid());
  EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
  EXPECT_TRUE(answerer->OnOfferReceived(offer));
  EXPECT_TRUE(offerer->SetLocalSessionDescription(std::move(offer)));

  // Create and send answer
  SessionDescriptionWrapper answer = answerer->CreateAnswer();
  ASSERT_TRUE(answer.IsValid());
  EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer);
  EXPECT_TRUE(offerer->OnAnswerReceived(answer));
  EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));

  // Retrieve Data Channels
  ExceptionOr<WebRtcSocketWrapper> offerer_socket =
      offerer_socket_future.Get(absl::Seconds(1));
  EXPECT_TRUE(offerer_socket.ok());
  ExceptionOr<WebRtcSocketWrapper> answerer_socket =
      answerer_socket_future.Get(absl::Seconds(1));
  EXPECT_TRUE(answerer_socket.ok());

  // Send message on data channel
  absl::string_view message = "Test";
  offerer_socket.result().GetImpl().GetOutputStream().Write(message);
  ExceptionOr<ByteArray> received_message =
      answerer_socket.result().GetImpl().GetInputStream().Read(4);
  EXPECT_TRUE(received_message.ok());
  EXPECT_EQ(received_message.result(), ByteArray{message.data()});
}

TEST_F(ConnectionFlowTest, CreateAnswerBeforeOfferReceived) {
  WebRtcMedium webrtc_medium;

  std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium);
  ASSERT_NE(answerer, nullptr);

  SessionDescriptionWrapper answer = answerer->CreateAnswer();
  EXPECT_FALSE(answer.IsValid());
}

TEST_F(ConnectionFlowTest, SetAnswerBeforeOffer) {
  WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;

  std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium_offerer);
  ASSERT_NE(offerer, nullptr);
  std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium_answerer);
  ASSERT_NE(answerer, nullptr);

  SessionDescriptionWrapper offer = offerer->CreateOffer();
  ASSERT_TRUE(offer.IsValid());
  EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
  // Did not set offer as local session description
  EXPECT_TRUE(answerer->OnOfferReceived(offer));

  SessionDescriptionWrapper answer = answerer->CreateAnswer();
  ASSERT_TRUE(answer.IsValid());
  EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer);
  EXPECT_FALSE(offerer->OnAnswerReceived(answer));
}

TEST_F(ConnectionFlowTest, CannotCreateOfferAfterClose) {
  WebRtcMedium webrtc_medium;

  std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium);
  ASSERT_NE(offerer, nullptr);

  EXPECT_TRUE(offerer->CloseIfNotConnected());

  EXPECT_FALSE(offerer->CreateOffer().IsValid());
}

TEST_F(ConnectionFlowTest, CannotSetSessionDescriptionAfterClose) {
  WebRtcMedium webrtc_medium;

  std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium);
  ASSERT_NE(offerer, nullptr);

  SessionDescriptionWrapper offer = offerer->CreateOffer();
  ASSERT_TRUE(offer.IsValid());
  EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);

  EXPECT_TRUE(offerer->CloseIfNotConnected());

  EXPECT_FALSE(offerer->SetLocalSessionDescription(offer));
}

TEST_F(ConnectionFlowTest, CannotReceiveOfferAfterClose) {
  WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;

  std::unique_ptr<ConnectionFlow> offerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium_offerer);
  ASSERT_NE(offerer, nullptr);
  std::unique_ptr<ConnectionFlow> answerer = ConnectionFlow::Create(
      LocalIceCandidateListener(), DataChannelListener(),
      ConnectionFlow::AdapterTypeListener(), webrtc_medium_answerer);
  ASSERT_NE(answerer, nullptr);

  EXPECT_TRUE(answerer->CloseIfNotConnected());

  SessionDescriptionWrapper offer = offerer->CreateOffer();
  ASSERT_TRUE(offer.IsValid());
  EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);

  EXPECT_FALSE(answerer->OnOfferReceived(offer));
}

TEST_F(ConnectionFlowTest, NullPeerConnection) {
  MediumEnvironment::Instance().SetUseValidPeerConnection(
      /*use_valid_peer_connection=*/false);

  WebRtcMedium medium;
  std::unique_ptr<ConnectionFlow> answerer =
      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
                             ConnectionFlow::AdapterTypeListener(), medium);
  EXPECT_EQ(answerer, nullptr);
}

TEST_F(ConnectionFlowTest, PeerConnectionTimeout) {
  MediumEnvironment::Instance().SetUseValidPeerConnection(
      /*use_valid_peer_connection=*/true);
  WebRtcMedium medium1;
  std::unique_ptr<ConnectionFlow> flow1 =
      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
                             ConnectionFlow::AdapterTypeListener(), medium1);
  EXPECT_NE(flow1, nullptr);

  // Attempt to trigger the 2.5s peer connection timeout.
  MediumEnvironment::Instance().SetPeerConnectionLatency(absl::Seconds(5));
  WebRtcMedium medium2;
  std::unique_ptr<ConnectionFlow> flow2 =
      ConnectionFlow::Create(LocalIceCandidateListener(), DataChannelListener(),
                             ConnectionFlow::AdapterTypeListener(), medium2);
  EXPECT_EQ(flow2, nullptr);
}

TEST_F(ConnectionFlowTest, TerminateAnswerer) {
  WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;

  Future<ByteArray> message_received_future;

  Future<WebRtcSocketWrapper> offerer_socket_future, answerer_socket_future;

  std::unique_ptr<ConnectionFlow> offerer, answerer;

  // Send Ice Candidates immediately when you retrieve them
  offerer = ConnectionFlow::Create(
      {.local_ice_candidate_found_cb =
           [&answerer](const webrtc::IceCandidate* candidate) {
             std::vector<std::unique_ptr<webrtc::IceCandidate>> vec;
             vec.push_back(CopyCandidate(candidate));
             // The callback might be alive while the objects in test are
             // destroyed.
             if (answerer)
               answerer->OnRemoteIceCandidatesReceived(std::move(vec));
           }},
      {.data_channel_open_cb =
           [&offerer_socket_future](WebRtcSocketWrapper socket) {
             offerer_socket_future.Set(std::move(socket));
           }},
      {.adapter_type_changed_cb =
           [](webrtc::AdapterType adapter_type) {
             // Do nothing
           }},
      webrtc_medium_offerer);
  ASSERT_NE(offerer, nullptr);
  answerer = ConnectionFlow::Create(
      {.local_ice_candidate_found_cb =
           [&offerer](const webrtc::IceCandidate* candidate) {
             std::vector<std::unique_ptr<webrtc::IceCandidate>> vec;
             vec.push_back(CopyCandidate(candidate));
             // The callback might be alive while the objects in test are
             // destroyed.
             if (offerer)
               offerer->OnRemoteIceCandidatesReceived(std::move(vec));
           }},
      {.data_channel_open_cb =
           [&answerer_socket_future](WebRtcSocketWrapper wrapper) {
             answerer_socket_future.Set(std::move(wrapper));
           }},
      {.adapter_type_changed_cb =
           [](webrtc::AdapterType adapter_type) {
             EXPECT_GE(adapter_type, webrtc::ADAPTER_TYPE_UNKNOWN);
             EXPECT_LE(adapter_type, webrtc::ADAPTER_TYPE_CELLULAR_5G);
           }},
      webrtc_medium_answerer);
  ASSERT_NE(answerer, nullptr);

  // Create and send offer
  SessionDescriptionWrapper offer = offerer->CreateOffer();
  ASSERT_TRUE(offer.IsValid());
  EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
  EXPECT_TRUE(answerer->OnOfferReceived(offer));
  EXPECT_TRUE(offerer->SetLocalSessionDescription(std::move(offer)));

  // Create and send answer
  SessionDescriptionWrapper answer = answerer->CreateAnswer();
  ASSERT_TRUE(answer.IsValid());
  EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer);
  EXPECT_TRUE(offerer->OnAnswerReceived(answer));
  EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));

  // Retrieve Data Channels
  ExceptionOr<WebRtcSocketWrapper> offerer_socket =
      offerer_socket_future.Get(absl::Seconds(1));
  EXPECT_TRUE(offerer_socket.ok());
  ExceptionOr<WebRtcSocketWrapper> answerer_socket =
      answerer_socket_future.Get(absl::Seconds(1));
  EXPECT_TRUE(offerer_socket.ok());

  CountDownLatch latch(1);
  auto pc = answerer->GetPeerConnection();
  pc->signaling_thread()->PostTask([pc, latch]() mutable {
    pc->Close();
    latch.CountDown();
  });
  latch.Await();

  // Send message on data channel
  absl::string_view message = "Test";
  offerer_socket.result().GetOutputStream().Write(message);
  ExceptionOr<ByteArray> received_message =
      answerer_socket.result().GetInputStream().Read(4);
  EXPECT_TRUE(received_message.GetResult().Empty());
}

TEST_F(ConnectionFlowTest, TerminateOfferer) {
  WebRtcMedium webrtc_medium_offerer, webrtc_medium_answerer;

  Future<ByteArray> message_received_future;

  Future<WebRtcSocketWrapper> offerer_socket_future, answerer_socket_future;

  std::unique_ptr<ConnectionFlow> offerer, answerer;

  // Send Ice Candidates immediately when you retrieve them
  offerer = ConnectionFlow::Create(
      {.local_ice_candidate_found_cb =
           [&answerer](const webrtc::IceCandidate* candidate) {
             std::vector<std::unique_ptr<webrtc::IceCandidate>> vec;
             vec.push_back(CopyCandidate(candidate));
             // The callback might be alive while the objects in test are
             // destroyed.
             if (answerer)
               answerer->OnRemoteIceCandidatesReceived(std::move(vec));
           }},
      {.data_channel_open_cb =
           [&offerer_socket_future](WebRtcSocketWrapper socket) {
             offerer_socket_future.Set(std::move(socket));
           }},
      {.adapter_type_changed_cb =
           [](webrtc::AdapterType adapter_type) {
             EXPECT_GE(adapter_type, webrtc::ADAPTER_TYPE_UNKNOWN);
             EXPECT_LE(adapter_type, webrtc::ADAPTER_TYPE_CELLULAR_5G);
           }},
      webrtc_medium_offerer);
  ASSERT_NE(offerer, nullptr);
  answerer = ConnectionFlow::Create(
      {.local_ice_candidate_found_cb =
           [&offerer](const webrtc::IceCandidate* candidate) {
             std::vector<std::unique_ptr<webrtc::IceCandidate>> vec;
             vec.push_back(CopyCandidate(candidate));
             // The callback might be alive while the objects in test are
             // destroyed.
             if (offerer)
               offerer->OnRemoteIceCandidatesReceived(std::move(vec));
           }},
      {.data_channel_open_cb =
           [&answerer_socket_future](WebRtcSocketWrapper wrapper) {
             answerer_socket_future.Set(std::move(wrapper));
           }},
      {.adapter_type_changed_cb =
           [](webrtc::AdapterType adapter_type) {
             EXPECT_GE(adapter_type, webrtc::ADAPTER_TYPE_UNKNOWN);
             EXPECT_LE(adapter_type, webrtc::ADAPTER_TYPE_CELLULAR_5G);
           }},
      webrtc_medium_answerer);
  ASSERT_NE(answerer, nullptr);

  // Create and send offer
  SessionDescriptionWrapper offer = offerer->CreateOffer();
  ASSERT_TRUE(offer.IsValid());
  EXPECT_EQ(offer.GetType(), webrtc::SdpType::kOffer);
  EXPECT_TRUE(answerer->OnOfferReceived(offer));
  EXPECT_TRUE(offerer->SetLocalSessionDescription(std::move(offer)));

  // Create and send answer
  SessionDescriptionWrapper answer = answerer->CreateAnswer();
  ASSERT_TRUE(answer.IsValid());
  EXPECT_EQ(answer.GetType(), webrtc::SdpType::kAnswer);
  EXPECT_TRUE(offerer->OnAnswerReceived(answer));
  EXPECT_TRUE(answerer->SetLocalSessionDescription(std::move(answer)));

  // Retrieve Data Channels
  ExceptionOr<WebRtcSocketWrapper> offerer_socket =
      offerer_socket_future.Get(absl::Seconds(1));
  EXPECT_TRUE(offerer_socket.ok());
  ExceptionOr<WebRtcSocketWrapper> answerer_socket =
      answerer_socket_future.Get(absl::Seconds(1));
  EXPECT_TRUE(offerer_socket.ok());

  CountDownLatch latch(1);
  auto pc = offerer->GetPeerConnection();
  pc->signaling_thread()->PostTask([pc, latch]() mutable {
    pc->Close();
    latch.CountDown();
  });
  latch.Await();

  // Send message on data channel
  absl::string_view message = "Test";
  offerer_socket.result().GetOutputStream().Write(message);
  ExceptionOr<ByteArray> received_message =
      answerer_socket.result().GetInputStream().Read(4);
  EXPECT_TRUE(received_message.GetResult().Empty());
}

}  // namespace
}  // namespace mediums
}  // namespace connections
}  // namespace nearby