TransactionManagementError“在使用信号时,直到“原子”块结束时才能执行查询”,但只能在单元测试期间执行

在试图保存Django用户模型实例时,我遇到TransactionManagementError,在它的post_save信号中,我正在保存一些将用户作为外键的模型

上下文和错误与这个问题非常相似
django TransactionManagement使用信号时出错

但是,在这种情况下,只有在单元测试时才会发生错误

它在手动测试中运行良好,但单元测试失败

有什么我遗漏的吗

以下是代码片段:

views.py

@csrf\u
def移动注册表(请求):
如果request.method==“GET”:
响应={“错误”:“获取请求未被接受!!”}
返回HttpResponse(json.dumps(response),content_type=“application/json”,status=500)
elif request.method==“POST”:
postdata=json.loads(request.body)
尝试:
#获取用于保存用户的POST数据
username=postdata.get('phone')
password=postdata.get('password')
email=postdata.get('email',“”)
first\u name=postdata.get('first\u name',“”)
last\u name=postdata.get('last\u name',“”)
用户=用户(用户名=用户名,电子邮件=电子邮件,
first_name=first_name,last_name=last_name)
用户._company=postdata.get('company',None)
用户._country_code=postdata.get('country_code',“+91”)
user.is_verified=True
user.\u gcm\u reg\u id=postdata.get('reg\u id',无)
user.\u gcm\u device\u id=postdata.get('device\u id',None)
#为用户设置密码
user.set_密码(password)
#保存用户
user.save()

signal.py

def create_user_profile(发送方、实例、已创建,**kwargs):
如果创建:
公司=无
公司联系人=无
try:#尝试使用提供的公司和国家代码创建用户配置文件
user=user.objects.get(id=instance.id)
rand_pass=random.randint(10009999)
公司=公司.对象.获取或创建(名称=实例.\u公司,用户=用户)
companycontact=companycontact.objects.get\u或\u create(contact\u type=“Owner”,company=company,contact\u number=instance.username)
profile=UserProfile.objects.get\u或创建(user=instance,phone=instance.username,verification\u code=rand\u pass,company=company,country\u code=instance.\u country\u code)
gcmDevice=gcmDevice.objects.create(注册号=instance.\gcm注册号=id,设备号=instance.\gcm注册号=id,用户号=instance)
除例外情况外,e:
通过

tests.py

类AuthTestCase(TestCase):
fixtures=['nextgencatalogs/fixtures.json']
def设置(自):
self.user\u数据={
“电话”:“0000000000”,
“密码”:“123”,
“名字”:“Gaurav”,
“姓氏”:“Toshniwal”
}
def测试注册api获取(自我):
response=self.client.get(“/mobileRegister/”)
self.assertEqual(响应状态代码,500)
def测试\注册\ api \ post(自我):
response=self.client.post(path=“/mobileRegister/”,
data=json.dumps(self.user_数据),
content_type=“application/json”)
self.assertEqual(响应状态代码201)
self.user_数据['username']=self.user_数据['phone']
user=user.objects.get(username=self.user\u数据['username'])
#检查公司是否已创建
company=company.objects.get(用户\用户名=self.user\数据['phone'])
自我主张(公司,公司)
#检查所有者的联系人是否与用户的电话号码相同
company\u contact=CompanyContact.objects.get(company=company,contact\u type=“owner”)
self.assertEqual(用户.用户名,公司\联系人[0].联系人\号码)

回溯:

======================================================================
错误:test\u registration\u api\u post(nextgencatalogs.apps.catalogsapp.tests.AuthTestCase)
----------------------------------------------------------------------
回溯(最近一次呼叫最后一次):
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/nextgencatalogs/apps/catalogsapp/tests.py”,第29行,在测试注册api帖子中
user=user.objects.get(username=self.user\u数据['username'])
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/manager.py”,get中第151行
返回self.get_queryset().get(*args,**kwargs)
get中第301行的文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/query.py”
num=len(克隆)
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/query.py”,第77行,在__
self._fetch_all()
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/query.py”,第854行,全部
self.\u result\u cache=list(self.iterator())
迭代器中的文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/query.py”,第220行
对于编译器.results\u iter()中的行:
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/sql/compiler.py”,第710行,结果文件
对于self.execute_sql(多)中的行:
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/models/sql/compiler.py”,第781行,在execute_sql中
cursor.execute(sql,params)
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/backends/util.py”,执行中第47行
self.db.validate\u no\u中断的\u事务()
文件“/Users/gauravtoshniwal1989/Developer/Web/Server/ngc/ngcvenv/lib/python2.7/site packages/django/db/backends/___________.py”,第365行,在validate_no_break_事务中
“当前事务中发生错误。您无法”
TransactionManagementError:当前事务中发生错误。在“原子”块结束之前,无法执行查询。
----------------------------------------------------------------------

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

我有一个unittest,它检查以确保通过故意触发IntegrityError异常来强制执行唯一列约束:

定义测试约束(自身):
尝试:
#应防止重复。
models.Question.objects.create(domain=self.domain,slug=’barks’)
self.fail(’允许重复问题’)
除完整性错误外:
通过
做更多的模特

在Django 1.4中,这工作得很好。但是,在Django 1.5/1.6中,每个测试都包装在一个事务中,因此如果发生异常,它会中断事务,直到您显式回滚它。因此,该事务中的任何进一步ORM操作,如mydo\u more\u model\u stuff(),都将失败,并出现django.db.transaction.TransactionManagementError异常

与评论中提到的caio一样,解决方案是使用transaction.atomic捕获异常,如:

来自django.db导入事务的


def测试_约束(自):
尝试:
#应防止重复。
使用transaction.atomic():
models.Question.objects.create(domain=self.domain,slug='barks')
self.fail('允许重复问题')
除完整性错误外:
通过

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

发表评论