你不能执行查询,直到'原子'block"而只在单元测试期间使用信号

当我试图保存一个Django用户模型实例时,我得到了TransactionManagementError,在它的post_save信号中,我保存了一些有用户作为外键的模型。

上下文和错误与这个问题非常相似 django使用信号时的TransactionManagementError < / p >

然而,在这种情况下,错误发生仅在单元测试时

它在手动测试中工作得很好,但单元测试失败。

我还遗漏了什么吗?

下面是代码片段:

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.


----------------------------------------------------------------------
113410 次浏览

我自己也遇到了同样的问题。这是由于新版本的Django中处理事务的方式有一个怪癖,再加上有意触发异常的unittest。

我有一个单元测试,检查确保一个唯一的列约束被强制触发一个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异常。

就像评论中提到的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

这将防止故意抛出的异常破坏整个unittest的事务。

由于@mkoistinen从未做出他的评论的答案,我将发布他的建议,这样人们就不必在评论中挖掘。

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

Django文档: A TransactionTestCase可以调用commit和rollback,并观察这些调用对数据库的影响。

我也有同样的问题。

在《我的情况》中,我就是这么做的

author.tasks.add(tasks)

那么把它转换成

author.tasks.add(*tasks)

删除了这个错误。

我在使用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='test@test.com', address='test', postcode='test', telephone='test')
cls.chamber,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeC, name='chamber for test', email_general='test@test.com', address='test', postcode='test', telephone='test')
cls.lawfirm,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeL, name='lawfirm for test', email_general='test@test.com', 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='test@test.com', 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='test@test.com', 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='test@test.com', address='test', postcode='test', telephone='test', defaults={'active': True})

我有同样的问题,但with transaction.atomic()TransactionTestCase不适合我。

python manage.py test -r代替python manage.py test对我来说是可以的,也许执行的顺序是至关重要的

然后我找到了一个关于测试执行的顺序的文档,它提到了哪个测试将首先运行。

所以,我使用TestCase数据库交互,unittest.TestCase为其他简单的测试,它现在工作!

对我来说,提议的修复没有起作用。在我的测试中,我用Popen打开了一些子进程来分析/lint迁移(例如,一个测试检查是否没有模型更改)。

对我来说,从SimpleTestCase而不是TestCase继承子类确实奏效了。

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

虽然这并没有回答最初的问题,但我希望这能帮助到一些人。

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

TestCase类将测试包装在两个嵌套的原子()块中: 全班一份,每个测试一份。这就是 应该使用TransactionTestCase。它不会用 原子()块,因此您可以测试您的特殊方法,需要

编辑:它没有工作,我想是的,但不是。

在4年可以固定 .......................................

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

下面是另一种方法,基于这个问题的答案:

with transaction.atomic():
self.assertRaises(IntegrityError, models.Question.objects.create, **{'domain':self.domain, 'slug':'barks'})
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

在我的情况下,它是造成的,但没有调用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()