View Source Oban.Testing (Oban v2.15.3)
This module simplifies testing workers and making assertions about enqueued jobs when testing in
:manual
mode.
Assertions may be made on any property of a job, but you'll typically want to check by args
,
queue
or worker
. If you're using namespacing through PostgreSQL schemas, also called
"prefixes" in Ecto, you should use the prefix
option when doing assertions about enqueued
jobs during testing. By default the prefix
option is public
.
Using in Tests
The most convenient way to use Oban.Testing
is to use
the module:
use Oban.Testing, repo: MyApp.Repo
That will define the helper functions you'll use to make assertions on the jobs that should (or should not) be inserted in the database while testing.
Along with the repo
you can also specify an alternate prefix to use in all assertions:
use Oban.Testing, repo: MyApp.Repo, prefix: "business"
Some example assertions:
# Assert that a job was already enqueued
assert_enqueued worker: MyWorker, args: %{id: 1}
# Assert that a job was enqueued or will be enqueued in the next 100ms
assert_enqueued [worker: MyWorker, args: %{id: 1}], 100
# Refute that a job was already enqueued
refute_enqueued queue: "special", args: %{id: 2}
# Refute that a job was already enqueued or would be enqueued in the next 100ms
refute_enqueued queue: "special", args: %{id: 2}, 100
# Make assertions on a list of all jobs matching some options
assert [%{args: %{"id" => 1}}] = all_enqueued(worker: MyWorker)
# Assert that no jobs are enqueued in any queues
assert [] = all_enqueued()
Note that the final example, using all_enqueued/1
, returns a raw list of matching jobs and
does not make an assertion by itself. This makes it possible to test using pattern matching at
the expense of being more verbose.
Example
Given a simple module that enqueues a job:
defmodule MyApp.Business do
def work(args) do
args
|> Oban.Job.new(worker: MyApp.Worker, queue: :special)
|> Oban.insert!()
end
end
The behaviour can be exercised in your test code:
defmodule MyApp.BusinessTest do
use ExUnit.Case, async: true
use Oban.Testing, repo: MyApp.Repo
alias MyApp.Business
test "jobs are enqueued with provided arguments" do
Business.work(%{id: 1, message: "Hello!"})
assert_enqueued worker: MyApp.Worker, args: %{id: 1, message: "Hello!"}
end
end
Matching Scheduled Jobs and Timestamps
In order to assert a job has been scheduled at a certain time, you will need to match against
the scheduled_at
attribute of the enqueued job.
in_an_hour = DateTime.add(DateTime.utc_now(), 3600, :second)
assert_enqueued worker: MyApp.Worker, scheduled_at: in_an_hour
By default, Oban will apply a 1 second delta to all timestamp fields of jobs, so that small
deviations between the actual value and the expected one are ignored. You may configure this
delta by passing a tuple of value and a delta
option (in seconds) to corresponding keyword:
assert_enqueued worker: MyApp.Worker, scheduled_at: {in_an_hour, delta: 10}
Adding to Case Templates
To include helpers in all of your tests you can add it to your case template:
defmodule MyApp.DataCase do
use ExUnit.CaseTemplate
using do
quote do
use Oban.Testing, repo: MyApp.Repo
import Ecto
import Ecto.Changeset
import Ecto.Query
import MyApp.DataCase
alias MyApp.Repo
end
end
end
Summary
Functions
Retrieve all currently enqueued jobs matching a set of options.
Assert that a job with particular options has been enqueued.
Assert that a job with particular options is or will be enqueued within a timeout period.
Construct a job and execute it with a worker module.
Refute that a job with particular options has been enqueued.
Refute that a job with particular options is or will be enqueued within a timeout period.
Change the testing mode within the context of a function.
Types
@type perform_opts() :: Oban.Job.option() | {:log, Logger.level()} | {:prefix, binary()} | {:repo, module()}
Functions
@spec all_enqueued(repo :: module(), opts :: Keyword.t()) :: [Oban.Job.t()]
Retrieve all currently enqueued jobs matching a set of options.
Only jobs matching all of the provided arguments will be returned. Additionally, jobs are returned in descending order where the most recently enqueued job will be listed first.
Examples
Assert based on only some of a job's args:
assert [%{args: %{"id" => 1}}] = all_enqueued(worker: MyWorker)
Assert that exactly one job was inserted for a queue:
assert [%Oban.Job{}] = all_enqueued(queue: :alpha)
Assert that there aren't any jobs enqueued for any queues or workers:
assert [] = all_enqueued()
Assert that a job with particular options has been enqueued.
Only values for the provided arguments will be checked. For example, an assertion made on
worker: "MyWorker"
will match any jobs for that worker, regardless of the queue or args.
@spec assert_enqueued(repo :: module(), opts :: Keyword.t(), timeout :: pos_integer()) :: true
Assert that a job with particular options is or will be enqueued within a timeout period.
See assert_enqueued/2
for additional details.
Examples
Assert that a job will be enqueued in the next 100ms:
assert_enqueued [worker: MyWorker], 100
@spec perform_job(worker :: Oban.Worker.t(), args :: term(), [perform_opts()]) :: Oban.Worker.result()
Construct a job and execute it with a worker module.
This reduces boilerplate when constructing jobs for unit tests and checks for common pitfalls.
For example, it automatically converts args
to string keys before calling perform/1
,
ensuring that perform clauses aren't erroneously trying to match atom keys.
The helper makes the following assertions:
- That the worker implements the
Oban.Worker
behaviour - That the options provided build a valid job
- That the return is valid, e.g.
:ok
,{:ok, value}
,{:error, value}
etc.
If all of the assertions pass then the function returns the result of perform/1
for you to
make additional assertions on.
Examples
Successfully execute a job with some string arguments:
assert :ok = perform_job(MyWorker, %{"id" => 1})
Successfully execute a job and assert that it returns an error tuple:
assert {:error, _} = perform_job(MyWorker, %{"bad" => "arg"})
Execute a job with the args keys automatically stringified:
assert :ok = perform_job(MyWorker, %{id: 1})
Exercise custom attempt handling within a worker by passing options:
assert :ok = perform_job(MyWorker, %{}, attempt: 42)
Cause a test failure because the provided worker isn't real:
assert :ok = perform_job(Vorker, %{"id" => 1})
Refute that a job with particular options has been enqueued.
See assert_enqueued/2
for additional details.
@spec refute_enqueued(repo :: module(), opts :: Keyword.t(), timeout :: pos_integer()) :: false
Refute that a job with particular options is or will be enqueued within a timeout period.
The minimum refute timeout is 10ms.
See assert_enqueued/2
for additional details.
Examples
Refute that a job will be enqueued in the next 100ms:
refute_enqueued [worker: MyWorker], 100
Change the testing mode within the context of a function.
Only :manual
and :inline
mode are supported, as :disabled
implies that supervised queues
and plugins are running and this function won't start any processes.
Examples
Switch to :manual
mode when an Oban instance is configured for :inline
testing:
Oban.Testing.with_testing_mode(:manual, fn ->
Oban.insert(MyWorker.new(%{id: 123}))
assert_enqueued worker: MyWorker, args: %{id: 123}
end)
Visa-versa, switch to :inline
mode:
Oban.Testing.with_testing_mode(:inline, fn ->
{:ok, %Job{state: "completed"}} = Oban.insert(MyWorker.new(%{id: 123}))
end)