TransactionManagementError “No puede ejecutar consultas hasta el final del bloque ‘atómico’ mientras usa señales, pero solo durante la prueba de unidad

Estoy obteniendo TransactionManagementError al intentar guardar una instancia del modelo de usuario de Django y, en su señal de post_save, estoy guardando algunos modelos que tienen al usuario como clave externa.

El contexto y el error son bastante similares a esta pregunta django TransactionManagementError cuando se usan señales

Sin embargo, en este caso, el error se produce solo durante la prueba de la unidad .

Funciona bien en las pruebas manuales, pero las pruebas unitarias fallan.

¿Hay algo que me esté perdiendo?

Aquí están los fragmentos de código:

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

señal.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) 

Rastrear:

 ====================================================================== 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. ---------------------------------------------------------------------- 

Me encontré con este mismo problema yo mismo. Esto se debe a una peculiaridad en cómo se manejan las transacciones en las versiones más nuevas de Django junto con una prueba de unidad que intencionalmente desencadena una excepción.

Tuve una prueba de unidad que se verificó para asegurarse de que se impusiera una restricción de columna única activando a propósito una excepción 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() 

En Django 1.4, esto funciona bien. Sin embargo, en Django 1.5 / 1.6, cada prueba se envuelve en una transacción, por lo que si se produce una excepción, se rompe la transacción hasta que se deshaga explícitamente. Por lo tanto, cualquier otra operación ORM en esa transacción, como mi do_more_model_stuff() , fallará con esa excepción django.db.transaction.TransactionManagementError .

Como lo mencionó Caio en los comentarios, la solución es capturar su excepción con transaction.atomic como:

 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 

Eso evitará que la excepción lanzada a propósito rompa la transacción de la prueba de unidad completa.

Ya que @mkoistinen nunca hizo su comentario , una respuesta, publicaré su sugerencia para que la gente no tenga que escarbar en los comentarios.

considere simplemente declarar su clase de prueba como TransactionTestCase en lugar de solo TestCase.

De los documentos : Un TransactionTestCase puede llamar a commit y rollback y observar los efectos de estas llamadas en la base de datos.

Para mí, las correcciones propuestas no funcionaron. En mis pruebas, abro algunos subprocesos con Popen para analizar migraciones / pelusas (por ejemplo, una prueba verifica si no hay cambios en el modelo).

Para mí, la subclasificación de SimpleTestCase lugar de TestCase hizo el truco.

Tenga en cuenta que SimpleTestCase no permite utilizar la base de datos.

Si bien esto no responde a la pregunta original, espero que esto ayude a algunas personas de todos modos.

Si usa pytest-django, puede pasar transaction=True al decorador django_db para evitar este error.

Consulte https://pytest-django.readthedocs.io/en/latest/database.html#testing-transactions

Django tiene el TransactionTestCase que le permite probar transacciones y limpiará la base de datos entre las pruebas para aislarlas. La desventaja de esto es que estas pruebas son mucho más lentas de configurar debido a la limpieza necesaria de la base de datos. pytest-django también admite este estilo de pruebas, que puede seleccionar usando un argumento para la marca django_db:

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

Recibía este error al ejecutar pruebas de unidad en mi función create_test_data usando django 1.9.7. Funcionó en versiones anteriores de django.

Se veía así:

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

Mi solución fue usar update_or_create en su lugar:

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

Tengo el mismo problema, pero with transaction.atomic() y TransactionTestCase no me funcionó.

python manage.py test -r lugar de python manage.py test está bien para mí, tal vez el orden de ejecución sea crucial

Luego encuentro un documento sobre Orden en el que se ejecutan las pruebas . Menciona qué prueba se ejecutará primero.

Por lo tanto, utilizo TestCase para la interacción con la base de datos, unittest.TestCase para otra prueba simple, ¡ahora funciona!

La respuesta de @kdazzle es correcta. No lo intenté porque la gente dijo que ‘la clase TestCase de Django es una subclase de TransactionTestCase más utilizada’, así que pensé que era el mismo uso que uno u otro. Pero el blog de Jahongir Rahmonov lo explicó mejor:

La clase TestCase envuelve las pruebas dentro de dos bloques atómicos () nesteds: uno para toda la clase y uno para cada prueba. Aquí es donde se debe utilizar TransactionTestCase. No envuelve las pruebas con el bloque atómico () y, por lo tanto, puede probar sus métodos especiales que requieren una transacción sin ningún problema.

EDIT: no funcionó, pensé que sí, pero no.

En 4 años pudieron arreglar esto …………………………………

Tuve el mismo problema.

En mi caso estaba haciendo esto.

 author.tasks.add(tasks) 

convirtiéndolo así a

 author.tasks.add(*tasks) 

Eliminado ese error.