come contare SQLAlchemy query in unit test

Question

In Django spesso affermano il numero di query che dovrebbe essere fatto in modo che i test di unità catturare nuovi N + 1 problemi di query

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) 

In in SQLAlchemy tracciare STDOUT è abilitata impostando il flag echo su motore

engine.echo=True 

Qual è il modo migliore per scrivere test che contano il numero di ricerche effettuate dagli SQLAlchemy?

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

Answer 1

Usa SQLAlchemy Core Events il login/query eseguite pista (è possibile allegare da test di unità in modo che non influenzano le prestazioni sulla domanda effettiva:

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

Ora si scrive la funzione di . catch_queries, in cui il modo in cui dipende da come si prova, ad esempio, è possibile definire questa funzione nella vostra dichiarazione di prova:

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 

L'abov Il metodo è solo un'ispirazione. Per casi estesi probabilmente ti piacerebbe avere una cache globale di eventi che svuoti dopo ogni test. Il motivo è che prima di 0.9 (current dev) non ci sono API per rimuovere i listener di eventi. Così fai un ascoltatore globale che accede a un elenco globale.

Answer 2

Ho creato una classe contesto manager per questo scopo:

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