1from sqlalchemy.testing import assert_raises, assert_raises_message
2import sqlalchemy as sa
3from sqlalchemy import testing
4from sqlalchemy.orm import scoped_session
5from sqlalchemy import Integer, String, ForeignKey
6from sqlalchemy.testing.schema import Table, Column
7from sqlalchemy.orm import mapper, relationship, query
8from sqlalchemy.testing import eq_
9from sqlalchemy.testing import fixtures
10
11
12
13class _ScopedTest(fixtures.MappedTest):
14    """Adds another lookup bucket to emulate Session globals."""
15
16    run_setup_mappers = 'once'
17
18    @classmethod
19    def setup_class(cls):
20        cls.scoping = _base.adict()
21        super(_ScopedTest, cls).setup_class()
22
23    @classmethod
24    def teardown_class(cls):
25        cls.scoping.clear()
26        super(_ScopedTest, cls).teardown_class()
27
28
29class ScopedSessionTest(fixtures.MappedTest):
30
31    @classmethod
32    def define_tables(cls, metadata):
33        Table('table1', metadata,
34              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
35              Column('data', String(30)))
36        Table('table2', metadata,
37              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
38              Column('someid', None, ForeignKey('table1.id')))
39
40    def test_basic(self):
41        table2, table1 = self.tables.table2, self.tables.table1
42
43        Session = scoped_session(sa.orm.sessionmaker())
44
45        class CustomQuery(query.Query):
46            pass
47
48        class SomeObject(fixtures.ComparableEntity):
49            query = Session.query_property()
50        class SomeOtherObject(fixtures.ComparableEntity):
51            query = Session.query_property()
52            custom_query = Session.query_property(query_cls=CustomQuery)
53
54        mapper(SomeObject, table1, properties={
55            'options':relationship(SomeOtherObject)})
56        mapper(SomeOtherObject, table2)
57
58        s = SomeObject(id=1, data="hello")
59        sso = SomeOtherObject()
60        s.options.append(sso)
61        Session.add(s)
62        Session.commit()
63        Session.refresh(sso)
64        Session.remove()
65
66        eq_(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]),
67            Session.query(SomeObject).one())
68        eq_(SomeObject(id=1, data="hello", options=[SomeOtherObject(someid=1)]),
69            SomeObject.query.one())
70        eq_(SomeOtherObject(someid=1),
71            SomeOtherObject.query.filter(
72                SomeOtherObject.someid == sso.someid).one())
73        assert isinstance(SomeOtherObject.query, query.Query)
74        assert not isinstance(SomeOtherObject.query, CustomQuery)
75        assert isinstance(SomeOtherObject.custom_query, query.Query)
76
77    def test_config_errors(self):
78        Session = scoped_session(sa.orm.sessionmaker())
79
80        s = Session()
81        assert_raises_message(
82            sa.exc.InvalidRequestError,
83            "Scoped session is already present",
84            Session, bind=testing.db
85        )
86
87        assert_raises_message(
88            sa.exc.SAWarning,
89            "At least one scoped session is already present. ",
90            Session.configure, bind=testing.db
91        )
92
93
94
95