Mock scorecard api and update tests to not depend on a running kafka instance

Author: callmekatootie

README.md

@@ -37,6 +37,7 @@ Also note that there is a `/health` endpoint that checks for the health of the a
 
 Configuration for the tests is at `config/test.js`, only add such new configurations different from `config/default.js`
 - WAIT_TIME: wait time used in test, default is 2000 or 2 seconds
+- SCORECARD_API_URL: the scorecard api url used in testing
 
 ## Local Kafka setup
 

config/test.js

@@ -3,5 +3,6 @@
  */
 
 module.exports = {
-  WAIT_TIME: process.env.WAIT_TIME ? Number(process.env.WAIT_TIME) : 2000
+  WAIT_TIME: process.env.WAIT_TIME ? Number(process.env.WAIT_TIME) : 2000,
+  SCORECARD_API_URL: 'https://api.topcoder-dev.com/scorecards'
 }

package.json

@@ -15,10 +15,10 @@
   "devDependencies": {
     "mocha": "^5.2.0",
     "mocha-prepare": "^0.1.0",
+    "nock": "^11.3.5",
     "nyc": "^13.3.0",
     "should": "^13.2.3",
-    "standard": "^12.0.1",
-    "nock": "^11.3.5"
+    "standard": "^12.0.1"
   },
   "dependencies": {
     "@topcoder-platform/topcoder-submission-api-wrapper": "^1.1.0",

src/app.js

@@ -13,9 +13,12 @@ const SubmissionProcessorService = require('./services/SubmissionProcessorServic
 
 // Start kafka consumer
 logger.info('Starting kafka consumer')
-// create consumer
 
-const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
+// create consumer
+let consumer
+if (process.env.NODE_ENV !== 'test') {
+  consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
+}
 
 /*
  * Data handler linked with Kafka consumer
@@ -52,7 +55,11 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
     .then(() => { logger.debug('Successfully processed message') })
     .catch((err) => { logger.logFullError(err) })
     // commit offset regardless of errors
-    .finally(() => consumer.commitOffset({ topic, partition, offset: m.offset }))
+    .finally(() => {
+      if (consumer) {
+        consumer.commitOffset({ topic, partition, offset: m.offset })
+      }
+    })
 })
 
 // check if there is kafka connection alive
@@ -68,23 +75,25 @@ function check () {
   return connected
 }
 
-const topics = [config.REVIEW_TOPIC, config.CREATE_SUBMISSION_TOPIC, config.UPDATE_SUBMISSION_TOPIC]
+if (consumer) {
+  const topics = [config.REVIEW_TOPIC, config.CREATE_SUBMISSION_TOPIC, config.UPDATE_SUBMISSION_TOPIC]
 
-consumer
-  .init([{
-    subscriptions: topics,
-    handler: dataHandler
-  }])
-  // consume configured topics
-  .then(() => {
-    logger.info('Initialized.......')
-    healthcheck.init([check])
-    logger.info('Adding topics successfully.......')
-    logger.info(topics)
-    logger.info('Kick Start.......')
-  })
-  .catch((err) => logger.error(err))
+  consumer
+    .init([{
+      subscriptions: topics,
+      handler: dataHandler
+    }])
+    // consume configured topics
+    .then(() => {
+      logger.info('Initialized.......')
+      healthcheck.init([check])
+      logger.info('Adding topics successfully.......')
+      logger.info(topics)
+      logger.info('Kick Start.......')
+    })
+    .catch((err) => logger.error(err))
+}
 
 if (process.env.NODE_ENV === 'test') {
-  module.exports = consumer
+  module.exports = dataHandler
 }

src/services/SubmissionProcessorService.js

@@ -81,25 +81,30 @@ function generateScoreSystemMessage (message, scoreSystem) {
  * @param {Object} message the kafka message to send.
  */
 async function sendMessage (message) {
-  // init producer if needed
-  if (!producer) {
-    producer = new Kafka.Producer(helper.getKafkaOptions())
-    // init kafka producer
-    try {
-      await producer.init()
-    } catch (e) {
-      // if there is any error, reset producer to null so that it will be re-created next time
-      producer = null
-      throw e
+  if (process.env.NODE_ENV === 'test') {
+    logger.debug(`Send Kafka message to topic: ${message.topic}`)
+    logger.debug(`Payload: ${JSON.stringify(message)}`)
+  } else {
+    // init producer if needed
+    if (!producer) {
+      producer = new Kafka.Producer(helper.getKafkaOptions())
+      // init kafka producer
+      try {
+        await producer.init()
+      } catch (e) {
+        // if there is any error, reset producer to null so that it will be re-created next time
+        producer = null
+        throw e
+      }
     }
+    // send message
+    await producer.send({
+      topic: message.topic,
+      message: {
+        value: JSON.stringify(message)
+      }
+    })
   }
-  // send message
-  await producer.send({
-    topic: message.topic,
-    message: {
-      value: JSON.stringify(message)
-    }
-  })
 }
 
 /**

test/common/prepare.js

@@ -35,6 +35,8 @@ prepare(function (done) {
     .reply(200)
     .post('/v5/reviews')
     .reply(200)
+    .get(/\/scorecards\/.+/)
+    .reply(200, { 'id': 'b9956051-4749-4912-abd5-bf4d701af891', 'scorecardDetails': [{ 'id': 'b9956051-4749-4912-abd1-af4d701af891', 'name': 'AV Scan', 'phase': 'submission', 'isActive': true, 'topic': 'avscan.action.scan', 'weight': 25 }, { 'id': 'b9956051-4749-4912-abd1-af4d701af892', 'name': 'OR', 'phase': 'review', 'isActive': true, 'topic': 'or.action.review', 'weight': 25 }, { 'id': 'b9956051-4749-4912-abd1-af4d701af893', 'name': 'AV Scan', 'phase': 'registration', 'isActive': true, 'topic': 'avscan.action.scan', 'weight': 25 }, { 'id': 'b9956051-4749-4912-abd1-af4d701af894', 'name': 'OR', 'phase': 'registration', 'isActive': true, 'topic': 'or.action.review', 'weight': 25 }] })
 
   done()
 }, function (done) {

test/e2e/review.processor.test.js

@@ -6,74 +6,42 @@ process.env.NODE_ENV = 'test'
 global.Promise = require('bluebird')
 
 const _ = require('lodash')
-const should = require('should')
-const request = require('superagent')
 
 const testHelper = require('../common/testHelper')
 const { testTopics } = require('../common/testData')
+const dataHandler = require('../../src/app')
 
 describe('Topcoder - Scorecard Review Processor E2E Test', () => {
   before(async () => {
-    // consume and commit existing messages if any
-    await testHelper.consumeMessages()
-    // init kafka producer
-    await testHelper.initProducer()
-
     // intercept logger
     testHelper.interceptLogger()
-
-    // start the application (kafka listener)
-    await testHelper.initApp()
   })
 
   after(async () => {
     testHelper.restoreLogger()
-
-    await testHelper.stopProducer()
   })
 
   beforeEach(() => {
     testHelper.clearInterceptedLogging()
   })
 
-  it('Should setup healthcheck with check on kafka connection', async () => {
-    const healthcheckEndpoint = `http://localhost:${process.env.PORT || 3000}/health`
-    let result = await request.get(healthcheckEndpoint)
-    should.equal(result.status, 200)
-    should.deepEqual(result.body, { checksRun: 1 })
-    testHelper.assertDebugMessage('connected=true')
-  })
-
   it('Should handle invalid json message', async () => {
     const { testMessage } = testTopics.create
-    await testHelper.getProducer().send({
-      topic: testMessage.topic,
-      message: {
-        value: '[ invalid'
-      }
-    })
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: '[ invalid' }, offset: 0 }], testMessage.topic, 0)
     testHelper.assertErrorMessage('Invalid message JSON.')
   })
 
   it('Should handle incorrect topic field message', async () => {
     const { testMessage } = testTopics.create
     let message = _.cloneDeep(testMessage)
     message.topic = 'invalid'
-    await testHelper.getProducer().send({
-      topic: testMessage.topic,
-      message: {
-        value: JSON.stringify(message)
-      }
-    })
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], testMessage.topic, 0)
     testHelper.assertErrorMessage(
       'The message topic invalid doesn\'t match the Kafka topic submission.notification.score.')
   })
 
   it('processor create review success', async () => {
-    await testHelper.sendMessage(testTopics.create.testMessage)
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: JSON.stringify(testTopics.create.testMessage) }, offset: 0 }], testTopics.create.testMessage.topic, 0)
 
     testHelper.assertDebugMessage('Get submission')
     testHelper.assertDebugMessage('Submission id: b91a0ca3-3988-4899-bab4-c789f22def39')
@@ -83,8 +51,7 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
   })
 
   it('processor update review success', async () => {
-    await testHelper.sendMessage(testTopics.update.testMessage)
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: JSON.stringify(testTopics.update.testMessage) }, offset: 0 }], testTopics.update.testMessage.topic, 0)
 
     testHelper.assertDebugMessage('Get submission')
     testHelper.assertDebugMessage('Submission id: b91a0ca3-3988-4899-bab4-c789f22def39')
@@ -98,24 +65,21 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
   it('test invalid parameters, userId is forbidden for create review message.', async () => {
     let message = _.cloneDeep(testTopics.create.testMessage)
     message.payload.userId = '12345'
-    await testHelper.sendMessage(message)
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
     testHelper.assertErrorMessage('"userId" is not allowed')
   })
 
   it('test invalid parameters, fail to retrieve review with given criteria.', async () => {
     let message = _.cloneDeep(testTopics.update.testMessage)
     message.payload.scorecardId = 300001611
-    await testHelper.sendMessage(message)
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
     testHelper.assertErrorMessage('Review doesn\'t exist under criteria')
   })
 
   it('test invalid eventType, fail to retrieve review with given criteria.', async () => {
     let message = _.cloneDeep(testTopics.create.testMessage)
     message.payload.eventType = 'INVALID_TYPE'
-    await testHelper.sendMessage(message)
-    await testHelper.waitJob()
+    await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
     testHelper.assertErrorMessage('Invalid or not supported eventType: INVALID_TYPE')
   })
 
@@ -125,16 +89,14 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
     it('test invalid parameters, field submissionId incorrect', async () => {
       let message = _.cloneDeep(testMessage)
       message.payload.submissionId = 111111111
-      await testHelper.sendMessage(message)
-      await testHelper.waitJob()
+      await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
       testHelper.assertErrorMessage('Incorrect submission id 111111111')
     })
 
     it('test invalid parameters, fail to get reviewType UUID.', async () => {
       let message = _.cloneDeep(testMessage)
       message.payload.reviewTypeId = 12
-      await testHelper.sendMessage(message)
-      await testHelper.waitJob()
+      await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
       testHelper.assertErrorMessage('Incorrect review type id 12')
     })
 
@@ -143,8 +105,7 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
         it(`test invalid parameters, required field ${requiredField} is missing`, async () => {
           let message = _.cloneDeep(testMessage)
           message = _.omit(message, requiredField)
-          await testHelper.sendMessage(message)
-          await testHelper.waitJob()
+          await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
           testHelper.assertErrorMessage(`"${_.last(requiredField.split('.'))}" is required`)
         })
       }
@@ -155,8 +116,7 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
         it(`test invalid parameters, invalid string type field ${stringField}`, async () => {
           let message = _.cloneDeep(testMessage)
           _.set(message, stringField, 123)
-          await testHelper.sendMessage(message)
-          await testHelper.waitJob()
+          await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
           testHelper.assertErrorMessage(`"${_.last(stringField.split('.'))}" must be a string`)
         })
       }
@@ -166,16 +126,14 @@ describe('Topcoder - Scorecard Review Processor E2E Test', () => {
       it(`test invalid parameters, invalid integer type field ${integerField}(wrong number)`, async () => {
         let message = _.cloneDeep(testMessage)
         _.set(message, integerField, 'string')
-        await testHelper.sendMessage(message)
-        await testHelper.waitJob()
+        await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
         testHelper.assertErrorMessage(`"${_.last(integerField.split('.'))}" must be a number`)
       })
 
       it(`test invalid parameters, invalid integer type field ${integerField}(wrong integer)`, async () => {
         let message = _.cloneDeep(testMessage)
         _.set(message, integerField, 1.1)
-        await testHelper.sendMessage(message)
-        await testHelper.waitJob()
+        await dataHandler([{ message: { value: JSON.stringify(message) }, offset: 0 }], message.topic, 0)
         if (integerField === 'payload.reviewTypeId') {
           testHelper.assertErrorMessage(`"${_.last(integerField.split('.'))}" must be one of [1, 2, 5, 6, 7, 8, 9, 10, 11, 12]`)
         } else {