Assert on what your code enqueued, and run job handlers directly, without a server.
Start the recorder once, in test/test_helper.exs:
Zizq.Testing.start_link()
ExUnit.start()Then in a case, name the client your code enqueues through:
defmodule MyApp.SignupTest do
use ExUnit.Case, async: true
use Zizq.Testing, client: MyApp.Zizq
test "signing up sends a welcome email" do
MyApp.Signup.run("ada@example.com")
assert_enqueued(type: "send_email", payload: %{"template" => "welcome"})
end
enduse Zizq.Testing sets up the client for each test and imports the
assertions bound to it, so nothing needs passing around.
Enqueues are recorded, not sent
A client set up this way handles enqueues in memory instead of
contacting a server, and hands back a Zizq.Job as the server would
— with a synthetic id, status: :ready, and attempts: 0.
Everything upstream of the request runs unchanged, so option
validation, payload-derived unique keys and batch keys, and
telemetry all behave as they do in production.
Endpoints other than enqueuing raise: there is no server to have taken a job, so acknowledging or streaming one has no meaning here.
Recordings are per test
Recordings belong to the test process that made them, not to the
client, so a fixed client name is safe under async: true. Enqueues
from a Task or a spawned process count too, as long as the process
was started from the test — $callers is followed to find the
owner. A process started outside that chain, such as a supervised
worker, is not attributable and its enqueues are not recorded.
Assertions
assert_enqueued/1 and refute_enqueued/1 take a subset of the
fields to match on. A job matches when every field given matches:
assert_enqueued(type: "send_email")
assert_enqueued(type: "send_email", queue: "emails")
assert_enqueued(payload: %{"user_id" => 42})A :payload matches when the recorded payload contains what was
given, so a test names the keys it cares about and ignores the rest.
all_enqueued/1 returns the matching enqueues for anything the
assertions do not cover.
clear_enqueued/0 forgets what has been recorded so far, so a test
that acts twice can assert on the second action alone.
Payloads are what the server would have stored
A payload is normalised through JSON on the way in, exactly as it is
on the way to a real server, so a payload enqueued as %{user_id: 1}
records and matches as %{"user_id" => 1}. The same applies to a
payload handed to perform_job/3: a handler receives string keys
here because it receives string keys in production, and a test
passing atom keys would otherwise match a clause that can never run.
Summary
Functions
Every enqueue this test made, most recent last, optionally filtered.
Assert that a job matching filters was enqueued.
Returns a specification to start this module under a supervisor.
Forget everything recorded so far.
Run every enqueued job through handler, and return how many ran.
Run a job's handler directly, without a worker or a server.
Assert that no job matching filters was enqueued.
Point client at the recorder, and own its recordings for this test.
Start the recorder. Call once, from test/test_helper.exs.
Functions
Every enqueue this test made, most recent last, optionally filtered.
@spec assert_enqueued(keyword()) :: true
Assert that a job matching filters was enqueued.
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec clear_enqueued() :: :ok
Forget everything recorded so far.
Assertions after this see only what was enqueued from here on, which
is what a test that acts twice needs: the first action's enqueues
would otherwise still be sitting there for refute_enqueued/1 to
find, and a refutation cannot tell the two apart.
MyApp.Sitemap.scan(sitemap)
assert_enqueued(type: "check_url")
clear_enqueued()
# The second scan should find nothing new to do.
MyApp.Sitemap.scan(sitemap)
refute_enqueued(type: "check_url")Only this test's recordings go, so it is safe under async: true.
Drain bookkeeping is dropped alongside them, so "cleared" means cleared rather than half-cleared. That has no effect you can observe — every enqueue gets a fresh id, so an already-drained one could never come up again — it just keeps the two halves of a test's state from disagreeing.
@spec drain_enqueued( module() | Zizq.Router.t() | (Zizq.Job.t() -> term()), keyword() ) :: non_neg_integer()
Run every enqueued job through handler, and return how many ran.
handler is what a worker takes: a one-argument function over a
Zizq.Job, or a Zizq.Router.
MyApp.Signup.run("ada@example.com")
assert drain_enqueued(MyApp.Router.build()) == 1A job is drained once. Calling again runs only what has been enqueued since.
Options
:recursive— keep going until nothing new is enqueued, so a handler that enqueues further jobs drains those too. Defaults tofalse, which runs only what was already enqueued when the call began.:max_iterations— how many rounds:recursivemay take before giving up, guarding against a handler that always enqueues and would otherwise hang the suite. Defaults to1_000.:type,:queue— drain only matching jobs, asassert_enqueued/1matches.
Handlers run in the calling process, one at a time, in the order the jobs were enqueued — so a test can reason about ordering in a way a real worker's concurrency would not allow. Anything a handler raises propagates, after the job is marked drained so a retry cannot loop on it.
@spec perform_job(module() | Zizq.Router.t(), term(), keyword()) :: term()
Run a job's handler directly, without a worker or a server.
Takes a module using Zizq.JobKind, or a Zizq.Router:
assert :ok = perform_job(MyApp.SendEmail, %{"user_id" => 42})
assert :ok = perform_job(router, %{"user_id" => 42}, type: "send_email")Returns whatever the handler returned, verbatim — including
{:error, reason} — so a test asserts on the outcome rather than on
what a worker would have done with it. Anything the handler raises
propagates.
The return value must be one a worker recognises, and the test fails if it is not. A worker would acknowledge an unrecognised value as complete and log a warning; a test is where that is cheapest to catch, so it is an assertion failure here.
Options
:type,:queue,:id,:attempts— fields of theZizq.Jobthe handler receives.:attemptscounts attempts already finished, so it defaults to0.
@spec refute_enqueued(keyword()) :: true
Assert that no job matching filters was enqueued.
@spec setup_client(atom()) :: :ok
Point client at the recorder, and own its recordings for this test.
Called for you by use Zizq.Testing. Safe to call from several tests
at once: the client is shared, the recordings are not.
@spec start_link(keyword()) :: GenServer.on_start()
Start the recorder. Call once, from test/test_helper.exs.