Add delayed queue tests

jest.config.js

@@ -123,7 +123,7 @@ module.exports = {
   // setupFiles: [],
 
   // The path to a module that runs some code to configure or set up the testing framework before each test
-  // setupTestFrameworkScriptFile: null,
+  setupFilesAfterEnv: ["jest-extended"],
 
   // A list of paths to snapshot serializer modules Jest should use for snapshot testing
   // snapshotSerializers: [],

package.json

@@ -133,6 +133,7 @@
     "husky": "^1.3.1",
     "jest": "^24.8.0",
     "jest-environment-node": "^24.8.0",
+    "jest-extended": "^0.11.5",
     "lint-staged": "^8.1.7",
     "mini-css-extract-plugin": "^0.4.0",
     "mock-express-request": "^0.2.2",

src/server/db/__tests__/UserPrivate.js

@@ -1,19 +1,61 @@
 /**
  * @jest-environment node
  */
-import UserDBPrivate from '../mongo/user-privat-provider'
+import storage from '../mongo/user-privat-provider'
 import mongoose from '../mongo-db'
-import _ from 'lodash'
+import { pick, keys } from 'lodash'
 
-const storage = UserDBPrivate
+import { DelayedTaskStatus } from '../mongo/models/delayed-task'
 
-const testUser = { identifier: '00', fullName: 'mongo_test', email: 'test@test.test', mobile: '123456789' }
+const testUserName = 'mongo_test'
+const testTaskName = 'mongo_test'
+const testTaskSubject = 'test_subject'
+const testUser = { identifier: '00', fullName: testUserName, email: 'test@test.test', mobile: '123456789' }
 
 jest.setTimeout(30000)
 
 describe('UserPrivate', () => {
+  const { model: userModel, taskModel } = storage
+
+  const testTasksExists = async isExists =>
+    expect(taskModel.exists({ taskName: testTaskName, subject: testTaskSubject })).resolves.toBe(isExists)
+
+  // check if result obtained from raw query is the same like from storage method
+  const testHasTasksQueued = async () =>
+    storage.hasTasksQueued(testTaskName, { subject: testTaskSubject }).then(testTasksExists)
+
+  const testTaskStatusSwitch = async status => {
+    const { _id } = await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    await storage.fetchTasksForProcessing(testTaskName)
+
+    switch (status) {
+      case DelayedTaskStatus.Complete:
+        await storage.completeDelayedTasks([_id])
+        break
+      case DelayedTaskStatus.Failed:
+        await storage.failDelayedTasks([_id])
+        break
+      default:
+        break
+    }
+
+    await expect(taskModel.find({ taskName: testTaskName })).resolves.toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({
+          status,
+          lockId: null
+        })
+      ])
+    )
+  }
+
+  beforeEach(async () => {
+    await taskModel.deleteMany({ taskName: testTaskName })
+  })
+
   afterAll(async () => {
-    await storage.model.deleteMany({ fullName: new RegExp('mongo_test', 'i') })
+    await userModel.deleteMany({ fullName: new RegExp(testUserName, 'i') })
   })
 
   it('Should monogo connect', async () => {
@@ -32,7 +74,7 @@ describe('UserPrivate', () => {
     let user = await storage.getByIdentifier(testUser.identifier)
     expect(user).toBeTruthy()
 
-    const userDb = _.pick(user, _.keys(testUser))
+    const userDb = pick(user, keys(testUser))
 
     expect(user.jwt === 'test jwt').toBeTruthy()
     expect(userDb).toMatchObject(testUser)
@@ -45,13 +87,13 @@ describe('UserPrivate', () => {
 
   it('Should getByIdentifier user', async () => {
     let user = await storage.getByIdentifier(testUser.identifier)
-    const userDb = _.pick(user, _.keys(testUser))
+    const userDb = pick(user, keys(testUser))
     expect(userDb).toMatchObject(testUser)
   })
 
   it('Should getUser user', async () => {
     let user = await storage.getUser(testUser.identifier)
-    const userDb = _.pick(user, _.keys(testUser))
+    const userDb = pick(user, keys(testUser))
     expect(userDb).toMatchObject(testUser)
   })
 
@@ -108,7 +150,7 @@ describe('UserPrivate', () => {
   it('Should getUserByEmail', async () => {
     let user = await storage.getUserByEmail(testUser.email)
     expect(user).toBeTruthy()
-    const userDb = _.pick(user, _.keys(testUser))
+    const userDb = pick(user, keys(testUser))
     expect(userDb).toMatchObject(testUser)
   })
 
@@ -150,4 +192,112 @@ describe('UserPrivate', () => {
 
     expect(users.length >= listUsers.length).toBeTruthy()
   })
+
+  it('Should add delayed task', async () => {
+    await testTasksExists(false)
+
+    const wrappedResponse = expect(storage.enqueueTask(testTaskName, testTaskSubject)).resolves
+
+    await wrappedResponse.toHaveProperty('_id')
+    await wrappedResponse.toHaveProperty('subject', testTaskSubject)
+    await wrappedResponse.toHaveProperty('taskName', testTaskName)
+    await testTasksExists(true)
+  })
+
+  it('Should check tasks for existence', async () => {
+    await expect(storage.hasTasksQueued(testTaskName, { subject: testTaskSubject })).resolves.toBeBoolean()
+
+    // check before add (both raw query and hasTasksQueued() should return false)
+    await testHasTasksQueued()
+
+    await storage.enqueueTask(testTaskName, testTaskSubject)
+    // check before add (both raw query and hasTasksQueued() should return true)
+    await testHasTasksQueued()
+  })
+
+  it('Should fetch tasks', async () => {
+    const { _id } = await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    const wrappedResponse = expect(storage.fetchTasksForProcessing(testTaskName)).resolves
+
+    await wrappedResponse.toBeArrayOfSize(1)
+
+    await wrappedResponse.toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({
+          _id,
+          taskName: testTaskName,
+          subject: testTaskSubject
+        })
+      ])
+    )
+  })
+
+  it('Should lock tasks fetched', async () => {
+    await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    // this call locks the tasks found and should set running status
+    await expect(storage.fetchTasksForProcessing(testTaskName)).resolves.toEqual(
+      expect.arrayContaining([
+        expect.objectContaining({
+          status: DelayedTaskStatus.Running,
+          lockId: expect.anything()
+        })
+      ])
+    )
+
+    // so calling fetchTasksForProcessing() again should return empty list
+    await expect(storage.fetchTasksForProcessing(testTaskName)).resolves.toBeArrayOfSize(0)
+  })
+
+  it('Should complete or fail tasks', async () => {
+    await testTaskStatusSwitch(DelayedTaskStatus.Complete)
+    await taskModel.deleteMany({ taskName: testTaskName })
+    await testTaskStatusSwitch(DelayedTaskStatus.Failed)
+  })
+
+  it('Should unlock failed tasks', async () => {
+    const { _id } = await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    await storage.fetchTasksForProcessing(testTaskName)
+    // this unlock running tasks
+    await storage.failDelayedTasks([_id])
+
+    // so the next fetchTasksForProcessing() call now should return tasks
+    await expect(storage.fetchTasksForProcessing(testTaskName)).resolves.toBeArrayOfSize(1)
+  })
+
+  it("Complete/fail shouldn't switch pending status", async () => {
+    const { Complete, Failed } = DelayedTaskStatus
+    const { _id } = await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    // we sholdn't be able to update status for task aren't locked via fetchTasksForProcessing()
+    await storage.failDelayedTasks([_id])
+    await storage.completeDelayedTasks([_id])
+
+    // no complete/failed tasks should be found despite we've called corresponding storage methods
+    await Promise.all(
+      [Complete, Failed].map(async status =>
+        expect(taskModel.find({ taskName: testTaskName, status })).resolves.toBeArrayOfSize(0)
+      )
+    )
+  })
+
+  it('Should remove delayed tasks', async () => {
+    const { _id } = await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    await testTasksExists(true)
+    await storage.fetchTasksForProcessing(testTaskName)
+
+    await expect(storage.removeDelayedTasks([_id])).resolves.toBeUndefined()
+    await testTasksExists(false)
+  })
+
+  it("Shouldn't remove pending tasks", async () => {
+    const { _id } = await storage.enqueueTask(testTaskName, testTaskSubject)
+
+    await testTasksExists(true)
+    await storage.removeDelayedTasks([_id])
+    await testTasksExists(true)
+  })
 })

src/server/db/mongo/models/delayed-task.js

@@ -22,13 +22,15 @@ export const DelayedTaskSchema = new Schema({
   userIdentifier: {
     type: ObjectId,
     ref: MODEL_USER_PRIVATE,
-    required: false
+    required: false,
+    unique: false
   },
   taskName: {
     // name (type, kind) of the delayed task. For 24h it should be 'verification/dispose_enrollments' (defined as constant in EnrollmentProcessor.js)
     type: String,
     required: true,
-    index: true
+    index: true,
+    unique: false
   },
   subject: {
     // some parameters (subject, options) of the task. Could be string number or object corresponding to the kind of the task.

src/server/db/mongo/user-privat-provider.js

@@ -217,7 +217,7 @@ class UserPrivate {
   async hasTasksQueued(taskName: string, filters: object = {}): Promise<boolean> {
     const { taskModel } = this
 
-    return taskModel.exists({ taskName, ...filters })
+    return taskModel.exists({ ...filters, taskName })
   }
 
   /**
@@ -283,10 +283,10 @@ class UserPrivate {
    */
   async removeDelayedTasks(tasksIdentifiers: string[]): Promise<void> {
     const { taskModel, logger } = this
-    const { Running } = DelayedTaskStatus
+    const { Running, Complete } = DelayedTaskStatus
 
     try {
-      await taskModel.deleteMany({ status: Running, _id: { $in: tasksIdentifiers } })
+      await taskModel.deleteMany({ status: { $in: [Running, Complete] }, _id: { $in: tasksIdentifiers } })
     } catch (exception) {
       const { message: errMessage } = exception
       const logPayload = { e: exception, errMessage, tasksIdentifiers }