Updated test

test/unit/connection.test.js

@@ -266,6 +266,64 @@ test('Body request', t => {
   })
 })
 
+test('Send body as buffer', t => {
+  t.plan(2)
+
+  function handler (req, res) {
+    var payload = ''
+    req.setEncoding('utf8')
+    req.on('data', chunk => { payload += chunk })
+    req.on('error', err => t.fail(err))
+    req.on('end', () => {
+      t.strictEqual(payload, 'hello')
+      res.end('ok')
+    })
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const connection = new Connection({
+      url: new URL(`http://localhost:${port}`)
+    })
+    connection.request({
+      path: '/hello',
+      method: 'POST',
+      body: Buffer.from('hello')
+    }, (err, res) => {
+      t.error(err)
+      server.stop()
+    })
+  })
+})
+
+test('Send body as stream', t => {
+  t.plan(2)
+
+  function handler (req, res) {
+    var payload = ''
+    req.setEncoding('utf8')
+    req.on('data', chunk => { payload += chunk })
+    req.on('error', err => t.fail(err))
+    req.on('end', () => {
+      t.strictEqual(payload, 'hello')
+      res.end('ok')
+    })
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const connection = new Connection({
+      url: new URL(`http://localhost:${port}`)
+    })
+    connection.request({
+      path: '/hello',
+      method: 'POST',
+      body: intoStream('hello')
+    }, (err, res) => {
+      t.error(err)
+      server.stop()
+    })
+  })
+})
+
 test('Should handle compression', t => {
   t.test('gzip', t => {
     t.plan(3)
@@ -494,3 +552,11 @@ test('asStream set to true', t => {
     })
   })
 })
+
+test('Connection id should not contain credentials', t => {
+  const connection = new Connection({
+    url: new URL('http://user:password@localhost:9200')
+  })
+  t.strictEqual(connection.id, 'http://localhost:9200/')
+  t.end()
+})

test/unit/transport.test.js

@@ -2,6 +2,7 @@
 
 const { test } = require('tap')
 const { URL } = require('url')
+const intoStream = require('into-stream')
 const {
   buildServer,
   connection: { MockConnection, MockConnectionTimeout, MockConnectionError }
@@ -152,6 +153,92 @@ test('Send POST (ndjson)', t => {
   })
 })
 
+test('Send stream', t => {
+  t.plan(4)
+  function handler (req, res) {
+    t.match(req.headers, {
+      'content-type': 'application/json'
+    })
+    var json = ''
+    req.setEncoding('utf8')
+    req.on('data', chunk => { json += chunk })
+    req.on('error', err => t.fail(err))
+    req.on('end', () => {
+      t.deepEqual(JSON.parse(json), { hello: 'world' })
+      res.setHeader('Content-Type', 'application/json;utf=8')
+      res.end(JSON.stringify({ hello: 'world' }))
+    })
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const pool = new ConnectionPool({ Connection })
+    pool.addConnection(`http://localhost:${port}`)
+
+    const transport = new Transport({
+      emit: () => {},
+      connectionPool: pool,
+      serializer: new Serializer(),
+      maxRetries: 3,
+      requestTimeout: 30000,
+      sniffInterval: false,
+      sniffOnStart: false
+    })
+
+    transport.request({
+      method: 'POST',
+      path: '/hello',
+      body: intoStream(JSON.stringify({ hello: 'world' }))
+    }, (err, { body }) => {
+      t.error(err)
+      t.deepEqual(body, { hello: 'world' })
+      server.stop()
+    })
+  })
+})
+
+test('Send stream (bulkBody)', t => {
+  t.plan(4)
+  function handler (req, res) {
+    t.match(req.headers, {
+      'content-type': 'application/x-ndjson'
+    })
+    var json = ''
+    req.setEncoding('utf8')
+    req.on('data', chunk => { json += chunk })
+    req.on('error', err => t.fail(err))
+    req.on('end', () => {
+      t.deepEqual(JSON.parse(json), { hello: 'world' })
+      res.setHeader('Content-Type', 'application/json;utf=8')
+      res.end(JSON.stringify({ hello: 'world' }))
+    })
+  }
+
+  buildServer(handler, ({ port }, server) => {
+    const pool = new ConnectionPool({ Connection })
+    pool.addConnection(`http://localhost:${port}`)
+
+    const transport = new Transport({
+      emit: () => {},
+      connectionPool: pool,
+      serializer: new Serializer(),
+      maxRetries: 3,
+      requestTimeout: 30000,
+      sniffInterval: false,
+      sniffOnStart: false
+    })
+
+    transport.request({
+      method: 'POST',
+      path: '/hello',
+      bulkBody: intoStream(JSON.stringify({ hello: 'world' }))
+    }, (err, { body }) => {
+      t.error(err)
+      t.deepEqual(body, { hello: 'world' })
+      server.stop()
+    })
+  })
+})
+
 test('Not JSON payload from server', t => {
   t.plan(2)
   function handler (req, res) {