TransactionManagementError“您不能在使用信号时执行查询，直到'原子'块结束”，但仅在单元测试期间

Posted 2023-02-23

【中文标题】TransactionManagementError“您不能在使用信号时执行查询，直到\'原子\'块结束”，但仅在单元测试期间

【英文标题】：TransactionManagementError "You can't execute queries until the end of the 'atomic' block" while using signals, but only during Unit TestingTransactionManagementError“您不能在使用信号时执行查询，直到'原子'块结束”，但仅在单元测试期间

【发布时间】：2014-02-22 20:49:42

【问题描述】：

我在尝试保存 Django User 模型实例时遇到 TransactionManagementError，并且在其 post_save 信号中，我正在保存一些将用户作为外键的模型。

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

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

在手动测试中效果很好，但是单元测试失败了。

我有什么遗漏的吗？

这里是sn-ps的代码：

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

信号.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.

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

【问题讨论】：

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

我发现了我的问题。有一个像这样的 IntegrityError 异常 "try: ... except IntegrityError: ..." 我必须做的是在 try-block 中使用 transaction.atomic: "try: with transaction.atomic(): .. . 除了 IntegrityError: ..." 现在一切正常。

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

code.djangoproject.com/ticket/21540

【参考方案1】：

由于@mkoistinen 从未给his comment 提供答案，所以我会发布他的建议，这样人们就不必深入研究 cmets。

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

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

【讨论】：

+1，但是，正如文档所说，“Django 的 TestCase 类是 TransactionTestCase 的更常用的子类”。要回答最初的问题，我们不应该使用 SimpleTestCase 代替 TestCase 吗？ SimpleTestCase 没有原子数据库功能。

@daigorocub 从SimpleTestCase继承时，allow_database_queries = True必须添加到测试类中，所以它不会吐出AssertionError("Database queries aren't allowed in SimpleTestCase...",)。

这是最适合我的答案，因为我试图测试完整性错误将被提出，然后我需要运行更多的数据库保存查询

请记住，TransactionTestCase 可能比传统的 TestCase 慢得多。

【参考方案2】：

在我的情况下，它是引起但没有调用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()

【参考方案3】：

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

【参考方案4】：

根据这个问题的答案，这是另一种方法：

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

【参考方案5】：

如果使用 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

【讨论】：

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

【参考方案6】：

@kdazzle 的答案是正确的。我没有尝试，因为人们说“Django 的 TestCase 类是 TransactionTestCase 的一个更常用的子类”，所以我认为它的用途相同。但blog of Jahongir Rahmonov 解释得更好：

TestCase 类将测试包装在两个嵌套的 atomic() 块中： 一个用于全班，一个用于每个测试。这是哪里 应该使用 TransactionTestCase。它不会用 atomic() 块，因此您可以测试需要的特殊方法 交易没有任何问题。

编辑：它不起作用，我认为是的，但不是。

他们可以在 4 年内解决这个问题..................................................

【参考方案7】：

对我来说，建议的修复无效。在我的测试中，我使用Popen 打开一些子进程来分析/检查迁移（例如，一项测试检查是否没有模型更改）。

对我来说，从 SimpleTestCase 继承而不是 TestCase 确实可以解决问题。

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

虽然这不能回答最初的问题，但我希望这对某些人有所帮助。

【参考方案8】：

我遇到了同样的问题。

就我而言，我正在这样做

author.tasks.add(tasks)

所以转换成

author.tasks.add(*tasks)

删除了那个错误。

【参考方案9】：

我自己也遇到了同样的问题。这是由于在较新版本的 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 异常而失败。

就像 cmets 中提到的 caio 一样，解决方案是使用 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

这将防止故意抛出的异常破坏整个单元测试的事务。

【讨论】：

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

哦，我从另一个question找到了相关文档。文档是here。

对我来说，我已经有一个transaction.atomic() 块，但是我收到了这个错误，我不知道为什么。我接受了这个答案的建议，并在问题区域周围的原子块内放置了一个 nested 原子块。之后，它给出了我遇到的完整性错误的详细错误，允许我修复我的代码并做我想做的事情。

@mkoistinen TestCase 继承自 TransactionTestCase，因此无需更改。如果您不在测试中对 DB 进行操作，请使用 SimpleTestCase。

@bns 你错过了评论的重点。是的 TestCase 继承自 TransactionTestCase 但它的行为完全不同：它将每个测试方法包装在一个事务中。另一方面，TransactionTestCase 的命名可能具有误导性：它会截断表以重置数据库——命名似乎反映了您可以在测试中测试事务，而不是测试被包装为事务！

【参考方案10】：

我有同样的问题，但 with transaction.atomic() 和 TransactionTestCase 对我不起作用。

python manage.py test -r 而不是python manage.py test 对我来说没问题，也许执行顺序很关键

然后我找到一个关于Order in which tests are executed 的文档，它提到了哪个测试将首先运行。

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

【参考方案11】：

我在使用 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()

我的解决方案是改用 u​​pdate_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)