SQLAlchemy中Query的单元测试

Question

在SQLAlchemy中，怎么测试查询呢？比如说，我们有这个 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)
        )

还有这个 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))

不过我觉得这样做很麻烦，而且我甚至还没开始测试那些复杂的过滤查询。有没有更优雅的解决方案，还是说我总得手动写一堆单独的查询呢？

Answer 1

你最开始的测试思路是对的，只需要做两件事中的一件：要么确保两个Panel对象的主键相同时比较结果是True：

import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from database.models import Base

class Panel(Base):
    # ...

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

要么你可以调整你的测试，确保你检查的是同一个Panel实例（因为这里我们利用了身份映射的概念）：

class TestQuery(unittest.TestCase):
    def setUp(self):
        self.engine = create_engine('sqlite:///:memory:')
        self.session = Session(self.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中。这个用法的例子可以在这个链接中找到。每个测试环境都有一个“:memory:”引擎会占用很多内存，并且不太适合除了SQLite以外的其他数据库。