कैसे इकाई में SQLAlchemy प्रश्नों गिनती करने के लिए परीक्षण

Question

Django में मैं अक्सर प्रश्नों की संख्या कि इतना है कि इकाई परीक्षण नई एन 1 क्वेरी समस्याओं

from django import db 
from django.conf import settings 
settings.DEBUG=True 

class SendData(TestCase): 
    def test_send(self): 
     db.connection.queries = [] 
     event = Events.objects.all()[1:] 
     s = str(event) # QuerySet is lazy, force retrieval 
     self.assertEquals(len(db.connection.queries), 2) 

में SQLAlchemy में STDOUT करने के लिए अनुरेखण सक्षम किया गया है को पकड़ने के प्रयास किए जाने चाहिए पर जोर इंजन

engine.echo=True 

सबसे अच्छा तरीका परीक्षण है कि SQLAlchemy द्वारा किए गए प्रश्नों की संख्या की गिनती लिखने के लिए क्या है पर echo ध्वज की स्थापना करके?

class SendData(TestCase): 
    def test_send(self): 
     event = session.query(Events).first() 
     s = str(event) 
     self.assertEquals(... , 2) 

Answer 1

लॉग इन करने/ट्रैक प्रश्नों निष्पादित उपयोग SQLAlchemy Core Events (यदि आप अपने इकाई परीक्षण से अनुलग्न कर सकते हैं ताकि वे वास्तविक आवेदन पर अपने प्रदर्शन से प्रभावित नहीं करतीं:

event.listen(engine, "before_cursor_execute", catch_queries) 

अब आप समारोह लिखने । catch_queries, जहां जिस तरह से आप कैसे परीक्षण पर निर्भर करता है उदाहरण के लिए, आप इस समारोह में अपने परीक्षण बयान में निर्धारित कर सकते हैं:

def test_something(self): 
    stmts = [] 
    def catch_queries(conn, cursor, statement, ...): 
     stmts.append(statement) 
    # Now attach it as a listener and work with the collected events after running your test 

abov ई विधि सिर्फ एक प्रेरणा है। विस्तारित मामलों के लिए आप शायद प्रत्येक परीक्षा के बाद खाली घटनाओं का वैश्विक कैश रखना चाहते हैं। कारण यह है कि 0.9 (वर्तमान देव) से पहले ईवेंट श्रोताओं को हटाने के लिए कोई एपीआई नहीं है। इस प्रकार एक वैश्विक श्रोता बनाएं जो वैश्विक सूची तक पहुंचता है।

Answer 2

मैं इस उद्देश्य के लिए एक संदर्भ प्रबंधक वर्ग बना लिया है:

class DBStatementCounter(object): 
    """ 
    Use as a context manager to count the number of execute()'s performed 
    against the given sqlalchemy connection. 

    Usage: 
     with DBStatementCounter(conn) as ctr: 
      conn.execute("SELECT 1") 
      conn.execute("SELECT 1") 
     assert ctr.get_count() == 2 
    """ 
    def __init__(self, conn): 
     self.conn = conn 
     self.count = 0 
     # Will have to rely on this since sqlalchemy 0.8 does not support 
     # removing event listeners 
     self.do_count = False 
     sqlalchemy.event.listen(conn, 'after_execute', self.callback) 

    def __enter__(self): 
     self.do_count = True 
     return self 

    def __exit__(self, *_): 
     self.do_count = False 

    def get_count(self): 
     return self.count 

    def callback(self, *_): 
     if self.do_count: 
      self.count += 1