TransactionManagementError “Você não pode executar consultas até o final do bloco 'atômico'” enquanto estiver usando sinais, mas apenas durante o teste de unidade

Estou recebendo TransactionManagementError ao tentar salvar uma instância de modelo de usuário do Django e em seu sinal post_save, estou salvando alguns modelos que têm o usuário como chave estrangeira.

O contexto e o erro são bastante semelhantes a esta pergunta django TransactionManagementError ao usar sinais

No entanto, nesse caso, o erro ocorre apenas durante o teste de unidade .

Funciona bem em testes manuais, mas os testes de unidade falham.

Falta alguma coisa?

Aqui estão os trechos 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)

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.

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

Eu me deparei com esse mesmo problema. Isso é causado por uma peculiaridade de como as transações são tratadas nas versões mais recentes do Django, juntamente com um mais unittest que intencionalmente aciona uma exceção.

Eu tive um teste mais unificado que verifiquei para garantir que uma restrição de coluna exclusiva fosse imposta ao acionar propositadamente uma exceção 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()

No Django 1.4, isso funciona bem. No entanto, no Django 1.5 / 1.6, cada teste é agrupado em uma transação; portanto, se ocorrer uma exceção, ele interrompe a transação até que você a revele explicitamente. Portanto, qualquer operação ORM adicional nessa transação, como a minha do_more_model_stuff(), falhará com essa django.db.transaction.TransactionManagementErrorexceção.

Como o caio mencionado nos comentários, a solução é capturar sua exceção com o seguinte 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

Isso impedirá que a exceção lançada propositalmente interrompa toda a transação do unittest.

Como @mkoistinen nunca fez seu comentário , responda, vou postar sua sugestão para que as pessoas não precisem procurar nos comentários.

considere apenas declarar sua classe de teste como TransactionTestCase em vez de apenas TestCase.

Dos documentos : Um TransactionTestCase pode chamar confirmação e reversão e observar os efeitos dessas chamadas no banco de dados.

Se estiver usando pytest-django, você pode passar transaction=Truepara o django_dbdecorador para evitar esse erro.

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

O próprio Django possui o TransactionTestCase, que permite testar transações e liberar o banco de dados entre testes para isolá-las. A desvantagem disso é que esses testes são muito mais lentos para configurar devido à liberação necessária do banco de dados. O pytest-django também suporta esse estilo de teste, que você pode selecionar usando um argumento para a marca django_db:

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

Para mim, as correções propostas não funcionaram. Nos meus testes, abro alguns subprocessos Popenpara analisar / lint migrações (por exemplo, um teste verifica se não há alterações no modelo).

Para mim, subclassificar de em SimpleTestCasevez de TestCasefez o truque.

Observe que SimpleTestCasenão permite usar o banco de dados.

Embora isso não responda à pergunta original, espero que isso ajude algumas pessoas de qualquer maneira.

Aqui está outra maneira de fazê-lo, com base na resposta a esta pergunta:

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

Eu estava recebendo esse erro ao executar testes de unidade na minha função create_test_data usando o django 1.9.7. Funcionou em versões anteriores do django.

Parecia assim:

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

Minha solução foi usar 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})

Eu tenho o mesmo problema, mas with transaction.atomic()e TransactionTestCasenão funcionou para mim.

python manage.py test -rem vez de python manage.py testestá bem para mim, talvez a ordem de execução seja crucial

então, encontro um documento sobre a ordem na qual os testes são executados . Ele menciona qual teste será executado primeiro.

Então, eu uso o TestCase para interação com o banco de dados, unittest.TestCasepara outro teste simples, ele funciona agora!

A resposta de @kdazzle está correta. Eu não tentei porque as pessoas diziam que 'a classe TestCase do Django é uma subclasse mais usada do TransactionTestCase', então pensei que era o mesmo uso de um ou outro. Mas o blog de Jahongir Rahmonov explicou melhor:

a classe TestCase agrupa os testes em dois blocos atômicos aninhados (): um para toda a classe e um para cada teste. É aqui que o TransactionTestCase deve ser usado. Ele não envolve os testes com o bloco atomic () e, portanto, você pode testar seus métodos especiais que exigem uma transação sem nenhum problema.

EDIT: Não deu certo, pensei que sim, mas NÃO.

Em 4 anos eles poderiam consertar isso .......................................

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

Eu tive o mesmo problema.

No meu caso, eu estava fazendo isso

author.tasks.add(tasks)

então convertendo para

author.tasks.add(*tasks)

Removido esse erro.