tests/src/integration/integration.cpp - cassandra-cpp-driver - Git at Google

 /*
   Licensed to the Apache Software Foundation (ASF) under one
   or more contributor license agreements.  See the NOTICE file
   distributed with this work for additional information
   regarding copyright ownership.  The ASF licenses this file
   to you 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 "integration.hpp"
 #include "options.hpp"

 #include "driver_info.hpp"
 #include "get_time.hpp"
 #include "murmur3.hpp"

 #include <algorithm>
 #include <cstdarg>
 #include <iostream>
 #include <sys/stat.h>

 #define FORMAT_BUFFER_SIZE 10240
 #define ENTITY_MAXIMUM_LENGTH 48
 #define SIMPLE_KEYSPACE_FORMAT "CREATE KEYSPACE IF NOT EXISTS %s WITH replication = %s"
 #define REPLICATION_STRATEGY "{ 'class': %s }"
 #define SELECT_SERVER_VERSION "SELECT release_version FROM system.local"
 #define LOGGER_MAXIMUM_WAIT_TIME_MS 10000u
 #define LOGGER_WAIT_FOR_NAP_MS 100

 // Initialize static variables
 bool Integration::skipped_message_displayed_ = false;

 using namespace datastax::internal;
 using namespace ::testing;

 Integration::Integration()
     : ccm_(NULL)
     , session_()
     , keyspace_name_("")
     , table_name_("")
     , system_schema_keyspaces_("system.schema_keyspaces")
     , uuid_generator_()
     , server_version_(Options::server_version())
     , number_dc1_nodes_(1)
     , number_dc2_nodes_(0)
     , replication_factor_(0)
     , replication_strategy_("")
     , contact_points_("")
     , is_password_authenticator_(false)
     , is_client_authentication_(false)
     , is_ssl_(false)
     , is_with_vnodes_(false)
     , is_randomized_contact_points_(false)
     , is_schema_metadata_(false)
     , is_ccm_requested_(true)
     , is_ccm_start_requested_(true)
     , is_ccm_start_node_individually_(false)
     , is_session_requested_(true)
     , is_keyspace_change_requested_(true)
     , is_test_chaotic_(false)
     , protocol_version_(CASS_PROTOCOL_VERSION_V4)
     , create_keyspace_query_("")
     , start_time_(0ull) {
   // Determine if the schema keyspaces table should be updated
   // TODO: Make cass_version (and dse_version) available for all tests
   CCM::CassVersion cass_version = server_version_;
   if (!Options::is_cassandra()) {
     cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version();
   }
   if (cass_version >= "3.0.0") {
     system_schema_keyspaces_ = "system_schema.keyspaces";
   }

   // Get the name of the test and the case/suite it belongs to
   const TestInfo* test_information = UnitTest::GetInstance()->current_test_info();
   test_name_ = test_information->name();

   // Determine if this is a typed test (e.g. ends in a number)
   const char* type_param = test_information->type_param();
   if (type_param) {
     std::vector<std::string> tokens = explode(test_information->test_case_name(), '/');
     for (std::vector<std::string>::const_iterator iterator = tokens.begin();
          iterator < tokens.end(); ++iterator) {
       std::string token = *iterator;

       // Determine if we are looking at the last token
       if ((iterator + 1) == tokens.end()) {
         size_t number = 0;
         std::stringstream tokenStream(token);
         if (!(tokenStream >> number).fail()) {
           std::vector<std::string> type_param_tokens = explode(type_param, ':');
           size_t size = type_param_tokens.size();
           test_case_name_ += test::Utils::replace_all(type_param_tokens[size - 1], ">", "");
         }
       } else {
         test_case_name_ += token + "_";
       }
     }
   } else {
     test_case_name_ = test_information->test_case_name();
   }

   // Determine if file logging should be enabled for the integration tests
   if (Options::log_tests()) {
     logger_.initialize(test_case_name_, test_name_);
   }
 }

 Integration::~Integration() {
   try {
     session_.close(false);
   } catch (...) {
   }

   // Reset the skipped message displayed state
   skipped_message_displayed_ = false;
 }

 void Integration::SetUp() {
   // Initialize the DSE workload (iff not set)
   if (dse_workload_.empty()) {
     dse_workload_.push_back(CCM::DSE_WORKLOAD_CASSANDRA);
   }
   // Generate the default settings for most tests (handles overridden values)
   keyspace_name_ = default_keyspace();
   table_name_ = default_table();

   if (replication_factor_ == 0) {
     replication_factor_ = default_replication_factor();
   }
   replication_strategy_ = default_replication_strategy();

   // Generate the keyspace query
   create_keyspace_query_ =
       format_string(SIMPLE_KEYSPACE_FORMAT, keyspace_name_.c_str(), replication_strategy_.c_str());

   // Create the data center nodes vector
   std::vector<unsigned short> data_center_nodes;
   data_center_nodes.push_back(number_dc1_nodes_);
   data_center_nodes.push_back(number_dc2_nodes_);

   if (is_ccm_requested_) {
     try {
       // Create and start the CCM cluster (if not already created)
       ccm_ = new CCM::Bridge(
           server_version_, Options::use_git(), Options::branch_tag(), Options::use_install_dir(),
           Options::install_dir(), Options::server_type(), dse_workload_, Options::cluster_prefix(),
           Options::dse_credentials(), Options::dse_username(), Options::dse_password(),
           Options::deployment_type(), Options::authentication_type(), Options::host(),
           Options::port(), Options::username(), Options::password(), Options::public_key(),
           Options::private_key(), Options::is_verbose_ccm());
       if (ccm_->create_cluster(data_center_nodes, is_with_vnodes_, is_password_authenticator_,
                                is_ssl_, is_client_authentication_)) {
         if (is_ccm_start_requested_) {
           if (is_ccm_start_node_individually_) {
             for (unsigned short node = 1; node <= (number_dc1_nodes_ + number_dc2_nodes_); ++node) {
               if (is_password_authenticator_) {
                 ccm_->start_node(node, "-Dcassandra.superuser_setup_delay_ms=0");
               } else {
                 ccm_->start_node(node);
               }
             }
           } else {
             if (is_password_authenticator_) {
               ccm_->start_cluster("-Dcassandra.superuser_setup_delay_ms=0");
             } else {
               ccm_->start_cluster();
             }
           }
         }
       }

       // Generate the default contact points
       contact_points_ =
           generate_contact_points(ccm_->get_ip_prefix(), number_dc1_nodes_ + number_dc2_nodes_);

       // Determine if the session connection should be established
       if (is_session_requested_ && is_ccm_start_requested_) {
         connect();
       }
     } catch (CCM::BridgeException be) {
       // Issue creating the CCM bridge instance (force failure)
       FAIL() << be.what();
     }
   }
 }

 void Integration::TearDown() {
   // Restart and resume all stopped and paused nodes
   if (!is_test_chaotic_) { // No need to restart as cluster will be destroyed
     for (std::vector<unsigned int>::iterator it = stopped_nodes_.begin(),
                                              end = stopped_nodes_.end();
          it != end; ++it) {
       TEST_LOG("Restarting Node Stopped in " << test_name_ << ": " << *it);
       ccm_->start_node(*it);
     }
     for (std::vector<unsigned int>::iterator it = paused_nodes_.begin(), end = paused_nodes_.end();
          it != end; ++it) {
       TEST_LOG("Resuming Node Paused in " << test_name_ << ": " << *it);
       ccm_->resume_node(*it);
     }
   }
   stopped_nodes_.clear();
   paused_nodes_.clear();

   // Drop keyspace for integration test (may or may have not been created)
   if (!is_test_chaotic_) { // No need to drop keyspace as cluster will be destroyed
     std::stringstream use_keyspace_query;
     use_keyspace_query << "DROP KEYSPACE " << keyspace_name_;
     try {
       session_.execute(use_keyspace_query.str(), CASS_CONSISTENCY_ANY, false, false);
     } catch (...) {
     }
   }

   // Determine if the CCM cluster should be destroyed
   if (is_test_chaotic_) {
     // Destroy the current cluster and reset the chaos flag for the next test
     if (!Options::keep_clusters()) {
       ccm_->remove_cluster();
     } else {
       TEST_LOG_ERROR("Chaotic test detected and keep clusters requested. Cluster will not be "
                      "removed and may result in later failures.");
     }
     is_test_chaotic_ = false;
   }
 }

 std::string Integration::default_keyspace() {
   if (!keyspace_name_.empty()) {
     return keyspace_name_;
   }

   // Clean up the initial keyspace name (remove category information)
   keyspace_name_ = to_lower(test_case_name_) + "_" + to_lower(test_name_);
   keyspace_name_ = replace_all(keyspace_name_, "tests",
                                ""); // TODO: Rename integration tests (remove 's' or add 's')
   keyspace_name_ = replace_all(keyspace_name_, "test", "");
   keyspace_name_ = replace_all(keyspace_name_, "integration", "");
   for (TestCategory::iterator iterator = TestCategory::begin(); iterator != TestCategory::end();
        ++iterator) {
     keyspace_name_ = replace_all(keyspace_name_, "_" + to_lower(iterator->name()) + "_", "");
   }

   // Generate the keyspace name
   maybe_shrink_name(keyspace_name_);
   return keyspace_name_;
 }

 unsigned short Integration::default_replication_factor() {
   // Calculate and return the default replication factor
   return (number_dc1_nodes_ % 2 == 0) ? number_dc1_nodes_ / 2 : (number_dc1_nodes_ + 1) / 2;
 }

 std::string Integration::default_replication_strategy() {
   // Determine the replication strategy
   std::stringstream replication_strategy_s;
   if (number_dc2_nodes_ > 0) {
     replication_strategy_s << "'NetworkTopologyStrategy', 'dc1': " << number_dc1_nodes_ << ", "
                            << "'dc2': " << number_dc2_nodes_;
   } else {
     replication_strategy_s << "'SimpleStrategy', 'replication_factor': ";

     // Ensure the replication factor has not been overridden or already set
     if (replication_factor_ == 0) {
       replication_factor_ = default_replication_factor();
     }
     replication_strategy_s << replication_factor_;
   }

   // Return the default replication strategy
   std::string replication_strategy = replication_strategy_s.str();
   return format_string(REPLICATION_STRATEGY, replication_strategy.c_str());
 }

 std::string Integration::default_select_all() {
   std::stringstream cql;
   cql << "SELECT * FROM " << default_keyspace() << "." << default_table();
   return cql.str();
 }

 int64_t Integration::default_select_count() {
   Result result = session_.execute(format_string(SELECT_COUNT_FORMAT, table_name_.c_str()));
   EXPECT_EQ(CASS_OK, result.error_code()) << "Unable to get Row Count: " << result.error_message();
   return result.first_row().next().as<BigInteger>().value();
 }

 std::string Integration::default_table() {
   if (!table_name_.empty()) {
     return table_name_;
   }

   table_name_ = to_lower(test_name_);
   table_name_ = replace_all(table_name_, "integration_", "");
   maybe_shrink_name(table_name_);
   return table_name_;
 }

 void Integration::drop_table(const std::string& table_name) {
   // Drop table from the current keyspace
   std::stringstream drop_table_query;
   drop_table_query << "DROP TABLE " << table_name;
   session_.execute(drop_table_query.str(), CASS_CONSISTENCY_ANY, false, false);
 }

 void Integration::drop_type(const std::string& type_name) {
   // Drop type from the current keyspace
   std::stringstream drop_type_query;
   drop_type_query << "DROP TYPE " << type_name;
   session_.execute(drop_type_query.str(), CASS_CONSISTENCY_ANY, false, false);
 }

 bool Integration::use_keyspace(const std::string& keyspace_name) {
   std::stringstream use_keyspace_query;
   use_keyspace_query << "USE " << keyspace_name;
   session_.execute(use_keyspace_query.str());
   if (this->HasFailure()) {
     return false;
   }
   return true;
 }

 void Integration::connect(Cluster cluster) {
   // Establish the session connection
   cluster_ = cluster;
   session_ = cluster.connect();
   CHECK_FAILURE;

   // Update the server version if branch_tag was specified
   if (Options::use_git() && !Options::branch_tag().empty()) {
     if (Options::is_ddac()) {
       FAIL() << "Unable to build DDAC from Branch/Tag";
       return;
     }
     if (Options::is_dse()) {
       server_version_ = ccm_->get_dse_version();
     } else {
       server_version_ = ccm_->get_cassandra_version();
     }
     TEST_LOG("Branch/Tag Option was Used: Retrieved server version is "
              << server_version_.to_string());
   }

   // Create the keyspace for the integration test
   session_.execute(create_keyspace_query_);
   CHECK_FAILURE;

   // Update the session to use the new keyspace by default
   if (is_keyspace_change_requested_) {
     use_keyspace(keyspace_name_);
   }
 }

 void Integration::connect() {
   // Create the cluster configuration and establish the session connection
   cluster_ = default_cluster();
   connect(cluster_);
 }

 test::driver::Cluster Integration::default_cluster(bool is_with_default_contact_points /*= true*/) {
   // Create the default cluster object
   Cluster cluster = Cluster::build()
                         .with_randomized_contact_points(is_randomized_contact_points_)
                         .with_schema_metadata(is_schema_metadata_);
   if (is_with_default_contact_points) {
     cluster.with_contact_points(contact_points_);
   }
   cluster.with_protocol_version(protocol_version_);

   // Assign the execution profiles to the cluster object (if available)
   for (ExecutionProfile::Map::iterator it = profiles_.begin(); it != profiles_.end(); ++it) {
     cluster.with_execution_profile(it->first, it->second);
   }
   return cluster;
 }

 void Integration::enable_cluster_tracing(bool enable /*= true*/) {
   std::vector<std::string> active_nodes = ccm_->cluster_ip_addresses();
   for (std::vector<std::string>::iterator iterator = active_nodes.begin();
        iterator != active_nodes.end(); ++iterator) {
     // Get the node number from the IP address
     std::string node_ip_address = *iterator;
     std::stringstream node_value;
     node_value << node_ip_address[node_ip_address.length() - 1];

     // Enable tracing on the node
     unsigned int node;
     node_value >> node;
     ccm_->enable_node_trace(node);
   }
 }

 bool Integration::decommission_node(unsigned int node, bool is_force /*= false */) {
   // Decommission the requested node
   bool status = ccm_->decommission_node(node, is_force);
   if (status) {
     // Indicate the test is chaotic
     is_test_chaotic_ = true;
   }
   return status;
 }

 bool Integration::force_decommission_node(unsigned int node) {
   return decommission_node(node, true);
 }

 bool Integration::start_node(unsigned int node) {
   // Start the requested node and ensure paused nodes are ignored
   std::vector<unsigned>::iterator it = std::find(paused_nodes_.begin(), paused_nodes_.end(), node);
   if (it == paused_nodes_.end() && ccm_->is_node_down(node, true)) {
     bool status = ccm_->start_node(node);
     it = std::find(stopped_nodes_.begin(), stopped_nodes_.end(), node);
     if (it != stopped_nodes_.end()) {
       stopped_nodes_.erase(it);
     }
     return status;
   }
   return false;
 }

 bool Integration::stop_node(unsigned int node, bool is_kill /*= false*/) {
   if (ccm_->is_node_up(node, true)) {
     bool status = ccm_->stop_node(node, is_kill);
     if (status) {
       stopped_nodes_.push_back(node);
     }
     return status;
   }
   return false;
 }

 bool Integration::pause_node(unsigned int node) {
   std::vector<unsigned>::iterator it = std::find(paused_nodes_.begin(), paused_nodes_.end(), node);
   if (it == paused_nodes_.end() && ccm_->is_node_up(node, true)) {
     ccm_->pause_node(node);
     paused_nodes_.push_back(node);
     return true;
   }
   return false;
 }

 bool Integration::resume_node(unsigned int node) {
   std::vector<unsigned>::iterator it = std::find(paused_nodes_.begin(), paused_nodes_.end(), node);
   if (it != paused_nodes_.end()) {
     ccm_->resume_node(node);
     paused_nodes_.erase(it);
     return true;
   }
   return false;
 }

 std::string Integration::generate_contact_points(const std::string& ip_prefix,
                                                  size_t number_of_nodes) {
   // Iterate over the total number of nodes to create the contact list
   std::vector<std::string> contact_points;
   for (size_t i = 1; i <= number_of_nodes; ++i) {
     std::stringstream contact_point;
     contact_point << ip_prefix << i;
     contact_points.push_back(contact_point.str());
   }
   return implode(contact_points, ',');
 }

 std::string Integration::format_string(const char* format, ...) const {
   // Create a buffer for the formatting of the string
   char buffer[FORMAT_BUFFER_SIZE] = { '\0' };

   // Parse the arguments into the buffer
   va_list args;
   va_start(args, format);
   vsnprintf(buffer, FORMAT_BUFFER_SIZE, format, args);
   va_end(args);

   // Return the formatted string
   return buffer;
 }

 void Integration::maybe_shrink_name(std::string& name) {
   if (name.size() > ENTITY_MAXIMUM_LENGTH) {
     // Update the name with a UUID (first portions of v4 UUID)
     std::vector<std::string> uuid_octets = explode(uuid_generator_.generate_timeuuid().str(), '-');
     std::string id = uuid_octets[0] + uuid_octets[3];
     name = name.substr(0, ENTITY_MAXIMUM_LENGTH - id.size()) + id;
   }
 }

 bool Integration::wait_for_logger(size_t expected_count) {
   start_timer();
   while (elapsed_time() < LOGGER_MAXIMUM_WAIT_TIME_MS && logger_.count() < expected_count) {
     msleep(LOGGER_WAIT_FOR_NAP_MS);
   }
   return logger_.count() >= expected_count;
 }

 int64_t Integration::murmur3_hash(const std::string& value) {
   return MurmurHash3_x64_128(value.data(), value.size(), 0);
 }

 uint64_t Integration::time_since_epoch_in_ms() { return get_time_since_epoch_ms(); }

 uint64_t Integration::time_since_epoch_us() { return get_time_since_epoch_us(); }

 std::string Integration::driver_name() { return datastax::internal::driver_name(); }

 std::string Integration::driver_version() { return datastax::internal::driver_version(); }
	/*
	Licensed to the Apache Software Foundation (ASF) under one
	or more contributor license agreements. See the NOTICE file
	distributed with this work for additional information
	regarding copyright ownership. The ASF licenses this file
	to you 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 "integration.hpp"
	#include "options.hpp"

	#include "driver_info.hpp"
	#include "get_time.hpp"
	#include "murmur3.hpp"

	#include <algorithm>
	#include <cstdarg>
	#include <iostream>
	#include <sys/stat.h>

	#define FORMAT_BUFFER_SIZE 10240
	#define ENTITY_MAXIMUM_LENGTH 48
	#define SIMPLE_KEYSPACE_FORMAT "CREATE KEYSPACE IF NOT EXISTS %s WITH replication = %s"
	#define REPLICATION_STRATEGY "{ 'class': %s }"
	#define SELECT_SERVER_VERSION "SELECT release_version FROM system.local"
	#define LOGGER_MAXIMUM_WAIT_TIME_MS 10000u
	#define LOGGER_WAIT_FOR_NAP_MS 100

	// Initialize static variables
	bool Integration::skipped_message_displayed_ = false;

	using namespace datastax::internal;
	using namespace ::testing;

	Integration::Integration()
	: ccm_(NULL)
	, session_()
	, keyspace_name_("")
	, table_name_("")
	, system_schema_keyspaces_("system.schema_keyspaces")
	, uuid_generator_()
	, server_version_(Options::server_version())
	, number_dc1_nodes_(1)
	, number_dc2_nodes_(0)
	, replication_factor_(0)
	, replication_strategy_("")
	, contact_points_("")
	, is_password_authenticator_(false)
	, is_client_authentication_(false)
	, is_ssl_(false)
	, is_with_vnodes_(false)
	, is_randomized_contact_points_(false)
	, is_schema_metadata_(false)
	, is_ccm_requested_(true)
	, is_ccm_start_requested_(true)
	, is_ccm_start_node_individually_(false)
	, is_session_requested_(true)
	, is_keyspace_change_requested_(true)
	, is_test_chaotic_(false)
	, protocol_version_(CASS_PROTOCOL_VERSION_V4)
	, create_keyspace_query_("")
	, start_time_(0ull) {
	// Determine if the schema keyspaces table should be updated
	// TODO: Make cass_version (and dse_version) available for all tests
	CCM::CassVersion cass_version = server_version_;
	if (!Options::is_cassandra()) {
	cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version();
	}
	if (cass_version >= "3.0.0") {
	system_schema_keyspaces_ = "system_schema.keyspaces";
	}

	// Get the name of the test and the case/suite it belongs to
	const TestInfo* test_information = UnitTest::GetInstance()->current_test_info();
	test_name_ = test_information->name();

	// Determine if this is a typed test (e.g. ends in a number)
	const char* type_param = test_information->type_param();
	if (type_param) {
	std::vector<std::string> tokens = explode(test_information->test_case_name(), '/');
	for (std::vector<std::string>::const_iterator iterator = tokens.begin();
	iterator < tokens.end(); ++iterator) {
	std::string token = *iterator;

	// Determine if we are looking at the last token
	if ((iterator + 1) == tokens.end()) {
	size_t number = 0;
	std::stringstream tokenStream(token);
	if (!(tokenStream >> number).fail()) {
	std::vector<std::string> type_param_tokens = explode(type_param, ':');
	size_t size = type_param_tokens.size();
	test_case_name_ += test::Utils::replace_all(type_param_tokens[size - 1], ">", "");
	}
	} else {
	test_case_name_ += token + "_";
	}
	}
	} else {
	test_case_name_ = test_information->test_case_name();
	}

	// Determine if file logging should be enabled for the integration tests
	if (Options::log_tests()) {
	logger_.initialize(test_case_name_, test_name_);
	}
	}

	Integration::~Integration() {
	try {
	session_.close(false);
	} catch (...) {
	}

	// Reset the skipped message displayed state
	skipped_message_displayed_ = false;
	}

	void Integration::SetUp() {
	// Initialize the DSE workload (iff not set)
	if (dse_workload_.empty()) {
	dse_workload_.push_back(CCM::DSE_WORKLOAD_CASSANDRA);
	}
	// Generate the default settings for most tests (handles overridden values)
	keyspace_name_ = default_keyspace();
	table_name_ = default_table();

	if (replication_factor_ == 0) {
	replication_factor_ = default_replication_factor();
	}
	replication_strategy_ = default_replication_strategy();

	// Generate the keyspace query
	create_keyspace_query_ =
	format_string(SIMPLE_KEYSPACE_FORMAT, keyspace_name_.c_str(), replication_strategy_.c_str());

	// Create the data center nodes vector
	std::vector<unsigned short> data_center_nodes;
	data_center_nodes.push_back(number_dc1_nodes_);
	data_center_nodes.push_back(number_dc2_nodes_);

	if (is_ccm_requested_) {
	try {
	// Create and start the CCM cluster (if not already created)
	ccm_ = new CCM::Bridge(
	server_version_, Options::use_git(), Options::branch_tag(), Options::use_install_dir(),
	Options::install_dir(), Options::server_type(), dse_workload_, Options::cluster_prefix(),
	Options::dse_credentials(), Options::dse_username(), Options::dse_password(),
	Options::deployment_type(), Options::authentication_type(), Options::host(),
	Options::port(), Options::username(), Options::password(), Options::public_key(),
	Options::private_key(), Options::is_verbose_ccm());
	if (ccm_->create_cluster(data_center_nodes, is_with_vnodes_, is_password_authenticator_,
	is_ssl_, is_client_authentication_)) {
	if (is_ccm_start_requested_) {
	if (is_ccm_start_node_individually_) {
	for (unsigned short node = 1; node <= (number_dc1_nodes_ + number_dc2_nodes_); ++node) {
	if (is_password_authenticator_) {
	ccm_->start_node(node, "-Dcassandra.superuser_setup_delay_ms=0");
	} else {
	ccm_->start_node(node);
	}
	}
	} else {
	if (is_password_authenticator_) {
	ccm_->start_cluster("-Dcassandra.superuser_setup_delay_ms=0");
	} else {
	ccm_->start_cluster();
	}
	}
	}
	}

	// Generate the default contact points
	contact_points_ =
	generate_contact_points(ccm_->get_ip_prefix(), number_dc1_nodes_ + number_dc2_nodes_);

	// Determine if the session connection should be established
	if (is_session_requested_ && is_ccm_start_requested_) {
	connect();
	}
	} catch (CCM::BridgeException be) {
	// Issue creating the CCM bridge instance (force failure)
	FAIL() << be.what();
	}
	}
	}

	void Integration::TearDown() {
	// Restart and resume all stopped and paused nodes
	if (!is_test_chaotic_) { // No need to restart as cluster will be destroyed
	for (std::vector<unsigned int>::iterator it = stopped_nodes_.begin(),
	end = stopped_nodes_.end();
	it != end; ++it) {
	TEST_LOG("Restarting Node Stopped in " << test_name_ << ": " << *it);
	ccm_->start_node(*it);
	}
	for (std::vector<unsigned int>::iterator it = paused_nodes_.begin(), end = paused_nodes_.end();
	it != end; ++it) {
	TEST_LOG("Resuming Node Paused in " << test_name_ << ": " << *it);
	ccm_->resume_node(*it);
	}
	}
	stopped_nodes_.clear();
	paused_nodes_.clear();

	// Drop keyspace for integration test (may or may have not been created)
	if (!is_test_chaotic_) { // No need to drop keyspace as cluster will be destroyed
	std::stringstream use_keyspace_query;
	use_keyspace_query << "DROP KEYSPACE " << keyspace_name_;
	try {
	session_.execute(use_keyspace_query.str(), CASS_CONSISTENCY_ANY, false, false);
	} catch (...) {
	}
	}

	// Determine if the CCM cluster should be destroyed
	if (is_test_chaotic_) {
	// Destroy the current cluster and reset the chaos flag for the next test
	if (!Options::keep_clusters()) {
	ccm_->remove_cluster();
	} else {
	TEST_LOG_ERROR("Chaotic test detected and keep clusters requested. Cluster will not be "
	"removed and may result in later failures.");
	}
	is_test_chaotic_ = false;
	}
	}

	std::string Integration::default_keyspace() {
	if (!keyspace_name_.empty()) {
	return keyspace_name_;
	}

	// Clean up the initial keyspace name (remove category information)
	keyspace_name_ = to_lower(test_case_name_) + "_" + to_lower(test_name_);
	keyspace_name_ = replace_all(keyspace_name_, "tests",
	""); // TODO: Rename integration tests (remove 's' or add 's')
	keyspace_name_ = replace_all(keyspace_name_, "test", "");
	keyspace_name_ = replace_all(keyspace_name_, "integration", "");
	for (TestCategory::iterator iterator = TestCategory::begin(); iterator != TestCategory::end();
	++iterator) {
	keyspace_name_ = replace_all(keyspace_name_, "_" + to_lower(iterator->name()) + "_", "");
	}

	// Generate the keyspace name
	maybe_shrink_name(keyspace_name_);
	return keyspace_name_;
	}

	unsigned short Integration::default_replication_factor() {
	// Calculate and return the default replication factor
	return (number_dc1_nodes_ % 2 == 0) ? number_dc1_nodes_ / 2 : (number_dc1_nodes_ + 1) / 2;
	}

	std::string Integration::default_replication_strategy() {
	// Determine the replication strategy
	std::stringstream replication_strategy_s;
	if (number_dc2_nodes_ > 0) {
	replication_strategy_s << "'NetworkTopologyStrategy', 'dc1': " << number_dc1_nodes_ << ", "
	<< "'dc2': " << number_dc2_nodes_;
	} else {
	replication_strategy_s << "'SimpleStrategy', 'replication_factor': ";

	// Ensure the replication factor has not been overridden or already set
	if (replication_factor_ == 0) {
	replication_factor_ = default_replication_factor();
	}
	replication_strategy_s << replication_factor_;
	}

	// Return the default replication strategy
	std::string replication_strategy = replication_strategy_s.str();
	return format_string(REPLICATION_STRATEGY, replication_strategy.c_str());
	}

	std::string Integration::default_select_all() {
	std::stringstream cql;
	cql << "SELECT * FROM " << default_keyspace() << "." << default_table();
	return cql.str();
	}

	int64_t Integration::default_select_count() {
	Result result = session_.execute(format_string(SELECT_COUNT_FORMAT, table_name_.c_str()));
	EXPECT_EQ(CASS_OK, result.error_code()) << "Unable to get Row Count: " << result.error_message();
	return result.first_row().next().as<BigInteger>().value();
	}

	std::string Integration::default_table() {
	if (!table_name_.empty()) {
	return table_name_;
	}

	table_name_ = to_lower(test_name_);
	table_name_ = replace_all(table_name_, "integration_", "");
	maybe_shrink_name(table_name_);
	return table_name_;
	}

	void Integration::drop_table(const std::string& table_name) {
	// Drop table from the current keyspace
	std::stringstream drop_table_query;
	drop_table_query << "DROP TABLE " << table_name;
	session_.execute(drop_table_query.str(), CASS_CONSISTENCY_ANY, false, false);
	}

	void Integration::drop_type(const std::string& type_name) {
	// Drop type from the current keyspace
	std::stringstream drop_type_query;
	drop_type_query << "DROP TYPE " << type_name;
	session_.execute(drop_type_query.str(), CASS_CONSISTENCY_ANY, false, false);
	}

	bool Integration::use_keyspace(const std::string& keyspace_name) {
	std::stringstream use_keyspace_query;
	use_keyspace_query << "USE " << keyspace_name;
	session_.execute(use_keyspace_query.str());
	if (this->HasFailure()) {
	return false;
	}
	return true;
	}

	void Integration::connect(Cluster cluster) {
	// Establish the session connection
	cluster_ = cluster;
	session_ = cluster.connect();
	CHECK_FAILURE;

	// Update the server version if branch_tag was specified
	if (Options::use_git() && !Options::branch_tag().empty()) {
	if (Options::is_ddac()) {
	FAIL() << "Unable to build DDAC from Branch/Tag";
	return;
	}
	if (Options::is_dse()) {
	server_version_ = ccm_->get_dse_version();
	} else {
	server_version_ = ccm_->get_cassandra_version();
	}
	TEST_LOG("Branch/Tag Option was Used: Retrieved server version is "
	<< server_version_.to_string());
	}

	// Create the keyspace for the integration test
	session_.execute(create_keyspace_query_);
	CHECK_FAILURE;

	// Update the session to use the new keyspace by default
	if (is_keyspace_change_requested_) {
	use_keyspace(keyspace_name_);
	}
	}

	void Integration::connect() {
	// Create the cluster configuration and establish the session connection
	cluster_ = default_cluster();
	connect(cluster_);
	}

	test::driver::Cluster Integration::default_cluster(bool is_with_default_contact_points /= true/) {
	// Create the default cluster object
	Cluster cluster = Cluster::build()
	.with_randomized_contact_points(is_randomized_contact_points_)
	.with_schema_metadata(is_schema_metadata_);
	if (is_with_default_contact_points) {
	cluster.with_contact_points(contact_points_);
	}
	cluster.with_protocol_version(protocol_version_);

	// Assign the execution profiles to the cluster object (if available)
	for (ExecutionProfile::Map::iterator it = profiles_.begin(); it != profiles_.end(); ++it) {
	cluster.with_execution_profile(it->first, it->second);
	}
	return cluster;
	}

	void Integration::enable_cluster_tracing(bool enable /= true/) {
	std::vector<std::string> active_nodes = ccm_->cluster_ip_addresses();
	for (std::vector<std::string>::iterator iterator = active_nodes.begin();
	iterator != active_nodes.end(); ++iterator) {
	// Get the node number from the IP address
	std::string node_ip_address = *iterator;
	std::stringstream node_value;
	node_value << node_ip_address[node_ip_address.length() - 1];

	// Enable tracing on the node
	unsigned int node;
	node_value >> node;
	ccm_->enable_node_trace(node);
	}
	}

	bool Integration::decommission_node(unsigned int node, bool is_force /= false /) {
	// Decommission the requested node
	bool status = ccm_->decommission_node(node, is_force);
	if (status) {
	// Indicate the test is chaotic
	is_test_chaotic_ = true;
	}
	return status;
	}

	bool Integration::force_decommission_node(unsigned int node) {
	return decommission_node(node, true);
	}

	bool Integration::start_node(unsigned int node) {
	// Start the requested node and ensure paused nodes are ignored
	std::vector<unsigned>::iterator it = std::find(paused_nodes_.begin(), paused_nodes_.end(), node);
	if (it == paused_nodes_.end() && ccm_->is_node_down(node, true)) {
	bool status = ccm_->start_node(node);
	it = std::find(stopped_nodes_.begin(), stopped_nodes_.end(), node);
	if (it != stopped_nodes_.end()) {
	stopped_nodes_.erase(it);
	}
	return status;
	}
	return false;
	}

	bool Integration::stop_node(unsigned int node, bool is_kill /= false/) {
	if (ccm_->is_node_up(node, true)) {
	bool status = ccm_->stop_node(node, is_kill);
	if (status) {
	stopped_nodes_.push_back(node);
	}
	return status;
	}
	return false;
	}

	bool Integration::pause_node(unsigned int node) {
	std::vector<unsigned>::iterator it = std::find(paused_nodes_.begin(), paused_nodes_.end(), node);
	if (it == paused_nodes_.end() && ccm_->is_node_up(node, true)) {
	ccm_->pause_node(node);
	paused_nodes_.push_back(node);
	return true;
	}
	return false;
	}

	bool Integration::resume_node(unsigned int node) {
	std::vector<unsigned>::iterator it = std::find(paused_nodes_.begin(), paused_nodes_.end(), node);
	if (it != paused_nodes_.end()) {
	ccm_->resume_node(node);
	paused_nodes_.erase(it);
	return true;
	}
	return false;
	}

	std::string Integration::generate_contact_points(const std::string& ip_prefix,
	size_t number_of_nodes) {
	// Iterate over the total number of nodes to create the contact list
	std::vector<std::string> contact_points;
	for (size_t i = 1; i <= number_of_nodes; ++i) {
	std::stringstream contact_point;
	contact_point << ip_prefix << i;
	contact_points.push_back(contact_point.str());
	}
	return implode(contact_points, ',');
	}

	std::string Integration::format_string(const char* format, ...) const {
	// Create a buffer for the formatting of the string
	char buffer[FORMAT_BUFFER_SIZE] = { '\0' };

	// Parse the arguments into the buffer
	va_list args;
	va_start(args, format);
	vsnprintf(buffer, FORMAT_BUFFER_SIZE, format, args);
	va_end(args);

	// Return the formatted string
	return buffer;
	}

	void Integration::maybe_shrink_name(std::string& name) {
	if (name.size() > ENTITY_MAXIMUM_LENGTH) {
	// Update the name with a UUID (first portions of v4 UUID)
	std::vector<std::string> uuid_octets = explode(uuid_generator_.generate_timeuuid().str(), '-');
	std::string id = uuid_octets[0] + uuid_octets[3];
	name = name.substr(0, ENTITY_MAXIMUM_LENGTH - id.size()) + id;
	}
	}

	bool Integration::wait_for_logger(size_t expected_count) {
	start_timer();
	while (elapsed_time() < LOGGER_MAXIMUM_WAIT_TIME_MS && logger_.count() < expected_count) {
	msleep(LOGGER_WAIT_FOR_NAP_MS);
	}
	return logger_.count() >= expected_count;
	}

	int64_t Integration::murmur3_hash(const std::string& value) {
	return MurmurHash3_x64_128(value.data(), value.size(), 0);
	}

	uint64_t Integration::time_since_epoch_in_ms() { return get_time_since_epoch_ms(); }

	uint64_t Integration::time_since_epoch_us() { return get_time_since_epoch_us(); }

	std::string Integration::driver_name() { return datastax::internal::driver_name(); }

	std::string Integration::driver_version() { return datastax::internal::driver_version(); }