2013-02-06 27 views
7

如何测试SQLAlchemy中的查询?例如,假设我们有这个models.py在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) 
     ) 

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) 

当我们试图运行测试,它失败了,即使两个面板看起来相同。

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

一个解决方案,我发现是做一个查询为每个实例我希望在查询中发现:

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

这令我很丑陋,虽然和我甚至没有达到我想要测试的复杂过滤查询的程度。有没有更优雅的解决方案,还是我总是必须手动进行一堆个人查询?

回答

14

原始测试是在正确的轨道上,你只需要做两件事情之一:要么确保相同的主键标识两个Panel比较对象为True

class Panel(Base): 
    # ... 

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

,或者你可以组织测试,这样你确保你正在检查对同一Panel实例(因为在这里我们利用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) 

至于引擎/会话设置/拆解,我会选择使用单引擎进行所有测试的模式,并假设您的模式已修复,所有测试都使用单个模式,然后确保您使用的数据在事务中执行回滚。可以使Session以这种方式工作,使得调用commit()实际上不提交“真实”事务,通过将整个测试包装在明确的Transaction内。 https://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites的例子说明了这种用法。在每个测试夹具上都有一个“:memory:”引擎会占用大量内存,而且除了SQLite之外,还不会真正扩展到其他数据库。

+2

这里的关键思想是,你需要在安装过程中实例化所有对象,将它们赋值为'self'的属性,稍后再检索它们,而不是通过再次查询数据库,而是通过这些'self'属性。另外,执行'__eq__'是不必要的。似乎SQLAlchemy将返回完全相同的模型实例(即'created_model_instance is instance_from_query'返回'True')。最后,这将有助于修改使用事务回滚模式的答案,尽管可以通过阅读提供的链接中的SQLAlchemy文档来推断它。 – gotgenes