TransactionManagementError "Anda tidak dapat menjalankan query sampai akhir blok 'atom'" saat menggunakan sinyal, tetapi hanya selama Unit Testing

194

Saya mendapatkan TransactionManagementError ketika mencoba untuk menyimpan contoh model Pengguna Django dan dalam sinyal post_save, saya menyimpan beberapa model yang memiliki pengguna sebagai kunci asing.

Konteks dan kesalahannya cukup mirip dengan pertanyaan ini django TransactionManagementError saat menggunakan sinyal

Namun, dalam kasus ini, kesalahan hanya terjadi saat pengujian unit .

Ini bekerja dengan baik dalam pengujian manual, tetapi pengujian unit gagal.

Apakah ada sesuatu yang saya lewatkan?

Berikut cuplikan kode:

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)

Melacak kembali:

======================================================================
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
sumber
Dari dokumen: "TestCase, di sisi lain, tidak memotong tabel setelah tes. Sebagai gantinya, ia menyertakan kode tes dalam transaksi basis data yang digulirkan kembali di akhir tes. Keduanya secara eksplisit melakukan seperti transaction.commit () dan yang tersirat yang mungkin disebabkan oleh transaksi.atomic () diganti dengan operasi nop. Ini menjamin bahwa rollback pada akhir tes mengembalikan database ke keadaan awal. "
Gaurav Toshniwal
6
Saya menemukan masalah saya. Ada pengecualian IntegrityError seperti ini "coba: ... kecuali IntegrityError: ..." apa yang harus saya lakukan adalah menggunakan transaksi.otomik di dalam blok percobaan: "coba: dengan transaction.atomic (): .. kecuali IntegrityError: ... "sekarang semuanya bekerja dengan baik.
caio
docs.djangoproject.com/en/dev/topics/db/transaksi dan kemudian mencari "Membungkus atomik dalam blok coba / kecuali memungkinkan penanganan kesalahan integritas alami:"
CamHart

Jawaban:

236

Saya mengalami masalah yang sama sendiri. Hal ini disebabkan oleh kekhasan dalam bagaimana transaksi ditangani dalam versi baru Django ditambah dengan unittest yang dengan sengaja memicu pengecualian.

Saya memiliki unittest yang memeriksa untuk memastikan kendala kolom unik ditegakkan dengan sengaja memicu pengecualian 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()

Di Django 1.4, ini berfungsi dengan baik. Namun, dalam Django 1.5 / 1.6, setiap tes dibungkus dalam suatu transaksi, jadi jika pengecualian terjadi, itu akan memecah transaksi sampai Anda secara eksplisit mengembalikannya. Karenanya, setiap operasi ORM lebih lanjut dalam transaksi itu, seperti saya do_more_model_stuff(), akan gagal dengan django.db.transaction.TransactionManagementErrorpengecualian itu.

Seperti caio yang disebutkan dalam komentar, solusinya adalah menangkap pengecualian Anda dengan transaction.atomicseperti:

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

Itu akan mencegah pengecualian yang sengaja dilempar dari melanggar seluruh transaksi unittest.

Cerin
sumber
70
Juga pertimbangkan untuk mendeklarasikan kelas tes Anda sebagai TransactionTestCase daripada hanya TestCase.
mkoistinen
1
Oh, saya menemukan dokumen terkait dari pertanyaan lain . Dokumennya ada di sini .
yaobin
2
Bagi saya, saya sudah memiliki transaction.atomic()blok, tetapi saya mendapatkan kesalahan ini dan saya tidak tahu mengapa. Saya mengambil saran jawaban ini dan meletakkan blok atom bersarang di dalam blok atom saya di sekitar area masalah. Setelah itu, ia memberikan kesalahan rinci kesalahan integritas yang saya tekan, memungkinkan saya untuk memperbaiki kode saya dan melakukan apa yang saya coba lakukan.
AlanSE
5
@mkoistinen TestCasemewarisi dari TransactionTestCasejadi tidak perlu mengubahnya. Jika Anda tidak beroperasi pada DB dalam penggunaan tes SimpleTestCase.
bns
1
@ bns Anda kehilangan titik komentar. Ya TestCasemewarisi dari TransactionTestCasetetapi perilakunya sangat berbeda: ia membungkus setiap metode pengujian dalam suatu transaksi. TransactionTestCase, di sisi lain, mungkin diberi nama yang menyesatkan: ini memotong tabel untuk mereset db - penamaan tersebut tampaknya mencerminkan bahwa Anda dapat menguji transaksi dalam suatu tes, bukan bahwa tes tersebut dibungkus sebagai suatu transaksi!
CS
48

Karena @mkoistinen tidak pernah membuat komentar , jawaban, saya akan mengirim sarannya sehingga orang tidak perlu menggali komentar.

pertimbangkan hanya mendeklarasikan kelas tes Anda sebagai TransactionTestCase daripada hanya TestCase.

Dari dokumen : TransactionTestCase dapat memanggil commit dan rollback dan mengamati efek dari panggilan ini pada database.

kdazzle
sumber
2
Memberi +1 untuk ini, tetapi, seperti yang dikatakan oleh dokumen, "Kelas TestCase Django adalah subkelas TransactionTestCase yang lebih umum digunakan". Untuk menjawab pertanyaan awal, bukankah sebaiknya kita menggunakan SimpleTestCase alih-alih TestCase? SimpleTestCase tidak memiliki fitur basis data atom.
daigorocub
@daigorocub Saat mewarisi dari SimpleTestCase, allow_database_queries = Trueharus ditambahkan di dalam kelas tes, agar tidak dimuntahkan AssertionError("Database queries aren't allowed in SimpleTestCase...",).
CristiFati
Ini adalah jawaban yang paling cocok untuk saya karena saya mencoba menguji integritas akan muncul dan kemudian saya perlu menjalankan lebih banyak basis data menyimpan pertanyaan
Kim Stacks
8

Jika menggunakan pytest-django Anda dapat beralih transaction=Trueke django_dbdekorator untuk menghindari kesalahan ini.

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

Django sendiri memiliki TransactionTestCase yang memungkinkan Anda untuk menguji transaksi dan akan membuka basis data di antara tes untuk mengisolasi mereka. Kelemahan dari ini adalah bahwa tes ini jauh lebih lambat untuk diatur karena diperlukan pembilasan database. pytest-django juga mendukung gaya tes ini, yang dapat Anda pilih menggunakan argumen ke tanda django_db:

@pytest.mark.django_db(transaction=True)
def test_spam():
    pass  # test relying on transactions
frmdstryr
sumber
Saya punya masalah dengan solusi ini, saya punya data awal di DB saya (ditambahkan oleh migrasi). Solusi ini menggelontor basis data, jadi pengujian lain yang bergantung pada data awal ini mulai gagal.
abumalick
1

Bagi saya, perbaikan yang diusulkan tidak berhasil. Dalam pengujian saya, saya membuka beberapa subproses denganPopen untuk menganalisis / lint migrasi (mis. Satu tes memeriksa jika tidak ada perubahan model).

Bagi saya, subclassing dari SimpleTestCasebukannya TestCasememang berhasil.

Catatan yang SimpleTestCasetidak memungkinkan untuk menggunakan database.

Meskipun ini tidak menjawab pertanyaan awal, saya harap ini membantu beberapa orang.

flix
sumber
1

Berikut ini cara lain untuk melakukannya, berdasarkan jawaban atas pertanyaan ini:

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

Saya mendapatkan kesalahan ini saat menjalankan tes unit di fungsi create_test_data menggunakan django 1.9.7. Ini bekerja di versi django sebelumnya.

Itu terlihat seperti ini:

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

Solusi saya adalah menggunakan pembaruan_or_create sebagai gantinya:

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
sumber
1
get_or_create()berfungsi juga, tampaknya itu .save () tidak seperti di dalam fungsi transaksi.atomic () dihiasi (saya gagal hanya dengan 1 panggilan di sana).
Timothy Makobu
0

Saya memiliki masalah yang sama, tetapi with transaction.atomic()dan TransactionTestCasetidak bekerja untuk saya.

python manage.py test -rbukannya python manage.py testok untuk saya, mungkin urutan eksekusi sangat penting

kemudian saya menemukan dokumen tentang Urutan di mana tes dijalankan , Ini menyebutkan tes mana yang akan dijalankan pertama.

Jadi, saya menggunakan TestCase untuk interaksi basis data, unittest.TestCaseuntuk tes sederhana lainnya, berfungsi sekarang!

Leo
sumber
0

Jawaban @kdazzle benar. Saya tidak mencobanya karena orang mengatakan bahwa 'kelas TestCase Django adalah subkelas TransactionTestCase' yang lebih umum digunakan, jadi saya pikir itu adalah penggunaan yang sama satu atau yang lain. Tetapi blog Jahongir Rahmonov menjelaskannya dengan lebih baik:

kelas TestCase membungkus tes dalam dua blok bersarang atom (): satu untuk seluruh kelas dan satu untuk setiap tes. Di sinilah TransactionTestCase harus digunakan. Itu tidak membungkus tes dengan blok atom () dan dengan demikian Anda dapat menguji metode khusus Anda yang memerlukan transaksi tanpa masalah.

EDIT: Tidak berhasil, saya pikir ya, tapi TIDAK.

Dalam 4 tahun mereka dapat memperbaiki ini .......................................

Shil Nevado
sumber
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
sumber
-4

Saya memiliki masalah yang sama.

Dalam Kasus Saya, saya melakukan ini

author.tasks.add(tasks)

jadi mengubahnya menjadi

author.tasks.add(*tasks)

Menghapus kesalahan itu.

Diaa Mohamed Kasem
sumber