2013-02-06 20 views
7

Jak można testować zapytania w SQLAlchemy? Na przykład załóżmy, że mamy tę models.pyTesty jednostkowe dla zapytania w SQLAlchemy

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) 
     ) 

i to 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) 

gdy staramy uruchomiony test, to się nie powiedzie, mimo że dwa panele wyglądają identycznie.

$ 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) 

Jednym rozwiązaniem znalazłem jest dokonanie kwerendy dla każdej pojedynczej instancji spodziewam się znaleźć w zapytaniu:

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)) 

To wydaje mi się dość brzydki, choć i jestem nawet do punktu, w którym mam złożone filtrowane zapytanie, które próbuję przetestować. Czy istnieje bardziej eleganckie rozwiązanie, czy też zawsze muszę ręcznie tworzyć kilka indywidualnych zapytań?

Odpowiedz

14

oryginalny test jest na dobrej drodze, po prostu trzeba zrobić jedną z dwóch rzeczy: albo upewnić się, że dwa Panel obiekty o tej samej tożsamości klucza podstawowego porównać jak True:

class Panel(Base): 
    # ... 

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

lub można zorganizować badanie takie, że upewnij się, że sprawdzanie przeciwko tej samej Panel instancji (bo tu skorzystać z 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) 

miarę konfiguracji silnika/sesji/W przypadku wszystkich testów, zakładając, że Twój schemat jest stały, pojedynczy schemat dla wszystkich testów, upewnisz się, że dane, z którymi pracujesz, są wykonywane w ramach transakcji, która może być wyłączona. wycofana. Session może działać tak, że wywołanie commit() nie powoduje prawdziwej "prawdziwej" transakcji, zawijając cały test w wyraźnym Transaction. Przykład ten przedstawia przykład pod adresem https://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites. Posiadanie silnika ": memory:" na każdym urządzeniu testowym zajmuje dużo pamięci i nie jest tak naprawdę skalowalne do innych baz danych oprócz SQLite.

+2

Kluczową ideą jest to, że musisz utworzyć instancje wszystkich obiektów podczas instalacji, przytrzymaj je, przypisując je jako atrybuty do 'self' i odzyskaj je później, nie poprzez ponowne wysyłanie zapytań do bazy danych, ale przez te" ja " atrybuty. Również wdrożenie '__eq__' było niepotrzebne; wydaje się, że SQLAlchemy zwróci dokładnie to samo wystąpienie modelu (tzn. 'created_model_instance to instance_from_query' zwraca' True'). Wreszcie, pomógłoby to w poprawieniu odpowiedzi, aby użyć wzoru wycofywania transakcji, nawet jeśli można to wywnioskować z lektury dokumentacji SQLAlchemy pod podanym linkiem. – gotgenes