Django post_save signal enqueuing a Celery task, tested with eager mode and on_commit: working versions (tested 2026-10-15)

django==6.1.2
celery==5.6.3
pytest-django==4.14.0
pytest==9.1.1

SECRET_KEY = "test-only-not-secret"
DEBUG = False
USE_TZ = True
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"

INSTALLED_APPS = [
    "django.contrib.contenttypes",
    "notes.apps.NotesConfig",
]

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.sqlite3",
        "NAME": ":memory:",
    }
}

from celery import Celery

# In-memory broker and result backend: no Redis or network needed.
app = Celery("notes", broker="memory://", backend="cache+memory://")

from django.apps import AppConfig


class NotesConfig(AppConfig):
    name = "notes"

    def ready(self):
        # Import here, after models are loaded, to avoid circular imports.
        from notes import signals  # noqa: F401

from django.db import models


class Note(models.Model):
    title = models.CharField(max_length=200)
    indexed = models.BooleanField(default=False)

from functools import partial

from django.db import transaction
from django.db.models.signals import post_save
from django.dispatch import receiver

from notes.models import Note
from notes.tasks import index_note


@receiver(post_save, sender=Note)
def queue_indexing(sender, instance, created, **kwargs):
    if created:
        # Pass the pk, not the instance, and wait for commit so the worker
        # never runs before the row is visible to other connections.
        transaction.on_commit(partial(index_note.delay, instance.pk))

from celery_app import app
from notes.models import Note


@app.task
def index_note(note_id: int) -> None:
    # Fetch by pk and use .update() so the task never re-fires post_save.
    # A missing row simply matches zero rows instead of raising.
    Note.objects.filter(pk=note_id).update(indexed=True)

import pytest

from celery_app import app


@pytest.fixture(autouse=True)
def eager_celery():
    app.conf.update(task_always_eager=True, task_eager_propagates=True)
    yield
    app.conf.update(task_always_eager=False)

from unittest.mock import patch

import pytest
from django.db import transaction

from notes.models import Note
from notes.tasks import index_note

pytestmark = pytest.mark.django_db


def test_task_runs_after_commit(django_capture_on_commit_callbacks):
    # Eager mode runs the task inline once the on_commit callback fires.
    with django_capture_on_commit_callbacks(execute=True):
        note = Note.objects.create(title="a")
    note.refresh_from_db()
    assert note.indexed


def test_nothing_enqueued_before_commit(django_capture_on_commit_callbacks):
    # Inside the test transaction, on_commit callbacks are deferred.
    with patch.object(index_note, "delay") as delay:
        with django_capture_on_commit_callbacks() as callbacks:
            note = Note.objects.create(title="a")
        delay.assert_not_called()
        assert len(callbacks) == 1
        for cb in callbacks:
            cb()
        delay.assert_called_once_with(note.pk)


def test_rollback_discards_enqueue(django_capture_on_commit_callbacks):
    # Callbacks registered inside a rolled-back savepoint are dropped.
    with django_capture_on_commit_callbacks() as callbacks:
        with pytest.raises(RuntimeError):
            with transaction.atomic():
                Note.objects.create(title="a")
                raise RuntimeError("boom")
    assert callbacks == []
    assert not Note.objects.exists()


def test_task_tolerates_missing_row():
    # The task passes the pk, so a deleted row is a no-op, not an error.
    result = index_note.delay(999_999)
    assert result.successful()

[pytest]
pythonpath = .
DJANGO_SETTINGS_MODULE = settings