Testing tasks

Unit-test tasks in isolation, then exercise the full enqueue → execute → result path without a production backend.

PythonNode.jsJavaRust tasks are plain functions, so the fastest tests call them directly with no queue involved. For coverage of enqueue, dispatch, and result plumbing, each binding also gives you a way to run the whole path without standing up a production backend:

test_mode() patches enqueue() so every .delay() / .apply_async() call runs the task synchronously in the calling thread — no worker, no Rust scheduler, no SQLite. This makes tests fast, deterministic, and easy to write.

There's no synchronous test mode — tests run a real worker against a throwaway SQLite database, so the whole enqueue → execute → result path runs in-process with no mocks of the core.

The flexiq-test artifact ships a pure-Java, in-memory QueueBackend — the full client API with no native library and no disk — ideal for fast unit tests of producers, handlers, retries, and dead-lettering.

There is no test mode, and nothing to patch — nothing is in the way to begin with. #[flexiq::task] leaves the body as an ordinary function reachable as <name>::run(args…), and FlexiQ::in_memory() opens a private database that dies with the handle. Unit tests call the body; integration tests run a real scheduler and a real worker in the test process.

Quick example

python

from flexiq import Queue

queue = Queue()

@queue.task()
def add(a: int, b: int) -> int:
    return a + b

def test_add():
    with queue.test_mode() as results:
        add.delay(2, 3)

        assert len(results) == 1
        assert results[0].return_value == 5
        assert results[0].succeeded

import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, expect, it } from "vitest";
import { Queue, type Worker } from "@byteveda/flexiq";

let worker: Worker | undefined;
afterEach(() => {
  worker?.stop(); // always stop — a leaked worker keeps polling across tests
  worker = undefined;
});

function newQueue(): Queue {
  // a fresh temp DB per test keeps them independent
  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "test-")), "q.db") });
}

it("runs a task end to end", async () => {
  const queue = newQueue();
  queue.task("add", (a: number, b: number) => a + b);

  const id = queue.enqueue("add", [2, 3]);
  worker = queue.runWorker();

  expect(await queue.result(id)).toBe(5);
});

kotlin

testImplementation("org.byteveda:flexiq-test:2.1.0")

java

import org.byteveda.flexiq.test.InMemoryFlexiQ;

@Test
void runsATaskEndToEnd() throws Exception {
    try (FlexiQ flexiq = InMemoryFlexiQ.open()) {
        String id = flexiq.enqueue("add", List.of(2, 3));

        try (Worker worker = flexiq.worker()
                .handle("add", List.class, numbers ->
                        (int) numbers.get(0) + (int) numbers.get(1))
                .start()) {
            Job job = flexiq.awaitJob(id, Duration.ofSeconds(5)).orElseThrow();
            assertEquals(JobStatus.COMPLETE, job.status);
        }

        assertEquals(5, flexiq.getResult(id, Integer.class).orElseThrow());
    }
}

#[flexiq::task] leaves the body as an ordinary function under run, so the fast test calls it with no queue at all. FlexiQ::in_memory() covers the rest — a real scheduler and a real worker over a database that dies with the test.

rust

use std::time::{Duration, Instant};

use flexiq::{FlexiQ, Job, JobStatus};

#[flexiq::task]
fn add(a: i64, b: i64) -> flexiq::Outcome<i64> {
    Ok(a + b)
}

#[test]
fn adds() {
    assert_eq!(add::run(2, 3).expect("runs"), 5);
}

#[test]
fn runs_a_task_end_to_end() {
    let queue = FlexiQ::in_memory().expect("opens");
    let job = queue.enqueue(add::call(2, 3)).expect("enqueues");

    let worker = queue.worker().register::<add>().spawn().expect("spawns");
    let done = wait_terminal(&queue, &job.id);
    worker.shutdown().expect("clean shutdown"); // never leak a worker between tests

    assert_eq!(done.status, JobStatus::Complete);
    // `job.result` is the archived bytes; the shell does not decode them for you.
    assert!(done.result.is_some());
}

/// Poll to a terminal state on a budget — the scheduler's poll interval is its
/// own, and a test that sleeps exactly one interval flakes on a slower machine.
fn wait_terminal(queue: &FlexiQ, job_id: &str) -> Job {
    let deadline = Instant::now() + Duration::from_secs(20);
    while Instant::now() < deadline {
        let job = queue.get_job(job_id).expect("reads").expect("job exists");
        if matches!(
            job.status,
            JobStatus::Complete | JobStatus::Failed | JobStatus::Dead
        ) {
            return job;
        }
        std::thread::sleep(Duration::from_millis(25));
    }
    panic!("job {job_id} never reached a terminal state");
}

InMemoryFlexiQ.open(serializer) swaps the JSON default. To construct the backend explicitly (e.g. to share one across a custom builder), pass it to FlexiQ.builder().open(new InMemoryQueueBackend()).

Tasks that use durable steps run here too, as a real sequence rather than a bypass — a step is memoized across a retry, a changed sequence diverges, and step.sleep reschedules the job. What the harness does not model, and when to reach for a file-backed queue instead, is covered in that guide.

Not supported: workflows. Periodic registration is recorded but never fires.

Two things about run before you lean on it.

Its error type is Abort, not QueueError: "this attempt failed" and "the queue could not be reached" are different facts, and ? deliberately does not bridge them. A test that asserts a failure matches Abort::Fail(err) and reads err.message and err.retryable — the retryable/fatal choice is the task author's, and nothing guesses it for you.

And run is the body alone: no queue, no scheduler, and no durable-step session. A body that calls flexiq::current_step() under run gets a fatal error saying it was called outside a running task. It will not panic, but it will not memoize either, so steps have to be tested through the worker — where FlexiQ::in_memory() runs them for real, skipping a committed step on the retry and rescheduling the job on a step.sleep.

How it works

When you enter queue.test_mode(), flexiq patches the enqueue() method so that every .delay() or .apply_async() call:

Looks up the task function in the registry
Calls it immediately in the current thread
Captures the return value (or exception) in a TestResult
Appends the result to the TestResults list

No database is created. No worker threads are spawned. Tasks execute eagerly and synchronously.

`queue.test_mode()`

python

with queue.test_mode(propagate_errors=False, resources=None) as results:
    # tasks run synchronously here
    ...

Parameter	Type	Default	Description
`propagate_errors`	`bool`	`False`	If `True`, task exceptions are re-raised immediately instead of being captured in `TestResult.error`
`resources`	`dict[str, Any] \| None`	`None`	Map of resource name → mock instance or `MockResource` for injection. See Resource System.

The context manager yields a TestResults list that accumulates results as tasks execute.

`TestResult`

Each executed task produces a TestResult:

python

with queue.test_mode() as results:
    add.delay(2, 3)

    r = results[0]
    r.job_id        # "test-000001"
    r.task_name     # "mymodule.add"
    r.args          # (2, 3)
    r.kwargs        # {}
    r.return_value  # 5
    r.error         # None
    r.traceback     # None
    r.succeeded     # True
    r.failed        # False

Attribute	Type	Description
`job_id`	`str`	Synthetic ID like `"test-000001"`
`task_name`	`str`	Fully qualified task name
`args`	`tuple`	Positional arguments passed to the task
`kwargs`	`dict`	Keyword arguments passed to the task
`return_value`	`Any`	Return value on success, `None` on failure
`error`	`Exception \| None`	The exception if the task failed
`traceback`	`str \| None`	Formatted traceback if the task failed
`succeeded`	`bool`	`True` if no error
`failed`	`bool`	`True` if an error occurred

`TestResults`

TestResults is a list of TestResult with convenience methods:

python

with queue.test_mode() as results:
    add.delay(2, 3)
    failing_task.delay()
    add.delay(10, 20)

    # Filter by outcome
    results.succeeded   # TestResults with 2 items
    results.failed      # TestResults with 1 item

    # Filter by task name
    results.filter(task_name="mymodule.add")  # 2 items

    # Combine filters
    results.filter(task_name="mymodule.add", succeeded=True)  # 2 items

`.filter()`

python

results.filter(task_name=None, succeeded=None) -> TestResults

Parameter	Type	Description
`task_name`	`str \| None`	Filter by exact task name
`succeeded`	`bool \| None`	`True` for successes, `False` for failures

Testing failures

By default, task exceptions are captured — not raised:

python

@queue.task()
def risky():
    raise ValueError("something broke")

def test_failure_captured():
    with queue.test_mode() as results:
        risky.delay()

        assert len(results) == 1
        assert results[0].failed
        assert isinstance(results[0].error, ValueError)
        assert "something broke" in str(results[0].error)
        assert results[0].traceback is not None

Propagating errors

Use propagate_errors=True when you want exceptions to bubble up:

python

def test_failure_propagated():
    with queue.test_mode(propagate_errors=True) as results:
        with pytest.raises(ValueError, match="something broke"):
            risky.delay()

Testing workflows

Chains and groups work in test mode because they call enqueue() internally, which is intercepted by the test mode patch. Chords do too — the callback receives the list of group results as its first argument:

python

from flexiq import chord

@queue.task()
def sum_results(values: list[int]) -> int:
    return sum(values)

def test_chord():
    with queue.test_mode() as results:
        chord([double.s(1), double.s(2)], sum_results.s()).apply()

        # 2 group tasks + 1 callback = 3 results
        assert len(results) == 3
        assert results[-1].return_value == 6  # sum([double(1), double(2)]) = sum([2, 4])

Chains

python

from flexiq import chain

@queue.task()
def double(n: int) -> int:
    return n * 2

@queue.task()
def add_ten(n: int) -> int:
    return n + 10

def test_chain():
    with queue.test_mode() as results:
        chain(double.s(5), add_ten.s()).apply()

        assert len(results) == 2
        assert results[0].return_value == 10   # double(5)
        assert results[1].return_value == 20   # add_ten(10)

Groups

python

from flexiq import group

def test_group():
    with queue.test_mode() as results:
        group(double.s(1), double.s(2), double.s(3)).apply()

        assert len(results) == 3
        values = [r.return_value for r in results]
        assert values == [2, 4, 6]

Job context in tests

current_job works inside test mode. The context is set up before each task runs:

python

from flexiq import current_job

@queue.task()
def context_aware():
    return {
        "job_id": current_job.id,
        "task_name": current_job.task_name,
        "retry_count": current_job.retry_count,
        "queue_name": current_job.queue_name,
    }

def test_context():
    with queue.test_mode() as results:
        context_aware.delay()

        ctx = results[0].return_value
        assert ctx["job_id"].startswith("test-")
        assert ctx["retry_count"] == 0
        assert ctx["queue_name"] == "default"

Faking dependencies

Register a stub factory to swap a real dependency in tests — the worker injects whatever is registered by name, so tasks can't tell a fake from the real thing and no real connection is ever opened. See dependency injection for how resources are registered and scoped.

python

from unittest.mock import MagicMock

@queue.worker_resource("db")
def create_db():
    return real_sessionmaker

@queue.task(inject=["db"])
def create_user(name: str, db):
    session = db()
    session.add(User(name=name))
    session.commit()

def test_create_user():
    mock_db = MagicMock()

    with queue.test_mode(resources={"db": mock_db}) as results:
        create_user.delay("Alice")

    assert results[0].succeeded
    mock_db.return_value.add.assert_called_once()

import { useResource } from "@byteveda/flexiq";

queue.resource("db", () => fakeDb);
queue.task("sync", async () => {
  const db = await useResource("db");
  // ...
});

java

flexiq.resource("db", context -> fakeDb);

Nothing to fake at the queue: a Rust task takes its arguments and nothing else, so there is no resource registry to register a stub with and no injection for the worker to do. A dependency is swapped the way it would be in any other Rust code — behind a trait the body resolves, or a value the test installs before it starts the worker.

`MockResource`

MockResource adds call tracking to a mock value:

python

from flexiq import MockResource

spy = MockResource("db", wraps=real_db, track_calls=True)

with queue.test_mode(resources={"db": spy}) as results:
    create_user.delay("Alice")

assert spy.call_count == 1
assert results[0].succeeded

Parameter	Type	Description
`name`	`str`	Resource name (informational).
`return_value`	`Any`	Value returned when the resource is accessed.
`wraps`	`Any`	Wrap a real object — returned as-is when accessed.
`track_calls`	`bool`	Increment `call_count` each access.

`return_value` vs `wraps`

Use return_value when you want a simple stub:

python

mock_cache = MockResource("cache", return_value={"key": "value"})

Use wraps when you need the real object but want call tracking:

python

real_db = create_test_database()
spy_db = MockResource("db", wraps=real_db, track_calls=True)

Multiple resources

Pass multiple resources to test_mode:

python

with queue.test_mode(resources={
    "db": MockResource("db", return_value=mock_db),
    "cache": MockResource("cache", return_value={}),
    "mailer": MockResource("mailer", return_value=mock_smtp),
}) as results:
    process_order.delay(order_id=123)

Testing with `inject`

Tasks that use @queue.task(inject=["db"]) receive the mock resource automatically:

python

@queue.task(inject=["db"])
def create_user(name, db=None):
    db.execute("INSERT INTO users (name) VALUES (?)", (name,))

mock_db = MagicMock()
with queue.test_mode(resources={"db": mock_db}) as results:
    create_user.delay("Alice")

assert results[0].succeeded
mock_db.execute.assert_called_once()

When resources= is provided, proxy reconstruction is bypassed automatically. Proxy markers in arguments are passed through as-is so tests don't fail due to missing files or network connections.

`mockResource()`

mockResource(value) wraps a value as a factory and records how often it was built — register the factory and assert on resolutions to confirm a worker-scoped resource was built once, not once per job:

import { mockResource, Queue, useResource } from "@byteveda/flexiq";

const db = mockResource({ query: () => 7 });
queue.resource("db", db.factory);

queue.task("read", async () => {
  const conn = await useResource<{ query: () => number }>("db");
  return conn.query();
});

queue.enqueue("read");
queue.enqueue("read");
worker = queue.runWorker();

// worker-scoped singleton, built once even though two jobs ran
await waitFor(() => db.resolutions === 1);

Asserting resource construction

InMemoryFlexiQ paired with a counting factory confirms a resource was built the expected number of times, without touching a native backend:

java

try (FlexiQ queue = InMemoryFlexiQ.open()) {
    AtomicInteger built = new AtomicInteger();
    queue.resource("db", ctx -> {
        built.incrementAndGet();
        return new FakeDatabase();
    });
    // ... run the task through a worker ...
    assertEquals(1, built.get());
    assertEquals(1, queue.resourceMetrics().get("db").created());
}

Asserting what startup refuses

The registration checks that do exist run at spawn(), and they are worth a test rather than a mock. Registering one task name twice is refused there with an error naming it — one of the two bodies would otherwise silently never run — and so is registering any scheduled task on a namespaced handle. Both come back as Err from spawn():

rust

match queue.worker().register::<add>().register::<add>().spawn() {
    Ok(handle) => {
        // A handle that reaches here is a live worker, not a failed assertion.
        handle.shutdown().expect("clean shutdown");
        panic!("a duplicate registration must be refused");
    }
    Err(err) => assert!(err.to_string().contains("add"), "message: {err}"),
}

WorkerHandle does not implement Debug, so expect_err will not compile here — match, and shut down the handle on the arm that should not have happened.

Pytest integration

Fixture pattern

Create a reusable fixture for test mode:

python

# conftest.py
import pytest
from myapp import queue

@pytest.fixture
def task_results():
    with queue.test_mode() as results:
        yield results

# test_tasks.py
def test_add(task_results):
    add.delay(2, 3)
    assert task_results[0].return_value == 5

def test_email(task_results):
    send_email.delay("user@example.com", "Hello", "World")
    assert task_results[0].succeeded

Fixture with error propagation

python

@pytest.fixture
def strict_tasks():
    with queue.test_mode(propagate_errors=True) as results:
        yield results

Testing async code

Test mode works with async test functions — the tasks still execute synchronously:

python

import pytest

@pytest.mark.asyncio
async def test_async_enqueue(task_results):
    add.delay(1, 2)
    assert task_results[0].return_value == 3

What test mode does NOT cover

Test mode is designed for unit and integration testing of task logic. It does not exercise:

SQLite storage or queries
Retry/backoff scheduling
Rate limiting
Timeout reaping
Worker thread pool dispatch
Priority ordering

For end-to-end tests that exercise the full Rust scheduler, run a real worker in a background thread:

python

import threading
import time

def test_e2e():
    queue_e2e = Queue(db_path=":memory:")

    @queue_e2e.task()
    def add(a, b):
        return a + b

    t = threading.Thread(target=queue_e2e.run_worker, daemon=True)
    t.start()

    job = add.delay(2, 3)
    result = job.result(timeout=10)
    assert result == 5

Middleware in test mode

Per-task and queue-level TaskMiddleware hooks (before, after, on_retry) do fire in test mode, since they run in the Python wrapper around your task function. This lets you verify middleware behavior in tests without running a real worker.

Contributing: flexiq's own test suite

This section is for people contributing to flexiq itself — building the Rust extension and running its internal test suite. If you're testing tasks in your own application, the sections above (test_mode(), TestResult, MockResource) are what you want instead.

bash

# Rust tests
cargo test --workspace

# Rebuild the Python extension after Rust changes
uv run maturin develop

# Python tests
uv run python -m pytest tests/python/ -v

# Linting (run from sdks/python/)
uv run ruff check flexiq/ tests/
uv run mypy flexiq/ --no-incremental

To build with native async support:

bash

uv run maturin develop --features native-async

Synchronization

awaitJob(id, timeout) blocks until the job reaches a terminal state and is the simplest synchronization point — tests rarely need a sleep. Registering an event listener with a CountDownLatch works too:

java

CountDownLatch done = new CountDownLatch(1);
Worker worker = flexiq.worker()
        .handle("echo", String.class, payload -> payload.length())
        .on(EventName.SUCCESS, event -> done.countDown())
        .start();

Integration tests

For coverage of the real storage engine (scheduling, claims, locks), open a throwaway SQLite database — the whole enqueue → execute → result path runs in-process:

java

Path dir = Files.createTempDirectory("flexiq-test");
try (FlexiQ flexiq = FlexiQ.builder().sqlite(dir.resolve("q.db").toString()).open()) {
    // real backend, no server
}

Always close (or try-with-resources) workers — a leaked worker keeps polling across tests.

In-memory backend scope

The in-memory backend covers the queue contract, not the native engine: workflows are not supported in-memory, and storage-engine behavior (SQL claims, cron timing) is only exercised against a real backend. Keep a few SQLite-backed integration tests alongside the fast in-memory suite.

Polling for side effects

When you assert on a side effect rather than a return value, poll until it settles instead of sleeping a fixed time:

async function waitFor(predicate: () => boolean, timeoutMs = 4000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    if (predicate()) return true;
    await new Promise((r) => setTimeout(r, 20));
  }
  return false;
}

Awaiting queue.result(id) (shown in the quick example above) is the simplest synchronization point — it resolves when the job reaches a terminal state and rejects on failure, so most tests don't need waitFor at all. Reach for it when you're asserting on a side effect instead of a return value.

What the in-memory queue does not skip

FlexiQ::in_memory() is a real SQLite database rather than a bypass: the scheduler, the claim protocol, the retry policy and the timeout reap all run. That is the point of it, and it is also why these tests have to be written like tests against a real system.

Poll to a terminal state on a budget. Nothing blocks waiting for a result. The scheduler's poll interval is its own, so a test that sleeps exactly one interval flakes on a slower machine — loop against a deadline, as the example above does.
Always shutdown() the worker. Dropping the handle stops nothing. shutdown() is what stops dispatch, drains in-flight work, ends the heartbeat and unregisters the worker; skip it and the next test inherits a live one.
Give each test its own handle. An in-memory database is private to the handle that opened it, so two tests never see each other's jobs — but a process-global counter cannot tell which test's scheduler incremented it. When tests run in parallel, assert through that handle's own reads.
job.result is bytes. It is an Option<Vec<u8>> holding the encoded return value, and the crate does not decode it for you. Assert on the status, on a side effect, or on the bytes.

Testing a scheduled task

A cron task is registered at spawn(), so the registration is assertable without waiting for the schedule to come round:

rust

#[flexiq::task(cron = "0 0 3 * * *", timezone = "Europe/Stockholm")]
fn nightly() -> flexiq::Outcome<()> {
    Ok(())
}

#[test]
fn the_schedule_is_registered_at_startup() {
    let queue = FlexiQ::in_memory().expect("opens");
    let worker = queue.worker().register::<nightly>().spawn().expect("spawns");

    let registered = queue.list_periodic().expect("lists");
    worker.shutdown().expect("clean shutdown");

    assert_eq!(registered[0].name, "nightly");
    assert_eq!(registered[0].cron_expr, "0 0 3 * * *");
    assert!(registered[0].enabled);
}

Six fields, seconds first. Firing is per-scheduler and not leader-elected, so two workers that both registered the same schedule can both fire it — keep it to one worker per test.