Django atomic transaction unit test failing (commits not being rolled back)

I have a Django project with the following two models:

# models.py
from django.db import models, transaction

class Person(models.Model):
    name    = models.TextField()
    surname = models.TextField()

class EmployeeInfo(models.Model):
    person = models.OneToOneField(Person, on_delete=models.CASCADE)
    employee_id = models.TextField(null=True, blank=True)

    @transaction.atomic
    def provision_employee_id(self):
        """Make atomic to ensure that two employees being provisioned at the same time do not get
        the same employee number"""
        self.employee_id = new_employee_number()
        self.save()
        raise Exception("forcing an exception") # Always raise an exception (for testing only)

Here is the unit test that is failing:

# tests.py
class PersonTestCase(TestCase):
    def test_employee_id_atomic(self):
        person1 = Person(name="John", surname="Doe")
        employee_info = EmployeeInfo(person=person1)
        self.assertIsNone(employee_info.employee_id) # No employee_id yet.
        
        with self.assertRaises(Exception) as context:
            cluster_updated = employee_info.provision_employee_id()

        self.assertIsNone(employee_info.employee_id) # This FAILS

In other words, even though I have wrapped provision_employee_id() in an atomic transaction the save() is not rolled back when the subsequent exception is raised. Why?

Solution

This has nothing to do with the database. If the database fails, it does not rollback the Python function, it only rolls back all the database queries in the transaction.

If you thus fetch the Employee object from the database, it is still NULL. If you would have done two database changes in the function, these are both rolled back.

If you thus fetch the data from the database again, with .refresh_from_db(…) ^[Django-doc], you will see that it is indeed still NULL:

# tests.py
class PersonTestCase(TestCase):
    def test_employee_id_atomic(self):
        person1 = Person.objects.create(name='John', surname='Doe')
        employee_info = EmployeeInfo.objects.create(person=person1)
        self.assertIsNone(employee_info.employee_id)  # No employee_id yet.

        with self.assertRaises(Exception) as context:
            cluster_updated = employee_info.provision_employee_id()

        employee_info.refresh_from_db()
        self.assertIsNone(employee_info.employee_id)