TransactionManagementError «Вы не можете выполнять запросы до конца« атомарного »блока» при использовании сигналов, но только во время модульного тестирования

196

Я получаю TransactionManagementError при попытке сохранить экземпляр модели Django User, а в сигнале post_save я сохраняю некоторые модели, в которых пользователь использует внешний ключ.

Контекст и ошибка очень похожи на этот вопрос django TransactionManagementError при использовании сигналов

Однако в этом случае ошибка возникает только при юнит-тестировании .

Это хорошо работает при ручном тестировании, но модульные тесты не пройдены.

Есть ли что-то, что я пропускаю?

Вот фрагменты кода:

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)

Выслеживать:

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

----------------------------------------------------------------------
Гаурав Тошнивал
источник
Из документации: "TestCase, с другой стороны, не усекает таблицы после теста. Вместо этого он заключает тестовый код в транзакцию базы данных, которая откатывается в конце теста. Оба явных коммита, как транзакции транзакции .mit () и неявные, которые могут быть вызваны транзакцией.atomic (), заменены операцией nop. Это гарантирует, что откат в конце теста восстанавливает исходное состояние базы данных. "
Гаурав Тошнивал
6
Я нашел свою проблему. Было исключение IntegrityError, подобное этому «try: ... кроме IntegrityError: ...», что мне нужно было сделать, это использовать транзакцию.atomic внутри блока try: «try: with transaction.atomic (): .. кроме IntegrityError: ... "теперь все работает нормально.
Caio
docs.djangoproject.com/en/dev/topics/db/transactions, а затем выполните поиск «Обтекание атомарных элементов в блоке try / Кроме, что позволяет естественным образом обрабатывать ошибки целостности:»
CamHart

Ответы:

238

Я столкнулся с этой же проблемой сам. Это вызвано причудой в том, как транзакции обрабатываются в более новых версиях Django в сочетании с unittest, который преднамеренно вызывает исключение.

У меня был unittest, который проверял, чтобы убедиться, что ограничение уникального столбца было применено, целенаправленно вызывая исключение 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()

В Django 1.4 это работает нормально. Однако в Django 1.5 / 1.6 каждый тест заключен в транзакцию, поэтому, если возникает исключение, он прерывает транзакцию, пока вы явно не откатите ее. Поэтому любые дальнейшие операции ORM в этой транзакции, такие как my do_more_model_stuff(), завершатся с этим django.db.transaction.TransactionManagementErrorисключением.

Как и caio, упомянутый в комментариях, решение состоит в том, чтобы зафиксировать ваше исключение с помощью 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

Это предотвратит намеренно выброшенное исключение от прерывания транзакции всего юнит-теста.

Cerin
источник
71
Также рассмотрите возможность объявления вашего тестового класса как TransactionTestCase, а не просто TestCase.
mkoistinen
1
О, я нашел соответствующий документ из другого вопроса . Документ здесь .
Яобин
2
Для меня у меня уже был transaction.atomic()блок, но я получил эту ошибку, и я понятия не имел, почему. Я воспользовался советом этого ответа и поместил вложенный атомный блок внутри моего атомного блока вокруг проблемной зоны. После этого он дал подробную ошибку ошибки целостности, которую я ударил, что позволило мне исправить свой код и сделать то, что я пытался сделать.
AlanSE
5
@mkoistinen TestCaseнаследует, TransactionTestCaseтак что не нужно это менять. Если вы не работаете с БД в тестовом режиме SimpleTestCase.
BNS
1
@bns, ты упустил смысл комментария. Да TestCaseнаследуется от, TransactionTestCaseно его поведение совершенно иное: оно оборачивает каждый тестовый метод в транзакцию. TransactionTestCaseс другой стороны, возможно, его вводят в заблуждение: он обрезает таблицы для сброса БД - наименование, кажется, отражает то, что вы можете тестировать транзакции внутри теста, а не то, что тест обернут как транзакция!
CS
48

Так как @mkoistinen так и не сделал свой комментарий , ответ, я опубликую его предложение, чтобы людям не приходилось копаться в комментариях.

рассмотрите возможность объявления вашего тестового класса как TransactionTestCase, а не просто TestCase.

Из документов : TransactionTestCase может вызывать commit и rollback и наблюдать влияние этих вызовов на базу данных.

kdazzle
источник
2
+1 за это, но, как говорят документы, «класс TestCase Джанго является более часто используемым подклассом TransactionTestCase». Чтобы ответить на первоначальный вопрос, не должны ли мы использовать SimpleTestCase вместо TestCase? SimpleTestCase не имеет элементарных функций базы данных.
daigorocub
@daigorocub При наследовании от SimpleTestCase, allow_database_queries = Trueдолжен быть добавлен в тестовый класс, чтобы он не выплевывал AssertionError("Database queries aren't allowed in SimpleTestCase...",).
CristiFati
Это ответ, который лучше всего подходит для меня, так как я пытаюсь проверить целостность ошибки, а затем мне нужно будет выполнить больше запросов сохранения базы данных
Ким Стэкс
8

Если вы используете pytest-django, вы можете перейти transaction=Trueк django_dbдекоратору, чтобы избежать этой ошибки.

См. Https://pytest-django.readthedocs.io/en/latest/database.html#testing-transactions

Сам Django имеет TransactionTestCase, который позволяет вам тестировать транзакции и очищать базу данных между тестами, чтобы изолировать их. Недостатком этого является то, что эти тесты гораздо медленнее настраиваются из-за необходимой очистки базы данных. pytest-django также поддерживает этот стиль тестов, который вы можете выбрать, используя аргумент для метки django_db:

@pytest.mark.django_db(transaction=True)
def test_spam():
    pass  # test relying on transactions
frmdstryr
источник
У меня была проблема с этим решением, у меня были начальные данные в моей БД (добавленные миграциями). Это решение очищает базу данных, поэтому другие тесты, зависящие от этих исходных данных, начали давать сбой.
abumalick
2

Вот еще один способ сделать это, основываясь на ответе на этот вопрос:

with transaction.atomic():
    self.assertRaises(IntegrityError, models.Question.objects.create, **{'domain':self.domain, 'slug':'barks'})
Махди Хамзех
источник
1

Для меня предложенные исправления не сработали. В моих тестах я открываю несколько подпроцессов Popenдля анализа / миграции миграции (например, один тест проверяет, нет ли изменений в модели).

Для меня подклассы SimpleTestCaseвместо вместо этого TestCaseсделали свое дело.

Обратите внимание, что SimpleTestCaseне позволяет использовать базу данных.

Хотя это не отвечает на первоначальный вопрос, я надеюсь, что это все равно поможет некоторым людям.

Фликс
источник
0

Я получал эту ошибку при запуске модульных тестов в моей функции create_test_data с использованием django 1.9.7. Это работало в более ранних версиях Django.

Это выглядело так:

cls.localauth,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeLA, name='LA for test', email_general='test@test.com', address='test', postcode='test', telephone='test')
cls.chamber,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeC, name='chamber for test', email_general='test@test.com', address='test', postcode='test', telephone='test')
cls.lawfirm,_ = Organisation.objects.get_or_create(organisation_type=cls.orgtypeL, name='lawfirm for test', email_general='test@test.com', 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()

Мое решение было использовать вместо этого update_or_create:

cls.localauth,_ = Organisation.objects.update_or_create(organisation_type=cls.orgtypeLA, name='LA for test', email_general='test@test.com', 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='test@test.com', 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='test@test.com', address='test', postcode='test', telephone='test', defaults={'active': True})
PhoebeB
источник
1
get_or_create()работает также, кажется, что это .save (), что ему не нравится внутри украшенной транзакции.atomic () функции (моя не удалась только с одним вызовом там).
Тимоти Макобу
0

У меня та же проблема, но у меня так with transaction.atomic()и TransactionTestCaseне получилось.

python manage.py test -rвместо python manage.py testменя это нормально, может быть, порядок исполнения имеет решающее значение

затем я нахожу документ о порядке, в котором выполняются тесты , в котором упоминается, какой тест будет выполняться первым.

Итак, я использую TestCase для взаимодействия с базой данных, unittest.TestCaseдля другого простого теста, он работает сейчас!

Лео
источник
0

Ответ @kdazzle правильный. Я не пробовал, потому что люди говорили, что класс TestCase от Django является более часто используемым подклассом TransactionTestCase, поэтому я подумал, что это одно и то же использование. Но блог Джахонгира Рахмонова объяснил это лучше:

класс TestCase заключает тесты в два вложенных блока atomic (): один для всего класса и один для каждого теста. Это где TransactionTestCase должен быть использован. Он не оборачивает тесты блоком atomic (), и, таким образом, вы можете без особых проблем протестировать ваши специальные методы, которые требуют транзакции.

РЕДАКТИРОВАТЬ: Это не сработало, я думал, да, но нет.

Через 4 года они могли это исправить .......................................

Шил Невадо
источник
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
Алексей Хаткевич
источник
-4

Я была такая же проблема.

В моем случае я делал это

author.tasks.add(tasks)

так что преобразовать его в

author.tasks.add(*tasks)

Убрал эту ошибку.

Диа Мохамед Касем
источник