Jak liczyć sqlalchemy zapytań w jednostce testuje

Question

W Django często dochodzić liczbę zapytań, które powinny być wykonane tak, że testy jednostkowe złapać nowych N + 1 problemy zapytań

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) 

W w SQLAlchemy śledzenie na standardowe wyjście jest włączone ustawiając flagę echo na silnika

engine.echo=True 

Jaki jest najlepszy sposób, aby napisać testy, które liczą liczbę zapytań dokonywanych przez SQLAlchemy?

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

Answer 1

Zastosowanie SQLAlchemy Core Events do log/zapytań wykonywanych utworów (można dołączyć go od testów jednostkowych, tak aby nie wpływać na wydajność od rzeczywistej aplikacji:

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

Teraz piszesz funkcję . catch_queries, gdzie droga zależy od sposobu testowania na przykład, można zdefiniować tę funkcję w swoim oświadczeniu testu:

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 Metoda jest tylko inspiracją. W przypadku rozszerzonych przypadków prawdopodobnie będziesz chciał mieć globalną pamięć podręczną zdarzeń, które opróżnisz po każdym teście. Powodem jest to, że przed wersją 0.9 (obecna wersja deweloperska) nie ma API do usuwania detektorów zdarzeń. W ten sposób utwórz jednego globalnego odbiornika, który uzyskuje dostęp do globalnej listy.

Answer 2

I utworzeniu klasy menedżera kontekst do tego celu:

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