使用SQLAlchemy DBSession對金字塔視圖進行單元測試的正確方法是什麼？

Question

在編寫金字塔單元測試套件時，單元測試執行SQLAlchemy調用的視圖的正確或適當的方法是什麼？例如：

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