Testing

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

PythonNode.jsJava 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.

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:1.0.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());
    }
}

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()).

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);

`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());
}

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.