Testing Tasks¶

TestQueue runs tasks synchronously in the test process without a separate worker or redis. It is designed for simple testing tasks in unit tests by providing deterministic synchronous execution.

Basic testing¶

Here is a simple task to test:

tasks.py

from sheppy import task

@task
async def add(x: int, y: int) -> int:
    return x + y

And its test:

tests/test_tasks.py

from sheppy import TestQueue
from tasks import add

def test_add():
    q = TestQueue()

    # instantiate task
    t = add(1, 2)

    # add to the test queue
    q.add(t)

    # process the next task in the queue
    processed_task = q.process_next()

    # verify the task result
    assert processed_task.status == 'completed'
    assert processed_task.exception is None
    assert processed_task.result == 3

Differences from "full" Queue:

Synchronous API, so no await and event loops
Tasks only run when you call process_next() or process_all()
Everything happens in the test process and stored in-memory

Testing task failures¶

When a task raises, the exception is captured in the exception attribute instead of propagating, so you can assert on it:

tests/test_failure.py

from sheppy import TestQueue, task


@task
def divide(x: int, y: int) -> float:
    return x / y


def test_divide_by_zero():
    q = TestQueue()

    # instantiate two tasks
    t1 = divide(1, 2)
    t2 = divide(1, 0)

    # add both tasks to the test queue
    q.add([t1, t2])

    # process all tasks in the queue (processed in order)
    processed_tasks = q.process_all()

    # verify the first task result
    assert processed_tasks[0].status == 'completed'
    assert processed_tasks[0].exception is None
    assert processed_tasks[0].result == 0.5

    # verify the second task result (should fail)
    assert processed_tasks[1].status == 'failed'
    assert str(processed_tasks[1].exception) == "ZeroDivisionError: division by zero"

Testing retry logic¶

Retries in TestQueue happen immediately, with no delay, so retry behavior is fast to test:

tests/test_retry_logic.py

from sheppy import TestQueue, task

FAIL_COUNTER = 0

@task(retry=2, retry_delay=0)
def fail_once() -> str:
    global FAIL_COUNTER

    if FAIL_COUNTER < 1:
        FAIL_COUNTER += 1
        raise ValueError("task failed")

    return "success"

def test_fail_once():
    q = TestQueue()

    # instantiate the task
    t = fail_once()

    # add the task to the test queue
    q.add(t)

    assert q.size() == 1  # one task in the queue

    # process all tasks in the queue
    processed = q.process_all()

    # there should be two processed tasks: the original + one retry
    assert len(processed) == 2

    # verify the first attempt failed and is marked for retry
    assert processed[0].status == 'retrying'
    assert str(processed[0].exception) == "ValueError: task failed"

    # retry should succeed
    assert processed[1].status == 'completed'
    assert processed[1].exception is None
    assert processed[1].result == "success"

    # both processed tasks should have the same id
    assert processed[0].id == processed[1].id

Note the first processed attempt has status retrying, not failed. A task is only failed once no attempts remain. Both attempts share the same task id.

Overriding dependencies¶

Tasks that use Depends can swap their dependencies in tests with dependency_overrides:

example_dependency.py

from sheppy import Depends, TestQueue, task


def get_db():
    return "PROD DATABASE"



@task
def my_task(db: str = Depends(get_db)):
    print("my database is: ", db)
    return db


def test_dependency_override():
    def alternative_get_db():
        return "TESTING DATABASE"

    q = TestQueue(dependency_overrides={get_db: alternative_get_db})

    q.add(my_task())

    processed = q.process_next()

    assert processed.status == 'completed'
    assert processed.result == "TESTING DATABASE"

The task resolves get_db through the override map and receives "TESTING DATABASE". The same parameter exists on Worker and InMemoryBackend for integration tests.

examples/example_timeout.py shows the same style of test for timeout and retry_on_timeout.