SQLAlchemy 2.0 पर pytest में नेस्टेड ट्रांज़ैक्शन के साथ टेस्ट आइसोलेशन

SQLAlchemy 2.0 में pytest नेस्टेड ट्रांज़ैक्शन को स्थिर करें: Postgres SAVEPOINT और स्पष्ट Connection से साझा डेटा सुरक्षित रखें, तेज़ व निर्धारक टेस्ट चलाएँ.

जब आप SQLAlchemy 1.3/1.4 से 2.0 पर टेस्ट सूट माइग्रेट करते हैं, तो Postgres SAVEPOINT के जरिए नेस्टेड ट्रांज़ैक्शन पर आधारित पहले के पैटर्न असंगत नतीजे देने लगते हैं। आम तौर पर एक सेटअप यह होता है कि हर मॉड्यूल के लिए साझा फ़िक्स्चर्स एक बार भर दिए जाते हैं, और फिर हर टेस्ट को ऐसे नेस्टेड ट्रांज़ैक्शन में चलाया जाता है जिसे बाद में रोलबैक किया जा सके। 1.4 में यह तरीका अक्सर एक साधारण मंकीपैच से काम कर जाता था जो session.commit को session.flush में बदल देता था। लेकिन 2.0 पर स्विच करने के बाद, लो-लेवल SQLAlchemy की कोई गलती न दिखने पर भी टेस्ट्स साझा डेटा के गायब होने की शिकायत करने लगे।

समस्या का सेटअप

नीचे दिया पैटर्न एक पैकेज-स्कोप्ड सेशन दिखाता है जो साझा डेटा सीड करता है, और एक फ़ंक्शन-स्कोप्ड नेस्टेड ट्रांज़ैक्शन जो हर टेस्ट को अलग रखता है। commit को flush में बदलने वाला मंकीपैच यह सुनिश्चित करता है कि रैपिंग ट्रांज़ैक्शन के बाहर कुछ भी न निकले।

@pytest.fixture(scope="package")
def pkg_sess(suite_engine: Engine, pkg_patch) -> Session:
    link = suite_engine.connect()
    outer_tx = link.begin()
    db_sess = Session(bind=link)
    pkg_patch.setattr(db_sess, "commit", db_sess.flush)
    seed_shared_state(db_sess)
    try:
        yield db_sess
    finally:
        outer_tx.rollback()
        link.close()
@pytest.fixture(scope="function")
def txn_sess(pkg_sess):
    """Create a SAVEPOINT so per-test changes can be rolled back without losing shared data"""
    pkg_sess.begin_nested()
    try:
        yield pkg_sess
    finally:
        pkg_sess.rollback()

असली दिक्कत क्या होती है

SQLAlchemy 2.0 में सूट साझा पंक्तियों के गायब होने की सूचनाओं के साथ फेल होने लगता है। जो तरीका नेस्टेड टेस्ट्स के अंदर साझा डेटा को दिखाई देता रखता था, वह अब पहले जैसा व्यवहार नहीं करता। काफी कोशिशों के बाद एक भरोसेमंद बात सामने आई: कई परतों वाले SAVEPOINT पहले जैसी तरह से दोहराए नहीं जा सके। पैकेज-स्तर की एक नेस्टिंग और प्रति-टेस्ट रोलबैक काम करती है, लेकिन गहरी परतें—जैसे पैकेज-स्तर का साझा डेटा, उसके ऊपर मॉड्यूल-स्तर का साझा डेटा, और फिर फ़ंक्शन-स्तर का रोलबैक—पहले की तरह संचालित नहीं हो पाईं। नतीजतन, एक मॉड्यूल जो इस तिहरी परत पर निर्भर था, उसे प्रति टेस्ट थोड़ा अधिक डेटा सीड करना पड़ा।

SQLAlchemy 2.0 पर काम करने वाला तरीका

मूल उद्देश्य को लगातार बहाल करने वाला समाधान यह है कि एक स्पष्ट Connection रखें, ज़रूरत के मुताबिक उसे पैकेज या टेस्ट-स्तर के बाहरी ट्रांज़ैक्शन में लपेटें, और Session को निर्देश दें कि वह SAVEPOINTs के साथ जॉइन करे। अब commit को मंकीपैच करने की जरूरत नहीं पड़ती।

tests/conftest.py

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session
from sqlalchemy_utils import create_database, database_exists, drop_database
from myapp.db import DeclarativeBase as OrmBase
@pytest.fixture(scope="session")
def qa_engine():
    """Create a dedicated test database, run migrations, and tear it down after the session"""
    eng = create_engine(
        "postgresql+psycopg2://USER:PASSWORD@HOST:PORT/test",
        echo=False,
        future=True,
    )
    if database_exists(eng.url):
        drop_database(eng.url)
    create_database(eng.url)
    OrmBase.metadata.create_all(bind=eng)
    try:
        yield eng
    finally:
        drop_database(eng.url)
@pytest.fixture(scope="function")
def db_session(qa_engine: Engine) -> Session:
    """Open a transaction and use SAVEPOINT for per-test isolation"""
    link = qa_engine.connect()
    tx = link.begin()
    try:
        with Session(bind=link, join_transaction_mode="create_savepoint") as db:
            yield db
    finally:
        tx.rollback()
        link.close()

tests/module_a/conftest.py

from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session
@pytest.fixture(scope="package")
def pkg_link(qa_engine: Engine):
    """Seed the database once per package with the data needed for Module A tests.
    The same connection and outer transaction are reused by tests in this package.
    """
    link = qa_engine.connect()
    top_tx = link.begin()
    top_sess = Session(bind=link, join_transaction_mode="create_savepoint")
    # इस पैकेज के लिए साझा डेटा यहाँ बनाएँ
    try:
        yield link
    finally:
        top_tx.rollback()
        link.close()
@pytest.fixture(scope="function")
def db_session(pkg_link):
    """Use a nested SAVEPOINT so individual tests can roll back without touching shared data"""
    sp = pkg_link.begin_nested()
    try:
        with Session(bind=pkg_link, join_transaction_mode="create_savepoint") as db:
            yield db
    finally:
        sp.rollback()

यह कैसे हल करता है

मुख्य बदलाव यह है कि commit-to-flush मंकीपैच हटाकर Session को साफ-साफ बताया जाए कि उसे पहले से शुरू हो चुके ट्रांज़ैक्शन में कैसे शामिल होना है। join_transaction_mode="create_savepoint" के साथ हर टेस्ट एक SAVEPOINT के तहत चलता है, जबकि पैकेज-स्तर का बाहरी ट्रांज़ैक्शन खुला रहता है। टेस्ट पूरा होने पर SAVEPOINT रोलबैक हो जाता है और पहले से सीड किया गया पैकेज डेटा जस का तस रहता है। इस सेटअप में commit ओवरराइड करने की ज़रूरत नहीं पड़ती; पैकेज-स्तर पर सीडिंग करते समय commit किया जा सकता है और फ़ंक्शन-स्तर का SAVEPOINT रोलबैक सिर्फ उसी टेस्ट के बदलाव वापस करेगा।

टॉप-लेवल के टेस्ट खाली डेटाबेस की अपेक्षा करते हैं। Module A में टेस्ट पहले से भरे हुए डेटा की उम्मीद करते हैं। उसी के समानांतर Module B में टेस्ट एक अलग साझा डेटा सेट चाहते हैं। पैकेज-स्तर की तैयारी से हर मॉड्यूल में साझा डेटा एक बार बनाया जा सकता है, जबकि प्रति-टेस्ट आइसोलेशन बना रहता है।

देखी गई सीमाएँ

कई-स्तरीय नेस्टेड SAVEPOINTs को 2.0 में पहले की तरह दोहराया नहीं जा सका। पैकेज-स्तर की नेस्टिंग के साथ प्रति-टेस्ट SAVEPOINT रोलबैक विश्वसनीय रूप से काम करता है। यदि एक अतिरिक्त परत की जरूरत हो, तो उस मॉड्यूल में कुछ डेटा प्रति टेस्ट सीड करना पड़ सकता है।

यह क्यों महत्वपूर्ण है

सुसंगत ट्रांज़ैक्शनल आइसोलेशन ही टेस्ट्स को निर्धारक और तेज़ रखता है। साझा फ़िक्स्चर्स को प्रति मॉड्यूल एक बार भरना दोहराव और रनटाइम घटाता है, जबकि SAVEPOINT-आधारित रोलबैक हर टेस्ट की लिखत को सीमित रखता है। SQLAlchemy 2.0 के लिए टेस्ट हार्नेस को स्पष्ट कनेक्शन और join_transaction_mode के साथ ढालने से बिना Session के मूल व्यवहार को मंकीपैच किए वही विशेषताएँ वापस मिल जाती हैं।

व्यावहारिक सुझाव

जिस स्कोप में साझा डेटा का स्वामित्व हो, वहाँ एक स्पष्ट Connection के साथ घेरने वाला ट्रांज़ैक्शन रखें। हर टेस्ट का Session join_transaction_mode="create_savepoint" के साथ बनाएँ ताकि रोलबैक सिर्फ उसी टेस्ट की लिखत को प्रभावित करे। commit-to-flush मंकीपैच छोड़ दें; पैकेज-स्तर की सीडिंग के दौरान किए गए commit ठीक हैं, और फ़ंक्शन-स्तर का SAVEPOINT रोलबैक साझा पंक्तियों को अप्रभावित छोड़ेगा। अगर पहले आप एक से अधिक अतिरिक्त SAVEPOINT परत पर निर्भर थे, तो उन मॉड्यूल्स में प्रति टेस्ट थोड़ा डेटा सीड करने के लिए तैयार रहें।

इन समायोजनों के साथ, 1.3/1.4-स्टाइल नेस्टेड-ट्रांज़ैक्शन टेस्ट रणनीति को SQLAlchemy 2.0 पर न्यूनतम रुकावट के साथ आगे बढ़ाया जा सकता है और भरोसेमंद परिणाम मिलते हैं।