Updated test

test/integration/index.js

@@ -8,7 +8,7 @@ const Git = require('simple-git')
 const ora = require('ora')
 const minimist = require('minimist')
 const tap = require('tap')
-const elasticsearch = require('../../index')
+const { Client } = require('../../index')
 const TestRunner = require('./test-runner')
 
 const esRepo = 'https://github.com/elastic/elasticsearch.git'
@@ -24,9 +24,7 @@ function Runner (opts) {
 
   assert(opts.node, 'Missing base node')
   this.bailout = opts.bailout
-  this.client = new elasticsearch.Client({
+  this.client = new Client({ node: opts.node })
-    node: opts.node
-  })
   this.log = ora('Loading yaml suite').start()
 }
 
@@ -239,7 +237,7 @@ if (require.main === module) {
     default: {
       // node: 'http://elastic:passw0rd@localhost:9200',
       node: process.env.TEST_ES_SERVER || 'http://localhost:9200',
-      version: '6.4',
+      version: '6.5',
       bailout: false
     }
   })

test/unit/connection-pool.test.js

@@ -4,11 +4,11 @@ const { test } = require('tap')
 const { URL } = require('url')
 const ConnectionPool = require('../../lib/ConnectionPool')
 const Connection = require('../../lib/Connection')
-const { buildServer } = require('../utils')
+const { connection: { MockConnection, MockConnectionTimeout } } = require('../utils')
 
 test('API', t => {
   t.test('addConnection', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const href = 'http://localhost:9200/'
     pool.addConnection(href)
     t.ok(pool.connections.get(href) instanceof Connection)
@@ -18,7 +18,7 @@ test('API', t => {
   })
 
   t.test('addConnection should throw with two connections with the same id', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const href = 'http://localhost:9200/'
     pool.addConnection(href)
     try {
@@ -31,7 +31,7 @@ test('API', t => {
   })
 
   t.test('markDead', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const href = 'http://localhost:9200/'
     var connection = pool.addConnection(href)
     pool.markDead(connection)
@@ -43,7 +43,7 @@ test('API', t => {
   })
 
   t.test('markDead should sort the dead queue by deadTimeout', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const href1 = 'http://localhost:9200/1'
     const href2 = 'http://localhost:9200/2'
     const conn1 = pool.addConnection(href1)
@@ -57,7 +57,7 @@ test('API', t => {
   })
 
   t.test('markAlive', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const href = 'http://localhost:9200/'
     var connection = pool.addConnection(href)
     pool.markDead(connection)
@@ -73,16 +73,12 @@ test('API', t => {
   t.test('resurrect', t => {
     t.test('ping strategy', t => {
       t.test('alive', t => {
-        function handler (req, res) {
-          res.end()
-        }
-
-        buildServer(handler, ({ port }, server) => {
         const pool = new ConnectionPool({
           resurrectStrategy: 'ping',
-            pingTimeout: 3000
+          pingTimeout: 3000,
+          Connection: MockConnection
         })
-          const href = `http://localhost:${port}/`
+        const href = 'http://localhost:9200/'
         var connection = pool.addConnection(href)
         pool.markDead(connection)
         pool.resurrect(Date.now() + 1000 * 60 * 3, (isAlive, connection) => {
@@ -92,20 +88,17 @@ test('API', t => {
           t.strictEqual(connection.resurrectTimeout, 0)
           t.strictEqual(connection.status, Connection.statuses.ALIVE)
           t.deepEqual(pool.dead, [])
-            server.stop()
           t.end()
         })
       })
-      })
 
       t.test('dead', t => {
-        buildServer(() => {}, ({ port }, server) => {
-          server.close()
         const pool = new ConnectionPool({
           resurrectStrategy: 'ping',
-            pingTimeout: 3000
+          pingTimeout: 3000,
+          Connection: MockConnectionTimeout
         })
-          const href = `http://localhost:${port}/`
+        const href = 'http://localhost:9200/'
         var connection = pool.addConnection(href)
         pool.markDead(connection)
         pool.resurrect(Date.now() + 1000 * 60 * 3, (isAlive, connection) => {
@@ -115,18 +108,17 @@ test('API', t => {
           t.true(connection.resurrectTimeout > 0)
           t.strictEqual(connection.status, Connection.statuses.DEAD)
           t.deepEqual(pool.dead, [href])
-            server.stop()
           t.end()
         })
       })
-      })
 
       t.end()
     })
 
     t.test('optimistic strategy', t => {
       const pool = new ConnectionPool({
-        resurrectStrategy: 'optimistic'
+        resurrectStrategy: 'optimistic',
+        Connection
       })
       const href = 'http://localhost:9200/'
       var connection = pool.addConnection(href)
@@ -144,7 +136,8 @@ test('API', t => {
 
     t.test('none strategy', t => {
       const pool = new ConnectionPool({
-        resurrectStrategy: 'none'
+        resurrectStrategy: 'none',
+        Connection
       })
       const href = 'http://localhost:9200/'
       var connection = pool.addConnection(href)
@@ -166,7 +159,7 @@ test('API', t => {
 
   t.test('getConnection', t => {
     t.test('Should return a connection', t => {
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       const href = 'http://localhost:9200/'
       pool.addConnection(href)
       t.ok(pool.getConnection() instanceof Connection)
@@ -174,7 +167,7 @@ test('API', t => {
     })
 
     t.test('filter option', t => {
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       const href1 = 'http://localhost:9200/'
       const href2 = 'http://localhost:9200/other'
       pool.addConnection([href1, href2])
@@ -184,9 +177,9 @@ test('API', t => {
       t.end()
     })
 
-    t.test('filter and weighter should get Connection objects', t => {
+    t.test('filter should get Connection objects', t => {
       t.plan(2)
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       const href1 = 'http://localhost:9200/'
       const href2 = 'http://localhost:9200/other'
       pool.addConnection([href1, href2])
@@ -200,7 +193,7 @@ test('API', t => {
 
     t.test('filter should get alive connections', t => {
       t.plan(2)
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       const href1 = 'http://localhost:9200/'
       const href2 = 'http://localhost:9200/other'
       const conn = pool.addConnection(href1)
@@ -220,6 +213,7 @@ test('API', t => {
       const href1 = 'http://localhost:9200/'
       const href2 = 'http://localhost:9200/other'
       const pool = new ConnectionPool({
+        Connection,
         nodeFilter: node => {
           t.ok('called')
           return true
@@ -233,7 +227,7 @@ test('API', t => {
   })
 
   t.test('removeConnection', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const href = 'http://localhost:9200/'
     var connection = pool.addConnection(href)
     t.ok(pool.getConnection() instanceof Connection)
@@ -243,7 +237,7 @@ test('API', t => {
   })
 
   t.test('empty', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection('http://localhost:9200/')
     pool.addConnection('http://localhost:9201/')
     pool.empty()
@@ -253,7 +247,7 @@ test('API', t => {
   })
 
   t.test('urlToHost', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const url = 'http://localhost:9200'
     t.deepEqual(
       pool.urlToHost(url),
@@ -263,7 +257,7 @@ test('API', t => {
   })
 
   t.test('nodesToHost', t => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     const nodes = {
       a1: {
         http: {
@@ -307,7 +301,7 @@ test('API', t => {
           t.fail('Should not be called')
         }
       }
-      const pool = new CustomConnectionPool()
+      const pool = new CustomConnectionPool({ Connection })
       pool.addConnection([{
         url: new URL('http://127.0.0.1:9200'),
         id: 'a1',
@@ -348,7 +342,7 @@ test('API', t => {
           super.markAlive(connection)
         }
       }
-      const pool = new CustomConnectionPool()
+      const pool = new CustomConnectionPool({ Connection })
       const conn1 = pool.addConnection({
         url: new URL('http://127.0.0.1:9200'),
         id: 'a1',
@@ -388,7 +382,7 @@ test('API', t => {
 
     t.test('Add a new connection', t => {
       t.plan(2)
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       pool.addConnection({
         url: new URL('http://127.0.0.1:9200'),
         id: 'a1',
@@ -415,7 +409,7 @@ test('API', t => {
 
     t.test('Remove old connections', t => {
       t.plan(3)
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       pool.addConnection({
         url: new URL('http://127.0.0.1:9200'),
         id: 'a1',

test/unit/events.test.js

@@ -0,0 +1,106 @@
+'use strict'
+
+const { test } = require('tap')
+const { Client, events } = require('../../index')
+const { TimeoutError } = require('../../lib/errors')
+const { connection: { MockConnection, MockConnectionTimeout } } = require('../utils')
+
+test('Should emit a request event when a request is performed', t => {
+  t.plan(3)
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection
+  })
+
+  client.on(events.REQUEST, (connection, request) => {
+    t.match(connection, {
+      id: 'http://localhost:9200'
+    })
+    t.match(request, {
+      method: 'GET',
+      path: '/test/doc/_search',
+      querystring: 'q=foo%3Abar'
+    })
+  })
+
+  client.search({
+    index: 'test',
+    type: 'doc',
+    q: 'foo:bar'
+  }, (err, result) => {
+    t.error(err)
+  })
+})
+
+test('Should emit a response event in case of a successful response', t => {
+  t.plan(4)
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection
+  })
+
+  client.on(events.RESPONSE, (connection, request, response) => {
+    t.match(connection, {
+      id: 'http://localhost:9200'
+    })
+    t.match(request, {
+      method: 'GET',
+      path: '/test/doc/_search',
+      querystring: 'q=foo%3Abar'
+    })
+    t.match(response, {
+      body: { hello: 'world' },
+      statusCode: 200,
+      headers: {
+        'content-type': 'application/json;utf=8',
+        'connection': 'keep-alive'
+      },
+      warnings: null
+    })
+  })
+
+  client.search({
+    index: 'test',
+    type: 'doc',
+    q: 'foo:bar'
+  }, (err, result) => {
+    t.error(err)
+  })
+})
+
+test('Should emit an error event in case of a failing response', t => {
+  t.plan(4)
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnectionTimeout,
+    maxRetries: 0
+  })
+
+  client.on(events.RESPONSE, (connection, request, response) => {
+    t.fail('This should not be called')
+  })
+
+  client.on(events.ERROR, (error, connection, request) => {
+    t.ok(error instanceof TimeoutError)
+    t.match(connection, {
+      id: 'http://localhost:9200'
+    })
+    t.match(request, {
+      method: 'GET',
+      path: '/test/doc/_search',
+      querystring: 'q=foo%3Abar'
+    })
+  })
+
+  client.search({
+    index: 'test',
+    type: 'doc',
+    q: 'foo:bar',
+    requestTimeout: 500
+  }, (err, result) => {
+    t.ok(err instanceof TimeoutError)
+  })
+})

test/unit/transport.test.js

@@ -2,7 +2,10 @@
 
 const { test } = require('tap')
 const { URL } = require('url')
-const { buildServer } = require('../utils')
+const {
+  buildServer,
+  connection: { MockConnection, MockConnectionTimeout, MockConnectionError }
+} = require('../utils')
 const {
   NoLivingConnectionsError,
   SerializationError,
@@ -13,6 +16,7 @@ const {
 } = require('../../lib/errors')
 
 const ConnectionPool = require('../../lib/ConnectionPool')
+const Connection = require('../../lib/Connection')
 const Serializer = require('../../lib/Serializer')
 const Transport = require('../../lib/Transport')
 
@@ -24,7 +28,7 @@ test('Basic', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -67,7 +71,7 @@ test('Send POST', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -123,7 +127,7 @@ test('Send POST (ndjson)', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -156,7 +160,7 @@ test('Not JSON payload from server', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -182,7 +186,7 @@ test('Not JSON payload from server', t => {
 
 test('NoLivingConnectionsError', t => {
   t.plan(1)
-  const pool = new ConnectionPool()
+  const pool = new ConnectionPool({ Connection })
 
   const transport = new Transport({
     emit: () => {},
@@ -204,7 +208,7 @@ test('NoLivingConnectionsError', t => {
 
 test('SerializationError', t => {
   t.plan(1)
-  const pool = new ConnectionPool()
+  const pool = new ConnectionPool({ Connection })
   pool.addConnection('http://localhost:9200')
 
   const transport = new Transport({
@@ -236,7 +240,7 @@ test('DeserializationError', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -269,17 +273,9 @@ test('TimeoutError (should call markDead on the failing connection)', t => {
     }
   }
 
-  function handler (req, res) {
+  const pool = new CustomConnectionPool({ Connection: MockConnectionTimeout })
-    setTimeout(() => {
-      res.setHeader('Content-Type', 'application/json;utf=8')
-      res.end(JSON.stringify({ hello: 'world' }))
-    }, 1000)
-  }
-
-  buildServer(handler, ({ port }, server) => {
-    const pool = new CustomConnectionPool()
   pool.addConnection({
-      url: new URL(`http://localhost:${port}`),
+    url: new URL('http://localhost:9200'),
     id: 'node1'
   })
 
@@ -298,8 +294,6 @@ test('TimeoutError (should call markDead on the failing connection)', t => {
     path: '/hello'
   }, (err, { body }) => {
     t.ok(err instanceof TimeoutError)
-      server.stop()
-    })
   })
 })
 
@@ -313,11 +307,9 @@ test('ConnectionError (should call markDead on the failing connection)', t => {
     }
   }
 
-  buildServer(() => {}, ({ port }, server) => {
+  const pool = new CustomConnectionPool({ Connection: MockConnectionError })
-    server.close()
-    const pool = new CustomConnectionPool()
   pool.addConnection({
-      url: new URL(`http://localhost:${port}`),
+    url: new URL('http://localhost:9200'),
     id: 'node1'
   })
 
@@ -336,8 +328,6 @@ test('ConnectionError (should call markDead on the failing connection)', t => {
     path: '/hello'
   }, (err, { body }) => {
     t.ok(err instanceof ConnectionError)
-      server.stop()
-    })
   })
 })
 
@@ -358,7 +348,7 @@ test('Retry mechanism', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection([{
       url: new URL(`http://localhost:${port}`),
       id: 'node1'
@@ -401,15 +391,9 @@ test('Should call markAlive with a successful response', t => {
     }
   }
 
-  function handler (req, res) {
+  const pool = new CustomConnectionPool({ Connection: MockConnection })
-    res.setHeader('Content-Type', 'application/json;utf=8')
-    res.end(JSON.stringify({ hello: 'world' }))
-  }
-
-  buildServer(handler, ({ port }, server) => {
-    const pool = new CustomConnectionPool()
   pool.addConnection({
-      url: new URL(`http://localhost:${port}`),
+    url: new URL('http://localhost:9200'),
     id: 'node1'
   })
 
@@ -429,8 +413,6 @@ test('Should call markAlive with a successful response', t => {
   }, (err, { body }) => {
     t.error(err)
     t.deepEqual(body, { hello: 'world' })
-      server.stop()
-    })
   })
 })
 
@@ -443,15 +425,9 @@ test('Should call resurrect on every request', t => {
     }
   }
 
-  function handler (req, res) {
+  const pool = new CustomConnectionPool({ Connection: MockConnection })
-    res.setHeader('Content-Type', 'application/json;utf=8')
-    res.end(JSON.stringify({ hello: 'world' }))
-  }
-
-  buildServer(handler, ({ port }, server) => {
-    const pool = new CustomConnectionPool()
   pool.addConnection({
-      url: new URL(`http://localhost:${port}`),
+    url: new URL('http://localhost:9200'),
     id: 'node1'
   })
 
@@ -471,25 +447,15 @@ test('Should call resurrect on every request', t => {
   }, (err, { body }) => {
     t.error(err)
     t.deepEqual(body, { hello: 'world' })
-      server.stop()
-    })
   })
 })
 
 test('Should return a request aborter utility', t => {
   t.plan(1)
 
-  function handler (req, res) {
+  const pool = new ConnectionPool({ Connection, MockConnection })
-    setTimeout(() => {
-      res.setHeader('Content-Type', 'application/json;utf=8')
-      res.end(JSON.stringify({ hello: 'world' }))
-    }, 1000)
-  }
-
-  buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
   pool.addConnection({
-      url: new URL(`http://localhost:${port}`),
+    url: new URL('http://localhost:9200'),
     id: 'node1'
   })
 
@@ -511,9 +477,7 @@ test('Should return a request aborter utility', t => {
   })
 
   request.abort()
-    server.stop()
   t.pass('ok')
-  })
 })
 
 test('ResponseError', t => {
@@ -526,7 +490,7 @@ test('ResponseError', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -561,7 +525,7 @@ test('Override requestTimeout', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -609,7 +573,7 @@ test('sniff', t => {
     }
 
     buildServer(handler, ({ port }, server) => {
-      const pool = new CustomConnectionPool()
+      const pool = new CustomConnectionPool({ Connection })
       pool.addConnection(`http://localhost:${port}`)
 
       // eslint-disable-next-line
@@ -653,7 +617,7 @@ test('sniff', t => {
     }
 
     buildServer(handler, ({ port }, server) => {
-      const pool = new CustomConnectionPool()
+      const pool = new CustomConnectionPool({ Connection })
       pool.addConnection(`http://localhost:${port}`)
       pool.addConnection(`http://localhost:${port}/other`)
 
@@ -700,7 +664,7 @@ test('sniff', t => {
     }
 
     buildServer(handler, ({ port }, server) => {
-      const pool = new CustomConnectionPool()
+      const pool = new CustomConnectionPool({ Connection })
       pool.addConnection(`http://localhost:${port}`)
 
       const transport = new Transport({
@@ -741,10 +705,8 @@ test('sniff', t => {
       }
     }
 
-    buildServer(() => {}, ({ port }, server) => {
+    const pool = new CustomConnectionPool({ Connection: MockConnectionError })
-      server.close()
+    pool.addConnection('http://localhost:9200')
-      const pool = new CustomConnectionPool()
-      pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
       emit: () => {},
@@ -758,8 +720,6 @@ test('sniff', t => {
 
     transport.sniff((err, hosts) => {
       t.ok(err instanceof ConnectionError)
-        server.stop()
-      })
     })
   })
 
@@ -773,11 +733,6 @@ test(`Should mark as dead connections where the statusCode is 502/3/4
   function runTest (statusCode) {
     t.test(statusCode, t => {
       t.plan(3)
-      function handler (req, res) {
-        res.statusCode = statusCode
-        res.setHeader('Content-Type', 'application/json;utf=8')
-        res.end(JSON.stringify({ hello: 'world' }))
-      }
 
       class CustomConnectionPool extends ConnectionPool {
         markDead (connection) {
@@ -786,9 +741,8 @@ test(`Should mark as dead connections where the statusCode is 502/3/4
         }
       }
 
-      buildServer(handler, ({ port }, server) => {
+      const pool = new CustomConnectionPool({ Connection: MockConnection })
-        const pool = new CustomConnectionPool()
+      pool.addConnection('http://localhost:9200')
-        pool.addConnection(`http://localhost:${port}`)
 
       const transport = new Transport({
         emit: () => {},
@@ -802,7 +756,7 @@ test(`Should mark as dead connections where the statusCode is 502/3/4
 
       transport.request({
         method: 'GET',
-          path: '/hello'
+        path: `/${statusCode}`
       }, (err, { body }) => {
         t.ok(err instanceof ResponseError)
         t.match(err, {
@@ -810,8 +764,6 @@ test(`Should mark as dead connections where the statusCode is 502/3/4
           headers: { 'content-type': 'application/json;utf=8' },
           statusCode: statusCode
         })
-          server.stop()
-        })
       })
     })
   }
@@ -843,7 +795,7 @@ test('Should retry the request if the statusCode is 502/3/4', t => {
       }
 
       buildServer(handler, ({ port }, server) => {
-        const pool = new CustomConnectionPool()
+        const pool = new CustomConnectionPool({ Connection })
         pool.addConnection(`http://localhost:${port}`)
 
         const transport = new Transport({
@@ -873,15 +825,9 @@ test('Should retry the request if the statusCode is 502/3/4', t => {
 
 test('Ignore status code', t => {
   t.plan(4)
-  function handler (req, res) {
-    res.statusCode = 404
-    res.setHeader('Content-Type', 'application/json;utf=8')
-    res.end(JSON.stringify({ hello: 'world' }))
-  }
 
-  buildServer(handler, ({ port }, server) => {
+  const pool = new ConnectionPool({ Connection: MockConnection })
-    const pool = new ConnectionPool()
+  pool.addConnection('http://localhost:9200')
-    pool.addConnection(`http://localhost:${port}`)
 
   const transport = new Transport({
     emit: () => {},
@@ -895,7 +841,7 @@ test('Ignore status code', t => {
 
   transport.request({
     method: 'GET',
-      path: '/hello',
+    path: '/404',
     ignore: [404]
   }, (err, { body }) => {
     t.error(err)
@@ -904,21 +850,18 @@ test('Ignore status code', t => {
 
   transport.request({
     method: 'GET',
-      path: '/hello'
+    path: '/404'
   }, (err, { body }) => {
     t.ok(err instanceof ResponseError)
   })
 
   transport.request({
     method: 'GET',
-      path: '/hello',
+    path: '/404',
     ignore: [403, 405]
   }, (err, { body }) => {
     t.ok(err instanceof ResponseError)
   })
-
-    setTimeout(() => server.stop(), 100)
-  })
 })
 
 test('Should serialize the querystring', t => {
@@ -930,7 +873,7 @@ test('Should serialize the querystring', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -970,7 +913,7 @@ test('timeout option', t => {
       t.plan(1)
 
       buildServer(handler, ({ port }, server) => {
-        const pool = new ConnectionPool()
+        const pool = new ConnectionPool({ Connection })
         pool.addConnection({
           url: new URL(`http://localhost:${port}`),
           id: 'node1'
@@ -1000,7 +943,7 @@ test('timeout option', t => {
       t.plan(1)
 
       buildServer(handler, ({ port }, server) => {
-        const pool = new ConnectionPool()
+        const pool = new ConnectionPool({ Connection })
         pool.addConnection({
           url: new URL(`http://localhost:${port}`),
           id: 'node1'
@@ -1035,7 +978,7 @@ test('timeout option', t => {
       t.plan(1)
 
       buildServer(handler, ({ port }, server) => {
-        const pool = new ConnectionPool()
+        const pool = new ConnectionPool({ Connection })
         pool.addConnection({
           url: new URL(`http://localhost:${port}`),
           id: 'node1'
@@ -1065,7 +1008,7 @@ test('timeout option', t => {
       t.plan(1)
 
       buildServer(handler, ({ port }, server) => {
-        const pool = new ConnectionPool()
+        const pool = new ConnectionPool({ Connection })
         pool.addConnection({
           url: new URL(`http://localhost:${port}`),
           id: 'node1'
@@ -1100,15 +1043,9 @@ test('timeout option', t => {
 
 test('Should cast to boolean HEAD request', t => {
   t.test('2xx response', t => {
-    t.plan(2)
+    t.plan(3)
-    function handler (req, res) {
+    const pool = new ConnectionPool({ Connection: MockConnection })
-      res.setHeader('Content-Type', 'application/json;utf=8')
+    pool.addConnection('http://localhost:9200')
-      res.end('')
-    }
-
-    buildServer(handler, ({ port }, server) => {
-      const pool = new ConnectionPool()
-      pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
       emit: () => {},
@@ -1122,26 +1059,18 @@ test('Should cast to boolean HEAD request', t => {
 
     transport.request({
       method: 'HEAD',
-        path: '/hello'
+      path: '/200'
-      }, (err, { body }) => {
+    }, (err, { body, statusCode }) => {
       t.error(err)
+      t.strictEqual(statusCode, 200)
       t.strictEqual(body, true)
-        server.stop()
-      })
     })
   })
 
   t.test('404 response', t => {
-    t.plan(2)
+    t.plan(3)
-    function handler (req, res) {
+    const pool = new ConnectionPool({ Connection: MockConnection })
-      res.statusCode = 404
+    pool.addConnection('http://localhost:9200')
-      res.setHeader('Content-Type', 'application/json;utf=8')
-      res.end('')
-    }
-
-    buildServer(handler, ({ port }, server) => {
-      const pool = new ConnectionPool()
-      pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
       emit: () => {},
@@ -1155,26 +1084,19 @@ test('Should cast to boolean HEAD request', t => {
 
     transport.request({
       method: 'HEAD',
-        path: '/hello'
+      path: '/404'
-      }, (err, { body }) => {
+    }, (err, { body, statusCode }) => {
       t.error(err)
+      t.strictEqual(statusCode, 404)
       t.strictEqual(body, false)
-        server.stop()
-      })
     })
   })
 
   t.test('4xx response', t => {
-    t.plan(1)
+    t.plan(2)
-    function handler (req, res) {
-      res.statusCode = 400
-      res.setHeader('Content-Type', 'application/json;utf=8')
-      res.end(JSON.stringify({ hello: 'world' }))
-    }
 
-    buildServer(handler, ({ port }, server) => {
+    const pool = new ConnectionPool({ Connection: MockConnection })
-      const pool = new ConnectionPool()
+    pool.addConnection('http://localhost:9200')
-      pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
       emit: () => {},
@@ -1188,25 +1110,17 @@ test('Should cast to boolean HEAD request', t => {
 
     transport.request({
       method: 'HEAD',
-        path: '/hello'
+      path: '/400'
-      }, (err, { body }) => {
+    }, (err, { body, statusCode }) => {
       t.ok(err instanceof ResponseError)
-        server.stop()
+      t.strictEqual(statusCode, 400)
-      })
     })
   })
 
   t.test('5xx response', t => {
-    t.plan(1)
+    t.plan(2)
-    function handler (req, res) {
+    const pool = new ConnectionPool({ Connection: MockConnection })
-      res.statusCode = 500
+    pool.addConnection('http://localhost:9200')
-      res.setHeader('Content-Type', 'application/json;utf=8')
-      res.end(JSON.stringify({ hello: 'world' }))
-    }
-
-    buildServer(handler, ({ port }, server) => {
-      const pool = new ConnectionPool()
-      pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
       emit: () => {},
@@ -1220,11 +1134,10 @@ test('Should cast to boolean HEAD request', t => {
 
     transport.request({
       method: 'HEAD',
-        path: '/hello'
+      path: '/500'
-      }, (err, { body }) => {
+    }, (err, { body, statusCode }) => {
       t.ok(err instanceof ResponseError)
-        server.stop()
+      t.strictEqual(statusCode, 500)
-      })
     })
   })
 
@@ -1242,7 +1155,7 @@ test('Suggest compression', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({
@@ -1278,7 +1191,7 @@ test('Warning header', t => {
     }
 
     buildServer(handler, ({ port }, server) => {
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       pool.addConnection(`http://localhost:${port}`)
 
       const transport = new Transport({
@@ -1315,7 +1228,7 @@ test('Warning header', t => {
     }
 
     buildServer(handler, ({ port }, server) => {
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       pool.addConnection(`http://localhost:${port}`)
 
       const transport = new Transport({
@@ -1349,7 +1262,7 @@ test('Warning header', t => {
     }
 
     buildServer(handler, ({ port }, server) => {
-      const pool = new ConnectionPool()
+      const pool = new ConnectionPool({ Connection })
       pool.addConnection(`http://localhost:${port}`)
 
       const transport = new Transport({
@@ -1384,7 +1297,7 @@ test('asStream set to true', t => {
   }
 
   buildServer(handler, ({ port }, server) => {
-    const pool = new ConnectionPool()
+    const pool = new ConnectionPool({ Connection })
     pool.addConnection(`http://localhost:${port}`)
 
     const transport = new Transport({

test/utils/MockConnection.js

@@ -0,0 +1,65 @@
+'use strict'
+
+const { Connection } = require('../../index')
+const { TimeoutError } = require('../../lib/errors')
+const intoStream = require('into-stream')
+
+class MockConnection extends Connection {
+  request (params, callback) {
+    var aborted = false
+    const stream = intoStream(JSON.stringify({ hello: 'world' }))
+    stream.statusCode = setStatusCode(params.path)
+    stream.headers = {
+      'content-type': 'application/json;utf=8',
+      'date': new Date().toISOString(),
+      'connection': 'keep-alive',
+      'content-length': '17'
+    }
+    process.nextTick(() => {
+      if (!aborted) {
+        callback(null, stream)
+      }
+    })
+    return {
+      abort: () => { aborted = true }
+    }
+  }
+}
+
+class MockConnectionTimeout extends Connection {
+  request (params, callback) {
+    var aborted = false
+    process.nextTick(() => {
+      if (!aborted) {
+        callback(new TimeoutError('Request timed out', params), null)
+      }
+    })
+    return {
+      abort: () => { aborted = true }
+    }
+  }
+}
+
+class MockConnectionError extends Connection {
+  request (params, callback) {
+    var aborted = false
+    process.nextTick(() => {
+      if (!aborted) {
+        callback(new Error('Kaboom'), null)
+      }
+    })
+    return {
+      abort: () => { aborted = true }
+    }
+  }
+}
+
+function setStatusCode (path) {
+  const statusCode = Number(path.slice(1))
+  if (Number.isInteger(statusCode)) {
+    return statusCode
+  }
+  return 200
+}
+
+module.exports = { MockConnection, MockConnectionTimeout, MockConnectionError }

test/utils/index.js

@@ -1,7 +1,9 @@
 'use strict'
 
 const buildServer = require('./buildServer')
+const connection = require('./MockConnection')
 
 module.exports = {
-  buildServer
+  buildServer,
+  connection
 }