Python TransactionManagementError“在使用信号时，您不能在‘原子’块结束之前执行查询”，但仅限在单元测试期间
声明：本页面是StackOverFlow热门问题的中英对照翻译，遵循CC BY-SA 4.0协议，如果您需要使用它，必须同样遵循CC BY-SA许可，注明原文地址和作者信息，同时你必须将它归于原作者（不是我）：StackOverFlow
原文地址: http://stackoverflow.com/questions/21458387/
Warning: these are provided under cc-by-sa 4.0 license. You are free to use/share it, But you must attribute it to the original authors (not me):
StackOverFlow
TransactionManagementError "You can't execute queries until the end of the 'atomic' block" while using signals, but only during Unit Testing
提问by Gaurav Toshniwal
I am getting TransactionManagementError when trying to save a Django User model instance and in its post_save signal, I'm saving some models that have the user as the foreign key.
我在尝试保存 Django User 模型实例时遇到 TransactionManagementError 并且在其 post_save 信号中，我正在保存一些将用户作为外键的模型。
The context and error is pretty similar to this question django TransactionManagementError when using signals
上下文和错误与使用信号时的这个问题django TransactionManagementError非常相似
However, in this case, the error occurs only while unit testing.
但是，在这种情况下，错误仅在单元测试时发生。
It works well in manual testing, but unit tests fails.
它在手动测试中运行良好，但单元测试失败。
Is there anything that I'm missing?
有什么我想念的吗？
Here are the code snippets:
以下是代码片段：
views.py
视图.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
信号.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
测试.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)
Traceback:
追溯：
======================================================================
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.
----------------------------------------------------------------------
采纳答案by Cerin
I ran into this same problem myself. This is caused by a quirk in how transactions are handled in the newer versions of Django coupled with a unittest that intentionally triggers an exception.
我自己也遇到了同样的问题。这是由于在较新版本的 Django 中处理事务的方式的一个怪癖加上有意触发异常的单元测试。
I had a unittest that checked to make sure a unique column constraint was enforced by purposefully triggering an IntegrityError exception:
我有一个单元测试，通过有目的地触发一个 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()
In Django 1.4, this works fine. However, in Django 1.5/1.6, each test is wrapped in a transaction, so if an exception occurs, it breaks the transaction until you explicitly roll it back. Therefore, any further ORM operations in that transaction, such as my do_more_model_stuff(), will fail with that django.db.transaction.TransactionManagementErrorexception.
在 Django 1.4 中，这很好用。但是，在 Django 1.5/1.6 中，每个测试都包含在一个事务中，因此如果发生异常，它会中断事务，直到您明确回滚它为止。因此，该事务中的任何进一步 ORM 操作（例如 my do_more_model_stuff()）都将因该django.db.transaction.TransactionManagementError异常而失败。
Like caio mentioned in the comments, the solution is to capture your exception with transaction.atomiclike:
就像评论中提到的 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
That will prevent the purposefully-thrown exception from breaking the entire unittest's transaction.
这将防止故意抛出的异常破坏整个单元测试的事务。
回答by kdazzle
Since @mkoistinen never made his comment, an answer, I'll post his suggestion so people won't have to dig through comments.
由于@mkoistinen 从未发表过他的评论，一个答案，我会发布他的建议，这样人们就不必深入研究评论。
consider just declaring your test class as a TransactionTestCase rather than just TestCase.
考虑将您的测试类声明为 TransactionTestCase 而不仅仅是 TestCase。
From the docs: A TransactionTestCase may call commit and rollback and observe the effects of these calls on the database.
来自文档： TransactionTestCase 可以调用 commit 和 rollback 并观察这些调用对数据库的影响。
回答by Diaa Mohamed Kasem
I had the same issue.
我遇到过同样的问题。
In My Case I was doing this
在我的情况下，我正在这样做
author.tasks.add(tasks)
so converting it to
所以将其转换为
author.tasks.add(*tasks)
Removed that error.
删除了那个错误。
回答by PhoebeB
I was getting this error on running unit tests in my create_test_data function using django 1.9.7. It worked in earlier versions of django.
我在使用 django 1.9.7 在 create_test_data 函数中运行单元测试时遇到此错误。它适用于早期版本的 django。
It looked like this:
它看起来像这样：
cls.localauth,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeLA, name='LA for test', email_general='[email protected]', address='test', postcode='test', telephone='test')
cls.chamber,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeC, name='chamber for test', email_general='[email protected]', address='test', postcode='test', telephone='test')
cls.lawfirm,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeL, name='lawfirm for test', email_general='[email protected]', 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()
My solution was to use update_or_create instead:
我的解决方案是使用 update_or_create 代替：
cls.localauth,_ = Organisation.objects.update_or_create(organisation_type=cls.orgtypeLA, name='LA for test', email_general='[email protected]', 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='[email protected]', 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='[email protected]', address='test', postcode='test', telephone='test', defaults={'active': True})
回答by Leo
I have the same issue, but with transaction.atomic()and TransactionTestCasedidn't work for me.
我有同样的问题，但with transaction.atomic()和TransactionTestCase我没有工作。
python manage.py test -rinstead of python manage.py testis ok for me, maybe the order of execution is crucial
python manage.py test -r而不是python manage.py test对我来说可以，也许执行顺序至关重要
then i find a doc about Order in which tests are executed, It mentions which test will run first.
然后我找到了一个关于执行测试的顺序的文档，它提到了哪个测试将首先运行。
So, i use TestCase for database interaction, unittest.TestCasefor other simple test, it works now!
所以，我使用 TestCase 进行数据库交互，unittest.TestCase对于其他简单的测试，它现在可以工作了！
回答by flix
For me, the proposed fixes did not work. In my tests, I open some subprocesses with Popento analyze/lint migrations (e.g. one test checks if there are no model changes).
对我来说，建议的修复方法不起作用。在我的测试中，我打开了一些子进程Popen来分析/lint 迁移（例如，一个测试检查是否没有模型更改）。
For me, subclassing from SimpleTestCaseinstead of TestCasedid do the trick.
对我来说，从而SimpleTestCase不是子类化TestCase确实可以解决问题。
Note that SimpleTestCasedoesn't allow to use the database.
请注意，SimpleTestCase不允许使用数据库。
While this does not answer the original question, I hope this helps some people anyway.
虽然这不能回答最初的问题，但我希望这对某些人有所帮助。
回答by Shil Nevado
The answer of @kdazzle is correct. I didnt try it because people said that 'Django's TestCase class is a more commonly used subclass of TransactionTestCase' so I thought it was the same use one or another. But the blog of Jahongir Rahmonovexplained it better:
@kdazzle 的答案是正确的。我没有尝试，因为人们说“Django 的 TestCase 类是 TransactionTestCase 的一个更常用的子类”，所以我认为它是相同的用法。但是Jahongir Rahmonov的博客解释得更好：
the TestCase class wraps the tests within two nested atomic() blocks: one for the whole class and one for each test. This is where TransactionTestCase should be used. It does not wrap the tests with atomic() block and thus you can test your special methods that require a transaction without any problem.
TestCase 类将测试包装在两个嵌套的 atomic() 块中：一个用于整个类，一个用于每个测试。这是应该使用 TransactionTestCase 的地方。它不使用 atomic() 块包装测试，因此您可以毫无问题地测试需要事务的特殊方法。
EDIT: It didn't work, I thought yes, but NO.
编辑：它没有用，我想是的，但不是。
In 4 years they could fixed this.......................................
在 4 年内他们可以解决这个问题......................................
回答by frmdstryr
If using pytest-django you can pass transaction=Trueto the django_dbdecorator to avoid this error.
如果使用 pytest-django，您可以传递transaction=True给django_db装饰器以避免此错误。
See https://pytest-django.readthedocs.io/en/latest/database.html#testing-transactions
见https://pytest-django.readthedocs.io/en/latest/database.html#testing-transactions
Django itself has the TransactionTestCase which allows you to test transactions and will flush the database between tests to isolate them. The downside of this is that these tests are much slower to set up due to the required flushing of the database. pytest-django also supports this style of tests, which you can select using an argument to the django_db mark:
Django 本身有 TransactionTestCase，它允许您测试事务并在测试之间刷新数据库以隔离它们。这样做的缺点是，由于需要刷新数据库，这些测试的设置速度要慢得多。pytest-django 也支持这种风格的测试，你可以使用 django_db 标记的参数来选择：
@pytest.mark.django_db(transaction=True)
def test_spam():
pass # test relying on transactions
回答by Mahdi Hamzeh
Here is another way to do it, based on the answer to this question:
根据对这个问题的回答，这是另一种方法：
with transaction.atomic():
self.assertRaises(IntegrityError, models.Question.objects.create, **{'domain':self.domain, 'slug':'barks'})
回答by Aleksei Khatkevich
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
