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

194

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.

----------------------------------------------------------------------
Gaurav Toshniwal
fonte
Dos documentos: "Um TestCase, por outro lado, não trunca tabelas após um teste. Em vez disso, inclui o código de teste em uma transação de banco de dados que é revertida no final do teste. Ambos os commit explícitos, como transaction.commit () e implícitos que podem ser causados ​​por transaction.atomic () são substituídos por uma operação nop. Isso garante que a reversão no final do teste restaure o banco de dados ao seu estado inicial. "
Gaurav Toshniwal
6
Eu encontrei o meu problema. Houve uma exceção IntegrityError como esta "try: ... exceto IntegrityError: ..." o que eu fiz foi usar o transaction.atomic dentro do bloco try: "try: with transaction.atomic (): .. "exceto IntegrityError: ..." agora tudo funciona bem.
caio
docs.djangoproject.com/en/dev/topics/db/transactions e, em seguida, procure por "Embrulho atômica em um try / exceto bloco permite a movimentação natural de erros de integridade:"
CamHart

Respostas:

236

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.

Cerin
fonte
70
Considere também apenas declarar sua classe de teste como um TransactionTestCase em vez de apenas TestCase.
precisa saber é o seguinte
1
Encontrei o documento relacionado de outra pergunta . O documento está aqui .
yaobin
2
Para mim, eu tinha um transaction.atomic()bloqueio, mas recebi esse erro e não fazia ideia do porquê. Eu segui o conselho desta resposta e coloquei um bloco atômico aninhado dentro do meu bloco atômico em torno da área problemática. Depois disso, deu um erro detalhado do erro de integridade que acertei, permitindo que eu corrigisse meu código e fizesse o que estava tentando fazer.
AlanSE
5
@mkoistinen TestCaseestá herdando, TransactionTestCaseentão não há necessidade de mudar isso. Se você não operar no DB em uso de teste SimpleTestCase.
bns
1
@bns você está perdendo o objetivo do comentário. Sim TestCaseherda, TransactionTestCasemas seu comportamento é bem diferente: envolve cada método de teste em uma transação. TransactionTestCase, por outro lado, talvez tenha um nome enganador: trunca tabelas para redefinir o banco de dados - a nomeação parece refletir que você pode testar transações dentro de um teste, não que o teste seja encapsulado como uma transação!
CS
48

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.

kdazzle
fonte
2
+1 para isso, mas, como dizem os documentos, "a classe TestCase do Django é uma subclasse de TransactionTestCase mais usada". Para responder à pergunta original, não devemos usar o SimpleTestCase em vez do TestCase? SimpleTestCase não possui os recursos atômicos do banco de dados.
Daigorocub 2/16
@daigorocub Ao herdar de SimpleTestCase, allow_database_queries = Truedeve ser adicionado dentro da classe de teste, para que não cuspa um AssertionError("Database queries aren't allowed in SimpleTestCase...",).
CristiFati 25/01
Esta é a resposta que funciona melhor para mim enquanto eu estava tentando teste para IntegrityError será levantada e posteriormente eu precisava correr mais banco de dados guardar Pesquisas
Kim Stacks
8

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
frmdstryr
fonte
Eu tive um problema com esta solução, tinha dados iniciais no meu banco de dados (adicionados por migrações). Essa solução liberou o banco de dados; portanto, outros testes dependentes desses dados iniciais começaram a falhar.
Abumalick 24/05/19
1

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.

flix
fonte
1

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'})
Mahdi Hamzeh
fonte
0

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})
PhoebeB
fonte
1
get_or_create()funciona bem, parece que é a .save () que não gosta dentro de uma função decorada transaction.atomic () (a minha falhou com apenas uma chamada).
Timothy Makobu
0

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!

Leo
fonte
0

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

Shil Nevado
fonte
0
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
Aleksei Khatkevich
fonte
-4

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.

Diaa Mohamed Kasem
fonte