wake-up-neo.com

TransactionManagementError "Sie können Abfragen nicht bis zum Ende des 'atomaren' Blocks ausführen", während Sie Signale verwenden, sondern nur während des Unit-Tests

Beim Versuch, eine Instanz eines Django-Benutzermodells zu speichern, erhalte ich TransactionManagementError. Ich speichere in seinem post_save-Signal einige Modelle, die den Benutzer als Fremdschlüssel haben. 

Der Kontext und der Fehler ist dieser Frage ziemlich ähnlich Django TransactionManagementError bei Verwendung von Signalen

In diesem Fall tritt der Fehler jedoch auf nur beim Komponententest.

Beim manuellen Testen funktioniert es gut, aber Unit-Tests schlagen fehl.

Gibt es etwas, was mir fehlt?

Hier sind die Codeausschnitte:

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)

Zurück verfolgen:

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

----------------------------------------------------------------------
142

Ich bin selbst auf dieses Problem gestoßen. Dies wird durch eine merkwürdige Art und Weise verursacht, wie Transaktionen in den neueren Versionen von Django gehandhabt werden, zusammen mit einem einzigen Test, der absichtlich eine Ausnahme auslöst.

Ich hatte einen einzigen Test, der überprüft hat, ob eine eindeutige Spalteneinschränkung durch gezieltes Auslösen einer IntegrityError-Ausnahme erzwungen wurde:

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

In Django 1.4 funktioniert das gut. In Django 1.5/1.6 ist jeder Test jedoch in eine Transaktion eingebettet. Wenn also eine Ausnahme auftritt, wird die Transaktion unterbrochen, bis Sie sie explizit zurücksetzen. Daher schlagen alle weiteren ORM-Vorgänge in dieser Transaktion, wie z. B. meine do_more_model_stuff(), mit dieser Django.db.transaction.TransactionManagementError-Ausnahme fehl.

Wie bei caio in den Kommentaren erwähnt, besteht die Lösung darin, Ihre Ausnahme mit transaction.atomic wie folgt zu erfassen:

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

Dadurch wird verhindert, dass die absichtlich geworfene Ausnahme die gesamte Transaktion der Transaktion abbricht.

185
Cerin

Da @mkoistinen nie seinen Kommentar abgegeben hat, eine Antwort, werde ich seinen Vorschlag posten, damit die Leute nicht durch Kommentare suchen müssen.

erwägen Sie, Ihre Testklasse nur als TransactionTestCase und nicht nur als TestCase zu deklarieren.

Aus den docs : Eine TransactionTestCase kann Commit und Rollback aufrufen und die Auswirkungen dieser Aufrufe auf die Datenbank beobachten.

30
kdazzle

Für mich funktionierten die vorgeschlagenen Korrekturen nicht. In meinen Tests öffne ich einige Unterprozesse mit Popen, um Migrationen von/lint zu analysieren (z. B. prüft ein Test, ob keine Modelländerungen vorliegen).

Für mich war die Unterklasse von SimpleTestCase anstelle von TestCase der Trick.

Beachten Sie, dass SimpleTestCase die Verwendung der Datenbank nicht zulässt.

Obwohl dies die ursprüngliche Frage nicht beantwortet, hoffe ich, dass dies einigen Leuten trotzdem hilft.

1
flix

Ich habe das gleiche Problem, aber with transaction.atomic() und TransactionTestCase haben bei mir nicht funktioniert.

python manage.py test -r anstelle von python manage.py test ist für mich in Ordnung, vielleicht ist die Reihenfolge der Ausführung entscheidend

dann finde ich ein Dokument über Reihenfolge, in der Tests ausgeführt werden , Es wird erwähnt, welcher Test zuerst ausgeführt wird.

Also benutze ich TestCase für die Datenbankinteraktion, unittest.TestCase für andere einfache Tests. Jetzt funktioniert es!

0
Leo

Wenn Sie pytest-Django verwenden, können Sie transaction=True an den Django_db-Dekorator übergeben, um diesen Fehler zu vermeiden.

Siehe https://pytest-Django.readthedocs.io/de/latest/database.html#testing-transactions

Django selbst hat die TransactionTestCase, mit der Sie .__ testen können. Transaktionen und löscht die Datenbank zwischen den Tests, um .__ zu isolieren. Sie. Der Nachteil ist, dass diese Tests aufgrund des erforderlichen Leeren der Datenbank viel langsamer eingerichtet werden. pytest-Django unterstützt auch diesen Teststil, den Sie mit einem Argument für die Marke Django_db auswählen können:

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

Die Antwort von @kdazzle ist richtig. Ich habe es nicht ausprobiert, weil die Leute sagten, dass die TestCase-Klasse von Django eine häufig verwendete Unterklasse von TransactionTestCase ist. Aber das Blog von Jahongir Rahmonov hat es besser erklärt:

die TestCase-Klasse fasst die Tests in zwei verschachtelten atomic () - Blöcken zusammen: eine für die ganze Klasse und eine für jeden Test. Das ist wo TransactionTestCase sollte verwendet werden. Die Tests werden nicht mit .__ umbrochen. atomic () -Block und damit können Sie Ihre speziellen Methoden testen, die .__ erfordern. eine Transaktion ohne Probleme.

EDIT: Es hat nicht funktioniert, ich dachte ja, aber nein.

In 4 Jahren konnten sie dies beheben .......................................

0
Shil Nevado

Ich habe diesen Fehler beim Ausführen von Komponententests in meiner Funktion create_test_data mit Django 1.9.7 erhalten. In früheren Versionen von Django hat es funktioniert.

Es sah so aus:

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

Meine Lösung bestand darin, stattdessen update_or_create zu verwenden:

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})
0
PhoebeB