Merge pull request #290 from riptano/CPP-813

CPP-813 - Detect CaaS and change consistency defaults

Author: mikefero

cpp-driver/gtests/src/integration/integration.cpp

@@ -325,6 +325,7 @@ void Integration::connect(Cluster cluster) {
   std::stringstream use_keyspace_query;
   use_keyspace_query << "USE " << keyspace_name_;
   session_.execute(use_keyspace_query.str());
+  CHECK_FAILURE;
 }
 
 void Integration::connect() {

cpp-driver/gtests/src/integration/integration.hpp

@@ -107,7 +107,8 @@
 
 #define CHECK_CONTINUE(flag, message) ASSERT_TRUE(flag) << message;
 
-#define CASSANDRA_KEY_VALUE_TABLE_FORMAT "CREATE TABLE %s (key %s PRIMARY KEY, value %s)"
+#define CASSANDRA_KEY_VALUE_TABLE_FORMAT \
+  "CREATE TABLE IF NOT EXISTS %s (key %s PRIMARY KEY, value %s)"
 #define CASSANDRA_KEY_VALUE_INSERT_FORMAT "INSERT INTO %s (key, value) VALUES(%s, %s)"
 #define CASSANDRA_SELECT_VALUE_FORMAT "SELECT value FROM %s WHERE key=%s"
 #define CASSANDRA_DELETE_ROW_FORMAT "DELETE FROM %s WHERE key=%s"

cpp-driver/gtests/src/integration/tests/test_dbaas.cpp

@@ -33,6 +33,7 @@
 #endif
 
 using datastax::internal::bind_callback;
+using datastax::internal::Map;
 using datastax::internal::Memory;
 using datastax::internal::OStringStream;
 using datastax::internal::ScopedMutex;
@@ -1088,6 +1089,15 @@ inline int32_t encode_uuid(CassUuid uuid, String* output) {
   return 16;
 }
 
+int32_t encode_string_map(const Map<String, Vector<String> >& value, String* output) {
+  int32_t size = encode_uint16(value.size(), output);
+  for (Map<String, Vector<String> >::const_iterator it = value.begin(); it != value.end(); ++it) {
+    size += encode_string(it->first, output);
+    size += encode_string_list(it->second, output);
+  }
+  return size;
+}
+
 static String encode_header(int8_t version, int8_t flags, int16_t stream, int8_t opcode,
                             int32_t len) {
   String header;
@@ -1993,7 +2003,17 @@ int32_t ProtocolHandler::decode_frame(ClientConnection* client, const char* fram
         } else {
           return len - remaining;
         }
-        state_ = BODY;
+
+        if (length_ == 0) {
+          decode_body(client, pos, 0);
+          version_ = 0;
+          flags_ = 0;
+          opcode_ = 0;
+          length_ = 0;
+          state_ = PROTOCOL_VERSION;
+        } else {
+          state_ = BODY;
+        }
         break;
       case BODY:
         if (remaining >= length_) {

cpp-driver/gtests/src/unit/mockssandra.cpp

@@ -28,6 +28,7 @@
 #include "address.hpp"
 #include "event_loop.hpp"
 #include "list.hpp"
+#include "map.hpp"
 #include "ref_counted.hpp"
 #include "scoped_ptr.hpp"
 #include "string.hpp"
@@ -46,6 +47,7 @@
 using datastax::String;
 using datastax::internal::Atomic;
 using datastax::internal::List;
+using datastax::internal::Map;
 using datastax::internal::RefCounted;
 using datastax::internal::ScopedPtr;
 using datastax::internal::SharedRefPtr;
@@ -360,6 +362,8 @@ struct QueryParameters {
   String keyspace;
 };
 
+int32_t encode_string_map(const Map<String, Vector<String> >& value, String* output);
+
 class Type {
 public:
   static Type text();

cpp-driver/gtests/src/unit/mockssandra.hpp

@@ -303,7 +303,7 @@ TEST_F(ConnectionUnitTest, SslCancel) {
 }
 
 TEST_F(ConnectionUnitTest, Timeout) {
-  mockssandra::RequestHandler::Builder builder;
+  mockssandra::SimpleRequestHandlerBuilder builder;
   builder.on(mockssandra::OPCODE_STARTUP).no_result(); // Don't return a response
   mockssandra::SimpleCluster cluster(builder.build());
   ASSERT_EQ(cluster.start_all(), 0);

cpp-driver/gtests/src/unit/tests/test_connection.cpp

@@ -740,7 +740,7 @@ TEST_F(DecoderUnitTest, DecodeStringMultiMap) {
                            0, 6, 80, 121, 116, 104, 111, 110,                // Python
                            0, 4, 82, 117, 98,  121 };                        // Ruby
   TestDecoder decoder(input, 58);
-  Map<String, Vector<String> > value;
+  StringMultimap value;
 
   // SUCCESS
   ASSERT_TRUE(decoder.decode_string_multimap(value));

cpp-driver/gtests/src/unit/tests/test_decoder.cpp

@@ -41,8 +41,8 @@ TEST(ExecutionProfileUnitTest, Consistency) {
   Config copy_config = config.new_instance();
   ExecutionProfile profile_lookup;
   ASSERT_TRUE(execution_profile(copy_config, "profile", profile_lookup));
-  ASSERT_EQ(CASS_DEFAULT_CONSISTENCY, profile_lookup.consistency());
-  ASSERT_EQ(CASS_DEFAULT_CONSISTENCY, copy_config.default_profile().consistency());
+  ASSERT_EQ(CASS_CONSISTENCY_UNKNOWN, profile_lookup.consistency());
+  ASSERT_EQ(CASS_CONSISTENCY_UNKNOWN, copy_config.default_profile().consistency());
 }
 
 TEST(ExecutionProfileUnitTest, SerialConsistency) {

cpp-driver/gtests/src/unit/tests/test_exec_profile.cpp

@@ -120,6 +120,21 @@ class SessionUnitTest : public EventLoopTest {
     query(session, true);
   }
 
+  bool check_consistency(const Session& session, CassConsistency expected_consistency,
+                         CassConsistency expected_profile_consistency) {
+    Config session_config = session.config();
+    EXPECT_EQ(expected_consistency, session_config.consistency());
+
+    const ExecutionProfile::Map& profiles = session_config.profiles();
+    for (ExecutionProfile::Map::const_iterator it = profiles.begin(), end = profiles.end();
+         it != end; ++it) {
+      if (expected_profile_consistency != it->second.consistency()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   class HostEventFuture : public Future {
   public:
     typedef SharedRefPtr<HostEventFuture> Ptr;
@@ -241,6 +256,21 @@ class SessionUnitTest : public EventLoopTest {
   private:
     String local_dc_;
   };
+
+  class SupportedDbaasOptions : public mockssandra::Action {
+  public:
+    virtual void on_run(mockssandra::Request* request) const {
+      Vector<String> product_type;
+      product_type.push_back("DATASTAX_APOLLO");
+
+      StringMultimap supported;
+      supported["PRODUCT_TYPE"] = product_type;
+
+      String body;
+      mockssandra::encode_string_map(supported, &body);
+      request->write(mockssandra::OPCODE_SUPPORTED, body);
+    }
+  };
 };
 
 TEST_F(SessionUnitTest, ExecuteQueryNotConnected) {
@@ -859,3 +889,95 @@ TEST_F(SessionUnitTest, NoContactPoints) {
   ASSERT_TRUE(connect_future->error());
   EXPECT_EQ(connect_future->error()->code, CASS_ERROR_LIB_NO_HOSTS_AVAILABLE);
 }
+
+TEST_F(SessionUnitTest, DefaultConsistency) {
+  mockssandra::SimpleCluster cluster(simple());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Session session;
+  {
+    Config session_config = session.config();
+    EXPECT_EQ(CASS_CONSISTENCY_UNKNOWN, session_config.consistency());
+  }
+
+  ExecutionProfile profile;
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.1", 9042));
+  config.set_execution_profile("profile", &profile);
+  connect(config, &session);
+
+  EXPECT_TRUE(check_consistency(session, CASS_DEFAULT_CONSISTENCY, CASS_DEFAULT_CONSISTENCY));
+
+  close(&session);
+}
+
+TEST_F(SessionUnitTest, DefaultConsistencyExecutionProfileNotUpdated) {
+  mockssandra::SimpleCluster cluster(simple());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Session session;
+  {
+    Config session_config = session.config();
+    EXPECT_EQ(CASS_CONSISTENCY_UNKNOWN, session_config.consistency());
+  }
+
+  ExecutionProfile profile;
+  profile.set_consistency(CASS_CONSISTENCY_LOCAL_QUORUM);
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.1", 9042));
+  config.set_execution_profile("profile", &profile);
+  connect(config, &session);
+
+  EXPECT_TRUE(check_consistency(session, CASS_DEFAULT_CONSISTENCY, CASS_CONSISTENCY_LOCAL_QUORUM));
+
+  close(&session);
+}
+
+TEST_F(SessionUnitTest, DbaasDetectionUpdateDefaultConsistency) {
+  mockssandra::SimpleRequestHandlerBuilder builder;
+  builder.on(mockssandra::OPCODE_OPTIONS).execute(new SupportedDbaasOptions());
+  mockssandra::SimpleCluster cluster(builder.build());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Session session;
+  {
+    Config session_config = session.config();
+    EXPECT_EQ(CASS_CONSISTENCY_UNKNOWN, session_config.consistency());
+  }
+
+  ExecutionProfile profile;
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.1", 9042));
+  config.set_execution_profile("profile", &profile);
+  connect(config, &session);
+
+  EXPECT_TRUE(
+      check_consistency(session, CASS_DEFAULT_DBAAS_CONSISTENCY, CASS_DEFAULT_DBAAS_CONSISTENCY));
+
+  close(&session);
+}
+
+TEST_F(SessionUnitTest, DbaasDefaultConsistencyExecutionProfileNotUpdate) {
+  mockssandra::SimpleRequestHandlerBuilder builder;
+  builder.on(mockssandra::OPCODE_OPTIONS).execute(new SupportedDbaasOptions());
+  mockssandra::SimpleCluster cluster(builder.build());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  Session session;
+  {
+    Config session_config = session.config();
+    EXPECT_EQ(CASS_CONSISTENCY_UNKNOWN, session_config.consistency());
+  }
+
+  ExecutionProfile profile;
+  profile.set_consistency(CASS_CONSISTENCY_LOCAL_ONE);
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.1", 9042));
+  config.set_execution_profile("profile", &profile);
+  connect(config, &session);
+
+  EXPECT_TRUE(
+      check_consistency(session, CASS_DEFAULT_DBAAS_CONSISTENCY, CASS_CONSISTENCY_LOCAL_ONE));
+
+  close(&session);
+}

cpp-driver/gtests/src/unit/tests/test_session.cpp

@@ -0,0 +1,131 @@
+/*
+  Copyright (c) DataStax, Inc.
+
+  This software can be used solely with DataStax Enterprise. Please consult the
+  license at http://www.datastax.com/terms/datastax-dse-driver-license-terms
+*/
+
+#include "loop_test.hpp"
+
+#include "options_request.hpp"
+#include "request_callback.hpp"
+#include "supported_response.hpp"
+
+using namespace datastax;
+using namespace datastax::internal;
+using namespace datastax::internal::core;
+
+class SupportedResponseUnitTest : public LoopTest {
+public:
+  const mockssandra::RequestHandler* simple_cluster_with_options() {
+    mockssandra::SimpleRequestHandlerBuilder builder;
+    builder.on(mockssandra::OPCODE_OPTIONS).execute(new SupportedOptions());
+    return builder.build();
+  }
+
+public:
+  static void on_connect(Connector* connector, StringMultimap* supported_options) {
+    ASSERT_TRUE(connector->is_ok());
+    *supported_options = connector->supported_options();
+  }
+
+private:
+  class SupportedOptions : public mockssandra::Action {
+  public:
+    virtual void on_run(mockssandra::Request* request) const {
+      Vector<String> compression;
+      Vector<String> cql_version;
+      Vector<String> protocol_versions;
+      compression.push_back("snappy");
+      compression.push_back("lz4");
+      cql_version.push_back("3.4.5");
+      protocol_versions.push_back("3/v3");
+      protocol_versions.push_back("4/v4");
+
+      StringMultimap supported;
+      supported["COMPRESSION"] = compression;
+      supported["CQL_VERSION"] = cql_version;
+      supported["PROTOCOL_VERSIONS"] = protocol_versions;
+
+      String body;
+      mockssandra::encode_string_map(supported, &body);
+      request->write(mockssandra::OPCODE_SUPPORTED, body);
+    }
+  };
+};
+
+TEST_F(SupportedResponseUnitTest, Simple) {
+  mockssandra::SimpleCluster cluster(simple_cluster_with_options());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  StringMultimap supported_options;
+  ASSERT_EQ(0, supported_options.size());
+  Connector::Ptr connector(new Connector(Host::Ptr(new Host(Address("127.0.0.1", PORT))),
+                                         PROTOCOL_VERSION,
+                                         bind_callback(on_connect, &supported_options)));
+  connector->connect(loop());
+  uv_run(loop(), UV_RUN_DEFAULT);
+
+  ASSERT_EQ(3u, supported_options.size());
+  {
+    Vector<String> compression = supported_options.find("COMPRESSION")->second;
+    ASSERT_EQ(2u, compression.size());
+    EXPECT_EQ("snappy", compression[0]);
+    EXPECT_EQ("lz4", compression[1]);
+  }
+  {
+    Vector<String> cql_version = supported_options.find("CQL_VERSION")->second;
+    ASSERT_EQ(1u, cql_version.size());
+    EXPECT_EQ("3.4.5", cql_version[0]);
+  }
+  {
+    Vector<String> protocol_versions = supported_options.find("PROTOCOL_VERSIONS")->second;
+    ASSERT_EQ(2u, protocol_versions.size());
+    EXPECT_EQ("3/v3", protocol_versions[0]);
+    EXPECT_EQ("4/v4", protocol_versions[1]);
+  }
+
+  { // Non-existent key
+    EXPECT_EQ(supported_options.end(), supported_options.find("invalid"));
+  }
+}
+
+TEST_F(SupportedResponseUnitTest, UppercaseKeysOnly) {
+  class CaseInsensitiveSupportedOptions : public mockssandra::Action {
+  public:
+    virtual void on_run(mockssandra::Request* request) const {
+      Vector<String> camel_key;
+      camel_key.push_back("success");
+
+      StringMultimap supported;
+      supported["CamEL_KeY"] = camel_key;
+
+      String body;
+      mockssandra::encode_string_map(supported, &body);
+      request->write(mockssandra::OPCODE_SUPPORTED, body);
+    }
+  };
+
+  mockssandra::SimpleRequestHandlerBuilder builder;
+  builder.on(mockssandra::OPCODE_OPTIONS).execute(new CaseInsensitiveSupportedOptions());
+  mockssandra::SimpleCluster cluster(builder.build());
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  StringMultimap supported_options;
+  ASSERT_EQ(0, supported_options.size());
+  Connector::Ptr connector(new Connector(Host::Ptr(new Host(Address("127.0.0.1", PORT))),
+                                         PROTOCOL_VERSION,
+                                         bind_callback(on_connect, &supported_options)));
+  connector->connect(loop());
+  uv_run(loop(), UV_RUN_DEFAULT);
+
+  ASSERT_EQ(1u, supported_options.size());
+  { // Uppercase
+    Vector<String> uppercase = supported_options.find("CAMEL_KEY")->second;
+    ASSERT_EQ(1u, uppercase.size());
+    EXPECT_EQ("success", uppercase[0]);
+  }
+  { // Exact key
+    EXPECT_EQ(supported_options.end(), supported_options.find("CamEL_KeY"));
+  }
+}