CODEWITHSUNDEEP
pythonunit-testingmockingsqlalchemypyramid
Thomas Farvour
Thomas Farvour

Reputation: 1103

What's the correct way to unit test Pyramid views with a SQLAlchemy DBSession?

When writing a Pyramid unit test suite, what is the proper or appropriate way to unit test a view that does a SQLAlchemy call. Example:

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

Would I use Mock() and patch to override the scope of DBSession to a DummyDB class of the sorts?

Upvotes: 3

Views: 1106

Answers (1)

zzzeek
zzzeek

Reputation: 75117

You can, and I'll be blogging/speaking/exampling about this soon. This is totally new stuff. Here's a sneak peek:

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

I've actually assigned this into the global ScopedSession within setup/teardown and it works amazingly:

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

Upvotes: 2

Related Questions