SQLAlchemy中Query的单元测试

got*_*nes 18 python unit-testing sqlalchemy

如何在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)
        )
Run Code Online (Sandbox Code Playgroud)

还有这个 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)
Run Code Online (Sandbox Code Playgroud)

当我们尝试运行测试时,它会失败,即使两个面板看起来相同.

$ 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)
Run Code Online (Sandbox Code Playgroud)

我发现的一个解决方案是对我希望在查询中找到的每个实例进行查询:

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))
Run Code Online (Sandbox Code Playgroud)

这让我觉得非常难看,而且我甚至没有达到我正在尝试测试的复杂过滤查询的程度.是否有更优雅的解决方案,还是我总是需要手动制作一堆单独的查询?

zzz*_*eek 23

您的原始测试是在正确的轨道上,您只需要执行以下两项操作之一:确保将Panel具有相同主键标识的两个对象进行比较True:

class Panel(Base):
    # ...

    def __eq__(self, other):
        return isinstance(other, Panel) and other.id == self.id
Run Code Online (Sandbox Code Playgroud)

或者你可以组织你的测试,以确保你检查同一个Panel实例(因为我们利用身份图):

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)
Run Code Online (Sandbox Code Playgroud)

就引擎/会话设置/拆解而言,我会选择一种模式,在这种模式下,您使用单个引擎进行所有测试,并假设您的模式已修复,所有测试都使用单个模式,然后确保您使用的数据with在可以回滚的事务中执行.该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之外的其他数据库.

  • 这里的关键思想是你需要在设置过程中实例化所有对象,通过将它们作为属性分配给`self`来保持它们,并在以后检索它们,而不是通过再次查询数据库,而是通过那些`self`属性.而且,实现`__eq__`是不必要的; 似乎SQLAlchemy将返回完全相同的模型实例(即,`created_model_instance is instance_from_query`返回'True`).最后,有助于修改使用事务回滚模式的答案,即使可以通过在提供的链接上阅读SQLAlchemy文档来推断它. (2认同)