test: Fixing and stabilizing integration tests

 - DSE server version checking
   - Value type server version checking
   - Authentication log message checking
   - EACH_QUORUM
   - System schema keyspaces
 - Ensure reconnect test doesn't restart node after chaotic test
 - Stabilized external host listener tests
 - Speculative execution tests required larger function timeout for UDF
 - Proxy authentication
   - Invalid YAML string for configuration
   - Updated server error message validation for unauthorized
 - Added missing C* versions to CCM bridge
 - JVM argument usage when starting a node
 - Ensure restarting nodes are performed once
 - Ensuring start/stop node operations do not cause test start failures

Author: Michael Fero

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

@@ -17,6 +17,7 @@
 #include "integration.hpp"
 #include "options.hpp"
 
+#include <algorithm>
 #include <cstdarg>
 #include <iostream>
 #include <sys/stat.h>
@@ -60,7 +61,12 @@ Integration::Integration()
   , create_keyspace_query_("")
   , start_time_(0ull) {
   // Determine if the schema keyspaces table should be updated
-  if (server_version_ >= "3.0.0") {
+  // TODO: Make cass_version (and dse_version) available for all tests
+  CCM::CassVersion cass_version = server_version_;
+  if (Options::is_dse()) {
+    cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version();
+  }
+  if (cass_version >= "3.0.0") {
     system_schema_keyspaces_ = "system_schema.keyspaces";
   }
 
@@ -391,13 +397,31 @@ bool Integration::force_decommission_node(unsigned int node) {
   return decommission_node(node, true);
 }
 
-bool Integration::stop_node(unsigned int node) {
+bool Integration::start_node(unsigned int node) {
   // Stop the requested node
-  bool status = ccm_->stop_node(node);
-  if (status) {
-    stopped_nodes_.push_back(node);
+  if (ccm_->is_node_down(node, true)) {
+    bool status = ccm_->start_node(node);
+    std::vector<unsigned int>::iterator it = std::find(stopped_nodes_.begin(),
+                                                       stopped_nodes_.end(),
+                                                       node);
+    if (it != stopped_nodes_.end()) {
+      stopped_nodes_.erase(it);
+    }
+    return status;
   }
-  return status;
+  return false;
+}
+
+bool Integration::stop_node(unsigned int node, bool is_kill /*= false*/) {
+  // Stop the requested node
+  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;
 }
 
 std::string Integration::generate_contact_points(const std::string& ip_prefix,

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

@@ -99,8 +99,12 @@
   }
 
 #define CHECK_VALUE_TYPE_VERSION(type) \
-  if (this->server_version_ < type::supported_server_version()) { \
-    SKIP_TEST_VERSION(this->server_version_.to_string(), \
+  CCM::CassVersion cass_version = this->server_version_; \
+  if (Options::is_dse()) { \
+    cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version(); \
+  } \
+  if (cass_version < type::supported_server_version()) { \
+    SKIP_TEST_VERSION(cass_version.to_string(), \
                       type::supported_server_version()) \
   }
 
@@ -423,14 +427,26 @@ class Integration : public testing::Test {
    */
   virtual bool force_decommission_node(unsigned int node);
 
+  /**
+   * Start a node that was previously stopped to ensure that it is not restarted
+   * after test is completed
+   *
+   * @param node Node that should be started
+   * @return True if node was started; false otherwise (the node is invalid or
+   *         was already started)
+   */
+  virtual bool start_node(unsigned int node);
+
   /**
    * Stop a node that should be restarted after test is completed
    *
    * @param node Node that should be stopped
+   * @param is_kill True if forced termination requested; false otherwise
+   *                (default: false)
    * @return True if node was stopped; false otherwise (the node is invalid or
    *         was already stopped)
    */
-  virtual bool stop_node(unsigned int node);
+  virtual bool stop_node(unsigned int node, bool is_kill = false);
 
   /**
    * Generate the contact points for the cluster

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

@@ -203,7 +203,11 @@ CASSANDRA_INTEGRATION_TEST_F(AuthenticationTests, BadCredentials) {
   CHECK_FAILURE;
 
   // Add the proper logging criteria (based on server version)
-  if (server_version_ >= "3.10") {
+  CCM::CassVersion cass_version = this->server_version_;
+  if (Options::is_dse()) {
+    cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version();
+  }
+  if (cass_version >= "3.10") {
     logger_.add_critera("Provided username invalid and/or password are incorrect");
   } else {
     logger_.add_critera("Username and/or password are incorrect");
@@ -241,7 +245,11 @@ CASSANDRA_INTEGRATION_TEST_F(AuthenticationTests, AuthenticatorSetErrorNull) {
   CHECK_FAILURE;
 
   // Add the proper logging criteria (based on server version)
-  if (server_version_ >= "3.10") {
+  CCM::CassVersion cass_version = this->server_version_;
+  if (Options::is_dse()) {
+    cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version();
+  }
+  if (cass_version >= "3.10") {
     logger_.add_critera("Provided username invalid and/or password are incorrect");
   } else {
     logger_.add_critera("Username and/or password are incorrect");

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

@@ -196,6 +196,7 @@ CASSANDRA_INTEGRATION_TEST_F(ConsistencyTwoNodeClusterTests, SimpleLocalQuorum)
  */
 CASSANDRA_INTEGRATION_TEST_F(ConsistencyTwoNodeClusterTests, SimpleEachQuorum) {
   CHECK_FAILURE;
+  CHECK_VERSION(3.0.0);
 
   // Assign the consistency level for the test
   insert_.set_consistency(CASS_CONSISTENCY_EACH_QUORUM);

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

@@ -336,7 +336,7 @@ CASSANDRA_INTEGRATION_TEST_F(ControlConnectionTests, TopologyChange) {
 
   // Bootstrap a second node and ensure all hosts are actively used
   logger_.add_critera("New node " + (ccm_->get_ip_prefix() + "2") + " added");
-  EXPECT_EQ(2, ccm_->bootstrap_node()); // Triggers a `NEW_NODE` event
+  EXPECT_EQ(2u, ccm_->bootstrap_node()); // Triggers a `NEW_NODE` event
   EXPECT_TRUE(wait_for_logger(1));
   std::set<unsigned short> expected_nodes;
   expected_nodes.insert(1);
@@ -395,7 +395,7 @@ CASSANDRA_INTEGRATION_TEST_F(ControlConnectionTwoNodeClusterTests,
 
   // Restart the second node and ensure all hosts are actively used
   reset_logger_criteria("Node ", logger_nodes, " is up");
-  ccm_->start_node(2); // Triggers a `UP` event
+  start_node(2); // Triggers a `UP` event
   ASSERT_TRUE(wait_for_logger(logger_nodes.size()));
   expected_nodes.insert(2);
   check_hosts(session, expected_nodes);

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

@@ -98,8 +98,8 @@ CASSANDRA_INTEGRATION_TEST_F(DcAwarePolicyTest, UsedHostsRemoteDc) {
   }
 
   // Stop the whole local DC
-  ccm_->stop_node(1, true);
-  ccm_->stop_node(2, true);
+  stop_node(1, true);
+  stop_node(2, true);
 
   { // Run queries using the remote DC
     std::vector<std::string> attempted_hosts = validate();

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

@@ -15,35 +15,83 @@
 */
 
 #include "integration.hpp"
+
 #include "cassandra.h"
+#include "scoped_lock.hpp"
+
+#include <queue>
+
+#define EVENT_MAXIMUM_WAIT_TIME_MS 5000
+#define EVENT_WAIT_FOR_NAP_MS 100
 
 class SessionTest : public Integration {
 public:
   typedef std::pair<CassHostListenerEvent, std::string> Event;
-  typedef std::vector<Event> Events;
+  typedef std::queue<Event> Events;
+
+  SessionTest() {
+    uv_mutex_init(&mutex_);
+  }
+  ~SessionTest() {
+    uv_mutex_destroy(&mutex_);
+  }
 
   void SetUp() {
     is_session_requested_ = false;
     Integration::SetUp();
   }
 
-  size_t event_count() { return events_.size(); }
-  const Events& events() const { return events_; }
+  void check_event(CassHostListenerEvent expected_event, short expected_node) {
+    cass::ScopedMutex l(&mutex_);
+    std::stringstream expected_address;
+    expected_address << ccm_->get_ip_prefix() << expected_node;
+    Event event = events_.front();
+    EXPECT_EQ(expected_event, event.first);
+    EXPECT_EQ(expected_address.str().c_str(), event.second);
+    events_.pop();
+  }
+  bool wait_for_event(size_t expected_count) {
+    start_timer();
+    while (elapsed_time() < EVENT_MAXIMUM_WAIT_TIME_MS
+           && event_count() < expected_count) {
+      msleep(EVENT_WAIT_FOR_NAP_MS);
+    }
+    return event_count() >= expected_count;
+  }
 
 protected:
+  size_t event_count() {
+    cass::ScopedMutex l(&mutex_);
+    return events_.size();
+  }
+  void add_event(CassHostListenerEvent event, CassInet inet) {
+    cass::ScopedMutex l(&mutex_);
+    char address[CASS_INET_STRING_LENGTH];
+
+    cass_inet_string(inet, address);
+    if (event == CASS_HOST_LISTENER_EVENT_ADD) {
+      TEST_LOG("Host " << address << " has been ADDED");
+    } else if (event == CASS_HOST_LISTENER_EVENT_REMOVE) {
+      TEST_LOG("Host " << address << " has been REMOVED");
+    } else if (event == CASS_HOST_LISTENER_EVENT_UP) {
+      TEST_LOG("Host " << address << " is UP");
+    } else if (event == CASS_HOST_LISTENER_EVENT_DOWN) {
+      TEST_LOG("Host " << address << " is DOWN");
+    } else {
+      TEST_LOG_ERROR("Invalid event [" << event << "] for " << address);
+    }
+
+    events_.push(Event(event, address));
+  }
   static void on_host_listener(CassHostListenerEvent event,
                                CassInet inet,
                                void* data) {
     SessionTest* instance = static_cast<SessionTest*>(data);
-
-    char address[CASS_INET_STRING_LENGTH];
-    uv_inet_ntop(inet.address_length == CASS_INET_V4_LENGTH ? AF_INET : AF_INET6,
-                 inet.address,
-                 address, CASS_INET_STRING_LENGTH);
-    instance->events_.push_back(Event(event, address));
+    instance->add_event(event, inet);
   }
 
 private:
+  uv_mutex_t mutex_;
   Events events_;
 };
 
@@ -78,26 +126,42 @@ CASSANDRA_INTEGRATION_TEST_F(SessionTest, ExternalHostListener) {
     .with_host_listener_callback(on_host_listener, this);
   Session session = cluster.connect();
 
-  ASSERT_EQ(0u, event_count());
+  // Initial node 1 events (add and up)
+  ASSERT_TRUE(wait_for_event(2u));
+  check_event(CASS_HOST_LISTENER_EVENT_ADD, 1);
+  check_event(CASS_HOST_LISTENER_EVENT_UP, 1);
+
+  // Bootstrap node 2 (add and up events)
+  EXPECT_EQ(2u, ccm_->bootstrap_node());
+  ASSERT_TRUE(wait_for_event(2u));
+  check_event(CASS_HOST_LISTENER_EVENT_ADD, 2);
+  check_event(CASS_HOST_LISTENER_EVENT_UP, 2);
 
-  EXPECT_EQ(2, ccm_->bootstrap_node());
+  // Stop node 1 (down event)
   stop_node(1);
+  ASSERT_TRUE(wait_for_event(1u));
+  check_event(CASS_HOST_LISTENER_EVENT_DOWN, 1);
+
+  // Restart node 1 (up event)
   ccm_->start_node(1);
+  CCM::CassVersion cass_version = this->server_version_;
+  if (Options::is_dse()) {
+    cass_version = static_cast<CCM::DseVersion>(cass_version).get_cass_version();
+  }
+  if (cass_version >= "2.2") {
+    ASSERT_TRUE(wait_for_event(1u));
+  } else {
+    ASSERT_TRUE(wait_for_event(3u)); // C* <= 2.1 fires remove and add event on restart
+    check_event(CASS_HOST_LISTENER_EVENT_REMOVE, 1);
+    check_event(CASS_HOST_LISTENER_EVENT_ADD, 1);
+  }
+  check_event(CASS_HOST_LISTENER_EVENT_UP, 1);
+
+  // Decomission node 1 (down and remove events)
   force_decommission_node(1);
+  ASSERT_TRUE(wait_for_event(1u));
+  check_event(CASS_HOST_LISTENER_EVENT_DOWN, 1);
+  check_event(CASS_HOST_LISTENER_EVENT_REMOVE, 1);
 
   session.close();
-
-  ASSERT_EQ(6u, event_count());
-  EXPECT_EQ(CASS_HOST_LISTENER_EVENT_ADD, events()[0].first);
-  EXPECT_EQ(ccm_->get_ip_prefix() + "2", events()[0].second);
-  EXPECT_EQ(CASS_HOST_LISTENER_EVENT_UP, events()[1].first);
-  EXPECT_EQ(ccm_->get_ip_prefix() + "2", events()[1].second);
-  EXPECT_EQ(CASS_HOST_LISTENER_EVENT_DOWN, events()[2].first);
-  EXPECT_EQ(ccm_->get_ip_prefix() + "1", events()[2].second);
-  EXPECT_EQ(CASS_HOST_LISTENER_EVENT_UP, events()[3].first);
-  EXPECT_EQ(ccm_->get_ip_prefix() + "1", events()[3].second);
-  EXPECT_EQ(CASS_HOST_LISTENER_EVENT_DOWN, events()[4].first);
-  EXPECT_EQ(ccm_->get_ip_prefix() + "1", events()[4].second);
-  EXPECT_EQ(CASS_HOST_LISTENER_EVENT_REMOVE, events()[5].first);
-  EXPECT_EQ(ccm_->get_ip_prefix() + "1", events()[5].second);
 }

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

@@ -35,7 +35,7 @@ CASSANDRA_INTEGRATION_TEST_F(TracingTests, Simple) {
     Statement statement("SELECT * FROM system_traces.sessions WHERE session_id = ?", 1);
     statement.bind(0, tracing_id);
     Result result = session_.execute(statement);
-    ASSERT_GT(result.row_count(),  0);
+    ASSERT_GT(result.row_count(),  0u);
     Uuid session_id = result.first_row().column_by_name<Uuid>("session_id");
     ASSERT_FALSE(session_id.is_null());
     EXPECT_EQ(tracing_id, session_id);