TransactionManagementError "Vous ne pouvez pas exécuter de requêtes avant la fin du bloc" atomique "" lors de l'utilisation de signaux, mais uniquement pendant les tests unitaires

J'obtiens TransactionManagementError en essayant de sauvegarder une instance de modèle Django User et dans son signal post_save, j'enregistre certains modèles qui ont l'utilisateur comme clé étrangère.

Le contexte et l'erreur sont assez similaires à cette question django TransactionManagementError lors de l'utilisation de signaux

Cependant, dans ce cas, l'erreur se produit uniquement lors des tests unitaires .

Cela fonctionne bien dans les tests manuels, mais les tests unitaires échouent.

Y a-t-il quelque chose qui me manque?

Voici les extraits de code:

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)

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.

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

J'ai moi-même rencontré le même problème. Cela est dû à une bizarrerie dans la façon dont les transactions sont gérées dans les nouvelles versions de Django, couplées à un test unitaire qui déclenche intentionnellement une exception.

J'ai eu un test unitaire qui a vérifié pour s'assurer qu'une contrainte de colonne unique était appliquée en déclenchant délibérément une 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()

Dans Django 1.4, cela fonctionne très bien. Cependant, dans Django 1.5 / 1.6, chaque test est encapsulé dans une transaction, donc si une exception se produit, il interrompt la transaction jusqu'à ce que vous l'annuliez explicitement. Par conséquent, toute autre opération ORM dans cette transaction, telle que my do_more_model_stuff(), échouera avec cette django.db.transaction.TransactionManagementErrorexception.

Comme caio mentionné dans les commentaires, la solution est de capturer votre exception avec transaction.atomiccomme:

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

Cela empêchera l'exception lancée délibérément de rompre toute la transaction d'unittest.

Puisque @mkoistinen n'a jamais fait son commentaire , une réponse, je posterai sa suggestion pour que les gens n'aient pas à fouiller dans les commentaires.

envisagez simplement de déclarer votre classe de test en tant que TransactionTestCase plutôt que simplement TestCase.

À partir de la documentation : Un TransactionTestCase peut appeler commit et rollback et observer les effets de ces appels sur la base de données.

Si vous utilisez pytest-django, vous pouvez passer transaction=Trueau django_dbdécorateur pour éviter cette erreur.

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

Django lui-même a le TransactionTestCase qui vous permet de tester les transactions et videra la base de données entre les tests pour les isoler. L'inconvénient est que ces tests sont beaucoup plus lents à mettre en place en raison du vidage nécessaire de la base de données. pytest-django prend également en charge ce style de tests, que vous pouvez sélectionner en utilisant un argument de la marque django_db:

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

Voici une autre façon de le faire, basée sur la réponse à cette question:

with transaction.atomic():
    self.assertRaises(IntegrityError, models.Question.objects.create, **{'domain':self.domain, 'slug':'barks'})

Pour moi, les correctifs proposés n'ont pas fonctionné. Dans mes tests, j'ouvre des sous-processus avecPopen pour analyser / lint les migrations (par exemple un test vérifie s'il n'y a pas de changement de modèle).

Pour moi, sous-classer de SimpleTestCaseau lieu deTestCase faire l'affaire.

Notez que SimpleTestCase cela ne permet pas d'utiliser la base de données.

Bien que cela ne réponde pas à la question initiale, j'espère que cela aidera certaines personnes de toute façon.

J'obtenais cette erreur lors de l'exécution de tests unitaires dans ma fonction create_test_data en utilisant django 1.9.7. Cela fonctionnait dans les versions antérieures de django.

Cela ressemblait à ceci:

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

Ma solution était d'utiliser à la place 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})

J'ai le même problème, mais with transaction.atomic()et TransactionTestCasen'a pas fonctionné pour moi.

python manage.py test -rau lieu de python manage.py testça me va, peut-être que l'ordre d'exécution est crucial

puis je trouve un document sur l' ordre dans lequel les tests sont exécutés , il mentionne quel test sera exécuté en premier.

Donc, j'utilise TestCase pour l'interaction avec la base de données, unittest.TestCasepour d'autres tests simples, cela fonctionne maintenant!

La réponse de @kdazzle est correcte. Je ne l'ai pas essayé parce que les gens ont dit que «la classe TestCase de Django est une sous-classe plus couramment utilisée de TransactionTestCase», alors j'ai pensé que c'était la même utilisation l'un ou l'autre. Mais le blog de Jahongir Rahmonov l' expliquait mieux:

la classe TestCase encapsule les tests dans deux blocs atomic () imbriqués: un pour toute la classe et un pour chaque test. C'est là que TransactionTestCase doit être utilisé. Il n'enveloppe pas les tests avec le bloc atomic () et vous pouvez donc tester vos méthodes spéciales qui nécessitent une transaction sans aucun problème.

EDIT: Cela n'a pas fonctionné, j'ai pensé que oui, mais NON.

En 4 ans, ils pourraient régler ce problème .......................................

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

J'ai eu le même problème.

Dans mon cas, je faisais ça

author.tasks.add(tasks)

alors le convertir en

author.tasks.add(*tasks)

Supprimé cette erreur.