Сбой модульного теста атомарных транзакций Django (коммиты не откатываются)

У меня есть проект Django со следующими двумя моделями:

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

Вот юнит-тест, который дает сбой:

# 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

Другими словами, несмотря на то, что я обернул provision_employee_id() в атомарную транзакцию, save() не откатывается при последующем возникновении исключения. Почему?

Это не имеет никакого отношения к базе данных. При сбое базы данных не происходит отката функции Python, откатываются только все запросы к базе данных в транзакции.

Если вы таким образом извлекаете объект Employee из базы данных, он все еще остается NULL. Если в функции было сделано два изменения базы данных, то оба они будут откачены.

Если таким образом получить данные из базы данных снова, с помощью .refresh_from_db(…) [Django-doc], вы увидите, что это действительно все еще 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)
Вернуться на верх