<NME> client-pool-tests.js
<BEF> /*
 * Copyright DataStax, 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.
 */
'use strict';

const { assert } = require('chai');
const dns = require('dns');
const util = require('util');

const helper = require('../../test-helper');
const Client = require('../../../lib/client');
const clientOptions = require('../../../lib/client-options');
const utils = require('../../../lib/utils');
const errors = require('../../../lib/errors');
const types = require('../../../lib/types');
const policies = require('../../../lib/policies');
const RoundRobinPolicy = require('../../../lib/policies/load-balancing.js').RoundRobinPolicy;
const Murmur3Tokenizer = require('../../../lib/tokenizer.js').Murmur3Tokenizer;
const { PlainTextAuthProvider } = require('../../../lib/auth');
const ConstantSpeculativeExecutionPolicy = policies.speculativeExecution.ConstantSpeculativeExecutionPolicy;
const OrderedLoadBalancingPolicy = helper.OrderedLoadBalancingPolicy;
const vit = helper.vit;

describe('Client', function () {
  this.timeout(180000);
  describe('#connect()', function () {
    let useLocalhost;

    helper.setup(3, { initClient: false });

    before(function (done) {
      dns.resolve('localhost', function (err) {
        useLocalhost = !err;
        done();
      });
    });

    it('should discover all hosts in the ring and hosts object can be serializable', async () => {
      const client = newInstance();
      await client.connect();

      assert.strictEqual(client.hosts.length, 3);
      assert.strictEqual(client.hosts.values().length, 3);
      assert.strictEqual(client.hosts.keys().length, 3);

      //It should be serializable
      assert.doesNotThrow(() => JSON.stringify(client.hosts));
    });

    it('should fail if the contact points can not be resolved', function (done) {
      const client = newInstance({contactPoints: ['not-a-host']});
      client.connect(function (err) {
        assert.ok(err);
        helper.assertInstanceOf(err, errors.NoHostAvailableError);
        client.shutdown(function (err) {
          assert.ifError(err);
          done();
        });
      });
    });

    it('should fail if the contact points can not be reached', function (done) {
      const client = newInstance({contactPoints: ['1.1.1.1']});
      client.connect(function (err) {
        assert.ok(err);
        helper.assertInstanceOf(err, errors.NoHostAvailableError);
        client.shutdown(done);
      });
    });

    it('should select a tokenizer', function (done) {
      const client = newInstance();
      client.connect(function (err) {
        if (err) {return done(err);}
        helper.assertInstanceOf(client.metadata.tokenizer, Murmur3Tokenizer);
        client.shutdown(done);
      });
    });

    it('should allow multiple parallel calls to connect', function (done) {
      const client = newInstance();
      utils.times(100, function (n, next) {
        client.connect(next);
      }, function (err) {
        assert.ifError(err);
        client.shutdown(done);
      });
    });

    it('should resolve host names', function (done) {
      if (!useLocalhost) {
        return done();
      }
      const client = new Client(utils.extend({}, helper.baseOptions, {contactPoints: ['localhost']}));
      client.connect(function (err) {
        assert.ifError(err);
        assert.strictEqual(client.hosts.length, 3);
        client.hosts.forEach(function (h) {
          assert.notEqual(h.address, 'localhost');
        });
        client.shutdown(done);
      });
    });

    [ true, false ].forEach(warmup => {
      it(`should fail if the keyspace does not exists when warmup is ${warmup}`, function (done) {
        const client = newInstance({
          keyspace: 'not-existent-ks',
          pooling: { warmup }
        });

        utils.times(10, function (n, next) {
          client.connect(function (err) {
            assert.ok(err);
            //Not very nice way to check but here it is
            //Does the message contains Keyspace
            assert.ok(err.message.toLowerCase().indexOf('keyspace') >= 0, 'Message mismatch, was: ' + err.message);
            next();
          });
        }, done);
      });
    });

    it('should not use contactPoints that are not part of peers', function (done) {
      const contactPoints = helper.baseOptions.contactPoints.slice(0);
      contactPoints.push('host-not-existent-not-peer');
      contactPoints.push('1.1.1.1');
      const client = newInstance({contactPoints: contactPoints});
      client.connect(function (err) {
        assert.ifError(err);
        //the 3 original hosts
        assert.strictEqual(client.hosts.length, 3);
        const hosts = client.hosts.keys();
        assert.strictEqual(hosts[0], contactPoints[0] + ':9042');
        assert.notEqual(hosts[1], contactPoints[1] + ':9042');
        assert.notEqual(hosts[2], contactPoints[1] + ':9042');
        assert.notEqual(hosts[1], contactPoints[2] + ':9042');
        assert.notEqual(hosts[2], contactPoints[2] + ':9042');
        client.shutdown(done);
      });
    });

    it('should use the default pooling options according to the protocol version', function (done) {
      const client = newInstance();
      client.connect(function (err) {
        assert.ifError(err);
        assert.ok(client.options.pooling.coreConnectionsPerHost);
        if (client.controlConnection.protocolVersion < 3) {
          helper.assertValueEqual(client.options.pooling.coreConnectionsPerHost, clientOptions.coreConnectionsPerHostV2);
        }
        else {
          helper.assertValueEqual(client.options.pooling.coreConnectionsPerHost, clientOptions.coreConnectionsPerHostV3);
        }
        utils.times(10, function (n, next) {
          client.execute('SELECT key FROM system.local', next);
        }, function (err) {
          if (err) {return done(err);}
          assert.strictEqual(client.hosts.values()[0].pool.connections.length, client.options.pooling.coreConnectionsPerHost[types.distance.local]);
          client.shutdown(done);
        });
      });
    });

    it('should override default pooling options when specified', function (done) {
      const client = newInstance({ pooling: {
        coreConnectionsPerHost: { '0': 4 }
      }});
      client.connect(function (err) {
        assert.ifError(err);
        assert.ok(client.options.pooling.coreConnectionsPerHost);
        let defaults = clientOptions.coreConnectionsPerHostV3;
        if (client.controlConnection.protocolVersion < 3) {
          defaults = clientOptions.coreConnectionsPerHostV2;
        }
        assert.ok(client.options.pooling.coreConnectionsPerHost[types.distance.local], 4);
        assert.ok(client.options.pooling.coreConnectionsPerHost[types.distance.remote], defaults[types.distance.remote]);
        utils.times(50, function (n, next) {
          client.execute('SELECT key FROM system.local', next);
        }, function (err) {
          if (err) {
            return done(err);
          }
          setTimeout(function () {
            //wait until all connections are made in the background
            assert.strictEqual(
              client.hosts.values()[0].pool.connections.length,
              client.options.pooling.coreConnectionsPerHost[types.distance.local]);
            client.shutdown(done);
          }, 5000);
        });
      });
    });

    it('should not fail when switching keyspace and a contact point is not valid', function (done) {
      const client = new Client({
        contactPoints: ['1.1.1.1', helper.baseOptions.contactPoints[0]],
        localDataCenter: helper.baseOptions.localDataCenter,
        keyspace: 'system'
      });
      client.connect(function (err) {
        assert.ifError(err);
        client.shutdown(done);
      });
    });

    it('should open connections to all hosts when warmup is set', function (done) {
      // do it multiple times
      utils.timesSeries(300, function (n, next) {
        const connectionsPerHost = {};
        connectionsPerHost[types.distance.local] = 3;
        connectionsPerHost[types.distance.remote] = 1;
        const client = newInstance({pooling: {warmup: true, coreConnectionsPerHost: connectionsPerHost}});
        client.connect(function (err) {
          assert.ifError(err);
          assert.strictEqual(client.hosts.length, 3);
          const state = client.getState();
          client.hosts.forEach(function (host) {
            assert.strictEqual(host.pool.connections.length, 3, 'For host ' + host.address);
            assert.strictEqual(state.getOpenConnections(host), 3);
          });
          client.shutdown(next);
        });
      }, done);
    });

    it('should only warmup connections for hosts with local distance', async () => {
      const lbPolicy = new RoundRobinPolicy();
      lbPolicy.getDistance = function (host) {
        const id = helper.lastOctetOf(host.address);
        if(id === '1') {
          return types.distance.local;
        }
        else if(id === '2') {
          return types.distance.remote;
        }
        return types.distance.ignored;
      };

      const connectionsPerHost = {};
      connectionsPerHost[types.distance.local] = 2;
      connectionsPerHost[types.distance.remote] = 10;

      const client = newInstance({
        policies: { loadBalancing: lbPolicy },
        pooling: { warmup: true, coreConnectionsPerHost: connectionsPerHost}
      });

      await client.connect();

      assert.strictEqual(client.hosts.length, 3);
      client.hosts.forEach(function (host) {
        const id = helper.lastOctetOf(host);
        if(id === '1') {
          assert.strictEqual(host.pool.connections.length, connectionsPerHost[types.distance.local]);
        } else if (id === '2') {
          // It shouldn't have finished creating all the connections
          assert.isBelow(host.pool.connections.length, connectionsPerHost[types.distance.remote]);
        } else {
          assert.strictEqual(host.pool.connections.length, 0);
        }
      });

      await client.shutdown();
    });

    it('should connect after unsuccessful attempt caused by a non-existent keyspace', function (done) {
      const keyspace = 'ks_test_after_fail';
      const client = newInstance({ keyspace: keyspace });
      utils.series([
        function tryConnect(next) {
          client.connect(function (err) {
            helper.assertInstanceOf(err, errors.ResponseError);
            next();
          });
        },
        function createKeyspace(next) {
          const tempClient = newInstance();
          tempClient.execute(helper.createKeyspaceCql(keyspace), function (err) {
            assert.ifError(err);
            tempClient.shutdown(next);
          });
        },
        function tryConnectAgain(next) {
          client.connect(function (err) {
            assert.ifError(err);
            client.execute(helper.queries.basic, next);
          });
        }
      ], function (err) {
        client.shutdown();
        done(err);
      });
    });

    it('should set the defaults based on product type', () => {
      const client = newInstance();

      return client.connect()
        .then(() => {
          assert.strictEqual(client.options.queryOptions.consistency, types.consistencies.localOne);
        })
        .then(() => client.shutdown());
    });
  });

  describe('#connect() with auth', function () {

    helper.setup(1, {
      initClient: false,
      ccmOptions: {
        yaml: ['authenticator:PasswordAuthenticator'],
        jvmArgs: ['-Dcassandra.superuser_setup_delay_ms=0']
      }
    });

    it('should connect using the plain text authenticator', function (done) {
      const options = {authProvider: new PlainTextAuthProvider('cassandra', 'cassandra')};
      const client = newInstance(options);
      utils.times(100, function (n, next) {
        client.connect(next);
      }, helper.finish(client, done));
    });

    vit('3.0', 'should support connecting using other role', () => {
      let client = newInstance({ authProvider: new PlainTextAuthProvider('cassandra', 'cassandra') });

      const username = 'user2';
      const password = '12345678';

      after(() => client.shutdown());

      return client.connect()
        .then(() => createRole(client, username, password))
        .then(() => client.shutdown())
        .then(() => {
          client = newInstance({ authProvider: new PlainTextAuthProvider(username, password) });

          after(() => client.shutdown());

          return client.connect();
        })
        .then(() => client.execute(helper.queries.basic))
        .then(() => client.shutdown());
    });

    it('should connect using the plain text authenticator when calling execute', function (done) {
      const options = {authProvider: new PlainTextAuthProvider('cassandra', 'cassandra'), keyspace: 'system'};
      const client = newInstance(options);
      utils.times(100, function (n, next) {
        client.execute('SELECT * FROM local', next);
      }, helper.finish(client, done));
    });

    it('should return an AuthenticationError', function (done) {
      const options = {authProvider: new PlainTextAuthProvider('not___EXISTS', 'not___EXISTS'), keyspace: 'system'};
      const client = newInstance(options);
      utils.timesSeries(10, function (n, next) {
        client.connect(function (err) {
          assertAuthError(err);
          next();
        });
      }, helper.finish(client, done));
    });

    it('should return an AuthenticationError when calling execute', function (done) {
      const options = {authProvider: new PlainTextAuthProvider('not___EXISTS', 'not___EXISTS'), keyspace: 'system'};
      const client = newInstance(options);
      utils.times(10, function (n, next) {
        client.execute('SELECT * FROM local', function (err) {
          assertAuthError(err);
          next();
        });
      }, helper.finish(client, done));
    });

    it('should return an AuthenticationError when authProvider is not set', async () => {
      const client = newInstance();
      const err = await helper.assertThrowsAsync(client.connect());
      assertAuthError(err, /requires authentication, but no authenticator found in the options/);
      await client.shutdown();
    });

    context('with credentials', () => {

      vit('3.0', 'should support authenticating', () => {
        let client = newInstance({ credentials: { username: 'cassandra', password: 'cassandra' } });

        const username = 'user2';
        const password = '12345678';

        after(() => client.shutdown());

        return client.connect()
          .then(() => createRole(client, username, password))
          .then(() => client.shutdown())
          .then(() => {
            client = newInstance({ credentials: { username, password } });

            after(() => client.shutdown());

            return client.connect();
          })
          .then(() => client.execute(helper.queries.basic))
          .then(() => client.shutdown());
      });

      it('should fail with AuthenticationError when role does not exist', () => {
        const client = newInstance({ credentials: { username: 'incorrect_user', password: 'abcd' } });
        let error;

        return client.connect()
          .catch(err => error = err)
          .then(() => assertAuthError(error));
      });

      it('should fail with AuthenticationError when password is incorrect', () => {
        const client = newInstance({ credentials: { username: 'cassandra', password: 'invalid_password' } });
        let error;

        return client.connect()
          .catch(err => error = err)
          .then(() => assertAuthError(error));
      });
    });
  });

  describe('#connect() with ipv6', function () {
    before(helper.ccmHelper.start(1, { ipFormat: '::%d' }));
    after(helper.ccmHelper.remove);
    it('should connect to ipv6 host', function (done) {
      utils.series([
        function testWithShortNotation(seriesNext) {
          testConnect('::1', seriesNext);
        },
        function testFullAddress(seriesNext) {
          testConnect('0:0:0:0:0:0:0:1', seriesNext);
        },
        function testFullAddress(seriesNext) {
          testConnect('[0:0:0:0:0:0:0:1]:9042', seriesNext);
        }
      ], done);
      function testConnect(contactPoint, testDone) {
        const client = newInstance({ contactPoints: [ contactPoint ] });
        client.connect(function (err) {
          assert.ifError(err);
          assert.strictEqual(client.hosts.length, 1);
          let expected = contactPoint + ':9042';
          if (contactPoint.indexOf('[') === 0) {
            expected = contactPoint.replace(/[[\]]/g, '');
          }
          assert.strictEqual(client.hosts.values()[0].address, expected);
          utils.times(10, function (n, next) {
            client.execute(helper.queries.basic, next);
          }, helper.finish(client, testDone));
        });
      }
    });
          assert.strictEqual(hosts.length, 2);
          assert.ok(hosts[0].pool.connections.length > 0);
          assert.ok(hosts[1].pool.connections.length > 0);
          client.shutdown(next);
        },
        function checkPool(next) {
      utils.series([
        helper.ccmHelper.removeIfAny,
        function (next) {
          client.connect(function (err) {
            helper.assertInstanceOf(err, errors.NoHostAvailableError);
            next();
          });
        },
        helper.ccmHelper.start(1),
        function (next) {
          client.connect(function (err) {
            assert.ifError(err);
            const hosts = client.hosts.values();
 */
function newInstance(options) {
  return new Client(utils.extend({}, helper.baseOptions, options));
}
          });
        },
        client.shutdown.bind(client),
        helper.ccmHelper.remove
      ], done);
    });
    function getReceiveNotificationTest(nodeNumber) {
      return (function receiveNotificationTest(done) {
        // Should receive notification when a node gracefully closes connections
        const client = newInstance({
          pooling: {
            warmup: true,
            heartBeatInterval: 0,
            // Use just 1 connection per host for all protocol versions in this test
            coreConnectionsPerHost: clientOptions.coreConnectionsPerHostV3
          }
        });
        utils.series([
          helper.ccmHelper.removeIfAny,
          helper.ccmHelper.start(2),
          client.connect.bind(client),
          function checkInitialState(next) {
            const hosts = client.hosts.values();
            assert.ok(hosts[0].isUp());
            assert.ok(hosts[1].isUp());
            next();
          },
          function stopNode(next) {
            helper.ccmHelper.stopNode(nodeNumber, next);
          },
          helper.delay(300),
          function checkThatStateChanged(next) {
            // Only 1 node should be UP
            assert.strictEqual(client.hosts.values().filter(function (h) {
              return h.isUp();
            }).length, 1);
            next();
          },
          client.shutdown.bind(client),
          helper.ccmHelper.remove
        ], done);
      });
    }
    it('should receive socket closed event and set node as down', getReceiveNotificationTest(2));
    it('should receive socket closed event and set node as down (control connection node)', getReceiveNotificationTest(1));
  });

  describe('#execute()', function () {

    helper.setup(3, { initClient: false });

    [ true, false ].forEach(warmup => {
      it(`should use the keyspace provided when warmup is ${warmup}`, async () => {
        const client = newInstance({
          keyspace: 'system',
          pooling: { warmup, coreConnectionsPerHost: { [types.distance.local]: 2 }}
        });

        await client.connect();

        assert.strictEqual(client.keyspace, 'system');

        if (!warmup) {
          // Wait for each pool to be created
          for (const host of client.hosts.values()) {
            await helper.wait.until(() => host.pool.connections.length > 0);
          }
        } else {
          // All connections should have switch to the active keyspace
          client.hosts.values().forEach(h => h.pool.connections.forEach(c =>
            assert.strictEqual(c.keyspace, client.keyspace)));
        }

        const promises = Array(30).fill(0).map(async () => {
          // Target a table on the system keyspace
          const result = await client.execute('SELECT * FROM local');
          assert.ok(result.rows);
          assert.ok(result.rows.length > 0);
        });

        await Promise.all(promises);
      });
    });

    it('should fail to execute if the keyspace does not exists', function (done) {
      const client = newInstance({ keyspace: 'NOT____EXISTS' });
      // Execute on all hosts, some executions in parallel and some serial
      utils.timesLimit(12, 6, function (n, next) {
        //No matter what, the keyspace does not exists
        client.execute(helper.queries.basic, function (err) {
          helper.assertInstanceOf(err, Error);
          next();
        });
      }, done);
    });

    it('should change the active keyspace after USE statement', function (done) {
      // Use a large amount of connections to make it more error prone
      const client = newInstance({ pooling: { coreConnectionsPerHost: { [types.distance.local]: 10 } } });
      client.execute('USE system', function (err) {
        if (err) {
          return done(err);
        }
        assert.strictEqual(client.keyspace, 'system');
        // all next queries, the instance should still "be" in the system keyspace
        utils.timesLimit(100, 50, function (n, next) {
          client.execute('SELECT * FROM local', [], next);
        }, helper.finish(client, done));
      });
    });

    it('should return ResponseError when executing USE with a wrong keyspace', function (done) {
      const client = newInstance();
      client.execute('USE ks_not_exist', function (err) {
        assert.ok(err instanceof errors.ResponseError);
        assert.equal(client.keyspace, null);
        client.shutdown();
        done();
      });
    });

    it('should create the amount of connections determined by the options', function (done) {
      const options = {
        pooling: {
          coreConnectionsPerHost: {
            '0': 3,
            '1': 0,
            '2': 0
          }
        }
      };
      const client = newInstance(options);

      // Execute a couple of queries
      utils.timesLimit(100, 50, function (n, next) {
        client.execute(helper.queries.basic, next);
      }, function (err) {
        if (err) {
          return done(err);
        }
        assert.strictEqual(client.hosts.length, 3);
        const hosts = client.hosts.slice(0);
        assert.strictEqual(hosts[0].pool.coreConnectionsLength, 3);
        assert.strictEqual(hosts[1].pool.coreConnectionsLength, 3);
        // wait for the pool to be the expected size
        helper.setIntervalUntil(function condition() {
          return (hosts[0].pool.connections.length === 3 && hosts[1].pool.connections.length === 3);
        }, 1000, 20, done);
      });
    });

    it('should wait for schema agreement before calling back', function (done) {
      const queries = [
        "CREATE KEYSPACE ks1 WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};",
        "CREATE TABLE ks1.tbl1 (id uuid PRIMARY KEY, value text)",
        "SELECT * FROM ks1.tbl1",
        "SELECT * FROM ks1.tbl1 where id = d54cb06d-d168-45a0-b1b2-9f5c75435d3d",
        "CREATE KEYSPACE ks2 WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 3};",
        "CREATE TABLE ks2.tbl2 (id uuid PRIMARY KEY, value text)",
        "SELECT * FROM ks2.tbl2",
        "SELECT * FROM ks2.tbl2",
        "CREATE TABLE ks2.tbl3 (id uuid PRIMARY KEY, value text)",
        "SELECT * FROM ks2.tbl3",
        "SELECT * FROM ks2.tbl3",
        "CREATE TABLE ks2.tbl4 (id uuid PRIMARY KEY, value text)",
        "SELECT * FROM ks2.tbl4",
        "SELECT * FROM ks2.tbl4",
        "SELECT * FROM ks2.tbl4"
      ];

      const client = newInstance();

      //warmup first
      utils.timesSeries(10, function (n, next) {
        client.execute('SELECT key FROM system.local', next);
      }, function (err) {
        assert.ifError(err);
        utils.eachSeries(queries, function (query, next) {
          client.execute(query, next);
        }, done);
      });
    });

    it('should handle distance changing load balancing policies', changingDistancesTest('2'));

    it('should handle distance changing load balancing policies for control connection host', changingDistancesTest('1'));
    [
      new policies.speculativeExecution.NoSpeculativeExecutionPolicy(),
      new ConstantSpeculativeExecutionPolicy(100, 1)
    ].forEach(function (policy) {
      context('with ' + policy.constructor.name, function () {
        afterEach(function (done) {
          helper.ccmHelper.resumeNode(1, done);
        });
        afterEach(function (done) {
          helper.ccmHelper.resumeNode(2, done);
        });
        it('should wait until is completed on the first node', function (done) {
          const client = newInstance({
            pooling: { warmup: true },
            policies: {
              speculativeExecution: policy,
              loadBalancing: new OrderedLoadBalancingPolicy(),
              retry: new helper.FallthroughRetryPolicy()
            },
            socketOptions: {
              readTimeout: 5000
            }
          });
          utils.series([
            client.connect.bind(client),
            helper.toTask(helper.ccmHelper.pauseNode, null, 1),
            function query(next) {
              if (!(policy instanceof ConstantSpeculativeExecutionPolicy)) {
                // Resume first node after a few ms
                setTimeout(function () {
                  helper.ccmHelper.resumeNode(1);
                }, 400);
              }
              utils.map([ false, true ], function execute(prepare, mapNext) {
                client.execute('SELECT * FROM system.local', null, { prepare: prepare, isIdempotent: true }, mapNext);
              }, function (err, results) {
                assert.ifError(err);
                assert.strictEqual(results.length, 2);
                let expectedHost;
                if (policy instanceof ConstantSpeculativeExecutionPolicy) {
                  // Use the next one in a speculative execution
                  expectedHost = client.hosts.keys()[1];
                }
                else {
                  // It should wait for it to yield the response without speculative execution
                  expectedHost = client.hosts.keys()[0];
                }
                assert.deepEqual(
                  results.map(function (rs) {
                    return rs.info.queriedHost;
                  }),
                  results.map(function () {
                    return expectedHost;
                  }));
                next();
              });
            },
            helper.toTask(helper.ccmHelper.resumeNode, null, 1),
          ], helper.finish(client, done));
        });
      });
    });

    function changingDistancesTest(address) {
      return (function doTest(done) {
        const lbp = new RoundRobinPolicy();
        let ignoredHost;
        let cc;
        lbp.getDistance = function (h) {
          return helper.lastOctetOf(h) === ignoredHost ? types.distance.ignored : types.distance.local;
        };
        const connectionsPerHost = 5;
        const client = newInstance({
          policies: { loadBalancing: lbp },
          pooling: { coreConnectionsPerHost: { '0': connectionsPerHost } }
        });
        function queryAndCheckPool(limit, assertions) {
          return (function executeSomeQueries(next) {
            const coordinators = {};
            utils.timesLimit(limit, 3, function (n, timesNext) {
              client.execute(helper.queries.basic, function (err, result) {
                if (!err) {
                  coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;
                }
                timesNext(err);
              });
            }, function (err) {
              assert.ifError(err);
              assertions(Object.keys(coordinators).sort());
              next();
            });
          });
        }
        utils.series([
          client.connect.bind(client),
          queryAndCheckPool(12, function (coordinators) {
            utils.objectValues(getPoolInfo(client)).forEach(function (poolSize) {
              assert.notStrictEqual(poolSize, 0);
            });
            assert.deepEqual(coordinators, [ '1', '2', '3']);
            cc = client.controlConnection.connection;
            // Set as the ignored host for the next queries
            ignoredHost = address;
          }),
          queryAndCheckPool(500, function (coordinators) {
            // The pool for 1st and 3rd host should have the appropriate size by now.
            const expectedPoolInfo = { '1': connectionsPerHost, '2': connectionsPerHost, '3': connectionsPerHost};
            expectedPoolInfo[ignoredHost] = 0;
            assert.deepEqual(getPoolInfo(client), expectedPoolInfo );
            assert.deepEqual(coordinators, [ '1', '2', '3'].filter(x => x !== ignoredHost));
          }),
          client.shutdown.bind(client),
          function checkPoolState(next) {
            const expectedState = { '1': 0, '2': 0, '3': 0};
            assert.deepEqual(getPoolInfo(client), expectedState);
            setTimeout(function checkPoolStateDelayed() {
              assert.deepEqual(getPoolInfo(client), expectedState);
              if (ignoredHost === '1') {
                // The control connection should have changed
                assert.ok(!cc.connected);
                assert.notStrictEqual(helper.lastOctetOf(client.controlConnection.host), '1');
              }
              else {
                assert.strictEqual(helper.lastOctetOf(client.controlConnection.host), '1');
              }
              next();
            }, 300);
          }
        ], done);
      });
    }
  });

  describe('failover', function () {
    beforeEach(helper.ccmHelper.start(3));
    afterEach(helper.ccmHelper.remove);

    it('should failover after a node goes down', function (done) {
      // treat queries as idempotent so they can be safely retried on another node
      const client = newInstance({ queryOptions: { isIdempotent: true } });

      after(() => client.shutdown());

      const hosts = {};
      const hostsDown = [];
      utils.series([
        client.connect.bind(client),
        function (next) {
          // wait for all initial events to ensure we don't incidentally get an 'UP' event for node 2
          // after we have killed it.
          setTimeout(next, 5000);
        },
        function warmUpPool(seriesNext) {
          client.on('hostDown', function (h) {
            hostsDown.push(h);
          });
          utils.times(100, function (n, next) {
            client.execute(helper.queries.basic, function (err, result) {
              assert.ifError(err);
              hosts[result.info.queriedHost] = true;
              next();
            });
          }, seriesNext);
        },
        function killNode(seriesNext) {
          assert.strictEqual(Object.keys(hosts).length, 3);
          setImmediate(function () {
            helper.ccmHelper.exec(['node1', 'stop', '--not-gently']);
            seriesNext();
          });
        },
        function executeWhileNodeGoingDown(seriesNext) {
          utils.timesLimit(1000, 10, function (i, next) {
            client.execute(helper.queries.basic, next);
          }, function (err) {
            assert.ifError(err);
            //delay the next queries
            setTimeout(seriesNext, 10000);
          });
        },
        function executeAfterNodeWentDown(seriesNext) {
          utils.timesSeries(20, function (i, next) {
            client.execute(helper.queries.basic, next);
          }, seriesNext);
        },
        function assertions(seriesNext) {
          //Only 2 hosts alive at the end
          assert.strictEqual(
            client.hosts.values().reduce(function (val, h) {
              return val + (h.isUp() ? 1 : 0);
            }, 0),
            2);
          assert.ok(hostsDown.length >= 1, "Expected at least 1 host down" +
            " event.");
          //Ensure each down event is for the stopped host.  We may get
          //multiple down events for the same host on a control connection.
          hostsDown.forEach(function (downHost) {
            assert.strictEqual(helper.lastOctetOf(downHost), '1');
          });
          seriesNext();
        },
        client.shutdown.bind(client)
      ], done);
    });

    it('should failover when a node goes down with some outstanding requests', function (done) {
      const options = utils.extend({ queryOptions: { isIdempotent: true } }, helper.baseOptions);
      options.pooling = {
        coreConnectionsPerHost: {
          '0': 1,
          '1': 1,
          '2': 0
        }
      };

      const client = new Client(options);
      const hosts = {};
      const query = helper.queries.basic;

      after(() => client.shutdown());

      utils.series([
        function (next) {
          // wait for all initial events to ensure we don't incidentally get an 'UP' event for node 2
          // after we have killed it.
          setTimeout(next, 5000);
        },
        function warmUpPool(seriesNext) {
          utils.times(20, function (n, next) {
            client.execute(query, function (err, result) {
              assert.ifError(err);
              hosts[result.info.queriedHost] = true;
              next();
            });
          }, seriesNext);
        },
        helper.setIntervalUntilTask(() => {
          const state = client.getState();
          return client.hosts.values().reduce((accumulator, h) => accumulator + state.getInFlightQueries(h), 0) === 0;
        }, 20, 100),
        function testCase(seriesNext) {
          // 3 hosts alive
          assert.strictEqual(Object.keys(hosts).length, 3);

          let killed = false;
          utils.timesLimit(500, 20, function (n, next) {
            if (n === 30) {
              //kill a node when there are some outstanding requests
              helper.ccmHelper.exec(['node2', 'stop', '--not-gently'], function (err) {
                killed = true;
                assert.ifError(err);
                //do a couple of more queries
                utils.timesSeries(10, function (n, next2) {
                  client.execute(query, next2);
                }, next);
              });
              return;
            }
            if (killed) {
              //Don't issue more requests
              return next();
            }
            client.execute(query, function (err) {
              assert.ifError(err);
              next();
            });
          }, seriesNext);
        },
        next => setImmediate(next),
        function assertHostState(next) {
          // Only 2 hosts are UP at the end
          assert.strictEqual(client.hosts.values().reduce((val, h) => val + (h.isUp() ? 1 : 0), 0), 2);

          // In-flight counter should be 0 for UP and DOWN nodes
          const state = client.getState();
          client.hosts.forEach(h => assert.strictEqual(state.getInFlightQueries(h), 0));
          next();
        }
      ], done);
    });

    it('should warn but not fail when warmup is enable and a node is down', async () => {
      await util.promisify(helper.ccmHelper.exec)(['node2', 'stop']);

      const warnings = [];
      const client = newInstance({ pooling: { warmup: true } });
      helper.afterThisTest(() => client.shutdown());

      client.on('log', (level, className, message) => {
        if (level !== 'warning' || className !== 'Client') {
          return;
        }

        warnings.push(message);
      });

      await client.connect();
      assert.lengthOf(warnings.filter(w => w.indexOf('pool') >= 0), 1);
    });
  });

  describe('#shutdown()', function () {
    helper.setup(2, { initClient: false });

    it('should close all connections to all hosts', function (done) {
      const client = newInstance();
      utils.series([
        client.connect.bind(client),
        function makeSomeQueries(next) {
          //to ensure that the pool is all up!
          utils.times(100, function (n, timesNext) {
            client.execute(helper.queries.basic, timesNext);
          }, next);
        },
        function shutDown(next) {
          const hosts = client.hosts.values();
          assert.strictEqual(hosts.length, 2);
          const state = client.getState();
          // Check the pools before shutting down
          hosts.forEach(function each(host) {
            assert.ok(state.getOpenConnections(host) > 0);
            assert.ok(host.pool.connections.length > 0);
            assert.ok(!host.pool.shuttingDown);
          });
          client.shutdown(next);
        },
        function checkPool(next) {
          const hosts = client.hosts.values();
          assert.strictEqual(hosts.length, 2);
          const state = client.getState();
          assert.deepEqual(state.getConnectedHosts(), []);
          hosts.forEach(function each(host) {
            assert.strictEqual(host.pool.connections.length, 0);
            assert.strictEqual(state.getOpenConnections(host), 0);
          });
          next();
        }
      ], done);
    });

    it('should not leak any connection when connection pool is still growing', function (done) {
      const client = newInstance({ pooling: { coreConnectionsPerHost: { '0': 4 }}});
      utils.series([
        client.connect.bind(client),
        function makeSomeQueries(next) {
          utils.times(10, function (n, timesNext) {
            client.execute(helper.queries.basic, timesNext);
          }, next);
        },
        function shutDown(next) {
          const hosts = client.hosts.values();
          assert.strictEqual(hosts.length, 2);
          assert.ok(hosts[0].pool.connections.length > 0);
          assert.ok(!hosts[0].pool.shuttingDown);
          assert.ok(!hosts[1].pool.shuttingDown);
          client.shutdown(next);
        },
        function checkPoolDelayed(next) {
          function checkNoConnections() {
            assert.deepEqual(getPoolInfo(client), { '1': 0, '2': 0 });
          }
          checkNoConnections();
          // Wait some time and check again to see if there is a new connection created in the background
          setTimeout(function checkNoConnectionsDelayed() {
            checkNoConnections();
            next();
          }, 1000);
        }
      ], done);
    });

    it('should callback after a NoHostAvailableError', function (done) {
      const client = newInstance({ contactPoints: [ '::1', '::2'] });
      client.connect(function (err) {
        helper.assertInstanceOf(err, errors.NoHostAvailableError);
        assert.strictEqual(client.hosts.length, 0);
        client.shutdown(function (err) {
          assert.strictEqual(client.hosts.length, 0);
          assert.ifError(err);
          done();
        });
      });
    });

    it('should close all connections after connecting with an invalid keyspace', function (done) {
      const client = newInstance({ keyspace: 'KS_DOES_NOT_EXIST' });
      client.connect(function (err) {
        helper.assertInstanceOf(err, errors.ResponseError);
        assert.strictEqual(client.hosts.length, 0);
        client.shutdown(function (err) {
          assert.ifError(err);
          assert.strictEqual(client.hosts.length, 0);
          done();
        });
      });
    });
  });
});

/**
 * @param {ClientOptions} [options]
 * @returns {Client}
 */
function newInstance(options) {
  const client = new Client(utils.deepExtend({}, helper.baseOptions, options));
  helper.shutdownAfterThisTest(client);
  return client;
}

/**
 * Returns a dictionary containing the last octet of the address as keys and the pool size as values.
 */
function getPoolInfo(client) {
  const info = {};
  client.hosts.forEach(function (h, address) {
    info[helper.lastOctetOf(address)] = h.pool.connections.length;
  });
  return info;
}

function createRole(client, role, password) {
  return client.execute(`CREATE ROLE IF NOT EXISTS ${role} WITH PASSWORD = '${password}' AND LOGIN = true`);
}

function assertAuthError(err, message) {
  helper.assertInstanceOf(err, errors.NoHostAvailableError);
  assert.ok(err.innerErrors);
  const firstErr = Object.values(err.innerErrors)[0];
  helper.assertInstanceOf(firstErr, errors.AuthenticationError);

  if (message) {
    assert.match(firstErr.message, message);
  }
}

<MSG> Merge branch 'master' of https://github.com/MCProHosting/nodejs-driver

<DFF> @@ -470,6 +470,8 @@ describe('Client', function () {
           assert.strictEqual(hosts.length, 2);
           assert.ok(hosts[0].pool.connections.length > 0);
           assert.ok(hosts[1].pool.connections.length > 0);
+          assert.ok(hosts[0].pool.shuttingDown === false);
+          assert.ok(hosts[1].pool.shuttingDown === false);
           client.shutdown(next);
         },
         function checkPool(next) {
@@ -489,4 +491,4 @@ describe('Client', function () {
  */
 function newInstance(options) {
   return new Client(utils.extend({}, helper.baseOptions, options));
-}
\ No newline at end of file
+}

{'commit': "diff --git a/lib/host.js b/lib/host.js\nindex fbf6cd7..b768c27 100644\n--- a/lib/host.js\n+++ b/lib/host.js\n@@ -204,6 +204,7 @@ HostConnectionPool.prototype.shutdown = function (callback) {\n     c.close(next);\n   }, function (err) {\n     self.connections = null;\n+    self.shuttingDown = false;\n     callback(err);\n   });\n };\ndiff --git a/test/integration/short/client-pool-tests.js b/test/integration/short/client-pool-tests.js\nindex 56bb1c5..2f98542 100644\n--- a/test/integration/short/client-pool-tests.js\n+++ b/test/integration/short/client-pool-tests.js\n@@ -470,6 +470,8 @@ describe('Client', function () {\n           assert.strictEqual(hosts.length, 2);\n           assert.ok(hosts[0].pool.connections.length > 0);\n           assert.ok(hosts[1].pool.connections.length > 0);\n+          assert.ok(hosts[0].pool.shuttingDown === false);\n+          assert.ok(hosts[1].pool.shuttingDown === false);\n           client.shutdown(next);\n         },\n         function checkPool(next) {\n@@ -489,4 +491,4 @@ describe('Client', function () {\n  */\n function newInstance(options) {\n   return new Client(utils.extend({}, helper.baseOptions, options));\n-}\n\\ No newline at end of file\n+}\n", 'commit_subject': "Merge branch 'master' of https://github.com/MCProHosting/nodejs-driver", 'repo_name': 'datastax/nodejs-driver', 'committer': {'name': 'Jorge Bay Gondra', 'email': '23d47d5953bdc691bf4fba8380b3c5ba06aeb1f9@datastax.com', 'time_sec': '1419587823', 'tz_offset': '60', 'date': {'seconds': '1419587823'}}, 'commit_tree': 'b9bc696d942186c47e78af405bedd097e8160366', 'commit_message': "Merge branch 'master' of https://github.com/MCProHosting/nodejs-driver\n", 'language_name': 'JavaScript', 'license': 'apache-2.0', 'hunks': "@@ -470,6 +470,8 @@ describe('Client', function () {\n           assert.strictEqual(hosts.length, 2);\n           assert.ok(hosts[0].pool.connections.length > 0);\n           assert.ok(hosts[1].pool.connections.length > 0);\n+          assert.ok(hosts[0].pool.shuttingDown === false);\n+          assert.ok(hosts[1].pool.shuttingDown === false);\n           client.shutdown(next);\n         },\n         function checkPool(next) {\n@@ -489,4 +491,4 @@ describe('Client', function () {\n  */\n function newInstance(options) {\n   return new Client(utils.extend({}, helper.baseOptions, options));\n-}\n\\ No newline at end of file\n+}\n", 'addition_count': 3, 'deletion_count': 1, 'src_file': 'a/test/integration/short/client-pool-tests.js', 'tgt_file': 'b/test/integration/short/client-pool-tests.js', 'file_extension': '.js', 'before_file': '/*\n * Copyright DataStax, Inc.\n *\n * Licensed under the Apache License, Version 2.0 (the "License");\n * you may not use this file except in compliance with the License.\n * You may obtain a copy of the License at\n *\n * http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\'use strict\';\n\nconst { assert } = require(\'chai\');\nconst dns = require(\'dns\');\nconst util = require(\'util\');\n\nconst helper = require(\'../../test-helper\');\nconst Client = require(\'../../../lib/client\');\nconst clientOptions = require(\'../../../lib/client-options\');\nconst utils = require(\'../../../lib/utils\');\nconst errors = require(\'../../../lib/errors\');\nconst types = require(\'../../../lib/types\');\nconst policies = require(\'../../../lib/policies\');\nconst RoundRobinPolicy = require(\'../../../lib/policies/load-balancing.js\').RoundRobinPolicy;\nconst Murmur3Tokenizer = require(\'../../../lib/tokenizer.js\').Murmur3Tokenizer;\nconst { PlainTextAuthProvider } = require(\'../../../lib/auth\');\nconst ConstantSpeculativeExecutionPolicy = policies.speculativeExecution.ConstantSpeculativeExecutionPolicy;\nconst OrderedLoadBalancingPolicy = helper.OrderedLoadBalancingPolicy;\nconst vit = helper.vit;\n\ndescribe(\'Client\', function () {\n  this.timeout(180000);\n  describe(\'#connect()\', function () {\n    let useLocalhost;\n\n    helper.setup(3, { initClient: false });\n\n    before(function (done) {\n      dns.resolve(\'localhost\', function (err) {\n        useLocalhost = !err;\n        done();\n      });\n    });\n\n    it(\'should discover all hosts in the ring and hosts object can be serializable\', async () => {\n      const client = newInstance();\n      await client.connect();\n\n      assert.strictEqual(client.hosts.length, 3);\n      assert.strictEqual(client.hosts.values().length, 3);\n      assert.strictEqual(client.hosts.keys().length, 3);\n\n      //It should be serializable\n      assert.doesNotThrow(() => JSON.stringify(client.hosts));\n    });\n\n    it(\'should fail if the contact points can not be resolved\', function (done) {\n      const client = newInstance({contactPoints: [\'not-a-host\']});\n      client.connect(function (err) {\n        assert.ok(err);\n        helper.assertInstanceOf(err, errors.NoHostAvailableError);\n        client.shutdown(function (err) {\n          assert.ifError(err);\n          done();\n        });\n      });\n    });\n\n    it(\'should fail if the contact points can not be reached\', function (done) {\n      const client = newInstance({contactPoints: [\'1.1.1.1\']});\n      client.connect(function (err) {\n        assert.ok(err);\n        helper.assertInstanceOf(err, errors.NoHostAvailableError);\n        client.shutdown(done);\n      });\n    });\n\n    it(\'should select a tokenizer\', function (done) {\n      const client = newInstance();\n      client.connect(function (err) {\n        if (err) {return done(err);}\n        helper.assertInstanceOf(client.metadata.tokenizer, Murmur3Tokenizer);\n        client.shutdown(done);\n      });\n    });\n\n    it(\'should allow multiple parallel calls to connect\', function (done) {\n      const client = newInstance();\n      utils.times(100, function (n, next) {\n        client.connect(next);\n      }, function (err) {\n        assert.ifError(err);\n        client.shutdown(done);\n      });\n    });\n\n    it(\'should resolve host names\', function (done) {\n      if (!useLocalhost) {\n        return done();\n      }\n      const client = new Client(utils.extend({}, helper.baseOptions, {contactPoints: [\'localhost\']}));\n      client.connect(function (err) {\n        assert.ifError(err);\n        assert.strictEqual(client.hosts.length, 3);\n        client.hosts.forEach(function (h) {\n          assert.notEqual(h.address, \'localhost\');\n        });\n        client.shutdown(done);\n      });\n    });\n\n    [ true, false ].forEach(warmup => {\n      it(`should fail if the keyspace does not exists when warmup is ${warmup}`, function (done) {\n        const client = newInstance({\n          keyspace: \'not-existent-ks\',\n          pooling: { warmup }\n        });\n\n        utils.times(10, function (n, next) {\n          client.connect(function (err) {\n            assert.ok(err);\n            //Not very nice way to check but here it is\n            //Does the message contains Keyspace\n            assert.ok(err.message.toLowerCase().indexOf(\'keyspace\') >= 0, \'Message mismatch, was: \' + err.message);\n            next();\n          });\n        }, done);\n      });\n    });\n\n    it(\'should not use contactPoints that are not part of peers\', function (done) {\n      const contactPoints = helper.baseOptions.contactPoints.slice(0);\n      contactPoints.push(\'host-not-existent-not-peer\');\n      contactPoints.push(\'1.1.1.1\');\n      const client = newInstance({contactPoints: contactPoints});\n      client.connect(function (err) {\n        assert.ifError(err);\n        //the 3 original hosts\n        assert.strictEqual(client.hosts.length, 3);\n        const hosts = client.hosts.keys();\n        assert.strictEqual(hosts[0], contactPoints[0] + \':9042\');\n        assert.notEqual(hosts[1], contactPoints[1] + \':9042\');\n        assert.notEqual(hosts[2], contactPoints[1] + \':9042\');\n        assert.notEqual(hosts[1], contactPoints[2] + \':9042\');\n        assert.notEqual(hosts[2], contactPoints[2] + \':9042\');\n        client.shutdown(done);\n      });\n    });\n\n    it(\'should use the default pooling options according to the protocol version\', function (done) {\n      const client = newInstance();\n      client.connect(function (err) {\n        assert.ifError(err);\n        assert.ok(client.options.pooling.coreConnectionsPerHost);\n        if (client.controlConnection.protocolVersion < 3) {\n          helper.assertValueEqual(client.options.pooling.coreConnectionsPerHost, clientOptions.coreConnectionsPerHostV2);\n        }\n        else {\n          helper.assertValueEqual(client.options.pooling.coreConnectionsPerHost, clientOptions.coreConnectionsPerHostV3);\n        }\n        utils.times(10, function (n, next) {\n          client.execute(\'SELECT key FROM system.local\', next);\n        }, function (err) {\n          if (err) {return done(err);}\n          assert.strictEqual(client.hosts.values()[0].pool.connections.length, client.options.pooling.coreConnectionsPerHost[types.distance.local]);\n          client.shutdown(done);\n        });\n      });\n    });\n\n    it(\'should override default pooling options when specified\', function (done) {\n      const client = newInstance({ pooling: {\n        coreConnectionsPerHost: { \'0\': 4 }\n      }});\n      client.connect(function (err) {\n        assert.ifError(err);\n        assert.ok(client.options.pooling.coreConnectionsPerHost);\n        let defaults = clientOptions.coreConnectionsPerHostV3;\n        if (client.controlConnection.protocolVersion < 3) {\n          defaults = clientOptions.coreConnectionsPerHostV2;\n        }\n        assert.ok(client.options.pooling.coreConnectionsPerHost[types.distance.local], 4);\n        assert.ok(client.options.pooling.coreConnectionsPerHost[types.distance.remote], defaults[types.distance.remote]);\n        utils.times(50, function (n, next) {\n          client.execute(\'SELECT key FROM system.local\', next);\n        }, function (err) {\n          if (err) {\n            return done(err);\n          }\n          setTimeout(function () {\n            //wait until all connections are made in the background\n            assert.strictEqual(\n              client.hosts.values()[0].pool.connections.length,\n              client.options.pooling.coreConnectionsPerHost[types.distance.local]);\n            client.shutdown(done);\n          }, 5000);\n        });\n      });\n    });\n\n    it(\'should not fail when switching keyspace and a contact point is not valid\', function (done) {\n      const client = new Client({\n        contactPoints: [\'1.1.1.1\', helper.baseOptions.contactPoints[0]],\n        localDataCenter: helper.baseOptions.localDataCenter,\n        keyspace: \'system\'\n      });\n      client.connect(function (err) {\n        assert.ifError(err);\n        client.shutdown(done);\n      });\n    });\n\n    it(\'should open connections to all hosts when warmup is set\', function (done) {\n      // do it multiple times\n      utils.timesSeries(300, function (n, next) {\n        const connectionsPerHost = {};\n        connectionsPerHost[types.distance.local] = 3;\n        connectionsPerHost[types.distance.remote] = 1;\n        const client = newInstance({pooling: {warmup: true, coreConnectionsPerHost: connectionsPerHost}});\n        client.connect(function (err) {\n          assert.ifError(err);\n          assert.strictEqual(client.hosts.length, 3);\n          const state = client.getState();\n          client.hosts.forEach(function (host) {\n            assert.strictEqual(host.pool.connections.length, 3, \'For host \' + host.address);\n            assert.strictEqual(state.getOpenConnections(host), 3);\n          });\n          client.shutdown(next);\n        });\n      }, done);\n    });\n\n    it(\'should only warmup connections for hosts with local distance\', async () => {\n      const lbPolicy = new RoundRobinPolicy();\n      lbPolicy.getDistance = function (host) {\n        const id = helper.lastOctetOf(host.address);\n        if(id === \'1\') {\n          return types.distance.local;\n        }\n        else if(id === \'2\') {\n          return types.distance.remote;\n        }\n        return types.distance.ignored;\n      };\n\n      const connectionsPerHost = {};\n      connectionsPerHost[types.distance.local] = 2;\n      connectionsPerHost[types.distance.remote] = 10;\n\n      const client = newInstance({\n        policies: { loadBalancing: lbPolicy },\n        pooling: { warmup: true, coreConnectionsPerHost: connectionsPerHost}\n      });\n\n      await client.connect();\n\n      assert.strictEqual(client.hosts.length, 3);\n      client.hosts.forEach(function (host) {\n        const id = helper.lastOctetOf(host);\n        if(id === \'1\') {\n          assert.strictEqual(host.pool.connections.length, connectionsPerHost[types.distance.local]);\n        } else if (id === \'2\') {\n          // It shouldn\'t have finished creating all the connections\n          assert.isBelow(host.pool.connections.length, connectionsPerHost[types.distance.remote]);\n        } else {\n          assert.strictEqual(host.pool.connections.length, 0);\n        }\n      });\n\n      await client.shutdown();\n    });\n\n    it(\'should connect after unsuccessful attempt caused by a non-existent keyspace\', function (done) {\n      const keyspace = \'ks_test_after_fail\';\n      const client = newInstance({ keyspace: keyspace });\n      utils.series([\n        function tryConnect(next) {\n          client.connect(function (err) {\n            helper.assertInstanceOf(err, errors.ResponseError);\n            next();\n          });\n        },\n        function createKeyspace(next) {\n          const tempClient = newInstance();\n          tempClient.execute(helper.createKeyspaceCql(keyspace), function (err) {\n            assert.ifError(err);\n            tempClient.shutdown(next);\n          });\n        },\n        function tryConnectAgain(next) {\n          client.connect(function (err) {\n            assert.ifError(err);\n            client.execute(helper.queries.basic, next);\n          });\n        }\n      ], function (err) {\n        client.shutdown();\n        done(err);\n      });\n    });\n\n    it(\'should set the defaults based on product type\', () => {\n      const client = newInstance();\n\n      return client.connect()\n        .then(() => {\n          assert.strictEqual(client.options.queryOptions.consistency, types.consistencies.localOne);\n        })\n        .then(() => client.shutdown());\n    });\n  });\n\n  describe(\'#connect() with auth\', function () {\n\n    helper.setup(1, {\n      initClient: false,\n      ccmOptions: {\n        yaml: [\'authenticator:PasswordAuthenticator\'],\n        jvmArgs: [\'-Dcassandra.superuser_setup_delay_ms=0\']\n      }\n    });\n\n    it(\'should connect using the plain text authenticator\', function (done) {\n      const options = {authProvider: new PlainTextAuthProvider(\'cassandra\', \'cassandra\')};\n      const client = newInstance(options);\n      utils.times(100, function (n, next) {\n        client.connect(next);\n      }, helper.finish(client, done));\n    });\n\n    vit(\'3.0\', \'should support connecting using other role\', () => {\n      let client = newInstance({ authProvider: new PlainTextAuthProvider(\'cassandra\', \'cassandra\') });\n\n      const username = \'user2\';\n      const password = \'12345678\';\n\n      after(() => client.shutdown());\n\n      return client.connect()\n        .then(() => createRole(client, username, password))\n        .then(() => client.shutdown())\n        .then(() => {\n          client = newInstance({ authProvider: new PlainTextAuthProvider(username, password) });\n\n          after(() => client.shutdown());\n\n          return client.connect();\n        })\n        .then(() => client.execute(helper.queries.basic))\n        .then(() => client.shutdown());\n    });\n\n    it(\'should connect using the plain text authenticator when calling execute\', function (done) {\n      const options = {authProvider: new PlainTextAuthProvider(\'cassandra\', \'cassandra\'), keyspace: \'system\'};\n      const client = newInstance(options);\n      utils.times(100, function (n, next) {\n        client.execute(\'SELECT * FROM local\', next);\n      }, helper.finish(client, done));\n    });\n\n    it(\'should return an AuthenticationError\', function (done) {\n      const options = {authProvider: new PlainTextAuthProvider(\'not___EXISTS\', \'not___EXISTS\'), keyspace: \'system\'};\n      const client = newInstance(options);\n      utils.timesSeries(10, function (n, next) {\n        client.connect(function (err) {\n          assertAuthError(err);\n          next();\n        });\n      }, helper.finish(client, done));\n    });\n\n    it(\'should return an AuthenticationError when calling execute\', function (done) {\n      const options = {authProvider: new PlainTextAuthProvider(\'not___EXISTS\', \'not___EXISTS\'), keyspace: \'system\'};\n      const client = newInstance(options);\n      utils.times(10, function (n, next) {\n        client.execute(\'SELECT * FROM local\', function (err) {\n          assertAuthError(err);\n          next();\n        });\n      }, helper.finish(client, done));\n    });\n\n    it(\'should return an AuthenticationError when authProvider is not set\', async () => {\n      const client = newInstance();\n      const err = await helper.assertThrowsAsync(client.connect());\n      assertAuthError(err, /requires authentication, but no authenticator found in the options/);\n      await client.shutdown();\n    });\n\n    context(\'with credentials\', () => {\n\n      vit(\'3.0\', \'should support authenticating\', () => {\n        let client = newInstance({ credentials: { username: \'cassandra\', password: \'cassandra\' } });\n\n        const username = \'user2\';\n        const password = \'12345678\';\n\n        after(() => client.shutdown());\n\n        return client.connect()\n          .then(() => createRole(client, username, password))\n          .then(() => client.shutdown())\n          .then(() => {\n            client = newInstance({ credentials: { username, password } });\n\n            after(() => client.shutdown());\n\n            return client.connect();\n          })\n          .then(() => client.execute(helper.queries.basic))\n          .then(() => client.shutdown());\n      });\n\n      it(\'should fail with AuthenticationError when role does not exist\', () => {\n        const client = newInstance({ credentials: { username: \'incorrect_user\', password: \'abcd\' } });\n        let error;\n\n        return client.connect()\n          .catch(err => error = err)\n          .then(() => assertAuthError(error));\n      });\n\n      it(\'should fail with AuthenticationError when password is incorrect\', () => {\n        const client = newInstance({ credentials: { username: \'cassandra\', password: \'invalid_password\' } });\n        let error;\n\n        return client.connect()\n          .catch(err => error = err)\n          .then(() => assertAuthError(error));\n      });\n    });\n  });\n\n  describe(\'#connect() with ipv6\', function () {\n    before(helper.ccmHelper.start(1, { ipFormat: \'::%d\' }));\n    after(helper.ccmHelper.remove);\n    it(\'should connect to ipv6 host\', function (done) {\n      utils.series([\n        function testWithShortNotation(seriesNext) {\n          testConnect(\'::1\', seriesNext);\n        },\n        function testFullAddress(seriesNext) {\n          testConnect(\'0:0:0:0:0:0:0:1\', seriesNext);\n        },\n        function testFullAddress(seriesNext) {\n          testConnect(\'[0:0:0:0:0:0:0:1]:9042\', seriesNext);\n        }\n      ], done);\n      function testConnect(contactPoint, testDone) {\n        const client = newInstance({ contactPoints: [ contactPoint ] });\n        client.connect(function (err) {\n          assert.ifError(err);\n          assert.strictEqual(client.hosts.length, 1);\n          let expected = contactPoint + \':9042\';\n          if (contactPoint.indexOf(\'[\') === 0) {\n            expected = contactPoint.replace(/[[\\]]/g, \'\');\n          }\n          assert.strictEqual(client.hosts.values()[0].address, expected);\n          utils.times(10, function (n, next) {\n            client.execute(helper.queries.basic, next);\n          }, helper.finish(client, testDone));\n        });\n      }\n    });\n          assert.strictEqual(hosts.length, 2);\n          assert.ok(hosts[0].pool.connections.length > 0);\n          assert.ok(hosts[1].pool.connections.length > 0);\n          client.shutdown(next);\n        },\n        function checkPool(next) {\n      utils.series([\n        helper.ccmHelper.removeIfAny,\n        function (next) {\n          client.connect(function (err) {\n            helper.assertInstanceOf(err, errors.NoHostAvailableError);\n            next();\n          });\n        },\n        helper.ccmHelper.start(1),\n        function (next) {\n          client.connect(function (err) {\n            assert.ifError(err);\n            const hosts = client.hosts.values();\n */\nfunction newInstance(options) {\n  return new Client(utils.extend({}, helper.baseOptions, options));\n}\n          });\n        },\n        client.shutdown.bind(client),\n        helper.ccmHelper.remove\n      ], done);\n    });\n    function getReceiveNotificationTest(nodeNumber) {\n      return (function receiveNotificationTest(done) {\n        // Should receive notification when a node gracefully closes connections\n        const client = newInstance({\n          pooling: {\n            warmup: true,\n            heartBeatInterval: 0,\n            // Use just 1 connection per host for all protocol versions in this test\n            coreConnectionsPerHost: clientOptions.coreConnectionsPerHostV3\n          }\n        });\n        utils.series([\n          helper.ccmHelper.removeIfAny,\n          helper.ccmHelper.start(2),\n          client.connect.bind(client),\n          function checkInitialState(next) {\n            const hosts = client.hosts.values();\n            assert.ok(hosts[0].isUp());\n            assert.ok(hosts[1].isUp());\n            next();\n          },\n          function stopNode(next) {\n            helper.ccmHelper.stopNode(nodeNumber, next);\n          },\n          helper.delay(300),\n          function checkThatStateChanged(next) {\n            // Only 1 node should be UP\n            assert.strictEqual(client.hosts.values().filter(function (h) {\n              return h.isUp();\n            }).length, 1);\n            next();\n          },\n          client.shutdown.bind(client),\n          helper.ccmHelper.remove\n        ], done);\n      });\n    }\n    it(\'should receive socket closed event and set node as down\', getReceiveNotificationTest(2));\n    it(\'should receive socket closed event and set node as down (control connection node)\', getReceiveNotificationTest(1));\n  });\n\n  describe(\'#execute()\', function () {\n\n    helper.setup(3, { initClient: false });\n\n    [ true, false ].forEach(warmup => {\n      it(`should use the keyspace provided when warmup is ${warmup}`, async () => {\n        const client = newInstance({\n          keyspace: \'system\',\n          pooling: { warmup, coreConnectionsPerHost: { [types.distance.local]: 2 }}\n        });\n\n        await client.connect();\n\n        assert.strictEqual(client.keyspace, \'system\');\n\n        if (!warmup) {\n          // Wait for each pool to be created\n          for (const host of client.hosts.values()) {\n            await helper.wait.until(() => host.pool.connections.length > 0);\n          }\n        } else {\n          // All connections should have switch to the active keyspace\n          client.hosts.values().forEach(h => h.pool.connections.forEach(c =>\n            assert.strictEqual(c.keyspace, client.keyspace)));\n        }\n\n        const promises = Array(30).fill(0).map(async () => {\n          // Target a table on the system keyspace\n          const result = await client.execute(\'SELECT * FROM local\');\n          assert.ok(result.rows);\n          assert.ok(result.rows.length > 0);\n        });\n\n        await Promise.all(promises);\n      });\n    });\n\n    it(\'should fail to execute if the keyspace does not exists\', function (done) {\n      const client = newInstance({ keyspace: \'NOT____EXISTS\' });\n      // Execute on all hosts, some executions in parallel and some serial\n      utils.timesLimit(12, 6, function (n, next) {\n        //No matter what, the keyspace does not exists\n        client.execute(helper.queries.basic, function (err) {\n          helper.assertInstanceOf(err, Error);\n          next();\n        });\n      }, done);\n    });\n\n    it(\'should change the active keyspace after USE statement\', function (done) {\n      // Use a large amount of connections to make it more error prone\n      const client = newInstance({ pooling: { coreConnectionsPerHost: { [types.distance.local]: 10 } } });\n      client.execute(\'USE system\', function (err) {\n        if (err) {\n          return done(err);\n        }\n        assert.strictEqual(client.keyspace, \'system\');\n        // all next queries, the instance should still "be" in the system keyspace\n        utils.timesLimit(100, 50, function (n, next) {\n          client.execute(\'SELECT * FROM local\', [], next);\n        }, helper.finish(client, done));\n      });\n    });\n\n    it(\'should return ResponseError when executing USE with a wrong keyspace\', function (done) {\n      const client = newInstance();\n      client.execute(\'USE ks_not_exist\', function (err) {\n        assert.ok(err instanceof errors.ResponseError);\n        assert.equal(client.keyspace, null);\n        client.shutdown();\n        done();\n      });\n    });\n\n    it(\'should create the amount of connections determined by the options\', function (done) {\n      const options = {\n        pooling: {\n          coreConnectionsPerHost: {\n            \'0\': 3,\n            \'1\': 0,\n            \'2\': 0\n          }\n        }\n      };\n      const client = newInstance(options);\n\n      // Execute a couple of queries\n      utils.timesLimit(100, 50, function (n, next) {\n        client.execute(helper.queries.basic, next);\n      }, function (err) {\n        if (err) {\n          return done(err);\n        }\n        assert.strictEqual(client.hosts.length, 3);\n        const hosts = client.hosts.slice(0);\n        assert.strictEqual(hosts[0].pool.coreConnectionsLength, 3);\n        assert.strictEqual(hosts[1].pool.coreConnectionsLength, 3);\n        // wait for the pool to be the expected size\n        helper.setIntervalUntil(function condition() {\n          return (hosts[0].pool.connections.length === 3 && hosts[1].pool.connections.length === 3);\n        }, 1000, 20, done);\n      });\n    });\n\n    it(\'should wait for schema agreement before calling back\', function (done) {\n      const queries = [\n        "CREATE KEYSPACE ks1 WITH replication = {\'class\': \'SimpleStrategy\', \'replication_factor\' : 3};",\n        "CREATE TABLE ks1.tbl1 (id uuid PRIMARY KEY, value text)",\n        "SELECT * FROM ks1.tbl1",\n        "SELECT * FROM ks1.tbl1 where id = d54cb06d-d168-45a0-b1b2-9f5c75435d3d",\n        "CREATE KEYSPACE ks2 WITH replication = {\'class\': \'SimpleStrategy\', \'replication_factor\' : 3};",\n        "CREATE TABLE ks2.tbl2 (id uuid PRIMARY KEY, value text)",\n        "SELECT * FROM ks2.tbl2",\n        "SELECT * FROM ks2.tbl2",\n        "CREATE TABLE ks2.tbl3 (id uuid PRIMARY KEY, value text)",\n        "SELECT * FROM ks2.tbl3",\n        "SELECT * FROM ks2.tbl3",\n        "CREATE TABLE ks2.tbl4 (id uuid PRIMARY KEY, value text)",\n        "SELECT * FROM ks2.tbl4",\n        "SELECT * FROM ks2.tbl4",\n        "SELECT * FROM ks2.tbl4"\n      ];\n\n      const client = newInstance();\n\n      //warmup first\n      utils.timesSeries(10, function (n, next) {\n        client.execute(\'SELECT key FROM system.local\', next);\n      }, function (err) {\n        assert.ifError(err);\n        utils.eachSeries(queries, function (query, next) {\n          client.execute(query, next);\n        }, done);\n      });\n    });\n\n    it(\'should handle distance changing load balancing policies\', changingDistancesTest(\'2\'));\n\n    it(\'should handle distance changing load balancing policies for control connection host\', changingDistancesTest(\'1\'));\n    [\n      new policies.speculativeExecution.NoSpeculativeExecutionPolicy(),\n      new ConstantSpeculativeExecutionPolicy(100, 1)\n    ].forEach(function (policy) {\n      context(\'with \' + policy.constructor.name, function () {\n        afterEach(function (done) {\n          helper.ccmHelper.resumeNode(1, done);\n        });\n        afterEach(function (done) {\n          helper.ccmHelper.resumeNode(2, done);\n        });\n        it(\'should wait until is completed on the first node\', function (done) {\n          const client = newInstance({\n            pooling: { warmup: true },\n            policies: {\n              speculativeExecution: policy,\n              loadBalancing: new OrderedLoadBalancingPolicy(),\n              retry: new helper.FallthroughRetryPolicy()\n            },\n            socketOptions: {\n              readTimeout: 5000\n            }\n          });\n          utils.series([\n            client.connect.bind(client),\n            helper.toTask(helper.ccmHelper.pauseNode, null, 1),\n            function query(next) {\n              if (!(policy instanceof ConstantSpeculativeExecutionPolicy)) {\n                // Resume first node after a few ms\n                setTimeout(function () {\n                  helper.ccmHelper.resumeNode(1);\n                }, 400);\n              }\n              utils.map([ false, true ], function execute(prepare, mapNext) {\n                client.execute(\'SELECT * FROM system.local\', null, { prepare: prepare, isIdempotent: true }, mapNext);\n              }, function (err, results) {\n                assert.ifError(err);\n                assert.strictEqual(results.length, 2);\n                let expectedHost;\n                if (policy instanceof ConstantSpeculativeExecutionPolicy) {\n                  // Use the next one in a speculative execution\n                  expectedHost = client.hosts.keys()[1];\n                }\n                else {\n                  // It should wait for it to yield the response without speculative execution\n                  expectedHost = client.hosts.keys()[0];\n                }\n                assert.deepEqual(\n                  results.map(function (rs) {\n                    return rs.info.queriedHost;\n                  }),\n                  results.map(function () {\n                    return expectedHost;\n                  }));\n                next();\n              });\n            },\n            helper.toTask(helper.ccmHelper.resumeNode, null, 1),\n          ], helper.finish(client, done));\n        });\n      });\n    });\n\n    function changingDistancesTest(address) {\n      return (function doTest(done) {\n        const lbp = new RoundRobinPolicy();\n        let ignoredHost;\n        let cc;\n        lbp.getDistance = function (h) {\n          return helper.lastOctetOf(h) === ignoredHost ? types.distance.ignored : types.distance.local;\n        };\n        const connectionsPerHost = 5;\n        const client = newInstance({\n          policies: { loadBalancing: lbp },\n          pooling: { coreConnectionsPerHost: { \'0\': connectionsPerHost } }\n        });\n        function queryAndCheckPool(limit, assertions) {\n          return (function executeSomeQueries(next) {\n            const coordinators = {};\n            utils.timesLimit(limit, 3, function (n, timesNext) {\n              client.execute(helper.queries.basic, function (err, result) {\n                if (!err) {\n                  coordinators[helper.lastOctetOf(result.info.queriedHost)] = true;\n                }\n                timesNext(err);\n              });\n            }, function (err) {\n              assert.ifError(err);\n              assertions(Object.keys(coordinators).sort());\n              next();\n            });\n          });\n        }\n        utils.series([\n          client.connect.bind(client),\n          queryAndCheckPool(12, function (coordinators) {\n            utils.objectValues(getPoolInfo(client)).forEach(function (poolSize) {\n              assert.notStrictEqual(poolSize, 0);\n            });\n            assert.deepEqual(coordinators, [ \'1\', \'2\', \'3\']);\n            cc = client.controlConnection.connection;\n            // Set as the ignored host for the next queries\n            ignoredHost = address;\n          }),\n          queryAndCheckPool(500, function (coordinators) {\n            // The pool for 1st and 3rd host should have the appropriate size by now.\n            const expectedPoolInfo = { \'1\': connectionsPerHost, \'2\': connectionsPerHost, \'3\': connectionsPerHost};\n            expectedPoolInfo[ignoredHost] = 0;\n            assert.deepEqual(getPoolInfo(client), expectedPoolInfo );\n            assert.deepEqual(coordinators, [ \'1\', \'2\', \'3\'].filter(x => x !== ignoredHost));\n          }),\n          client.shutdown.bind(client),\n          function checkPoolState(next) {\n            const expectedState = { \'1\': 0, \'2\': 0, \'3\': 0};\n            assert.deepEqual(getPoolInfo(client), expectedState);\n            setTimeout(function checkPoolStateDelayed() {\n              assert.deepEqual(getPoolInfo(client), expectedState);\n              if (ignoredHost === \'1\') {\n                // The control connection should have changed\n                assert.ok(!cc.connected);\n                assert.notStrictEqual(helper.lastOctetOf(client.controlConnection.host), \'1\');\n              }\n              else {\n                assert.strictEqual(helper.lastOctetOf(client.controlConnection.host), \'1\');\n              }\n              next();\n            }, 300);\n          }\n        ], done);\n      });\n    }\n  });\n\n  describe(\'failover\', function () {\n    beforeEach(helper.ccmHelper.start(3));\n    afterEach(helper.ccmHelper.remove);\n\n    it(\'should failover after a node goes down\', function (done) {\n      // treat queries as idempotent so they can be safely retried on another node\n      const client = newInstance({ queryOptions: { isIdempotent: true } });\n\n      after(() => client.shutdown());\n\n      const hosts = {};\n      const hostsDown = [];\n      utils.series([\n        client.connect.bind(client),\n        function (next) {\n          // wait for all initial events to ensure we don\'t incidentally get an \'UP\' event for node 2\n          // after we have killed it.\n          setTimeout(next, 5000);\n        },\n        function warmUpPool(seriesNext) {\n          client.on(\'hostDown\', function (h) {\n            hostsDown.push(h);\n          });\n          utils.times(100, function (n, next) {\n            client.execute(helper.queries.basic, function (err, result) {\n              assert.ifError(err);\n              hosts[result.info.queriedHost] = true;\n              next();\n            });\n          }, seriesNext);\n        },\n        function killNode(seriesNext) {\n          assert.strictEqual(Object.keys(hosts).length, 3);\n          setImmediate(function () {\n            helper.ccmHelper.exec([\'node1\', \'stop\', \'--not-gently\']);\n            seriesNext();\n          });\n        },\n        function executeWhileNodeGoingDown(seriesNext) {\n          utils.timesLimit(1000, 10, function (i, next) {\n            client.execute(helper.queries.basic, next);\n          }, function (err) {\n            assert.ifError(err);\n            //delay the next queries\n            setTimeout(seriesNext, 10000);\n          });\n        },\n        function executeAfterNodeWentDown(seriesNext) {\n          utils.timesSeries(20, function (i, next) {\n            client.execute(helper.queries.basic, next);\n          }, seriesNext);\n        },\n        function assertions(seriesNext) {\n          //Only 2 hosts alive at the end\n          assert.strictEqual(\n            client.hosts.values().reduce(function (val, h) {\n              return val + (h.isUp() ? 1 : 0);\n            }, 0),\n            2);\n          assert.ok(hostsDown.length >= 1, "Expected at least 1 host down" +\n            " event.");\n          //Ensure each down event is for the stopped host.  We may get\n          //multiple down events for the same host on a control connection.\n          hostsDown.forEach(function (downHost) {\n            assert.strictEqual(helper.lastOctetOf(downHost), \'1\');\n          });\n          seriesNext();\n        },\n        client.shutdown.bind(client)\n      ], done);\n    });\n\n    it(\'should failover when a node goes down with some outstanding requests\', function (done) {\n      const options = utils.extend({ queryOptions: { isIdempotent: true } }, helper.baseOptions);\n      options.pooling = {\n        coreConnectionsPerHost: {\n          \'0\': 1,\n          \'1\': 1,\n          \'2\': 0\n        }\n      };\n\n      const client = new Client(options);\n      const hosts = {};\n      const query = helper.queries.basic;\n\n      after(() => client.shutdown());\n\n      utils.series([\n        function (next) {\n          // wait for all initial events to ensure we don\'t incidentally get an \'UP\' event for node 2\n          // after we have killed it.\n          setTimeout(next, 5000);\n        },\n        function warmUpPool(seriesNext) {\n          utils.times(20, function (n, next) {\n            client.execute(query, function (err, result) {\n              assert.ifError(err);\n              hosts[result.info.queriedHost] = true;\n              next();\n            });\n          }, seriesNext);\n        },\n        helper.setIntervalUntilTask(() => {\n          const state = client.getState();\n          return client.hosts.values().reduce((accumulator, h) => accumulator + state.getInFlightQueries(h), 0) === 0;\n        }, 20, 100),\n        function testCase(seriesNext) {\n          // 3 hosts alive\n          assert.strictEqual(Object.keys(hosts).length, 3);\n\n          let killed = false;\n          utils.timesLimit(500, 20, function (n, next) {\n            if (n === 30) {\n              //kill a node when there are some outstanding requests\n              helper.ccmHelper.exec([\'node2\', \'stop\', \'--not-gently\'], function (err) {\n                killed = true;\n                assert.ifError(err);\n                //do a couple of more queries\n                utils.timesSeries(10, function (n, next2) {\n                  client.execute(query, next2);\n                }, next);\n              });\n              return;\n            }\n            if (killed) {\n              //Don\'t issue more requests\n              return next();\n            }\n            client.execute(query, function (err) {\n              assert.ifError(err);\n              next();\n            });\n          }, seriesNext);\n        },\n        next => setImmediate(next),\n        function assertHostState(next) {\n          // Only 2 hosts are UP at the end\n          assert.strictEqual(client.hosts.values().reduce((val, h) => val + (h.isUp() ? 1 : 0), 0), 2);\n\n          // In-flight counter should be 0 for UP and DOWN nodes\n          const state = client.getState();\n          client.hosts.forEach(h => assert.strictEqual(state.getInFlightQueries(h), 0));\n          next();\n        }\n      ], done);\n    });\n\n    it(\'should warn but not fail when warmup is enable and a node is down\', async () => {\n      await util.promisify(helper.ccmHelper.exec)([\'node2\', \'stop\']);\n\n      const warnings = [];\n      const client = newInstance({ pooling: { warmup: true } });\n      helper.afterThisTest(() => client.shutdown());\n\n      client.on(\'log\', (level, className, message) => {\n        if (level !== \'warning\' || className !== \'Client\') {\n          return;\n        }\n\n        warnings.push(message);\n      });\n\n      await client.connect();\n      assert.lengthOf(warnings.filter(w => w.indexOf(\'pool\') >= 0), 1);\n    });\n  });\n\n  describe(\'#shutdown()\', function () {\n    helper.setup(2, { initClient: false });\n\n    it(\'should close all connections to all hosts\', function (done) {\n      const client = newInstance();\n      utils.series([\n        client.connect.bind(client),\n        function makeSomeQueries(next) {\n          //to ensure that the pool is all up!\n          utils.times(100, function (n, timesNext) {\n            client.execute(helper.queries.basic, timesNext);\n          }, next);\n        },\n        function shutDown(next) {\n          const hosts = client.hosts.values();\n          assert.strictEqual(hosts.length, 2);\n          const state = client.getState();\n          // Check the pools before shutting down\n          hosts.forEach(function each(host) {\n            assert.ok(state.getOpenConnections(host) > 0);\n            assert.ok(host.pool.connections.length > 0);\n            assert.ok(!host.pool.shuttingDown);\n          });\n          client.shutdown(next);\n        },\n        function checkPool(next) {\n          const hosts = client.hosts.values();\n          assert.strictEqual(hosts.length, 2);\n          const state = client.getState();\n          assert.deepEqual(state.getConnectedHosts(), []);\n          hosts.forEach(function each(host) {\n            assert.strictEqual(host.pool.connections.length, 0);\n            assert.strictEqual(state.getOpenConnections(host), 0);\n          });\n          next();\n        }\n      ], done);\n    });\n\n    it(\'should not leak any connection when connection pool is still growing\', function (done) {\n      const client = newInstance({ pooling: { coreConnectionsPerHost: { \'0\': 4 }}});\n      utils.series([\n        client.connect.bind(client),\n        function makeSomeQueries(next) {\n          utils.times(10, function (n, timesNext) {\n            client.execute(helper.queries.basic, timesNext);\n          }, next);\n        },\n        function shutDown(next) {\n          const hosts = client.hosts.values();\n          assert.strictEqual(hosts.length, 2);\n          assert.ok(hosts[0].pool.connections.length > 0);\n          assert.ok(!hosts[0].pool.shuttingDown);\n          assert.ok(!hosts[1].pool.shuttingDown);\n          client.shutdown(next);\n        },\n        function checkPoolDelayed(next) {\n          function checkNoConnections() {\n            assert.deepEqual(getPoolInfo(client), { \'1\': 0, \'2\': 0 });\n          }\n          checkNoConnections();\n          // Wait some time and check again to see if there is a new connection created in the background\n          setTimeout(function checkNoConnectionsDelayed() {\n            checkNoConnections();\n            next();\n          }, 1000);\n        }\n      ], done);\n    });\n\n    it(\'should callback after a NoHostAvailableError\', function (done) {\n      const client = newInstance({ contactPoints: [ \'::1\', \'::2\'] });\n      client.connect(function (err) {\n        helper.assertInstanceOf(err, errors.NoHostAvailableError);\n        assert.strictEqual(client.hosts.length, 0);\n        client.shutdown(function (err) {\n          assert.strictEqual(client.hosts.length, 0);\n          assert.ifError(err);\n          done();\n        });\n      });\n    });\n\n    it(\'should close all connections after connecting with an invalid keyspace\', function (done) {\n      const client = newInstance({ keyspace: \'KS_DOES_NOT_EXIST\' });\n      client.connect(function (err) {\n        helper.assertInstanceOf(err, errors.ResponseError);\n        assert.strictEqual(client.hosts.length, 0);\n        client.shutdown(function (err) {\n          assert.ifError(err);\n          assert.strictEqual(client.hosts.length, 0);\n          done();\n        });\n      });\n    });\n  });\n});\n\n/**\n * @param {ClientOptions} [options]\n * @returns {Client}\n */\nfunction newInstance(options) {\n  const client = new Client(utils.deepExtend({}, helper.baseOptions, options));\n  helper.shutdownAfterThisTest(client);\n  return client;\n}\n\n/**\n * Returns a dictionary containing the last octet of the address as keys and the pool size as values.\n */\nfunction getPoolInfo(client) {\n  const info = {};\n  client.hosts.forEach(function (h, address) {\n    info[helper.lastOctetOf(address)] = h.pool.connections.length;\n  });\n  return info;\n}\n\nfunction createRole(client, role, password) {\n  return client.execute(`CREATE ROLE IF NOT EXISTS ${role} WITH PASSWORD = \'${password}\' AND LOGIN = true`);\n}\n\nfunction assertAuthError(err, message) {\n  helper.assertInstanceOf(err, errors.NoHostAvailableError);\n  assert.ok(err.innerErrors);\n  const firstErr = Object.values(err.innerErrors)[0];\n  helper.assertInstanceOf(firstErr, errors.AuthenticationError);\n\n  if (message) {\n    assert.match(firstErr.message, message);\n  }\n}\n', 'message': "Merge branch 'master' of https://github.com/MCProHosting/nodejs-driver\n", 'author_name': 'Jorge Bay Gondra', 'lang': 'js'}