mule/testing

Worker and enqueued-job testing helpers, mirroring Oban.Testing (testing.ex). The use Oban.Testing, repo: macro currying becomes explicit parameters: the assertion helpers take the ENGINE record (the repo analog — they work against both engines through engine.all_jobs, with or without a running instance), and perform_job takes the typed Worker(a) — the type system is the “implements Oban.Worker” assertion.

Assertion failures panic as their message (gleeunit fails on crash — the flunk substitute); the messages are built by the public *_failure_message functions so their formatting is unit-testable.

Hazard (true in Elixir too): a worker timeout arms timer:exit_after in the CALLING process, so a timed-out perform_job kills the test process.

Types

Match

</>

One assertion condition (testing.ex’s assertion keywords). Conditions AND together, and assertions only ever see ENQUEUED rows — state in available/scheduled/suspended (testing.ex base_query). MatchTimestamp applies the default ±1s window; MatchTimestampDelta is the {time, delta: n} override.

pub type Match {
  MatchArgs(args: json.Json)
  MatchAttempt(attempt: Int)
  MatchId(id: Int)
  MatchMaxAttempts(max_attempts: Int)
  MatchMeta(meta: json.Json)
  MatchPriority(priority: Int)
  MatchQueue(queue: String)
  MatchState(state: states.State)
  MatchTags(tags: List(String))
  MatchWorker(worker: String)
  MatchTimestamp(
    field: TimestampField,
    time: timestamp.Timestamp,
  )
  MatchTimestampDelta(
    field: TimestampField,
    time: timestamp.Timestamp,
    delta: duration.Duration,
  )
}

Constructors

```
MatchArgs(args: json.Json)
```
```
MatchAttempt(attempt: Int)
```
```
MatchId(id: Int)
```
```
MatchMaxAttempts(max_attempts: Int)
```
```
MatchMeta(meta: json.Json)
```
```
MatchPriority(priority: Int)
```
```
MatchQueue(queue: String)
```
```
MatchState(state: states.State)
```
```
MatchTags(tags: List(String))
```
```
MatchWorker(worker: String)
```

MatchTimestamp(field: TimestampField, time: timestamp.Timestamp)

MatchTimestampDelta(
  field: TimestampField,
  time: timestamp.Timestamp,
  delta: duration.Duration,
)

Override

</>

The execution-context fields build_job defaults (testing.ex’s put_new_change set — Job fields Job.new/2 can cast but the insert path never accepts). Duplicates keep the last, like job.new_with options.

pub type Override {
  Id(id: Int)
  Attempt(attempt: Int)
  AttemptedAt(attempted_at: timestamp.Timestamp)
  InsertedAt(inserted_at: timestamp.Timestamp)
  State(state: states.State)
}

Constructors

```
Id(id: Int)
```
```
Attempt(attempt: Int)
```

AttemptedAt(attempted_at: timestamp.Timestamp)

InsertedAt(inserted_at: timestamp.Timestamp)

```
State(state: states.State)
```

TimestampField

</>

The six timestamp columns assertions may window-match (testing.ex @timestamp_fields). By* prefix: bare names would collide with the Override constructors in this module.

pub type TimestampField {
  ByAttemptedAt
  ByCancelledAt
  ByCompletedAt
  ByDiscardedAt
  ByInsertedAt
  ByScheduledAt
}

Constructors

```
ByAttemptedAt
```
```
ByCancelledAt
```
```
ByCompletedAt
```
```
ByDiscardedAt
```
```
ByInsertedAt
```
```
ByScheduledAt
```

Values

all_enqueued

</>

pub fn all_enqueued(
  engine engine: engine.Engine,
  matches matches: List(Match),
) -> List(job.Job)

All currently enqueued jobs matching the conditions, most recently enqueued first — id-descending, testing.ex’s order_by(desc: :id) (the reverse of engine.all_jobs’ deterministic ascending order).

assert_enqueued

</>

pub fn assert_enqueued(
  engine engine: engine.Engine,
  matches matches: List(Match),
) -> Nil

Panic unless a job matching the conditions is enqueued (testing.ex assert_enqueued/1); the message lists the bare enqueued set, like Elixir’s available_jobs dump.

assert_enqueued_within

</>

pub fn assert_enqueued_within(
  engine engine: engine.Engine,
  matches matches: List(Match),
  timeout timeout: duration.Duration,
) -> Nil

Like assert_enqueued, but polls every 10ms until a match appears or the timeout elapses (testing.ex assert_enqueued/2). Returns on the first hit.

assert_enqueued_within_polling

</>

pub fn assert_enqueued_within_polling(
  engine engine: engine.Engine,
  matches matches: List(Match),
  timeout timeout: duration.Duration,
  poll_interval poll_interval: duration.Duration,
) -> Nil

assert_enqueued_within with a tunable poll interval — fast in-memory suites and slower Postgres-backed ones want different cadences.

assert_failure_message

</>

pub fn assert_failure_message(
  matches matches: List(Match),
  enqueued enqueued: List(job.Job),
) -> String

The assert_enqueued failure text, built pure so formatting is testable. enqueued is the bare enqueued set the message dumps (Elixir projects to the asserted keys; the fixed projection below is the typed substitute, with Dynamic args rendering Erlang-term-shaped).

assert_within_failure_message

</>

pub fn assert_within_failure_message(
  matches matches: List(Match),
  timeout timeout: duration.Duration,
) -> String

build_job

</>

pub fn build_job(
  worker worker: worker.Worker(a),
  args args: a,
) -> job.Job

Construct a Job from a worker and args with the execution-context defaults filled in (testing.ex build_job/3): attempt: 1, attempted_at/inserted_at/scheduled_at: now (an explicit ScheduledAt/ScheduleIn option keeps its value and lands Scheduled), a fabricated unique id, and args/meta JSON-recoded to the Dynamic a stored row carries — so perform_built_job decodes args back through the worker’s codec exactly as a fetched row would (the atom-keys pitfall check, typed).

build_job_with

</>

pub fn build_job_with(
  worker worker: worker.Worker(a),
  args args: a,
  options options: List(job.Option),
  overrides overrides: List(Override),
) -> job.Job

build_job with Job.new/2-style options and Overrides. An option set that fails job.new_with’s validations panics with the field and message (testing.ex’s assert_valid_changeset flunk).

match_worker

</>

pub fn match_worker(worker worker: worker.Worker(a)) -> Match

Build a MatchWorker from the typed worker — the analog of Elixir accepting a module and calling Worker.to_string/1.

perform_built_job

</>

pub fn perform_built_job(
  worker worker: worker.Worker(a),
  job built_job: job.Job,
) -> worker.WorkerResult

Execute an already-built job with the worker (testing.ex’s 2.19 built-job perform_job arity, renamed — Gleam has no overloading). Resolution is by job.worker: a name mismatch panics, like Elixir flunking on a bogus worker.

perform_job

</>

pub fn perform_job(
  worker worker: worker.Worker(a),
  args args: a,
) -> worker.WorkerResult

Construct a job and execute it with the worker, returning the worker’s result for further assertions (testing.ex perform_job/2). A crash, throw, or in-process exit re-raises with its original stacktrace (safe: False), so the test fails with the real crash; nothing is persisted or acked (ack: False).

perform_job_with

</>

pub fn perform_job_with(
  worker worker: worker.Worker(a),
  args args: a,
  options options: List(job.Option),
  overrides overrides: List(Override),
) -> worker.WorkerResult

perform_job with job options and overrides — e.g. exercise custom attempt handling with overrides: [Attempt(42)].

refute_enqueued

</>

pub fn refute_enqueued(
  engine engine: engine.Engine,
  matches matches: List(Match),
) -> Nil

Panic if a job matching the conditions is enqueued (testing.ex refute_enqueued/1).

refute_enqueued_within

</>

pub fn refute_enqueued_within(
  engine engine: engine.Engine,
  matches matches: List(Match),
  timeout timeout: duration.Duration,
) -> Nil

Like refute_enqueued, but keeps checking every 10ms for the whole window — panicking on the first match, returning only after surviving it (testing.ex refute_enqueued/2).

refute_enqueued_within_polling

</>

pub fn refute_enqueued_within_polling(
  engine engine: engine.Engine,
  matches matches: List(Match),
  timeout timeout: duration.Duration,
  poll_interval poll_interval: duration.Duration,
) -> Nil

refute_enqueued_within with a tunable poll interval.

refute_failure_message

</>

pub fn refute_failure_message(
  matches matches: List(Match),
) -> String

refute_within_failure_message

</>

pub fn refute_within_failure_message(
  matches matches: List(Match),
  timeout timeout: duration.Duration,
) -> String

wildcard

</>

pub fn wildcard() -> json.Json

The :_ wildcard for MatchArgs/MatchMeta: matches ANY value provided the key exists. Literally json.string("_") — the wire value Elixir’s :_ json-recodes to, quirk included (a stored literal "_" also matches).

with_testing_mode

</>

pub fn with_testing_mode(
  mode mode: mule.TestingMode,
  instance instance: mule.Mule,
  run run: fn(mule.Mule) -> value,
) -> value

testing.ex with_testing_mode/2. Elixir flips a process-dictionary key that Config.get_engine reads across self + $callers; Gleam has no ambient state, so the substitute is a DERIVED HANDLE: run receives an instance whose config resolves the requested mode, and everything using that handle runs in that mode. The registered instance and every other handle are untouched — strictly tighter scoping than Elixir’s (and mule.config(name) readers still see the started mode; parity §9). Only TestingManual and TestingInline are accepted (Elixir’s guard); TestingDisabled panics.