// Copyright 2017 Google Inc.
//
// 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
//
//     http://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 "google/cloud/bigtable/table.h"
#include "google/cloud/bigtable/testing/mock_read_rows_reader.h"
#include "google/cloud/bigtable/testing/table_test_fixture.h"
#include "google/cloud/testing_util/assert_ok.h"
#include "absl/memory/memory.h"

namespace google {
namespace cloud {
namespace bigtable {
inline namespace BIGTABLE_CLIENT_NS {
namespace {

namespace btproto = ::google::bigtable::v2;
using ::google::cloud::bigtable::testing::MockReadRowsReader;
using ::google::cloud::testing_util::IsContextMDValid;
using ::testing::Return;

class TableReadRowTest : public bigtable::testing::TableTestFixture {};

TEST_F(TableReadRowTest, ReadRowSimple) {
  auto const response = bigtable::testing::ReadRowsResponseFromString(R"(
      chunks {
        row_key: "r1"
        family_name { value: "fam" }
        qualifier { value: "col" }
        timestamp_micros: 42000
        value: "value"
        commit_row: true
      }
)");

  EXPECT_CALL(*client_, ReadRows)
      .WillOnce([&response, this](grpc::ClientContext* context,
                                  btproto::ReadRowsRequest const& req) {
        auto stream = absl::make_unique<MockReadRowsReader>(
            "google.bigtable.v2.Bigtable.ReadRows");
        EXPECT_CALL(*stream, Read)
            .WillOnce([response](btproto::ReadRowsResponse* r) {
              *r = response;
              return true;
            })
            .WillOnce(Return(false));
        EXPECT_CALL(*stream, Finish).WillOnce(Return(grpc::Status::OK));

        EXPECT_STATUS_OK(
            IsContextMDValid(*context, "google.bigtable.v2.Bigtable.ReadRows",
                             google::cloud::internal::ApiClientHeader()));
        EXPECT_EQ(1, req.rows().row_keys_size());
        EXPECT_EQ("r1", req.rows().row_keys(0));
        EXPECT_EQ(1, req.rows_limit());
        EXPECT_EQ(table_.table_name(), req.table_name());
        return stream;
      });

  auto result = table_.ReadRow("r1", bigtable::Filter::PassAllFilter());
  ASSERT_STATUS_OK(result);
  EXPECT_TRUE(std::get<0>(*result));
  auto row = std::get<1>(*result);
  EXPECT_EQ("r1", row.row_key());
}

TEST_F(TableReadRowTest, ReadRowMissing) {
  EXPECT_CALL(*client_, ReadRows)
      .WillOnce([this](grpc::ClientContext* context,
                       btproto::ReadRowsRequest const& req) {
        auto stream = absl::make_unique<MockReadRowsReader>(
            "google.bigtable.v2.Bigtable.ReadRows");
        EXPECT_CALL(*stream, Read).WillOnce(Return(false));
        EXPECT_CALL(*stream, Finish).WillOnce(Return(grpc::Status::OK));

        EXPECT_STATUS_OK(
            IsContextMDValid(*context, "google.bigtable.v2.Bigtable.ReadRows",
                             google::cloud::internal::ApiClientHeader()));
        EXPECT_EQ(1, req.rows().row_keys_size());
        EXPECT_EQ("r1", req.rows().row_keys(0));
        EXPECT_EQ(1, req.rows_limit());
        EXPECT_EQ(table_.table_name(), req.table_name());
        return stream;
      });

  auto result = table_.ReadRow("r1", bigtable::Filter::PassAllFilter());
  ASSERT_STATUS_OK(result);
  EXPECT_FALSE(std::get<0>(*result));
}

TEST_F(TableReadRowTest, UnrecoverableFailure) {
  EXPECT_CALL(*client_, ReadRows)
      .WillRepeatedly([](grpc::ClientContext* context,
                         btproto::ReadRowsRequest const&) {
        auto stream = absl::make_unique<MockReadRowsReader>(
            "google.bigtable.v2.Bigtable.ReadRows");
        EXPECT_CALL(*stream, Read).WillRepeatedly(Return(false));
        EXPECT_CALL(*stream, Finish)
            .WillRepeatedly(Return(
                grpc::Status(grpc::StatusCode::PERMISSION_DENIED, "uh oh")));

        EXPECT_STATUS_OK(
            IsContextMDValid(*context, "google.bigtable.v2.Bigtable.ReadRows",
                             google::cloud::internal::ApiClientHeader()));
        return stream;
      });

  auto row = table_.ReadRow("r1", bigtable::Filter::PassAllFilter());
  EXPECT_FALSE(row);
}

}  // anonymous namespace
}  // namespace BIGTABLE_CLIENT_NS
}  // namespace bigtable
}  // namespace cloud
}  // namespace google