sqlalchemy/test/orm/test_scoping.py

from sqlalchemy.test.testing import assert_raises, assert_raises_message
import sqlalchemy as sa
from sqlalchemy.test import testing
from sqlalchemy.orm import scoped_session
from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, query
from sqlalchemy.test.testing import eq_
from test.orm import _base



class _ScopedTest(_base.MappedTest):
    """Adds another lookup bucket to emulate Session globals."""

    run_setup_mappers = 'once'

    _artifact_registries = (
        _base.MappedTest._artifact_registries + ('scoping',))

    @classmethod
    def setup_class(cls):
        cls.scoping = _base.adict()
        super(_ScopedTest, cls).setup_class()

    @classmethod
    def teardown_class(cls):
        cls.scoping.clear()
        super(_ScopedTest, cls).teardown_class()


class ScopedSessionTest(_base.MappedTest):

    @classmethod
    def define_tables(cls, metadata):
        Table('table1', metadata,
              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('data', String(30)))
        Table('table2', metadata,
              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
              Column('someid', None, ForeignKey('table1.id')))

    @testing.resolve_artifact_names
    def test_basic(self):
        Session = scoped_session(sa.orm.sessionmaker())

        class CustomQuery(query.Query):
            pass

        class SomeObject(_base.ComparableEntity):
            query = Session.query_property()
        class SomeOtherObject(_base.ComparableEntity):
            query = Session.query_property()
            custom_query = Session.query_property(query_cls=CustomQuery)

        mapper(SomeObject, table1, properties={
            'options':relationship(SomeOtherObject)})
        mapper(SomeOtherObject, table2)

        s = SomeObject(id=1, data="hello")
        sso = SomeOtherObject()
        s.options.append(sso)
        Session.add(s)
        Session.commit()
        Session.refresh(sso)
        Session.remove()

        eq_(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]),
            Session.query(SomeObject).one())
        eq_(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]),
            SomeObject.query.one())
        eq_(SomeOtherObject(someid=1),
            SomeOtherObject.query.filter(
                SomeOtherObject.someid == sso.someid).one())
        assert isinstance(SomeOtherObject.query, query.Query)
        assert not isinstance(SomeOtherObject.query, CustomQuery)
        assert isinstance(SomeOtherObject.custom_query, query.Query)


class ScopedMapperTest(_ScopedTest):

    @classmethod
    def define_tables(cls, metadata):
        Table('table1', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)))
        Table('table2', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('someid', None, ForeignKey('table1.id')))

    @classmethod
    def setup_classes(cls):
        class SomeObject(_base.ComparableEntity):
            pass
        class SomeOtherObject(_base.ComparableEntity):
            pass

    @classmethod
    @testing.uses_deprecated()
    @testing.resolve_artifact_names
    def setup_mappers(cls):
        Session = scoped_session(sa.orm.create_session)
        Session.mapper(SomeObject, table1, properties={
            'options':relationship(SomeOtherObject)
        })
        Session.mapper(SomeOtherObject, table2)

        cls.scoping['Session'] = Session

    @classmethod
    @testing.resolve_artifact_names
    def insert_data(cls):
        s = SomeObject()
        s.id = 1
        s.data = 'hello'
        sso = SomeOtherObject()
        s.options.append(sso)
        Session.flush()
        Session.expunge_all()

    @testing.resolve_artifact_names
    def test_query(self):
        sso = SomeOtherObject.query().first()
        assert SomeObject.query.filter_by(id=1).one().options[0].id == sso.id

    @testing.uses_deprecated()
    @testing.resolve_artifact_names
    def test_query_compiles(self):
        class Foo(object):
            pass
        Session.mapper(Foo, table2)
        assert hasattr(Foo, 'query')

        ext = sa.orm.MapperExtension()

        class Bar(object):
            pass
        Session.mapper(Bar, table2, extension=[ext])
        assert hasattr(Bar, 'query')

        class Baz(object):
            pass
        Session.mapper(Baz, table2, extension=ext)
        assert hasattr(Baz, 'query')

    @testing.uses_deprecated()
    @testing.resolve_artifact_names
    def test_default_constructor_state_not_shared(self):
        scope = scoped_session(sa.orm.sessionmaker())

        class A(object):
            pass
        class B(object):
            def __init__(self):
                pass

        scope.mapper(A, table1)
        scope.mapper(B, table2)

        A(foo='bar')
        assert_raises(TypeError, B, foo='bar')

        scope = scoped_session(sa.orm.sessionmaker())

        class C(object):
            def __init__(self):
                pass
        class D(object):
            pass

        scope.mapper(C, table1)
        scope.mapper(D, table2)

        assert_raises(TypeError, C, foo='bar')
        D(foo='bar')

    @testing.uses_deprecated()
    @testing.resolve_artifact_names
    def test_validating_constructor(self):
        s2 = SomeObject(someid=12)
        s3 = SomeOtherObject(someid=123, bogus=345)

        class ValidatedOtherObject(object): pass
        Session.mapper(ValidatedOtherObject, table2, validate=True)

        v1 = ValidatedOtherObject(someid=12)
        assert_raises(sa.exc.ArgumentError, ValidatedOtherObject,
                          someid=12, bogus=345)

    @testing.uses_deprecated()
    @testing.resolve_artifact_names
    def test_dont_clobber_methods(self):
        class MyClass(object):
            def expunge(self):
                return "an expunge !"

        Session.mapper(MyClass, table2)

        assert MyClass().expunge() == "an expunge !"


class ScopedMapperTest2(_ScopedTest):

    @classmethod
    def define_tables(cls, metadata):
        Table('table1', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', String(30)),
            Column('type', String(30)))
        Table('table2', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('someid', None, ForeignKey('table1.id')),
            Column('somedata', String(30)))

    @classmethod
    def setup_classes(cls):
        class BaseClass(_base.ComparableEntity):
            pass
        class SubClass(BaseClass):
            pass

    @classmethod
    @testing.uses_deprecated()
    @testing.resolve_artifact_names
    def setup_mappers(cls):
        Session = scoped_session(sa.orm.sessionmaker())

        Session.mapper(BaseClass, table1,
                       polymorphic_identity='base',
                       polymorphic_on=table1.c.type)
        Session.mapper(SubClass, table2,
                       polymorphic_identity='sub',
                       inherits=BaseClass)

        cls.scoping['Session'] = Session

    @testing.resolve_artifact_names
    def test_inheritance(self):
        def expunge_list(l):
            for x in l:
                Session.expunge(x)
            return l

        b = BaseClass(data='b1')
        s =  SubClass(data='s1', somedata='somedata')
        Session.commit()
        Session.expunge_all()

        eq_(expunge_list([BaseClass(data='b1'),
                          SubClass(data='s1', somedata='somedata')]),
            BaseClass.query.all())
        eq_(expunge_list([SubClass(data='s1', somedata='somedata')]),
            SubClass.query.all())