pytest-django के साथ async Django टेस्ट में ट्रांज़ैक्शन जैसा रोलबैक पाएं

Async Django टेस्ट में pytest-django पर डेटा लीक क्यों होता है और ट्रांज़ैक्शन चालू किए बिना sync_to_async से विश्वसनीय ORM रोलबैक/आइसोलेशन कैसे पाएं।

डेटाबेस तक पहुँच करने वाले असिंक्रोनस Django टेस्ट अक्सर अनुभवी टीमों को भी हैरान कर देते हैं: pytest-django के django_db मार्कर इस्तेमाल करने के बावजूद, एक टेस्ट में बनी पंक्तियाँ अगले तक लीक हो जाती हैं। यदि आपका WebSocket या ASGI स्टैक async मार्गों पर निर्भर है, तो यह गाइड बताएगी कि ऐसा क्यों होता है और बिना धीमे, पूर्ण ट्रांज़ैक्शन रन का सहारा लिए इसे कैसे सुधारा जाए।

समस्या को पुन: उत्पन्न करना

नीचे दिए गए टेस्ट pytest-django और pytest-asyncio का उपयोग करते हैं। अपेक्षा सीधी है: एक टेस्ट के भीतर डेटा दिखना चाहिए; लेकिन टेस्टों के बीच वह रोलबैक से साफ हो जाना चाहिए। डिफॉल्ट मोड में django_db यही गारंटी देने वाला है।

import pytest
from cameras.models import CameraGroup as GroupUnit
@pytest.mark.django_db
@pytest.mark.asyncio
class AsyncSuite:
    async def test_a(self):  # ठीक है
        await GroupUnit.objects.acreate(name="alpha1")
        assert await GroupUnit.objects.acount() == 1
    async def test_b(self):  # विफल
        # अगर रोलबैक लागू होता, तो test_a का रिकॉर्ड दिखना नहीं चाहिए
        assert await GroupUnit.objects.acount() == 0
@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
class AsyncSuiteTxn:
    async def test_a(self):  # ठीक है
        await GroupUnit.objects.acreate(name="alpha2")
        assert await GroupUnit.objects.acount() == 1
    async def test_b(self):  # ठीक है
        # यहाँ आइसोलेशन बना रहता है, लेकिन टेस्ट धीमे हैं
        assert await GroupUnit.objects.acount() == 0
@pytest.mark.django_db
class SyncSuite:
    def test_a(self):  # ठीक है
        GroupUnit.objects.create(name="alpha3")
        assert GroupUnit.objects.count() == 1
    def test_b(self):  # ठीक है
        # सिंक पाथ रोलबैक के जरिए आइसोलेशन बनाए रखता है
        assert GroupUnit.objects.count() == 0

क्या हो रहा है

व्यवहार sync और async कोड पाथों के बीच बदल जाता है। synchronous टेस्ट में django_db, django.test.TestCase की तरह काम करता है और हर टेस्ट को एक ट्रांज़ैक्शन में लपेटता है जो अंत में रोलबैक हो जाता है। लेकिन async टेस्ट में यही धारणा लागू नहीं होती।

Async मोड में ट्रांज़ैक्शन अभी काम नहीं करते। यदि आपके किसी कोड हिस्से को ट्रांज़ैक्शन जैसा व्यवहार चाहिए, तो हम सलाह देते हैं कि उसे एक एकल synchronous फ़ंक्शन के रूप में लिखें और sync_to_async() के जरिए कॉल करें।

मतलब, ट्रांज़ैक्शनल आइसोलेशन—वह सुरक्षा जाल जो टेस्टों के बीच आपके बदलाव साफ करता है—शुद्ध async ORM कॉल पर सक्रिय नहीं होता। नतीजतन, acreate और संबंधित async queryset मेथड से बनाए गए ऑब्जेक्ट टिके रह सकते हैं और अगले टेस्ट में दिख सकते हैं। django_db(transaction=True) इस्तेमाल करने पर टेस्ट तो पास होंगे, पर यह वास्तविक कमिट करता है और सफाई धीरे होती है, जिससे कुल रनटाइम बढ़ता है।

बिना पूर्ण ट्रांज़ैक्शनों को चालू किए आइसोलेशन बचाने का उपाय

जिस हिस्से को ट्रांज़ैक्शनल सेमान्टिक्स चाहिए, उसे synchronous फ़ंक्शन में चलाएँ और अपने async टेस्ट से asgiref.sync.sync_to_async के जरिए कॉल करें। Django के अपने async टेस्ट भी चुनिंदा जगहों पर sync_to_async और async_to_sync पर निर्भर रहते हैं।

import pytest
from asgiref.sync import sync_to_async, async_to_sync
from cameras.models import CameraGroup as GroupUnit
@pytest.mark.django_db
@pytest.mark.asyncio
class AsyncSuiteFixed:
    async def test_a(self):
        # ट्रांज़ैक्शनल रोलबैक का लाभ पाने के लिए ORM कॉल्स को synchronous रूप से चलाया गया है
        await sync_to_async(GroupUnit.objects.create)(name="alpha1")
        assert await sync_to_async(GroupUnit.objects.count)() == 1
    async def test_b(self):
        # आइसोलेशन बहाल है क्योंकि पहले वाले बदलाव sync ट्रांज़ैक्शन के भीतर थे
        assert await sync_to_async(GroupUnit.objects.count)() == 0
# विकल्प के तौर पर, कुछ जगहों पर दिशा उलट सकते हैं
# पूरे async टेस्ट बॉडी को async_to_sync के जरिए चलाकर।
# @async_to_sync
# async def test_bulk_like_case():
#     ... यहाँ ORM से जुड़ी कॉल्स के लिए sync_to_async(...) का उपयोग करें ...

यह क्यों मायने रखता है

आइसोलेशन के बिना async टेस्ट सूट नाज़ुक होते हैं। टेस्टों के बीच डेटा लीक होने से कभी-कभी अस्थिर विफलताएँ आती हैं, असली दोष छिप जाते हैं, और डिबगिंग धीमी पड़ती है। WebSocket फ्लो या अन्य ASGI-चालित लॉजिक की जाँच करने वाली टीमों के लिए, Django ORM को विश्वसनीय टेस्ट आइसोलेशन में शामिल करना निर्धारणशीलता और गति, दोनों के लिए जरूरी है।

व्यावहारिक निष्कर्ष

अगर आपको async टेस्ट में ट्रांज़ैक्शन जैसा व्यवहार चाहिए, तो डेटाबेस से जुड़ा हिस्सा sync_to_async के जरिए synchronous चलाएँ। इससे आपको synchronous टेस्ट की तरह रोलबैक सेमान्टिक्स मिलते हैं, बिना पूरी क्लास को transaction=True पर चलाने की परफॉर्मेंस कीमत चुकाए। अगर आप सरल रास्ता चाहते हैं और लंबा रनटाइम स्वीकार्य है, तो ऊपर बताए अनुसार django_db(transaction=True) सक्षम करना भी काम करता है, लेकिन धीमी गति की उम्मीद रखें क्योंकि यह वास्तविक कमिट करता है।

किसी भी रास्ते में, यह स्पष्ट रखें कि आपकी ORM कॉल्स कहाँ निष्पादित होंगी। जिन हिस्सों को ट्रांज़ैक्शन पर निर्भर रहना है, उन्हें synchronous सीमा के भीतर रखें, और async टेस्ट से उस सीमा को sync_to_async के साथ कॉल करें। इससे टेस्ट तेज, अलग-थलग और अनुमानित बने रहते हैं।