Django 6.0 Tasks with TDD: three walls the happy path hides

Django 6.0 introduced a native tasks framework. I followed the tutorial, it worked on the first try — and that was exactly when I grew suspicious. I decided to rebuild everything with TDD, writing the test before every line of code. I hit three walls that the happy path doesn't reveal.

This article documents the decisions and stumbling blocks of cotton-desk-tasks, a fictional cotton trading desk built on top of Real Python's Django Tasks tutorial. The domain isn't just decoration: a trading desk is literally a place where fast-paced work and slow-running jobs compete for the same process — which is precisely the problem the framework was built to solve.

(All data is synthetic. No real contracts, lab reports, or prices.)

The Problem

An HVI report comes in from the lab, a contract is closed, a price index is ingested, a position report is generated. Four jobs with completely different criticality and latency:

• Summarizing an HVI report — the classifier is waiting on the screen. It needs to be fast.
• Confirming a contract — cannot trigger before the contract exists in the database.
• Crop report — scans thousands of HVI reports. Can wait.
• Price ingestion — once a day, after market close.

If all of this lands in a single queue, a slow report blocks a contract confirmation. The framework solves this with queue_name and priority:

@task(queue_name="hvi_reports", priority=50)
def summarize_hvi_report(report_id: int) -> str: ...

@task(queue_name="crop_reports", priority=-10)
def generate_crop_report(season: str) -> str: ...

With a dedicated worker per queue (db_worker --queue-name hvi_reports), isolation is structural: the reports worker remains blind to anything that isn't its own.

The Boundary: Permissive Storage, Strict Domain

Before tasks, one decision shaped the rest of the project: where business logic lives.

An HVI report has four parameters with commercial ranges — micronaire between 3.5 and 4.9, minimum length of 1.11", minimum strength of 28 gf/tex, minimum uniformity of 80%. The temptation is to validate this in the model's save(). But a lab measures what it measures: if micronaire came back as 2.1, that's a fact that must be recorded, not an error to reject.

So validation doesn't live in the model. It lives in an immutable value object, completely free of Django:

@dataclass(frozen=True)
class HVIParameters:
    micronaire: float
    length: float
    strength: float
    uniformity: float

    def __post_init__(self) -> None:
        self._validate_micronaire()
        self._validate_length()
        self._validate_strength()
        self._validate_uniformity()

And the model exposes an explicit crossing of that boundary:

def to_domain(self) -> HVIParameters:
    return HVIParameters(micronaire=float(self.micronaire), ...)

HVIReport.objects.create() with a micronaire of 2.1 saves without issue. The error is only raised when someone asks for the domain object. The test proving this is the most important one in the project:

report = HVIReport.objects.create(bale=bale, micronaire="2.00", ...)  # passes

with pytest.raises(MicronaireOutOfRange):
    report.to_domain()  # here it fails, as expected

This pays dividends later: when the task fails, it fails with desk.domain.MicronaireOutOfRange in the traceback — not with an anonymous ValueError.

Wall 1: The Test Backend Cannot Retrieve Results

The natural flow of an asynchronous API consists of two requests: a POST enqueues and returns the ID, followed by a GET checking the status later. I wrote the test, wrote the view, ran it:

NotImplementedError: This backend does not support retrieving or refreshing results.

ImmediateBackend — the built-in backend that runs tasks inline, ideal for testing — does not implement get_result(). And the reason is structural, not an overlooked feature: it never persists anything anywhere. The only TaskResult that exists is the object returned by .enqueue() at that exact moment. Querying by ID later means looking for something that was never stored.

This broke an assumption I had baked into conftest.py. I had forced ImmediateBackend across the entire test suite via an autouse fixture so tests wouldn't depend on a worker running in parallel. A good decision — yet completely broken for the exact view built for the scenario this backend doesn't support.

The solution was to separate the concerns. "Can the backend retrieve the result?" is the library's responsibility, already tested upstream. "Given a TaskResult in a specific state, does the view return the correct JSON and HTTP status?" is mine — and it can be tested without a database or backend:

def test_get_task_status_with_invalid_report_returns_failed(client):
    fake_error = MagicMock(exception_class_path="desk.domain.MicronaireOutOfRange")
    fake_result = MagicMock(status=TaskResultStatus.FAILED, errors=[fake_error])

    with patch("desk.views.default_task_backend.get_result", return_value=fake_result):
        response = client.get(reverse("task_status", args=["any-id"]))

    assert response.status_code == 422

422 Unprocessable Entity, not 500: the request itself is processable; the data just isn't commercially valid.

Wall 2: Decimal Dies in Serialization, and the Error Hits Before the Worker

Price is Decimal. On a trading desk, that's not a preference, it's mandatory — using float for prices is how you pay dearly for rounding errors. So the ingestion task seemed straightforward:

record_index_reading.enqueue("ICE-CT2", Decimal("82.35"), "2026-04-10")

TypeError: Unsupported type: <class 'decimal.Decimal'>

The crucial detail: the error blows up inside .enqueue(), during argument serialization, before ever touching the backend. It's not a database issue — task arguments must be JSON-serializable, and validation happens at enqueue time. A Decimal buried three levels deep in a dictionary will only surface right there.

The fix isn't technical; it's contractual. The function signature now documents the rule:

@task(queue_name="prices", priority=0)
def record_index_reading(code: str, value: str, trading_date: str) -> str:
    """`value` arrives as a string, not Decimal: task arguments undergo
    JSON serialization in `.enqueue()`, and Decimal does not survive this
    round-trip — callers of this task must convert beforehand.
    """
    reading, _ = PriceIndex.objects.update_or_create(
        code=code, trading_date=trading_date, defaults={"value": Decimal(value)},
    )

value: str is the type checker warning you at the call site before runtime. The conversion to Decimal happens across the internal boundary, where the proper context exists.

Wall 3: on_commit Never Fires in Tests — and Mocking Explodes

The classic gotcha: if the view creates the contract inside a transaction and enqueues the confirmation in the same breath, the worker might fetch the contract before the commit and find nothing. The solution is well-known:

with transaction.atomic():
    contract = Contract.objects.create(...)
    transaction.on_commit(partial(confirm_contract.enqueue, contract.id))

The problem is testing this. I deliberately wrote the naive test, mocking enqueue to check that it had been called:

with patch("desk.views.confirm_contract.enqueue") as enqueue_mock:

TypeError: super(type, obj): obj must be an instance or subtype of type

Bizarre error, simple cause: the @task decorator turns the function into an instance of Task, which is a frozen dataclass. unittest.mock.patch works by executing setattr on the target upon entering the with block and delattr upon exiting — and a frozen object rejects both operations. You cannot mock methods on a Task.

(Fittingly consistent: the framework applied the exact same principle to Task that I applied to HVIParameters. An object representing a fact shouldn't change after creation.)

Yet the bigger issue lay beneath that. @pytest.mark.django_db wraps each test in a transaction that is rolled back at the end — it never commits. And on_commit only fires when the transaction actually commits. Meaning: even with a flawless view, the callback would never run in the test. The naive test would have passed or failed for reasons completely unrelated to what it claimed to verify.

The right tool is a fixture provided by pytest-django itself:

with django_capture_on_commit_callbacks() as callbacks:
    response = client.post(reverse("checkout"), {...})

assert response.status_code == 201
assert len(callbacks) == 1

Without execute=True: the test proves the structure — exactly one callback scheduled. Neither zero (the classic bug of enqueuing before commit) nor executed immediately on the spot (which would mean not using on_commit at all).

Testing with a Real Worker, Without Opening a Terminal

The three gotchas above can be tested using the inline backend. But I wanted an end-to-end proof: a task truly enqueued, a real worker processing it, a genuine failure, and the HTTP endpoint returning 422 — zero mocks, and without relying on me remembering to spin up a background process in another terminal tab.

db_worker supports --batch: it processes whatever is ready and exits. You can call it right from inside the test. The first attempt broke:

OperationalError: cannot start a transaction within a transaction

The worker issues a BEGIN EXCLUSIVE to safely lock the queue against other workers. SQLite doesn't allow this inside an already open transaction — and @pytest.mark.django_db opens one. The fix is transaction=True, which disables the transaction wrapping and allows the test to commit just as production would:

@pytest.mark.django_db(transaction=True)
def test_invalid_report_actually_fails_with_real_worker(client):
    ...
    result = summarize_hvi_report.enqueue(report.id)
    assert result.status == TaskResultStatus.READY

    call_command("db_worker", queue_name="hvi_reports", batch=True, verbosity=0)

    final_result = default_task_backend.get_result(result.id)
    assert final_result.status == TaskResultStatus.FAILED
    assert final_result.errors[0].exception_class_path == "desk.domain.MicronaireOutOfRange"

It is the slowest test in the suite, and the only one of its kind. The price to pay for verifying behavior that only exists when transactions genuinely commit.

The Wall That Isn't the Framework's Fault: Changing QUEUES Doesn't Migrate Existing Tasks

This one didn't show up in tests. It appeared while actually running the project.

In the early stages, the only queue was default. Later, I migrated to four named queues. Weeks of commits later, I spun up a worker listening to everything (--queue-name '*') and got:

InvalidTaskError: Queue 'default' is not valid for backend.

Tasks enqueued weeks earlier with a queue name that no longer existed in configuration sat stranded in the table, waiting for a worker that would never come. Updating QUEUES changes what the backend accepts going forward — it does nothing to what is already stored.

In development, cleanup is a single line. In production, it requires a migration plan: drain the old queue before deploying, or keep the old queue name accepted throughout the transition window. It's the kind of thing you won't find in tutorials because tutorials don't have a history.

The Dashboard, and the Lie I Refused to Tell

I built a dashboard to see the queues at work — each queue is a conveyor belt, each task a badge changing color based on its state.

Then came the frustration: badges never showed up as "running". They jumped straight from READY to SUCCESSFUL. The reason is honest — summarize_hvi_report performs a single database read and some string formatting. It completes in milliseconds. The dashboard polls every 1.5 seconds. There is simply nothing to see.

The lazy fix would be dropping a time.sleep(2) inside the task. I refused: that would corrupt business logic for the sake of visual flair, and the dashboard would portray latency that doesn't actually exist.

First, I tried the legitimate route: increasing the worker's --interval. It barely helped, leading to an interesting realization — the interval controls the wait time when the queue is empty, not execution duration. The moment the worker finds seven reports, it processes all seven in one go.

The honest workaround was separating concerns: a fifth queue, demo, with a task whose name and docstring make it explicit that slowness is its purpose.

@task(queue_name="demo", priority=0)
def demo_task() -> str:
    """Artificially slow task, solely for the dashboard to display the 'running' state.

    The `sleep` here is intentional and honest: simulating latency IS the purpose
    of this task. It plays no role in the business domain — it exists solely to make
    the READY → RUNNING → SUCCESSFUL transition visible, which in real (fast)
    tasks happens too quickly for the human eye to track.
    """
    time.sleep(4)

The four business queues remain fast and truthful. The one that lies about time says so right in its name.

What the Framework Doesn't Do

There is no built-in scheduling. There is no @task(run_every="0 18 * * *") — and daily price ingestion requires exactly that. The gap is bridged with a management command invoked via cron:

0 18 * * * cd /project && uv run python manage.py record_price ICE-CT2 "$(price.sh)" "$(date +\%Y-\%m-\%d)"

The pattern is "cron invokes command, command enqueues task". It works, but it's an extra moving part to maintain — and it's the most concrete difference when compared to Celery Beat.

Testing and What Was Left Out

There are 34 tests, developed via TDD from the first to the last commit. Most run inline with ImmediateBackend in ~1s. One runs a real worker. None require a GOOGLE_API_KEY — the PydanticAI integration (a task extracting structured data from free-text trade confirmations) leverages TestModel + Agent.override(), the library's built-in testing mechanism, without touching the network.

Left out intentionally: Celery, WebSockets on the dashboard, deployment, authentication. Every omission is documented in an ADR.md along with its rationale and review trigger — the concrete condition that would prompt revisiting the decision. Swapping django-tasks-db for Celery right now without the workload to justify it would be premature optimization; documenting when to switch is far more valuable than switching prematurely.

Opting for 1.5s polling instead of SSE also became an ADR, complete with the first-hand trade-off: state transitions shorter than the poll interval remain invisible, requiring an artificial task just to observe RUNNING.

The Pattern

All three gotchas share the same pattern: the production code was correct, yet the test failed anyway. A backend that doesn't implement what the API suggests. A data type that doesn't survive an invisible round-trip. A transaction that never commits.

None of these would have surfaced on the happy path — and that's precisely why writing the tests first was worth it. TDD didn't uncover bugs in my code; it uncovered my mistaken assumptions about the framework.

The complete code is available at github.com/carvalhocaio/cotton-desk-tasks.