在使用信号（signals）时，在单元测试期间出现“TransactionManagementError，只有在‘原子’块结束之前才能执行查询”的错误。

Question

当我试图保存Django User模型实例时，出现了TransactionManagementError，在其post_save信号中，我保存了一些以该用户为外键的模型。

上下文和错误与此问题非常相似：django TransactionManagementError when using signals

但是，在这种情况下，错误仅在单元测试期间发生。

手动测试运行良好，但单元测试失败。

我有遗漏了什么吗？

以下是代码段：

views.py

@csrf_exempt
def mobileRegister(request):
    if request.method == 'GET':
        response = {"error": "GET request not accepted!!"}
        return HttpResponse(json.dumps(response), content_type="application/json",status=500)
    elif request.method == 'POST':
        postdata = json.loads(request.body)
        try:
            # Get POST data which is to be used to save the user
            username = postdata.get('phone')
            password = postdata.get('password')
            email = postdata.get('email',"")
            first_name = postdata.get('first_name',"")
            last_name = postdata.get('last_name',"")
            user = User(username=username, email=email,
                        first_name=first_name, last_name=last_name)
            user._company = postdata.get('company',None)
            user._country_code = postdata.get('country_code',"+91")
            user.is_verified=True
            user._gcm_reg_id = postdata.get('reg_id',None)
            user._gcm_device_id = postdata.get('device_id',None)
            # Set Password for the user
            user.set_password(password)
            # Save the user
            user.save()

signal.py

def create_user_profile(sender, instance, created, **kwargs):
    if created:
        company = None
        companycontact = None
        try:   # Try to make userprofile with company and country code provided
            user = User.objects.get(id=instance.id)
            rand_pass = random.randint(1000, 9999)
            company = Company.objects.get_or_create(name=instance._company,user=user)
            companycontact = CompanyContact.objects.get_or_create(contact_type="Owner",company=company,contact_number=instance.username)
            profile = UserProfile.objects.get_or_create(user=instance,phone=instance.username,verification_code=rand_pass,company=company,country_code=instance._country_code)
            gcmDevice = GCMDevice.objects.create(registration_id=instance._gcm_reg_id,device_id=instance._gcm_reg_id,user=instance)
        except Exception, e:
            pass

tests.py

class AuthTestCase(TestCase):
    fixtures = ['nextgencatalogs/fixtures.json']
    def setUp(self):
        self.user_data={
            "phone":"0000000000",
            "password":"123",
            "first_name":"Gaurav",
            "last_name":"Toshniwal"
            }

    def test_registration_api_get(self):
        response = self.client.get("/mobileRegister/")
        self.assertEqual(response.status_code,500)

    def test_registration_api_post(self):
        response = self.client.post(path="/mobileRegister/",
                                    data=json.dumps(self.user_data),
                                    content_type="application/json")
        self.assertEqual(response.status_code,201)
        self.user_data['username']=self.user_data['phone']
        user = User.objects.get(username=self.user_data['username'])
        # Check if the company was created
        company = Company.objects.get(user__username=self.user_data['phone'])
        self.assertIsInstance(company,Company)
        # Check if the owner's contact is the same as the user's phone number
        company_contact = CompanyContact.objects.get(company=company,contact_type="owner")
        self.assertEqual(user.username,company_contact[0].contact_number)

追踪：

======================================================================
ERROR: test_registration_api_post (nextgencatalogs.apps.catalogsapp.tests.AuthTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/nextgencatalogs/apps/catalogsapp/tests.py", line 29, in test_registration_api_post
    user = User.objects.get(username=self.user_data['username'])
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/manager.py", line 151, in get
    return self.get_queryset().get(*args, **kwargs)
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 301, in get
    num = len(clone)
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 77, in __len__
    self._fetch_all()
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 854, in _fetch_all
    self._result_cache = list(self.iterator())
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/query.py", line 220, in iterator
    for row in compiler.results_iter():
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 710, in results_iter
    for rows in self.execute_sql(MULTI):
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/models/sql/compiler.py", line 781, in execute_sql
    cursor.execute(sql, params)
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/backends/util.py", line 47, in execute
    self.db.validate_no_broken_transaction()
  File "/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site-packages/django/db/backends/__init__.py", line 365, in validate_no_broken_transaction
    "An error occurred in the current transaction. You can't "
TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block.

----------------------------------------------------------------------

- Gaurav Toshniwal

---

从文档中： “另一方面，TestCase在测试后不会截断表。相反，它将测试代码封装在数据库事务中，在测试结束时回滚。像transaction.commit()这样的显式提交和可能由transaction.atomic()引起的隐式提交都被替换为nop操作。这保证了测试结束时的回滚将数据库恢复到其初始状态。” - Gaurav Toshniwal

我找到了问题所在。有一个IntegrityError异常，代码如下："try: ... except IntegrityError: ..." 我需要在try块中使用transaction.atomic: "try: with transaction.atomic(): ... except IntegrityError: ..." 现在一切都正常了。 - caio

https://docs.djangoproject.com/en/dev/topics/db/transactions/ 然后搜索 “将原子包装在try / except块中允许自然处理完整性错误：” - CamHart

https://code.djangoproject.com/ticket/21540 - djvg

Answer 1

我自己也遇到了这个问题。这是因为Django较新版本中事务处理方式与故意触发异常的单元测试相结合所导致的奇怪情况。

我有一个单元测试，目的是通过故意触发IntegrityError异常来检查是否强制执行了唯一列约束：

def test_constraint(self):
    try:
        # Duplicates should be prevented.
        models.Question.objects.create(domain=self.domain, slug='barks')
        self.fail('Duplicate question allowed.')
    except IntegrityError:
        pass

    do_more_model_stuff()

在Django 1.4中，这个代码可以正常工作。然而，在Django 1.5/1.6中，每个测试都被包含在一个事务中，因此如果出现异常，除非你显式地回滚它，否则会中断事务。因此，在该事务中进行的任何其他ORM操作（例如我的do_more_model_stuff()）都会失败，并抛出django.db.transaction.TransactionManagementError异常。像评论中提到的那样，解决方法是使用transaction.atomic来捕获异常，如下所示：

from django.db import transaction
def test_constraint(self):
    try:
        # Duplicates should be prevented.
        with transaction.atomic():
            models.Question.objects.create(domain=self.domain, slug='barks')
        self.fail('Duplicate question allowed.')
    except IntegrityError:
        pass

这样可以防止有意抛出的异常破坏整个单元测试的事务。

- Cerin

---

还可以考虑将测试类声明为TransactionTestCase而不是仅仅是TestCase。 - mkoistinen

对我来说，我已经有了一个transaction.atomic()代码块，但我遇到了这个错误，而且不知道为什么会出现。我采纳了这个答案的建议，在我的原子块内部放置了一个嵌套的原子块，围绕着出问题的区域。之后，它给出了我遇到的完整性错误的详细信息，让我能够修复我的代码并完成我想做的事情。 - AlanSE

TestCase继承自TransactionTestCase，因此不需要更改。如果测试中不涉及数据库操作，请使用SimpleTestCase。 - bns

@bns，你没有理解评论的重点。是的，TestCase继承自TransactionTestCase，但它们的行为相当不同：它将每个测试方法包装在一个事务中。另一方面，TransactionTestCase的命名可能有些误导：它截断表以重置数据库——命名似乎反映了您可以在测试中测试事务，而不是将测试作为事务包装！ - C S

请注意，如果您需要使用setUpTestData，则无法与TransactionTestCase一起使用，它必须与TestCase一起使用。 在测试唯一组的完整性错误时，我不得不将save调用包装在with atomic()块中。 - Vincent

Answer 2

由于@mkoistinen从未将他们的评论作为答案发布，我将发布这个建议，这样人们就不必深入评论中查找了。

考虑将您的测试类声明为TransactionTestCase而不仅仅是TestCase。

来自Django文档：TransactionTestCase可以调用commit和rollback，并观察这些调用对数据库的影响。

- kdazzle

---

+1 对此表示赞同，但正如文档所说，“Django的TestCase类是TransactionTestCase的一个更常用的子类”。回答原始问题，我们应该使用SimpleTestCase而不是TestCase吗？SimpleTestCase没有原子数据库功能。 - daigorocub

@daigorocub 当从SimpleTestCase继承时，必须在测试类内添加allow_database_queries = True，这样它就不会抛出AssertionError("Database queries aren't allowed in SimpleTestCase...",)。 - CristiFati

这是对我来说最有效的答案，因为我在尝试测试IntegrityError是否会被引发，然后随后需要运行更多的数据库保存查询。 - Kim Stacks

需要记住的重点是，TransactionTestCase 可能比传统的 TestCase 慢得多。 - Dougyfresh

Answer 3

如果使用pytest-django，您可以将transaction=True传递给django_db装饰器，以避免此错误。

请参阅https://pytest-django.readthedocs.io/en/latest/database.html#testing-transactions Django本身有TransactionTestCase，它允许您测试事务并在测试之间刷新数据库以进行隔离。缺点是这些测试设置速度要慢得多，因为需要刷新数据库。 pytest-django也支持这种测试风格，您可以使用django_db标记的参数来选择它们。

@pytest.mark.django_db(transaction=True)
def test_spam():
    pass  # test relying on transactions

- frmdstryr

---

我在这个解决方案中遇到了问题，我的数据库中有初始数据（由迁移添加）。这个解决方案清空了数据库，因此依赖于这些初始数据的其他测试开始失败。 - abumalick

Answer 4

这里有另一种做法，基于对这个问题的回答：

with transaction.atomic():
    self.assertRaises(IntegrityError, models.Question.objects.create, **{'domain':self.domain, 'slug':'barks'})

- Mahdi Hamzeh

Answer 5

在我的情况下，这是由于未调用super().tearDownClass()引起的。

class TnsFileViewSetTestCase(APITestCase):
    @classmethod
    def tearDownClass(self):
        super().tearDownClass()    # without this line we will get TransactionManagementError
        for tnsfile in TnsFile.objects.all():
            tnsfile.file.delete()

- Artem Bernatskyi

Answer 6

我有同样的问题，但是使用transaction.atomic() 和 TransactionTestCase没有解决我的问题。

对我来说，使用python manage.py test -r 而不是python manage.py test可以正常工作，可能执行的顺序很关键。

然后我发现了一份关于测试执行顺序的文档，它提到了哪些测试会首先运行。

所以，我使用TestCase进行数据库交互，使用unittest.TestCase进行其他简单的测试，现在它可以正常工作了！

- Leo

Answer 7

对我来说，提出的修复措施没有起作用。在我的测试中，我使用 Popen 打开一些子进程来分析/检查迁移（例如，一个测试检查是否有模型更改）。

对我来说，从 TestCase 而不是 SimpleTestCase 派生子类确实起到了作用。

请注意，SimpleTestCase 不允许使用数据库。

虽然这并没有回答原始问题，但我希望这可以帮助一些人。

- flix

Answer 8

def test_wrong_user_country_db_constraint(self):
        """
        Check whether or not DB constraint doesnt allow to save wrong country code in DB.
        """
        self.test_user_data['user_country'] = 'XX'
        expected_constraint_name = "country_code_within_list_of_countries_check"

        with transaction.atomic():
            with self.assertRaisesRegex(IntegrityError, expected_constraint_name) as cm:
                get_user_model().objects.create_user(**self.test_user_data)

        self.assertFalse(
            get_user_model().objects.filter(email=self.test_user_data['email']).exists()
        )

with transaction.atomic() seems do the job correct

- Aleksei Khatkevich

Answer 9

class Pricing(models.Model):
price = models.DecimalField(max_digits=4, decimal_places=1)
is_active = models.BooleanField(default=False)

def __str__(self):
    return str(self.price)

class Services(models.Model):
name = models.CharField(max_length=30)
description = models.CharField(max_length=30)
pricing = models.OneToOneField(Pricing, on_delete=models.CASCADE, default=0.0)
categories = models.ManyToManyField(Category, related_name="services")
is_active = models.BooleanField(default=False)

def __str__(self):
    return self.name

在我的情况下，您在测试一个onetoonefield实例的函数中遇到了这个错误：在创建一个价格实例并将其添加到2个服务实例中时，出现了这个错误。

解决方案：在您的函数内部执行以下操作。

with transaction.atomic():
        with self.assertRaises(IntegrityError):
            service = Services.objects.create(name = "Some Service", pricing= self.pricing1)

- Abhay KanWasi

Answer 10

我在使用 Django 1.9.7 运行单元测试中的 create_test_data 函数时遇到了错误。在 Django 的早期版本中这个函数是可用的。

错误信息如下：

cls.localauth,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeLA, name='LA for test', email_general='[e-mail]', address='test', postcode='test', telephone='test')
cls.chamber,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeC, name='chamber for test', email_general='[e-mail]', address='test', postcode='test', telephone='test')
cls.lawfirm,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeL, name='lawfirm for test', email_general='[e-mail]', address='test', postcode='test', telephone='test')

cls.chamber.active = True
cls.chamber.save()

cls.localauth.active = True
cls.localauth.save()    <---- error here

cls.lawfirm.active = True
cls.lawfirm.save()

我的解决方案是使用update_or_create：

cls.localauth,_ = Organisation.objects.update_or_create(organisation_type=cls.orgtypeLA, name='LA for test', email_general='[e-mail]', address='test', postcode='test', telephone='test', defaults={'active': True})
cls.chamber,_ = Organisation.objects.update_or_create(organisation_type=cls.orgtypeC, name='chamber for test', email_general='[e-mail]', address='test', postcode='test', telephone='test', defaults={'active': True})
cls.lawfirm,_ = Organisation.objects.update_or_create(organisation_type=cls.orgtypeL, name='lawfirm for test', email_general='[e-mail]', address='test', postcode='test', telephone='test', defaults={'active': True})

- PhoebeB

---

似乎也可以使用get_or_create()，但是在一个使用transaction.atomic()修饰的函数内部进行.save()操作时会出现问题（我的函数即使只有1个调用也失败了）。 - belteshazzar