Wie sqlalchemy Abfragen in Einheit zählen testet

Question

In Django oft ich die Anzahl der Abfragen behaupten, so dass Unit-Tests neue N + 1 Abfrage Probleme fangen gemacht werden sollten

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 zu STDOUT Tracing aktiviert ist durch Setzen der echo Flagge auf Motor

engine.echo=True 

Was ist der beste Weg, Tests zu schreiben, die die Anzahl der Abfragen gemacht von SQLAlchemy zählen?

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

Answer 1

Verwenden SQLAlchemy Core Events log/Track-Abfragen ausgeführt (Sie es von Ihrem Unit-Tests anbringen können, damit sie nicht Ihre Leistung auf dem tatsächlichen Anwendung auswirken:

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

Jetzt schreiben Sie die Funktion . catch_queries, wo die Art und Weise hängt davon ab, wie Sie testen zum Beispiel könnten Sie diese Funktion in Ihrer Test-Anweisung definieren:

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 

die abov Die Methode ist nur eine Inspiration. Für erweiterte Fälle möchten Sie wahrscheinlich einen globalen Cache mit Ereignissen haben, die Sie nach jedem Test leeren. Der Grund dafür ist, dass es vor 0.9 (current dev) keine API gibt, um Event Listener zu entfernen. Machen Sie also einen globalen Listener, der auf eine globale Liste zugreift.

Answer 2

ich eine Kontext-Managerklasse für diesen Zweck erstellt habe:

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