const {
  createConnectionPool,
  connectionOpts,
  saslSCRAM256ConnectionOpts,
  newLogger,
  testIfKafkaAtLeast_1_1_0,
  describeIfOauthbearerDisabled,
  saslEntries,
} = require('testHelpers')

const Long = require('../../utils/long')
const Broker = require('../index')

describe('Broker > connect', () => {
  let broker, connectionPool

  beforeEach(() => {
    connectionPool = createConnectionPool(connectionOpts())
    broker = new Broker({ connectionPool, logger: newLogger() })
  })

  afterEach(async () => {
    broker && (await broker.disconnect())
  })

  test('establish the connection', async () => {
    await broker.connect()
    expect(broker.connectionPool.isConnected()).toEqual(true)
  })

  test('load api versions if not provided', async () => {
    expect(broker.versions).toEqual(null)
    await broker.connect()
    expect(broker.versions).toBeTruthy()
  })

  for (const e of saslEntries) {
    test(`authenticate with SASL ${e.name} if configured`, async () => {
      broker = new Broker({
        connectionPool: createConnectionPool(e.opts()),
        logger: newLogger(),
      })
      expect(broker.isConnected()).toEqual(false)
      await broker.connect()
      expect(broker.isConnected()).toEqual(true)
    })
  }

  describeIfOauthbearerDisabled('when SASL SCRAM is configured', () => {
    test('parallel calls to connect using SCRAM', async () => {
      broker = new Broker({
        connectionPool: createConnectionPool(saslSCRAM256ConnectionOpts()),
        logger: newLogger(),
      })

      expect(broker.isConnected()).toEqual(false)

      await Promise.all([
        broker.connect(),
        broker.connect(),
        broker.connect(),
        broker.connect(),
        broker.connect(),
      ])

      expect(broker.isConnected()).toEqual(true)
    })
  })

  describe('#isConnected', () => {
    test('returns false when not connected', () => {
      expect(broker.isConnected()).toEqual(false)
    })

    for (const e of saslEntries) {
      test(`returns false when connected but not authenticated on connections with SASL ${e.name}`, async () => {
        const connectionPool = createConnectionPool(e.opts())
        broker = new Broker({ connectionPool, logger: newLogger() })
        expect(broker.isConnected()).toEqual(false)
        await connectionPool.getConnection()
        expect(broker.isConnected()).toEqual(false)
      })
    }

    test('returns true when connected', async () => {
      await broker.connect()
      expect(broker.isConnected()).toEqual(true)
    })

    describe('when SaslAuthenticate protocol is available', () => {
      for (const e of saslEntries) {
        test(`returns true when connected and authenticated on connections with SASL ${e.name}`, async () => {
          const connectionPool = createConnectionPool(e.opts())
          broker = new Broker({ connectionPool, logger: newLogger() })
          await broker.connect()
          expect(broker.isConnected()).toEqual(true)
        })

        test('returns false when the session lifetime has expired', async () => {
          const sessionLifetime = 15000
          const reauthenticationThreshold = 10000
          const connectionPool = createConnectionPool({ ...e.opts(), reauthenticationThreshold })
          broker = new Broker({ connectionPool, logger: newLogger() })

          await broker.connect()
          expect(broker.isConnected()).toEqual(true)

          const connection = await connectionPool.getConnection()

          connection.sessionLifetime = Long.fromValue(sessionLifetime)
          const [seconds] = connection.authenticatedAt
          connection.authenticatedAt = [seconds - sessionLifetime / 1000, 0]

          expect(broker.isConnected()).toEqual(false)
        })

        test('returns true when the session lifetime is 0', async () => {
          const connectionPool = createConnectionPool(e.opts())
          broker = new Broker({ connectionPool, logger: newLogger() })

          await broker.connect()
          expect(broker.isConnected()).toEqual(true)

          const connection = await connectionPool.getConnection()
          connection.sessionLifetime = Long.ZERO
          connection.authenticatedAt = [0, 0]

          expect(broker.isConnected()).toEqual(true)
        })

        testIfKafkaAtLeast_1_1_0(`authenticate with SASL ${e.name} if configured`, async () => {
          const connectionPool = createConnectionPool(e.opts())
          broker = new Broker({ connectionPool, logger: newLogger() })
          expect(broker.isConnected()).toEqual(false)
          await broker.connect()
          expect(broker.isConnected()).toEqual(true)

          const connection = await connectionPool.getConnection()
          expect(connection.getSupportAuthenticationProtocol()).toEqual(true)
        })
      }
    })

    describeIfOauthbearerDisabled('when SASL SCRAM is configured', () => {
      testIfKafkaAtLeast_1_1_0('parallel calls to connect using SCRAM', async () => {
        broker = new Broker({
          connectionPool: createConnectionPool(saslSCRAM256ConnectionOpts()),
          logger: newLogger(),
        })

        expect(broker.isConnected()).toEqual(false)

        await Promise.all([
          broker.connect(),
          broker.connect(),
          broker.connect(),
          broker.connect(),
          broker.connect(),
        ])

        expect(broker.isConnected()).toEqual(true)
      })
    })
  })
})

@@ -73,16 +73,37 @@
 onOpts,%0A
+  sslConnectionOpts,%0A
   newLog
@@ -899,16 +899,644 @@
 )%0A  %7D)%0A%0A
+  test(%22throws if the mechanism isn't supported by the server%22, async () =%3E %7B%0A    broker = new Broker(%7B%0A      connectionPool: createConnectionPool(%0A        Object.assign(sslConnectionOpts(), %7B%0A          port: 9094,%0A          sasl: %7B%0A            mechanism: 'fake-mechanism',%0A            authenticationProvider: () =%3E (%7B%0A              authenticate: async () =%3E %7B%0A                throw new Error('%F0%9F%A5%B8')%0A              %7D,%0A            %7D),%0A          %7D,%0A        %7D)%0A      ),%0A      logger: newLogger(),%0A    %7D)%0A%0A    await expect(broker.connect()).rejects.toThrow(%0A      'The broker does not support the requested SASL mechanism'%0A    )%0A  %7D)%0A%0A
   for (c

add unsupported mechanism test