Test unitari per query in SQLAlchemy

Question

Come si fa a testare le query in SQLAlchemy? Per esempio supponiamo di avere questo models.py

from sqlalchemy import (
     Column, 
     Integer, 
     String, 
) 
from sqlalchemy.ext.declarative import declarative_base 

Base = declarative_base() 

class Panel(Base): 
    __tablename__ = 'Panels' 

    id = Column(Integer, primary_key=True) 
    category = Column(Integer, nullable=False) 
    platform = Column(String, nullable=False) 
    region = Column(String, nullable=False) 

    def __init__(self, category, platform, region): 
     self.category = category 
     self.platform = platform 
     self.region = region 


    def __repr__(self): 
     return (
      "<Panel('{self.category}', '{self.platform}', " 
      "'{self.region}')>".format(self=self) 
     ) 

e questo tests.py

import unittest 

from sqlalchemy import create_engine 
from sqlalchemy.orm import sessionmaker 

from models import Base, Panel 


class TestQuery(unittest.TestCase): 

    engine = create_engine('sqlite:///:memory:') 
    Session = sessionmaker(bind=engine) 
    session = Session() 

    def setUp(self): 
     Base.metadata.create_all(self.engine) 
     self.session.add(Panel(1, 'ion torrent', 'start')) 
     self.session.commit() 

    def tearDown(self): 
     Base.metadata.drop_all(self.engine) 

    def test_query_panel(self): 
     expected = [Panel(1, 'ion torrent', 'start')] 
     result = self.session.query(Panel).all() 
     self.assertEqual(result, expected) 

Quando abbiamo provare a eseguire il test, non riesce, anche se i due pannelli appaiono identici.

$ nosetests 
F 
====================================================================== 
FAIL: test_query_panel (tests.TestQuery) 
---------------------------------------------------------------------- 
Traceback (most recent call last): 
    File "/Users/clasher/tmp/tests.py", line 31, in test_query_panel 
    self.assertEqual(result, expected) 
AssertionError: Lists differ: [<Panel('1', 'ion torrent', 's... != [<Panel('1', 'ion torrent', 's... 

First differing element 0: 
<Panel('1', 'ion torrent', 'start')> 
<Panel('1', 'ion torrent', 'start')> 

    [<Panel('1', 'ion torrent', 'start')>, <Panel('2', 'ion torrent', 'end')>] 

---------------------------------------------------------------------- 
Ran 1 test in 0.063s 

FAILED (failures=1) 

Una soluzione che ho trovato è quello di fare una query per ogni singola istanza mi aspetto di trovare nella query:

class TestQuery(unittest.TestCase): 

    ... 

    def test_query_panel(self): 
     expected = [ 
      (1, 'ion torrent', 'start'), 
      (2, 'ion torrent', 'end') 
     ] 
     successful = True 
     # Check to make sure every expected item is in the query 
     try: 
      for category, platform, region in expected: 
       self.session.query(Panel).filter_by(
         category=category, platform=platform, 
         region=region).one() 
     except (NoResultFound, MultipleResultsFound): 
      successful = False 
     self.assertTrue(successful) 
     # Check to make sure no unexpected items are in the query 
     self.assertEqual(self.session.query(Panel).count(), 
         len(expected)) 

Questo mi sembra abbastanza brutto, però, e io sono nemmeno arrivando al punto in cui ho una query filtrata complessa che sto provando a testare. C'è una soluzione più elegante, o devo sempre fare manualmente una serie di singole query?

Answer 1

il test originale è sulla strada giusta, devi solo fare una delle due cose: o fare in modo che due Panel oggetti della stessa identità chiave primaria confrontano come True:

class Panel(Base): 
    # ... 

    def __eq__(self, other): 
     return isinstance(other, Panel) and other.id == self.id 

oppure è possibile organizzare il test in modo tale che si assicura che si sta controllando contro lo stesso Panel istanza (perché qui ci avvaliamo del identity map):

class TestQuery(unittest.TestCase): 
    def setUp(self): 
     self.engine = create_engine('sqlite:///:memory:') 
     self.session = Session(engine) 
     Base.metadata.create_all(self.engine) 
     self.panel = Panel(1, 'ion torrent', 'start') 
     self.session.add(self.panel) 
     self.session.commit() 

    def tearDown(self): 
     Base.metadata.drop_all(self.engine) 

    def test_query_panel(self): 
     expected = [self.panel] 
     result = self.session.query(Panel).all() 
     self.assertEqual(result, expected) 

per quanto riguarda la messa a punto del motore/session/teardown, sceglierei un pattern in cui si utilizza un singolo motore per tutti i test e, supponendo che lo schema sia corretto, un singolo schema per tutti i test, quindi ci si assicura che i dati con cui si lavora vengano eseguiti all'interno di una transazione che può essere tornato alla situazione precedente. È possibile eseguire Session in questo modo, in modo tale che la chiamata di commit() non impegni effettivamente la transazione "reale", avvolgendo l'intero test all'interno di un esplicito Transaction. L'esempio su https://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites illustra questo utilizzo. Avere un motore ": memory:" su ogni dispositivo di test occuperà molta memoria e non scalerà realmente su altri database oltre a SQLite.