Testing Jobs - how to start a worker?

[zevisert]

I don't see anything in the docs, and it seems like the package.json doesn't export the Worker or WorkerManager. I'd like to be able to start a worker with my functional tests, either in the test itself or during the suite's startup hooks. Is this possible? It seems that currently, my only option is to run node ace queue:work outside / alongside my tests.

[zevisert]

To add some more context, I can run tests with node ace queue:work - but since this is a separate process, I can't control the timing or order of jobs being executed, nor can I compute any code-coverage for the logic in my jobs (or health checks).

It'd be great if I could instantiate these classes as part of my test suite - especially from the code-coverage perspective:

adonis-jobs/packages/core/commands/queue/queue_work.ts

Lines 90 to 94 in 2dd9100

	const jobs = await new JobDiscoverer(this.app.appRoot).discoverAndLoadJobs()
	const resolver = new ConnectionResolver(this.#config, redis)
	this.#manager = new WorkerManager(this.app, emitter, this.#config as any, jobs, resolver)
	this.#healthCheckManager = new HealthCheckManager(this.#config, redis)

[zevisert]

I've been fiddling with this, and what I've come up with the following japa runner hooks:

configure(suite => {
  if (suite.name == "e2e") {
    // Start and stop a redis container to queue jobs on
    suite = suite.setup(async () => {
      const redis = await new RedisContainer('redis:8.4')
        .withExposedPorts(6379)
        .withWaitStrategy(Wait.forLogMessage('Ready to accept connections'))
        .start()

      // To not conflict with a running redis instance, set the redis
      // port to what testcontainers arranged
      redisConfig.connections.main.port = redis.getMappedPort(6379)
      env.set('REDIS_PORT', redisConfig.connections.main.port)

      return async () => {
        await redis.stop()
      }
    })

    // Start and stop a worker to handle jobs
    suite.setup(async () => {
      // Disable connection retries for testing purposes
      redisConfig.connections.main.retryStrategy = () => null

      // Set the port to something else for the worker's healthchecks to not
      // interfere with the main application during testing
      env.set('PORT', 3003)

      // Adonis-Jobs does not export the worker, only the
      // command to start it, so we have to start the worker
      // here in test setup and kill it in the teardown
      const commandClass = await getCommand({
        commandName: 'queue:work',
        description: '',
        args: [],
        flags: [],
        namespace: '',
        options: {},
        aliases: [],
      })

      // Start the worker within this process to allow
      // for coverage measurement
      const command: BaseCommand = await ace.create(commandClass, [])
      void command.exec()

      // Shutdown the worker with a SIGINT, but this is probably
      // a footgun. It doesn't seem to kill the test process.
      // See https://github.com/nemoengineering/adonis-jobs/issues/100
      return () => {
        process.kill(process.pid, 'SIGINT')
      }
    })
  }
})

I'm going through this effort because I care about test coverage. I have code like MyJob.dispatch({...}) and queuManager.useQueue().getJobState(id) that I want to have test coverage for, using dependency injection to swap these out results in uncovered code. Using mocks and testing MyJob by constructing it, then setting myJob.data = {...} before calling myJob.process() doesn't run dependency injection, so my code crashes when trying to use this.logger in the Job.

I'd love to have a test workflow where I can do something like:

import { test } from '@japa/runner'
import { Worker } from '@nemoengineering/adonis-jobs/testing'

import MyJob from '#app/jobs/my_job'

test('ideal', ({assert}) => {
  const job = MyJob.dispatch({...})

  // This is the ask for this issue
  const worker = new Worker()
  await worker.runOnce()
  // Or something like the above
  // eg, `.runUntilIdle()`, or maybe an AbortController?

  const result = await job.waitResult() // Should be instantaneous
  assert.deepEqual(result, ...)
})