Test helpers for Shigoto: run workers with no database, and assert on the jobs your code enqueued.

Running a worker (shigoto:perform_job/2,3)

shigoto:perform_job/2,3 runs a worker through the exact perform path production uses (the middleware chain and deps_results injection) and returns the raw worker result, with no database, no queue, no retry, and no resilience gating (rate limits, circuit breakers, concurrency caps are skipped). It is a pure, in-process unit-test runner — it must never be used on a production path.

ok = shigoto:perform_job(email_worker, #{~"to" => ~"a@b.com"}),
{ok, Total} = shigoto:perform_job(sum_worker, #{~"xs" => [1, 2, 3]}),
%% predecessors' results, exactly as a real dependent would see them:
{ok, _} = shigoto:perform_job(rollup_worker, #{}, #{deps_results => #{41 => #{~"n" => 7}}}).

Testing modes

Set a testing mode so insert/1,2 and insert_all/1,2 stop persisting. Both keys are required — see shigoto_config:testing_mode/0 for why:

application:set_env(shigoto, testing, manual),
application:set_env(shigoto, testing_confirm_no_persistence, true).

• inline runs each inserted job synchronously and keeps no row. A job that returns {error, _} raises {shigoto_inline_job_failed, Worker, Reason} so failures surface loudly in the test.
• manual captures inserted jobs in a process-local buffer without running or persisting them, for the assertions below.

manual capture is per-process: it sees enqueues made by the calling process, not by other processes it spawned. For cross-process expectations, run against a real test pool (the assertions fall back to querying the pool when not in manual mode) and clean the table between tests.

Assertions

my_code_that_enqueues(),
shigoto_testing:assert_enqueued(#{worker => email_worker}),
shigoto_testing:refute_enqueued(#{queue => ~"sms"}),
[_ | _] = shigoto_testing:all_enqueued().

Filters: worker, queue, args (containment), tags (containment), state. Call reset/0 in your test setup to clear the manual buffer.