grind/testing

Test support: run a worker’s handler without a database, and drain a queue on demand instead of waiting for consumers to poll.

import grind/testing
import grind/worker

pub fn mailer_sends_test() {
  let assert Ok(worker.Succeeded("msg:a@b.c")) =
    testing.perform(mailer(), Email("a@b.c", "hi"))
}

perform round-trips the input, output and error through the worker’s codecs, as a consumer does, so a codec defect shows up in the test. drain claims and runs a queue’s due jobs in the test’s own node, with a bound on how long it may take; configure the runtime without_consumers so nothing else claims them.

Types

DrainError

</>

Why drain stopped early.

pub type DrainError {
  UnknownQueue(queue: String)
  DrainTimedOut(processed: Int)
  DrainFailed(processed: Int, reason: String)
  DrainNotRunning
}

Constructors

```
UnknownQueue(queue: String)
```
No registered worker uses this queue.
```
DrainTimedOut(processed: Int)
```
within elapsed after processed jobs. A job still running then keeps its claim until its lease expires.
```
DrainFailed(processed: Int, reason: String)
```
Claiming or acknowledging failed after processed jobs.
```
DrainNotRunning
```
No runtime is running under this name on this node.

Error

</>

Why perform could not run the handler, or could not encode its result.

pub type Error {
  InputRejected(reason: String)
  InputUndecodable(reason: String)
  OutputRejected(reason: String)
  ErrorRejected(reason: String)
}

Constructors

```
InputRejected(reason: String)
```
The input codec rejected the input.
```
InputUndecodable(reason: String)
```
The encoded input did not decode with the input codec.
```
OutputRejected(reason: String)
```
The output codec rejected the handler’s output, or it did not decode.
```
ErrorRejected(reason: String)
```
The error codec rejected the handler’s error, or it did not decode.

Values

cancelled

</>

pub fn cancelled(context: worker.Context) -> worker.Context

A context whose cancellation has already fired, for the process that calls this (where perform runs the handler).

context

</>

pub fn context(
  job_id job_id: Int,
  attempt attempt: Int,
) -> worker.Context

A context for the given job and attempt, of 20, with no snoozes, queue "default", a cancellation that never fires, and a connection to no pool: set one with with_connection for a handler that queries.

drain

</>

pub fn drain(
  grind: grind.Grind,
  queue queue: String,
  limit limit: Int,
  within within: duration.Duration,
) -> Result(Int, DrainError)

Claims and runs up to limit due jobs from queue, one at a time, and returns how many ran. Stops early when no job is due. Fails with DrainTimedOut when within elapses first.

perform

</>

pub fn perform(
  worker: worker.Worker(input, output, error),
  input: input,
) -> Result(worker.Response(output, error), Error)

Runs the worker’s handler once, in the calling process, with context(job_id: 1, attempt: 1).

perform_with

</>

pub fn perform_with(
  worker: worker.Worker(input, output, error),
  context: worker.Context,
  input: input,
) -> Result(worker.Response(output, error), Error)

Runs the worker’s handler once, in the calling process, with context. The input is encoded and decoded first, and the output or error after.

with_connection

</>

pub fn with_connection(
  context: worker.Context,
  connection: pog.Connection,
) -> worker.Context

The connection worker.connection returns, such as grind.connection(jobs) or a test’s own pool.

with_max_attempts

</>

pub fn with_max_attempts(
  context: worker.Context,
  max_attempts: Int,
) -> worker.Context

with_queue

</>

pub fn with_queue(
  context: worker.Context,
  queue: String,
) -> worker.Context

with_snooze_count

</>

pub fn with_snooze_count(
  context: worker.Context,
  snooze_count: Int,
) -> worker.Context