text: // Copyright 2019 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
  //
  //     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/spanner/client.h"
  #include "google/cloud/spanner/database.h"
  #include "google/cloud/spanner/database_admin_client.h"
  #include "google/cloud/spanner/mutations.h"
  #include "google/cloud/spanner/testing/database_integration_test.h"
  #include "google/cloud/spanner/timestamp.h"
  #include "google/cloud/internal/getenv.h"
  #include "google/cloud/testing_util/status_matchers.h"
  #include "absl/memory/memory.h"
  #include "absl/time/time.h"
  #include <gmock/gmock.h>
  
  namespace google {
  namespace cloud {
  namespace spanner {
  inline namespace SPANNER_CLIENT_NS {
  namespace {
  
  using ::google::cloud::testing_util::IsOk;
  using ::google::cloud::testing_util::StatusIs;
  using ::testing::HasSubstr;
  using ::testing::UnorderedElementsAreArray;
  
  absl::Time MakeTime(std::time_t sec, int nanos) {
    return absl::FromTimeT(sec) + absl::Nanoseconds(nanos);
  }
  
  // A helper function used in the test fixtures below. This function writes the
  // given data to the DataTypes table, then it reads all the data back and
  // returns it to the caller.
  template <typename T>
  StatusOr<T> WriteReadData(Client& client, T const& data,
                            std::string const& column) {
    Mutations mutations;
    int id = 0;
    for (auto&& x : data) {
      mutations.push_back(MakeInsertMutation("DataTypes", {"Id", column},
                                             "Id-" + std::to_string(id++), x));
    }
    auto commit_result = client.Commit(std::move(mutations));
    if (!commit_result) return commit_result.status();
  
    T actual;
    auto rows = client.Read("DataTypes", KeySet::All(), {column});
    using RowType = std::tuple<typename T::value_type>;
    for (auto const& row : StreamOf<RowType>(rows)) {
      if (!row) return row.status();
      actual.push_back(std::get<0>(*row));
    }
    return actual;
  }
  
  class DataTypeIntegrationTest
      : public spanner_testing::DatabaseIntegrationTest {
   public:
    static void SetUpTestSuite() {
      spanner_testing::DatabaseIntegrationTest::SetUpTestSuite();
      client_ = absl::make_unique<Client>(MakeConnection(GetDatabase()));
    }
  
    void SetUp() override {
      auto commit_result = client_->Commit(Mutations{
          MakeDeleteMutation("DataTypes", KeySet::All()),
      });
      EXPECT_STATUS_OK(commit_result);
    }
  
    static void TearDownTestSuite() {
      client_ = nullptr;
      spanner_testing::DatabaseIntegrationTest::TearDownTestSuite();
    }
  
    static bool UsingEmulator() {
      return google::cloud::internal::GetEnv("SPANNER_EMULATOR_HOST").has_value();
    }
  
   protected:
    static std::unique_ptr<Client> client_;
  };
  
  std::unique_ptr<Client> DataTypeIntegrationTest::client_;
  
  TEST_F(DataTypeIntegrationTest, WriteReadBool) {
    std::vector<bool> const data = {true, false};
    auto result = WriteReadData(*client_, data, "BoolValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadInt64) {
    std::vector<std::int64_t> const data = {
        std::numeric_limits<std::int64_t>::min(), -123, -42, -1, 0, 1, 42, 123,
        std::numeric_limits<std::int64_t>::max(),
    };
    auto result = WriteReadData(*client_, data, "Int64Value");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadFloat64) {
    std::vector<double> const data = {
        -std::numeric_limits<double>::infinity(),
        std::numeric_limits<double>::lowest(),
        std::numeric_limits<double>::min(),
        -123.456,
        -123,
        -42.42,
        -42,
        -1.5,
        -1,
        -0.5,
        0,
        0.5,
        1,
        1.5,
        42,
        42.42,
        123,
        123.456,
        std::numeric_limits<double>::max(),
        std::numeric_limits<double>::infinity(),
    };
    auto result = WriteReadData(*client_, data, "Float64Value");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadFloat64NaN) {
    // Since NaN is not equal to anything, including itself, we need to handle
    // NaN separately from other Float64 values.
    std::vector<double> const data = {
        std::numeric_limits<double>::quiet_NaN(),
    };
    auto result = WriteReadData(*client_, data, "Float64Value");
    ASSERT_STATUS_OK(result);
    EXPECT_EQ(1, result->size());
    EXPECT_TRUE(std::isnan(result->front()));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadString) {
    std::vector<std::string> const data = {
        "",
        "a",
        "Hello World",
        "123456789012345678901234567890",
        std::string(1024, 'x'),
    };
    auto result = WriteReadData(*client_, data, "StringValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadBytes) {
    // Makes a blob containing unprintable characters.
    std::string blob;
    for (char c = std::numeric_limits<char>::min();
         c != std::numeric_limits<char>::max(); ++c) {
      blob.push_back(c);
    }
    std::vector<Bytes> const data = {
        Bytes(""),
        Bytes("a"),
        Bytes("Hello World"),
        Bytes("123456789012345678901234567890"),
        Bytes(blob),
    };
    auto result = WriteReadData(*client_, data, "BytesValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadTimestamp) {
    auto min = spanner_internal::TimestampFromRFC3339("0001-01-01T00:00:00Z");
    ASSERT_STATUS_OK(min);
    auto max =
        spanner_internal::TimestampFromRFC3339("9999-12-31T23:59:59.999999999Z");
    ASSERT_STATUS_OK(max);
    auto now = MakeTimestamp(std::chrono::system_clock::now());
    ASSERT_STATUS_OK(now);
  
    std::vector<Timestamp> const data = {
        *min,
        MakeTimestamp(MakeTime(-1, 0)).value(),
        MakeTimestamp(MakeTime(0, -1)).value(),
        MakeTimestamp(MakeTime(0, 0)).value(),
        MakeTimestamp(MakeTime(0, 1)).value(),
        MakeTimestamp(MakeTime(1, 0)).value(),
        *now,
        *max,
    };
    auto result = WriteReadData(*client_, data, "TimestampValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadDate) {
    std::vector<absl::CivilDay> const data = {
        absl::CivilDay(1, 1, 1),       //
        absl::CivilDay(161, 3, 8),     //
        absl::CivilDay(),              //
        absl::CivilDay(2019, 11, 21),  //
        absl::CivilDay(9999, 12, 31),  //
    };
    auto result = WriteReadData(*client_, data, "DateValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadJson) {
    // TODO(#6873): Remove this check when the emulator supports JSON.
    if (UsingEmulator()) GTEST_SKIP();
  
    std::vector<Json> const data = {
        Json(),                     //
        Json(R"("Hello world!")"),  //
        Json("42"),                 //
        Json("true"),               //
    };
    auto result = WriteReadData(*client_, data, "JsonValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadNumeric) {
    // TODO(#5024): Remove this check when the emulator supports NUMERIC.
    if (UsingEmulator()) GTEST_SKIP();
  
    auto min = MakeNumeric("-99999999999999999999999999999.999999999");
    ASSERT_STATUS_OK(min);
    auto max = MakeNumeric("99999999999999999999999999999.999999999");
    ASSERT_STATUS_OK(max);
  
    std::vector<Numeric> const data = {
        *min,                                //
        MakeNumeric(-999999999e-3).value(),  //
        MakeNumeric(-1).value(),             //
        MakeNumeric(0).value(),              //
        MakeNumeric(1).value(),              //
        MakeNumeric(999999999e-3).value(),   //
        *max,                                //
    };
    auto result = WriteReadData(*client_, data, "NumericValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayBool) {
    std::vector<std::vector<bool>> const data = {
        std::vector<bool>{},
        std::vector<bool>{true},
        std::vector<bool>{false},
        std::vector<bool>{true, false},
        std::vector<bool>{false, true},
    };
    auto result = WriteReadData(*client_, data, "ArrayBoolValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayInt64) {
    std::vector<std::vector<std::int64_t>> const data = {
        std::vector<std::int64_t>{},
        std::vector<std::int64_t>{-1},
        std::vector<std::int64_t>{-1, 0, 1},
    };
    auto result = WriteReadData(*client_, data, "ArrayInt64Value");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayFloat64) {
    std::vector<std::vector<double>> const data = {
        std::vector<double>{},
        std::vector<double>{-0.5},
        std::vector<double>{-0.5, 0.5, 1.5},
    };
    auto result = WriteReadData(*client_, data, "ArrayFloat64Value");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayString) {
    std::vector<std::vector<std::string>> const data = {
        std::vector<std::string>{},
        std::vector<std::string>{""},
        std::vector<std::string>{"", "foo", "bar"},
    };
    auto result = WriteReadData(*client_, data, "ArrayStringValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayBytes) {
    std::vector<std::vector<Bytes>> const data = {
        std::vector<Bytes>{},
        std::vector<Bytes>{Bytes("")},
        std::vector<Bytes>{Bytes(""), Bytes("foo"), Bytes("bar")},
    };
    auto result = WriteReadData(*client_, data, "ArrayBytesValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayTimestamp) {
    std::vector<std::vector<Timestamp>> const data = {
        std::vector<Timestamp>{},
        std::vector<Timestamp>{MakeTimestamp(MakeTime(-1, 0)).value()},
        std::vector<Timestamp>{
            MakeTimestamp(MakeTime(-1, 0)).value(),
            MakeTimestamp(MakeTime(0, 0)).value(),
            MakeTimestamp(MakeTime(1, 0)).value(),
        },
    };
    auto result = WriteReadData(*client_, data, "ArrayTimestampValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayDate) {
    std::vector<std::vector<absl::CivilDay>> const data = {
        std::vector<absl::CivilDay>{},
        std::vector<absl::CivilDay>{absl::CivilDay()},
        std::vector<absl::CivilDay>{
            absl::CivilDay(1, 1, 1),
            absl::CivilDay(),
            absl::CivilDay(9999, 12, 31),
        },
    };
    auto result = WriteReadData(*client_, data, "ArrayDateValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayJson) {
    // TODO(#6873): Remove this check when the emulator supports JSON.
    if (UsingEmulator()) GTEST_SKIP();
  
    std::vector<std::vector<Json>> const data = {
        std::vector<Json>{},
        std::vector<Json>{Json()},
        std::vector<Json>{
            Json(),
            Json(R"("Hello world!")"),
            Json("42"),
            Json("true"),
        },
    };
    auto result = WriteReadData(*client_, data, "ArrayJsonValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, WriteReadArrayNumeric) {
    // TODO(#5024): Remove this check when the emulator supports NUMERIC.
    if (UsingEmulator()) GTEST_SKIP();
  
    std::vector<std::vector<Numeric>> const data = {
        std::vector<Numeric>{},
        std::vector<Numeric>{Numeric()},
        std::vector<Numeric>{
            MakeNumeric(-1e+9).value(),
            MakeNumeric(1e-9).value(),
            MakeNumeric(1e+9).value(),
        },
    };
    auto result = WriteReadData(*client_, data, "ArrayNumericValue");
    ASSERT_STATUS_OK(result);
    EXPECT_THAT(*result, UnorderedElementsAreArray(data));
  }
  
  TEST_F(DataTypeIntegrationTest, InsertAndQueryWithNumericKey) {
    // TODO(#5024): Remove this check when the emulator supports NUMERIC.
    if (UsingEmulator()) GTEST_SKIP();
  
    auto& client = *client_;
    auto const key = MakeNumeric(42).value();
  
    auto commit_result = client.Commit(
        Mutations{InsertOrUpdateMutationBuilder("NumericKey", {"Key"})
                      .EmplaceRow(key)
                      .Build()});
    ASSERT_STATUS_OK(commit_result);
  
    auto rows = client.Read("NumericKey", KeySet::All(), {"Key"});
    using RowType = std::tuple<Numeric>;
    auto row = GetSingularRow(StreamOf<RowType>(rows));
    ASSERT_STATUS_OK(row);
    EXPECT_EQ(std::get<0>(*std::move(row)), key);
  }
  
  // This test differs a lot from the other tests since Spanner STRUCT types may
  // not be used as column types, and they may not be returned as top-level
  // objects in a select statement. See
  // https://cloud.google.com/spanner/docs/data-types#struct-type for more info.
  //
  // So the approach taken here instead is to use a STRUCT (i.e., a std::tuple<>
  // in the C++ world), as a bound SQL query parameter to insert some data into
  // the table. This, in a way, tests the "write" path.
  //
  // To test the "read" path, we create a query that returns an array of struct,
  // that we then compare to the original data.
  TEST_F(DataTypeIntegrationTest, InsertAndQueryWithStruct) {
    using StructType =
        std::tuple<std::pair<std::string, std::string>,
                   std::pair<std::string, std::vector<std::int64_t>>>;
    auto data = StructType{{"StringValue", "xx"}, {"ArrayInt64Value", {1, 2, 3}}};
  
    auto& client = *client_;
    auto commit_result = client.Commit(
        [&data, &client](Transaction const& txn) -> StatusOr<Mutations> {
          auto dml_result = client.ExecuteDml(
              txn,
              SqlStatement(
                  "INSERT INTO DataTypes (Id, StringValue, ArrayInt64Value)"
                  "VALUES(@id, @struct.StringValue, @struct.ArrayInt64Value)",
                  {{"id", Value("id-1")}, {"struct", Value(data)}}));
          if (!dml_result) return dml_result.status();
          return Mutations{};
        });
    ASSERT_STATUS_OK(commit_result);
  
    auto rows = client_->ExecuteQuery(
        SqlStatement("SELECT ARRAY(SELECT STRUCT(StringValue, ArrayInt64Value)) "
                     "FROM DataTypes"));
    using RowType = std::tuple<std::vector<StructType>>;
    auto row = GetSingularRow(StreamOf<RowType>(rows));
    ASSERT_STATUS_OK(row);
  
    auto const& v = std::get<0>(*row);
    EXPECT_EQ(1, v.size());
    EXPECT_EQ(data, v[0]);
  }
  
  // Verify maximum JSON nesting.
  TEST_F(DataTypeIntegrationTest, JsonMaxNesting) {
    // TODO(#6873): Remove this check when the emulator supports JSON.
    if (UsingEmulator()) GTEST_SKIP();
  
    // The default value of the backend max-nesting-level flag.
    int const k_spanner_json_max_nesting_level = 100;
  
    // Nested arrays that exceed `k_spanner_json_max_nesting_level` by one.
    std::string bad_json;
    for (int i = 0; i != k_spanner_json_max_nesting_level + 1; ++i)
      bad_json.append(1, '[');
    bad_json.append("null");
    for (int i = 0; i != k_spanner_json_max_nesting_level + 1; ++i)
      bad_json.append(1, ']');
  
    // Nested arrays that match `k_spanner_json_max_nesting_level`.
    std::string good_json = bad_json.substr(1, bad_json.size() - 2);
  
    std::vector<Json> const good_data = {Json(good_json)};
    auto result = WriteReadData(*client_, good_data, "JsonValue");
    ASSERT_THAT(result, IsOk());
    EXPECT_THAT(*result, UnorderedElementsAreArray(good_data));
  
    std::vector<Json> const bad_data = {Json(bad_json)};
    result = WriteReadData(*client_, bad_data, "JsonValue");
    // NOTE: The backend is currently dropping a more specific "Max nesting
    // of 100 had been exceeded [INVALID_ARGUMENT]" error, so expect this
    // expectation to change when that problem is fixed.
    EXPECT_THAT(result, StatusIs(StatusCode::kFailedPrecondition,
                                 HasSubstr("Expected JSON")));
  }
  
  }  // namespace
  }  // namespace SPANNER_CLIENT_NS
  }  // namespace spanner
  }  // namespace cloud
  }  // namespace google
  