python - 使用 SQLAlchemy DBSession 对 Pyramid View 进行单元测试的正确方法是什么？

Question

在编写 Pyramid 单元测试套件时，对执行 SQLAlchemy 调用的 View 进行单元测试的正确或适当方法是什么。示例:

def my_view(request):
    DBSession.query(DeclarativeBase).all()

我会使用 Mock() 和 patch 将 DBSession 的范围覆盖到 DummyDB 类吗？

Answer 1

你可以，我很快就会写博客/演讲/举例。这是全新的东西。先睹为快:

import mock

from sqlalchemy.sql import ClauseElement

class MockSession(mock.MagicMock):
    def __init__(self, *arg, **kw):
        kw.setdefault('side_effect', self._side_effect)
        super(MockSession, self).__init__(*arg, **kw)
        self._lookup = {}

    def _side_effect(self, *arg, **kw):
        if self._mock_return_value is not mock.sentinel.DEFAULT:
            return self._mock_return_value
        else:
            return self._generate(*arg, **kw)

    def _get_key(self, arg, kw):
        return tuple(self._hash(a) for a in arg) + \
                tuple((k, self._hash(kw[k])) for k in sorted(kw))

    def _hash(self, arg):
        if isinstance(arg, ClauseElement):
            expr = str(arg.compile(compile_kwargs={"literal_binds": True}))
            return expr
        else:
            assert hash(arg)
            return arg

    def _generate(self, *arg, **kw):
        key = self._get_key(arg, kw)
        if key in self._lookup:
            return self._lookup[key]
        else:
            self._lookup[key] = ret = MockSession()
            return ret


if __name__ == '__main__':

    from sqlalchemy import Column, Integer
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class Foo(Base):
        __tablename__ = 'foo'

        id = Column(Integer, primary_key=True)
        x = Column(Integer)
        y = Column(Integer)

    sess = MockSession()

    # write out queries as they would in the code, assign return_value
    sess.query(Foo.x).filter(Foo.x == 5).first.return_value = 5
    sess.query(Foo.x).filter(Foo.x == 2).first.return_value = 2
    sess.query(Foo).filter(Foo.x == 2).filter_by(y=5).all.return_value = [Foo(x=2, y=5)]
    sess.query(Foo).filter(Foo.x == 9).all.return_value = [Foo(x=9, y=1), Foo(x=9, y=2)]

    # those queries are now replayable and will return what was assigned.
    print sess.query(Foo.x).filter(Foo.x == 5).first()
    print sess.query(Foo.x).filter(Foo.x == 2).first()
    print sess.query(Foo).filter(Foo.x == 2).filter_by(y=5).all()
    print sess.query(Foo).filter(Foo.x == 9).all()

我实际上已经将它分配到设置/拆卸中的全局 ScopedSession 中，它的效果非常好:

from myapp.model import MyScopedSession

def setUp(self):
     MyScopedSession.registry.set(MockSession())

     # set up some queries, we can usually use scoped_session's proxying

     MyScopedSession.query(User).filter_by(id=1).first.return_value = User(id=1)

def tearDown(self):
     MyScopedSession.remove()


def some_test(self):
     # to get at mock methods and accessors, call the scoped_session to get at
     # the registry

     #  ... test some thing

     # test a User was added to the session
     self.assertEquals(
            MyScopedSession().add.mock_calls,
            [mock.call(User(name='someuser'))]
     )