Merge pull request #15488 from Budibase/automation-tests-3

Create an automation cron trigger test.

packages/backend-core/src/queue/inMemoryQueue.ts

@@ -1,45 +1,58 @@
 import events from "events"
 import { newid } from "../utils"
 import { Queue, QueueOptions, JobOptions } from "./queue"
+import { helpers } from "@budibase/shared-core"
+import { Job, JobId, JobInformation } from "bull"
 
-interface JobMessage {
+function jobToJobInformation(job: Job): JobInformation {
+  let cron = ""
+  let every = -1
+  let tz: string | undefined = undefined
+  let endDate: number | undefined = undefined
+
+  const repeat = job.opts?.repeat
+  if (repeat) {
+    endDate = repeat.endDate ? new Date(repeat.endDate).getTime() : Date.now()
+    tz = repeat.tz
+    if ("cron" in repeat) {
+      cron = repeat.cron
+    } else {
+      every = repeat.every
+    }
+  }
+
+  return {
+    id: job.id.toString(),
+    name: "",
+    key: job.id.toString(),
+    tz,
+    endDate,
+    cron,
+    every,
+    next: 0,
+  }
+}
+
+interface JobMessage<T = any> extends Partial<Job<T>> {
   id: string
   timestamp: number
-  queue: string
+  queue: Queue<T>
   data: any
   opts?: JobOptions
 }
 
 /**
- * Bull works with a Job wrapper around all messages that contains a lot more information about
- * the state of the message, this object constructor implements the same schema of Bull jobs
- * for the sake of maintaining API consistency.
- * @param queue The name of the queue which the message will be carried on.
- * @param message The JSON message which will be passed back to the consumer.
- * @returns A new job which can now be put onto the queue, this is mostly an
- * internal structure so that an in memory queue can be easily swapped for a Bull queue.
- */
-function newJob(queue: string, message: any, opts?: JobOptions): JobMessage {
-  return {
-    id: newid(),
-    timestamp: Date.now(),
-    queue: queue,
-    data: message,
-    opts,
-  }
-}
-
-/**
- * This is designed to replicate Bull (https://github.com/OptimalBits/bull) in memory as a sort of mock.
- * It is relatively simple, using an event emitter internally to register when messages are available
- * to the consumers - in can support many inputs and many consumers.
+ * This is designed to replicate Bull (https://github.com/OptimalBits/bull) in
+ * memory as a sort of mock.  It is relatively simple, using an event emitter
+ * internally to register when messages are available to the consumers - in can
+ * support many inputs and many consumers.
  */
 class InMemoryQueue implements Partial<Queue> {
   _name: string
   _opts?: QueueOptions
   _messages: JobMessage[]
   _queuedJobIds: Set<string>
-  _emitter: NodeJS.EventEmitter
+  _emitter: NodeJS.EventEmitter<{ message: [JobMessage]; completed: [Job] }>
   _runCount: number
   _addCount: number
 
@@ -69,34 +82,29 @@ class InMemoryQueue implements Partial<Queue> {
    */
   async process(concurrencyOrFunc: number | any, func?: any) {
     func = typeof concurrencyOrFunc === "number" ? func : concurrencyOrFunc
-    this._emitter.on("message", async () => {
-      if (this._messages.length <= 0) {
-        return
-      }
-      let msg = this._messages.shift()
-
-      let resp = func(msg)
+    this._emitter.on("message", async message => {
+      let resp = func(message)
 
       async function retryFunc(fnc: any) {
         try {
           await fnc
         } catch (e: any) {
-          await new Promise<void>(r => setTimeout(() => r(), 50))
-
-          await retryFunc(func(msg))
+          await helpers.wait(50)
+          await retryFunc(func(message))
         }
       }
 
       if (resp.then != null) {
         try {
           await retryFunc(resp)
+          this._emitter.emit("completed", message as Job)
         } catch (e: any) {
           console.error(e)
         }
       }
       this._runCount++
-      const jobId = msg?.opts?.jobId?.toString()
-      if (jobId && msg?.opts?.removeOnComplete) {
+      const jobId = message.opts?.jobId?.toString()
+      if (jobId && message.opts?.removeOnComplete) {
         this._queuedJobIds.delete(jobId)
       }
     })
@@ -130,9 +138,16 @@ class InMemoryQueue implements Partial<Queue> {
     }
 
     const pushMessage = () => {
-      this._messages.push(newJob(this._name, data, opts))
+      const message: JobMessage = {
+        id: newid(),
+        timestamp: Date.now(),
+        queue: this as unknown as Queue,
+        data,
+        opts,
+      }
+      this._messages.push(message)
       this._addCount++
-      this._emitter.emit("message")
+      this._emitter.emit("message", message)
     }
 
     const delay = opts?.delay
@@ -158,13 +173,6 @@ class InMemoryQueue implements Partial<Queue> {
     console.log(cronJobId)
   }
 
-  /**
-   * Implemented for tests
-   */
-  async getRepeatableJobs() {
-    return []
-  }
-
   async removeJobs(_pattern: string) {
     // no-op
   }
@@ -176,13 +184,31 @@ class InMemoryQueue implements Partial<Queue> {
     return []
   }
 
-  async getJob() {
+  async getJob(id: JobId) {
+    for (const message of this._messages) {
+      if (message.id === id) {
+        return message as Job
+      }
+    }
     return null
   }
 
-  on() {
-    // do nothing
-    return this as any
+  on(event: string, callback: (...args: any[]) => void): Queue {
+    // @ts-expect-error - this callback can be one of many types
+    this._emitter.on(event, callback)
+    return this as unknown as Queue
+  }
+
+  async count() {
+    return this._messages.length
+  }
+
+  async getCompletedCount() {
+    return this._runCount
+  }
+
+  async getRepeatableJobs() {
+    return this._messages.map(job => jobToJobInformation(job as Job))
   }
 }
 

packages/server/src/automations/bullboard.ts

@@ -5,9 +5,9 @@ import * as automation from "../threads/automation"
 import { backups } from "@budibase/pro"
 import { getAppMigrationQueue } from "../appMigrations/queue"
 import { createBullBoard } from "@bull-board/api"
-import BullQueue from "bull"
+import { AutomationData } from "@budibase/types"
 
-export const automationQueue: BullQueue.Queue = queue.createQueue(
+export const automationQueue = queue.createQueue<AutomationData>(
   queue.JobQueue.AUTOMATION,
   { removeStalledCb: automation.removeStalled }
 )
@@ -16,24 +16,20 @@ const PATH_PREFIX = "/bulladmin"
 
 export async function init() {
   // Set up queues for bull board admin
+  const queues = [new BullAdapter(automationQueue)]
+
   const backupQueue = backups.getBackupQueue()
-  const appMigrationQueue = getAppMigrationQueue()
-  const queues = [automationQueue]
   if (backupQueue) {
-    queues.push(backupQueue)
+    queues.push(new BullAdapter(backupQueue))
   }
+
+  const appMigrationQueue = getAppMigrationQueue()
   if (appMigrationQueue) {
-    queues.push(appMigrationQueue)
+    queues.push(new BullAdapter(appMigrationQueue))
   }
-  const adapters = []
-  const serverAdapter: any = new KoaAdapter()
-  for (let queue of queues) {
-    adapters.push(new BullAdapter(queue))
-  }
-  createBullBoard({
-    queues: adapters,
-    serverAdapter,
-  })
+
+  const serverAdapter = new KoaAdapter()
+  createBullBoard({ queues, serverAdapter })
   serverAdapter.setBasePath(PATH_PREFIX)
   return serverAdapter.registerPlugin()
 }

packages/server/src/automations/index.ts

@@ -1,7 +1,6 @@
 import { processEvent } from "./utils"
 import { automationQueue } from "./bullboard"
 import { rebootTrigger } from "./triggers"
-import BullQueue from "bull"
 import { automationsEnabled } from "../features"
 
 export { automationQueue } from "./bullboard"
@@ -25,6 +24,6 @@ export async function init() {
   return promise
 }
 
-export function getQueues(): BullQueue.Queue[] {
-  return [automationQueue]
+export function getQueue() {
+  return automationQueue
 }

packages/server/src/automations/tests/triggers/cron.spec.ts

@@ -0,0 +1,62 @@
+import { createAutomationBuilder } from "../utilities/AutomationTestBuilder"
+import TestConfiguration from "../../../tests/utilities/TestConfiguration"
+import { getQueue } from "../.."
+import { Job } from "bull"
+
+describe("cron trigger", () => {
+  const config = new TestConfiguration()
+
+  beforeEach(async () => {
+    await config.init()
+  })
+
+  afterAll(() => {
+    config.end()
+  })
+
+  it("should queue a Bull cron job", async () => {
+    const queue = getQueue()
+    expect(await queue.getCompletedCount()).toEqual(0)
+
+    const jobPromise = new Promise<Job>(resolve => {
+      queue.on("completed", async job => {
+        resolve(job)
+      })
+    })
+
+    await createAutomationBuilder({ config })
+      .cron({ cron: "* * * * *" })
+      .serverLog({
+        text: "Hello, world!",
+      })
+      .save()
+
+    await config.api.application.publish(config.getAppId())
+
+    expect(await queue.getCompletedCount()).toEqual(1)
+
+    const job = await jobPromise
+    const repeat = job.opts?.repeat
+    if (!repeat || !("cron" in repeat)) {
+      throw new Error("Expected cron repeat")
+    }
+    expect(repeat.cron).toEqual("* * * * *")
+  })
+
+  it("should fail if the cron expression is invalid", async () => {
+    await createAutomationBuilder({ config })
+      .cron({ cron: "* * * * * *" })
+      .serverLog({
+        text: "Hello, world!",
+      })
+      .save()
+
+    await config.api.application.publish(config.getAppId(), {
+      status: 500,
+      body: {
+        message:
+          'Deployment Failed: Invalid automation CRON "* * * * * *" - Expected 5 values, but got 6.',
+      },
+    })
+  })
+})

packages/server/src/automations/utils.ts

@@ -230,7 +230,7 @@ export async function enableCronTrigger(appId: any, automation: Automation) {
     // can't use getAppDB here as this is likely to be called from dev app,
     // but this call could be for dev app or prod app, need to just use what
     // was passed in
-    await dbCore.doWithDB(appId, async (db: any) => {
+    await dbCore.doWithDB(appId, async db => {
       const response = await db.put(automation)
       automation._id = response.id
       automation._rev = response.rev

packages/server/src/tests/utilities/api/application.ts

@@ -33,7 +33,10 @@ export class ApplicationAPI extends TestAPI {
     await this._delete(`/api/applications/${appId}`, { expectations })
   }
 
-  publish = async (appId: string): Promise<PublishResponse> => {
+  publish = async (
+    appId: string,
+    expectations?: Expectations
+  ): Promise<PublishResponse> => {
     return await this._post<PublishResponse>(
       `/api/applications/${appId}/publish`,
       {
@@ -42,14 +45,16 @@ export class ApplicationAPI extends TestAPI {
         headers: {
           [constants.Header.APP_ID]: appId,
         },
+        expectations,
       }
     )
   }
 
-  unpublish = async (appId: string): Promise<void> => {
-    await this._post(`/api/applications/${appId}/unpublish`, {
-      expectations: { status: 200 },
-    })
+  unpublish = async (
+    appId: string,
+    expectations?: Expectations
+  ): Promise<void> => {
+    await this._post(`/api/applications/${appId}/unpublish`, { expectations })
   }
 
   sync = async (
@@ -144,13 +149,20 @@ export class ApplicationAPI extends TestAPI {
     })
   }
 
-  fetch = async ({ status }: { status?: AppStatus } = {}): Promise<App[]> => {
+  fetch = async (
+    { status }: { status?: AppStatus } = {},
+    expectations?: Expectations
+  ): Promise<App[]> => {
     return await this._get<App[]>("/api/applications", {
       query: { status },
+      expectations,
     })
   }
 
-  addSampleData = async (appId: string): Promise<void> => {
-    await this._post(`/api/applications/${appId}/sample`)
+  addSampleData = async (
+    appId: string,
+    expectations?: Expectations
+  ): Promise<void> => {
+    await this._post(`/api/applications/${appId}/sample`, { expectations })
   }
 }