TransactionManagementError „Nie można wykonywać zapytań do końca bloku„ atomowego ”podczas korzystania z sygnałów, ale tylko podczas testowania jednostkowego

Dostaję TransactionManagementError podczas próby zapisania instancji modelu użytkownika Django i w jego sygnale post_save zapisuję niektóre modele, w których użytkownik jest kluczem obcym.

Kontekst i błąd są bardzo podobne do tego pytania django TransactionManagementError podczas używania sygnałów

Jednak w tym przypadku błąd występuje tylko podczas testowania jednostkowego .

Działa dobrze w testach ręcznych, ale testy jednostkowe kończą się niepowodzeniem.

Czy czegoś mi brakuje?

Oto fragmenty kodu:

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.

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

Sam napotkałem ten sam problem. Jest to spowodowane dziwactwem w obsłudze transakcji w nowszych wersjach Django w połączeniu z jednostką, która celowo wyzwala wyjątek.

Miałem unittest, który sprawdził, aby upewnić się, że wymuszono unikalne ograniczenie kolumny, celowo uruchamiając wyjątek 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()

W Django 1.4 to działa dobrze. Jednak w Django 1.5 / 1.6 każdy test jest zawinięty w transakcję, więc jeśli wystąpi wyjątek, przerywa transakcję do momentu jej jawnego wycofania. Dlatego wszelkie dalsze operacje ORM w tej transakcji, takie jak moja do_more_model_stuff(), zakończą się niepowodzeniem z tym django.db.transaction.TransactionManagementErrorwyjątkiem.

Podobnie jak caio wspomniany w komentarzach, rozwiązaniem jest uchwycenie wyjątku za pomocą 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

Zapobiegnie to celowo zgłoszonemu wyjątkowi, który złamie całą transakcję najinteligentniejszego.

Ponieważ @mkoistinen nigdy nie skomentował odpowiedzi, opublikuję jego sugestię, aby ludzie nie musieli przekopywać komentarzy.

rozważ po prostu zadeklarowanie klasy testowej jako TransactionTestCase, a nie tylko TestCase.

Z dokumentacji : TransactionTestCase może wywoływać zatwierdzanie i wycofywanie oraz obserwować wpływ tych wywołań na bazę danych.

Jeśli używasz pytest-django, możesz przejść transaction=Truedodjango_db dekoratora, aby uniknąć tego błędu.

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

Sam Django ma TransactionTestCase, który pozwala testować transakcje i opróżnia bazę danych między testami, aby je odizolować. Wadą tego jest to, że testy te są znacznie wolniejsze w konfiguracji z powodu wymaganego opróżnienia bazy danych. pytest-django obsługuje również ten styl testów, który można wybrać za pomocą argumentu do znaku django_db:

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

Dla mnie proponowane poprawki nie działały. W moich testach otwieram niektóre podprocesy za pomocąPopen aby analizować / migrować kłaczki (np. Jeden test sprawdza, czy nie ma zmian w modelu).

Dla mnie podklasa z SimpleTestCasezamiastTestCase wykonałem.

Zauważ, że SimpleTestCase nie zezwala na korzystanie z bazy danych.

Chociaż to nie odpowiada na pierwotne pytanie, mam nadzieję, że to i tak pomaga niektórym ludziom.

Oto inny sposób, aby to zrobić, w oparciu o odpowiedź na to pytanie:

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

Ten błąd występował podczas uruchamiania testów jednostkowych w mojej funkcji create_test_data przy użyciu django 1.9.7. Działa we wcześniejszych wersjach django.

Wyglądało to tak:

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

Moim rozwiązaniem było użycie zamiast tego 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})

Mam ten sam problem, ale with transaction.atomic()i TransactionTestCasenie działa dla mnie.

python manage.py test -rzamiast python manage.py testjest dla mnie ok, być może kolejność wykonania jest kluczowa

następnie znajduję dokument dotyczący kolejności wykonywania testów Wspomina, który test zostanie uruchomiony jako pierwszy.

Tak więc używam TestCase do interakcji z bazą danych, unittest.TestCasedo innego prostego testu, to działa teraz!

Odpowiedź @kdazzle jest poprawna. Nie próbowałem tego, ponieważ ludzie mówili, że „Klasa TestCase Django jest częściej używaną podklasą TransactionTestCase”, więc pomyślałem, że to samo lub inne zastosowanie. Ale blog Jahongira Rahmonova wyjaśnił to lepiej:

klasa TestCase otacza testy w dwa zagnieżdżone bloki atomic (): jeden dla całej klasy i jeden dla każdego testu. W tym miejscu należy użyć TransactionTestCase. Nie zawija testów blokiem atomic (), dzięki czemu możesz bez problemu przetestować swoje specjalne metody wymagające transakcji.

EDYCJA: Nie działało, myślałem tak, ale NIE.

Za 4 lata mogliby to naprawić .......................................

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

Miałem ten sam problem.

W moim przypadku to robiłem

author.tasks.add(tasks)

więc konwertując to na

author.tasks.add(*tasks)

Usunięto ten błąd.