python - TransactionManagementError "No se pueden ejecutar consultas hasta el final del bloque ''atómico''" mientras se utilizan señales, pero solo durante la prueba de la unidad

django unit-testing

Question

Estoy obteniendo TransactionManagementError al intentar guardar una instancia de modelo de usuario de Django y en su señal 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 al usar señales

Sin embargo, en este caso, el error ocurre solo durante la prueba unitaria .

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

¿Hay algo que me estoy perdiendo?

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

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)

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.

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

Answer 1

Como @mkoistinen nunca hizo su comentario , una respuesta, publicaré su sugerencia para que la gente no tenga que buscar comentarios.

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

De los docs : una TransactionTestCase puede invocar commit y rollback y observar los efectos de estas llamadas en la base de datos.

Answer 2

Estaba obteniendo este error al ejecutar pruebas unitarias 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=''[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()

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=''[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})

Answer 3

Me encontré con este mismo problema yo mismo. Esto se debe a una peculiaridad en la forma en que se manejan las transacciones en las versiones más nuevas de Django, junto con una prueba unitaria que activa intencionalmente una excepción.

Tuve una prueba unitaria que verificaba para asegurarme de que se aplicaba una restricción de columna única activando deliberadamente 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 la retrotraiga explícitamente. Por lo tanto, cualquier operación de ORM adicional en esa transacción, como mi do_more_model_stuff() , fallará con esa excepción django.db.transaction.TransactionManagementError .

Al igual que caio mencionado 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 intencionalmente lanzada rompa la transacción de la prueba unitaria.

Answer 4

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

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

luego encuentro un documento sobre el pedido en el que se ejecutan las pruebas , menciona qué prueba se ejecutará primero.

Entonces, uso TestCase para la interacción con la base de datos, unittest.TestCase para otra prueba simple, ¡funciona ahora!

Answer 5

Tuve el mismo problema.

En mi caso, yo estaba haciendo esto

author.tasks.add(tasks)

así que convertirlo a

author.tasks.add(*tasks)

Eliminado ese error.