text: from sqlalchemy.orm import Session, subqueryload, \
      mapper, relationship, lazyload, clear_mappers, backref, aliased, \
      Load, defaultload
  from sqlalchemy.testing import eq_, is_, is_not_
  from sqlalchemy.testing import assert_raises, assert_raises_message
  from sqlalchemy import testing
  from sqlalchemy import inspect
  from test.orm import _fixtures
  from sqlalchemy.ext.baked import BakedQuery
  from sqlalchemy.ext import baked
  from sqlalchemy import bindparam, func, literal_column
  from sqlalchemy.orm import exc as orm_exc
  import itertools
  from sqlalchemy.testing import mock
  from sqlalchemy.testing.assertsql import CompiledSQL
  
  
  class BakedTest(_fixtures.FixtureTest):
      run_setup_mappers = 'once'
      run_inserts = 'once'
      run_deletes = None
  
      def setup(self):
          self.bakery = baked.bakery()
  
  
  class StateChangeTest(BakedTest):
      @classmethod
      def setup_mappers(cls):
          User = cls.classes.User
  
          mapper(User, cls.tables.users)
  
      def _assert_cache_key(self, key, elements):
          eq_(
              key,
              tuple(elem.__code__ for elem in elements)
          )
  
      def test_initial_key(self):
          User = self.classes.User
          session = Session()
  
          def l1(): return session.query(User)
          q1 = self.bakery(l1)
          self._assert_cache_key(
              q1._cache_key,
              [l1]
          )
          eq_(q1.steps, [l1])
  
      def test_inplace_add(self):
          User = self.classes.User
          session = Session()
  
          def l1(): return session.query(User)
  
          def l2(q): return q.filter(User.name == bindparam('name'))
          q1 = self.bakery(l1)
          self._assert_cache_key(
              q1._cache_key,
              [l1]
          )
          eq_(q1.steps, [l1])
  
          q2 = q1.add_criteria(l2)
          is_(q2, q1)
  
          self._assert_cache_key(
              q1._cache_key,
              [l1, l2]
          )
          eq_(q1.steps, [l1, l2])
  
      def test_inplace_add_operator(self):
          User = self.classes.User
          session = Session()
  
          def l1(): return session.query(User)
  
          def l2(q): return q.filter(User.name == bindparam('name'))
          q1 = self.bakery(l1)
          self._assert_cache_key(
              q1._cache_key,
              [l1]
          )
  
          q1 += l2
  
          self._assert_cache_key(
              q1._cache_key,
              [l1, l2]
          )
  
      def test_chained_add(self):
          User = self.classes.User
          session = Session()
  
          def l1(): return session.query(User)
  
          def l2(q): return q.filter(User.name == bindparam('name'))
          q1 = self.bakery(l1)
  
          q2 = q1.with_criteria(l2)
          is_not_(q2, q1)
  
          self._assert_cache_key(
              q1._cache_key,
              [l1]
          )
          self._assert_cache_key(
              q2._cache_key,
              [l1, l2]
          )
  
      def test_chained_add_operator(self):
          User = self.classes.User
          session = Session()
  
          def l1(): return session.query(User)
  
          def l2(q): return q.filter(User.name == bindparam('name'))
          q1 = self.bakery(l1)
  
          q2 = q1 + l2
          is_not_(q2, q1)
  
          self._assert_cache_key(
              q1._cache_key,
              [l1]
          )
          self._assert_cache_key(
              q2._cache_key,
              [l1, l2]
          )
  
  
  class LikeQueryTest(BakedTest):
      @classmethod
      def setup_mappers(cls):
          User = cls.classes.User
  
          mapper(User, cls.tables.users)
  
      def test_first_no_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name == 'asdf')
  
          eq_(
              bq(Session()).first(),
              None
          )
  
      def test_first_multiple_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User.id))
          bq += lambda q: q.filter(User.name.like('%ed%')).order_by(User.id)
  
          eq_(
              bq(Session()).first(),
              (8, )
          )
  
      def test_one_or_none_no_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name == 'asdf')
  
          eq_(
              bq(Session()).one_or_none(),
              None
          )
  
      def test_one_or_none_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name == 'ed')
  
          u1 = bq(Session()).one_or_none()
          eq_(u1.name, 'ed')
  
      def test_one_or_none_multiple_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name.like('%ed%'))
  
          assert_raises_message(
              orm_exc.MultipleResultsFound,
              "Multiple rows were found for one_or_none()",
              bq(Session()).one_or_none
          )
  
      def test_one_no_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name == 'asdf')
  
          assert_raises_message(
              orm_exc.NoResultFound,
              "No row was found for one()",
              bq(Session()).one
          )
  
      def test_one_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name == 'ed')
  
          u1 = bq(Session()).one()
          eq_(u1.name, 'ed')
  
      def test_one_multiple_result(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
          bq += lambda q: q.filter(User.name.like('%ed%'))
  
          assert_raises_message(
              orm_exc.MultipleResultsFound,
              "Multiple rows were found for one()",
              bq(Session()).one
          )
  
      def test_get(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
  
          sess = Session()
  
          def go():
              u1 = bq(sess).get(7)
              eq_(u1.name, 'jack')
          self.assert_sql_count(testing.db, go, 1)
  
          u1 = sess.query(User).get(7)  # noqa
  
          def go():
              u2 = bq(sess).get(7)
              eq_(u2.name, 'jack')
          self.assert_sql_count(testing.db, go, 0)
  
          def go():
              u2 = bq(sess).get(8)
              eq_(u2.name, 'ed')
          self.assert_sql_count(testing.db, go, 1)
  
      def test_scalar(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User.id))
  
          sess = Session()
  
          bq += lambda q: q.filter(User.id == 7)
  
          eq_(
              bq(sess).scalar(), 7
          )
  
      def test_count(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
  
          sess = Session()
  
          eq_(
              bq(sess).count(),
              4
          )
  
          bq += lambda q: q.filter(User.id.in_([8, 9]))
  
          eq_(
              bq(sess).count(), 2
          )
  
          # original query still works
          eq_(
              set([(u.id, u.name) for u in bq(sess).all()]),
              set([(8, 'ed'), (9, 'fred')])
          )
  
      def test_count_with_bindparams(self):
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
  
          sess = Session()
  
          eq_(
              bq(sess).count(),
              4
          )
  
          bq += lambda q: q.filter(User.name == bindparam("uname"))
          # calling with *args
          eq_(
              bq(sess).params(uname='fred').count(), 1
          )
          # with multiple params, the **kwargs will be used
          bq += lambda q: q.filter(User.id == bindparam("anid"))
          eq_(
              bq(sess).params(uname='fred', anid=9).count(), 1
          )
          eq_(
              # wrong id, so 0 results:
              bq(sess).params(uname='fred', anid=8).count(), 0
          )
  
  
      def test_get_pk_w_null(self):
          """test the re-implementation of logic to do get with IS NULL."""
  
          class AddressUser(object):
              pass
          mapper(
              AddressUser,
              self.tables.users.outerjoin(self.tables.addresses),
              properties={
                  "id": self.tables.users.c.id,
                  "address_id": self.tables.addresses.c.id
              }
          )
  
          bq = self.bakery(lambda s: s.query(AddressUser))
  
          sess = Session()
  
          def go():
              u1 = bq(sess).get((10, None))
              eq_(u1.name, 'chuck')
          self.assert_sql_count(testing.db, go, 1)
  
          u1 = sess.query(AddressUser).get((10, None))  # noqa
  
          def go():
              u2 = bq(sess).get((10, None))
              eq_(u2.name, 'chuck')
          self.assert_sql_count(testing.db, go, 0)
  
      def test_get_includes_getclause(self):
          # test issue #3597
          User = self.classes.User
  
          bq = self.bakery(lambda s: s.query(User))
  
          for i in range(5):
              sess = Session()
              u1 = bq(sess).get(7)
              eq_(u1.name, 'jack')
              sess.close()
  
          eq_(len(bq._bakery), 2)
  
          # simulate race where mapper._get_clause
          # may be generated more than once
          from sqlalchemy import inspect
          del inspect(User).__dict__['_get_clause']
  
          for i in range(5):
              sess = Session()
              u1 = bq(sess).get(7)
              eq_(u1.name, 'jack')
              sess.close()
          eq_(len(bq._bakery), 4)
  
  
  class ResultTest(BakedTest):
      __backend__ = True
  
      @classmethod
      def setup_mappers(cls):
          User = cls.classes.User
          Address = cls.classes.Address
          Order = cls.classes.Order
  
          mapper(User, cls.tables.users, properties={
              "addresses": relationship(
                  Address, order_by=cls.tables.addresses.c.id),
              "orders": relationship(
                  Order, order_by=cls.tables.orders.c.id)
          })
          mapper(Address, cls.tables.addresses)
          mapper(Order, cls.tables.orders)
  
      def test_cachekeys_on_constructor(self):
          User = self.classes.User
  
          queue = [7, 8]
  
          def fn(s): return s.query(User.id).filter_by(id=queue.pop(0))
          bq1 = self.bakery(fn, 7)
          bq2 = self.bakery(fn, 8)
  
          for i in range(3):
              session = Session(autocommit=True)
              eq_(
                  bq1(session).all(),
                  [(7,)]
              )
  
              eq_(
                  bq2(session).all(),
                  [(8,)]
              )
  
      def test_no_steps(self):
          User = self.classes.User
  
          bq = self.bakery(
              lambda s: s.query(User.id, User.name).order_by(User.id))
  
          for i in range(3):
              session = Session(autocommit=True)
              eq_(
                  bq(session).all(),
                  [(7, 'jack'), (8, 'ed'), (9, 'fred'), (10, 'chuck')]
              )
  
      def test_different_limits(self):
          User = self.classes.User
  
          bq = self.bakery(
              lambda s: s.query(User.id, User.name).order_by(User.id))
  
          bq += lambda q: q.limit(bindparam('limit')).offset(bindparam('offset'))
          session = Session(autocommit=True)
  
          for i in range(4):
              for limit, offset, exp in [
                  (2, 1, [(8, 'ed'), (9, 'fred')]),
                  (3, 0, [(7, 'jack'), (8, 'ed'), (9, 'fred')]),
                  (1, 2, [(9, 'fred')])
              ]:
                  eq_(
                      bq(session).params(limit=limit, offset=offset).all(),
                      exp
                  )
  
      def test_disable_on_session(self):
          User = self.classes.User
  
          canary = mock.Mock()
  
          def fn1(s):
              canary.fn1()
              return s.query(User.id, User.name).order_by(User.id)
  
          def fn2(q):
              canary.fn2()
              return q.filter(User.id == bindparam('id'))
  
          def fn3(q):
              canary.fn3()
              return q
  
          for x in range(3):
              bq = self.bakery(fn1)
  
              bq += fn2
  
              sess = Session(autocommit=True, enable_baked_queries=False)
              eq_(
                  bq.add_criteria(fn3)(sess).params(id=7).all(),
                  [(7, 'jack')]
              )
  
          eq_(
              canary.mock_calls,
              [mock.call.fn1(), mock.call.fn2(), mock.call.fn3(),
               mock.call.fn1(), mock.call.fn2(), mock.call.fn3(),
               mock.call.fn1(), mock.call.fn2(), mock.call.fn3()]
          )
  
      def test_spoiled_full_w_params(self):
          User = self.classes.User
  
          canary = mock.Mock()
  
          def fn1(s):
              canary.fn1()
              return s.query(User.id, User.name).order_by(User.id)
  
          def fn2(q):
              canary.fn2()
              return q.filter(User.id == bindparam('id'))
  
          def fn3(q):
              canary.fn3()
              return q
  
          for x in range(3):
              bq = self.bakery(fn1)
  
              bq += fn2
  
              sess = Session(autocommit=True)
              eq_(
                  bq.spoil(full=True).add_criteria(fn3)(sess).params(id=7).all(),
                  [(7, 'jack')]
              )
  
          eq_(
              canary.mock_calls,
              [mock.call.fn1(), mock.call.fn2(), mock.call.fn3(),
               mock.call.fn1(), mock.call.fn2(), mock.call.fn3(),
               mock.call.fn1(), mock.call.fn2(), mock.call.fn3()]
          )
  
      def test_spoiled_half_w_params(self):
          User = self.classes.User
  
          canary = mock.Mock()
  
          def fn1(s):
              canary.fn1()
              return s.query(User.id, User.name).order_by(User.id)
  
          def fn2(q):
              canary.fn2()
              return q.filter(User.id == bindparam('id'))
  
          def fn3(q):
              canary.fn3()
              return q
  
          bq = self.bakery(fn1)
  
          bq += fn2
  
          for x in range(3):
              bq = self.bakery(fn1)
  
              bq += fn2
  
              sess = Session(autocommit=True)
              eq_(
                  bq.spoil().add_criteria(fn3)(sess).params(id=7).all(),
                  [(7, 'jack')]
              )
  
          eq_(
              canary.mock_calls,
              [mock.call.fn1(), mock.call.fn2(),
               mock.call.fn3(), mock.call.fn3(), mock.call.fn3()]
          )
  
      def test_w_new_entities(self):
          """Test that the query can have its entities modified in
          an arbitrary callable, and that this new entity list is preserved
          when the query is invoked.
  
          """
          User = self.classes.User
  
          bq = self.bakery(
              lambda s: s.query(User.id, User.name))
  
          bq += lambda q: q.from_self().with_entities(
              func.count(User.id))
  
          for i in range(3):
              session = Session(autocommit=True)
              eq_(
                  bq(session).all(),
                  [(4, )]
              )
  
      def test_conditional_step(self):
          """Test a large series of conditionals and assert that
          results remain correct between all of them within a series
          of loops.
  
          """
          User = self.classes.User
  
          base_bq = self.bakery(
              lambda s: s.query(User.id, User.name))
  
          base_bq += lambda q: q.order_by(User.id)
  
          for i in range(4):
              for cond1, cond2, cond3, cond4 in itertools.product(
                      *[(False, True) for j in range(4)]):
                  bq = base_bq._clone()
                  if cond1:
                      bq += lambda q: q.filter(User.name != 'jack')
                      if cond2:
                          bq += lambda q: q.join(User.addresses)
                      else:
                          bq += lambda q: q.outerjoin(User.addresses)
                  elif cond3:
                      bq += lambda q: q.filter(User.name.like('%ed%'))
                  else:
                      bq += lambda q: q.filter(User.name == 'jack')
  
                  if cond4:
                      bq += lambda q: q.from_self().with_entities(
                          func.count(User.id))
                  sess = Session(autocommit=True)
                  result = bq(sess).all()
                  if cond4:
                      if cond1:
                          if cond2:
                              eq_(result, [(4,)])
                          else:
                              eq_(result, [(5,)])
                      elif cond3:
                          eq_(result, [(2,)])
                      else:
                          eq_(result, [(1,)])
                  else:
                      if cond1:
                          if cond2:
                              eq_(
                                  result,
                                  [(8, 'ed'), (8, 'ed'), (8, 'ed'),
                                   (9, 'fred')]
                              )
                          else:
                              eq_(
                                  result,
                                  [(8, 'ed'), (8, 'ed'), (8, 'ed'),
                                   (9, 'fred'), (10, 'chuck')]
                              )
                      elif cond3:
                          eq_(result, [(8, 'ed'), (9, 'fred')])
                      else:
                          eq_(result, [(7, 'jack')])
  
                  sess.close()
  
      def test_conditional_step_oneline(self):
          User = self.classes.User
  
          base_bq = self.bakery(
              lambda s: s.query(User.id, User.name))
  
          base_bq += lambda q: q.order_by(User.id)
  
          for i in range(4):
              for cond1 in (False, True):
                  bq = base_bq._clone()
  
                  # we were using (filename, firstlineno) as cache key,
                  # which fails for this kind of thing!
                  bq += (lambda q: q.filter(User.name != 'jack')) if cond1 else (lambda q: q.filter(User.name == 'jack'))  # noqa
                  sess = Session(autocommit=True)
                  result = bq(sess).all()
  
                  if cond1:
                      eq_(result, [(8, u'ed'), (9, u'fred'), (10, u'chuck')])
                  else:
                      eq_(result, [(7, 'jack')])
  
                  sess.close()
  
      def test_subquery_eagerloading(self):
          User = self.classes.User
          Address = self.classes.Address
          Order = self.classes.Order
  
          # Override the default bakery for one with a smaller size. This used to
          # trigger a bug when unbaking subqueries.
          self.bakery = baked.bakery(size=3)
          base_bq = self.bakery(lambda s: s.query(User))
  
          base_bq += lambda q: q.options(subqueryload(User.addresses),
                                         subqueryload(User.orders))
          base_bq += lambda q: q.order_by(User.id)
  
          assert_result = [
              User(id=7,
                   addresses=[Address(id=1, email_address='jack@bean.com')],
                   orders=[Order(id=1), Order(id=3), Order(id=5)]),
              User(id=8, addresses=[
                  Address(id=2, email_address='ed@wood.com'),
                  Address(id=3, email_address='ed@bettyboop.com'),
                  Address(id=4, email_address='ed@lala.com'),
              ]),
              User(id=9,
                   addresses=[Address(id=5)],
                   orders=[Order(id=2), Order(id=4)]),
              User(id=10, addresses=[])
          ]
  
          for i in range(4):
              for cond1, cond2 in itertools.product(
                      *[(False, True) for j in range(2)]):
                  bq = base_bq._clone()
  
                  sess = Session()
  
                  if cond1:
                      bq += lambda q: q.filter(User.name == 'jack')
                  else:
                      bq += lambda q: q.filter(User.name.like('%ed%'))
  
                  if cond2:
                      ct = func.count(Address.id).label('count')
                      subq = sess.query(
                          ct,
                          Address.user_id).group_by(Address.user_id).\
                          having(ct > 2).subquery()
  
                      bq += lambda q: q.join(subq)
  
                  if cond2:
                      if cond1:
                          def go():
                              result = bq(sess).all()
                              eq_([], result)
                          self.assert_sql_count(testing.db, go, 1)
                      else:
                          def go():
                              result = bq(sess).all()
                              eq_(assert_result[1:2], result)
                          self.assert_sql_count(testing.db, go, 3)
                  else:
                      if cond1:
                          def go():
                              result = bq(sess).all()
                              eq_(assert_result[0:1], result)
                          self.assert_sql_count(testing.db, go, 3)
                      else:
                          def go():
                              result = bq(sess).all()
                              eq_(assert_result[1:3], result)
                          self.assert_sql_count(testing.db, go, 3)
  
                  sess.close()
  
  
  class LazyLoaderTest(testing.AssertsCompiledSQL, BakedTest):
      run_setup_mappers = 'each'
  
      def _o2m_fixture(self, lazy="select", **kw):
          User = self.classes.User
          Address = self.classes.Address
  
          mapper(User, self.tables.users, properties={
              'addresses': relationship(
                  Address, order_by=self.tables.addresses.c.id,
                  lazy=lazy, **kw)
          })
          mapper(Address, self.tables.addresses)
          return User, Address
  
      def _o2m_twolevel_fixture(self, lazy="select", **kw):
          User = self.classes.User
          Address = self.classes.Address
          Dingaling = self.classes.Dingaling
  
          mapper(User, self.tables.users, properties={
              'addresses': relationship(
                  Address, order_by=self.tables.addresses.c.id,
                  lazy=lazy, **kw)
          })
          mapper(Address, self.tables.addresses, properties={
              "dingalings": relationship(Dingaling, lazy=lazy)
          })
          mapper(Dingaling, self.tables.dingalings)
          return User, Address, Dingaling
  
      def _m2o_fixture(self):
          User = self.classes.User
          Address = self.classes.Address
  
          mapper(User, self.tables.users)
          mapper(Address, self.tables.addresses, properties={
              'user': relationship(User)
          })
          return User, Address
  
      def test_unsafe_unbound_option_cancels_bake(self):
          User, Address, Dingaling = self._o2m_twolevel_fixture(lazy="joined")
  
          class SubDingaling(Dingaling):
              pass
          mapper(SubDingaling, None, inherits=Dingaling)
  
          lru = Address.dingalings.property._lazy_strategy._bakery(
              lambda q: None)._bakery
          l1 = len(lru)
          for i in range(5):
              sess = Session()
              u1 = sess.query(User).options(
                  defaultload(User.addresses).lazyload(
                      Address.dingalings.of_type(aliased(SubDingaling)))).first()
              for ad in u1.addresses:
                  ad.dingalings
          l2 = len(lru)
          eq_(l1, 0)
          eq_(l2, 1)
  
      def test_unsafe_bound_option_cancels_bake(self):
          User, Address, Dingaling = self._o2m_twolevel_fixture(lazy="joined")
  
          class SubDingaling(Dingaling):
              pass
          mapper(SubDingaling, None, inherits=Dingaling)
  
          lru = Address.dingalings.property._lazy_strategy._bakery(
              lambda q: None)._bakery
          l1 = len(lru)
          for i in range(5):
              sess = Session()
              u1 = sess.query(User).options(
                  Load(User).defaultload(User.addresses).lazyload(
                      Address.dingalings.of_type(aliased(SubDingaling)))).first()
              for ad in u1.addresses:
                  ad.dingalings
          l2 = len(lru)
          eq_(l1, 0)
          eq_(l2, 1)
  
      def test_safe_unbound_option_allows_bake(self):
          User, Address, Dingaling = self._o2m_twolevel_fixture(lazy="joined")
  
          lru = Address.dingalings.property._lazy_strategy._bakery(
              lambda q: None)._bakery
          l1 = len(lru)
          for i in range(5):
              sess = Session()
              u1 = sess.query(User).options(
                  defaultload(User.addresses).lazyload(
                      Address.dingalings)).first()
              for ad in u1.addresses:
                  ad.dingalings
          l2 = len(lru)
          eq_(l1, 0)
          eq_(l2, 2)
  
      def test_safe_bound_option_allows_bake(self):
          User, Address, Dingaling = self._o2m_twolevel_fixture(lazy="joined")
  
          lru = Address.dingalings.property._lazy_strategy._bakery(
              lambda q: None)._bakery
          l1 = len(lru)
          for i in range(5):
              sess = Session()
              u1 = sess.query(User).options(
                  Load(User).defaultload(User.addresses).lazyload(
                      Address.dingalings)).first()
              for ad in u1.addresses:
                  ad.dingalings
          l2 = len(lru)
          eq_(l1, 0)
          eq_(l2, 2)
  
      def test_baked_lazy_loading_relationship_flag_true(self):
          self._test_baked_lazy_loading_relationship_flag(True)
  
      def test_baked_lazy_loading_relationship_flag_false(self):
          self._test_baked_lazy_loading_relationship_flag(False)
  
      def _test_baked_lazy_loading_relationship_flag(self, flag):
          User, Address = self._o2m_fixture(bake_queries=flag)
  
          sess = Session()
          u1 = sess.query(User).first()
  
          from sqlalchemy.orm import Query
  
          canary = mock.Mock()
  
          # I would think Mock can do this but apparently
          # it cannot (wrap / autospec don't work together)
          real_compile_context = Query._compile_context
  
          def _my_compile_context(*arg, **kw):
              if arg[0].column_descriptions[0]['entity'] is Address:
                  canary()
              return real_compile_context(*arg, **kw)
  
          with mock.patch.object(
              Query,
              "_compile_context",
              _my_compile_context
          ):
              u1.addresses
  
              sess.expire(u1)
              u1.addresses
  
          if flag:
              eq_(canary.call_count, 1)
          else:
              eq_(canary.call_count, 2)
  
      def test_baked_lazy_loading_option_o2m(self):
          User, Address = self._o2m_fixture()
          self._test_baked_lazy_loading(set_option=True)
  
      def test_baked_lazy_loading_mapped_o2m(self):
          User, Address = self._o2m_fixture(lazy="baked_select")
          self._test_baked_lazy_loading(set_option=False)
  
      def _test_baked_lazy_loading(self, set_option):
          User, Address = self.classes.User, self.classes.Address
  
          base_bq = self.bakery(
              lambda s: s.query(User))
  
          if set_option:
              base_bq += lambda q: q.options(lazyload(User.addresses))
  
          base_bq += lambda q: q.order_by(User.id)
  
          assert_result = self.static.user_address_result
  
          for i in range(4):
              for cond1, cond2 in itertools.product(
                      *[(False, True) for j in range(2)]):
                  bq = base_bq._clone()
  
                  sess = Session()
  
                  if cond1:
                      bq += lambda q: q.filter(User.name == 'jack')
                  else:
                      bq += lambda q: q.filter(User.name.like('%ed%'))
  
                  if cond2:
                      ct = func.count(Address.id).label('count')
                      subq = sess.query(
                          ct,
                          Address.user_id).group_by(Address.user_id).\
                          having(ct > 2).subquery()
  
                      bq += lambda q: q.join(subq)
  
                  if cond2:
                      if cond1:
                          def go():
                              result = bq(sess).all()
                              eq_([], result)
                          self.assert_sql_count(testing.db, go, 1)
                      else:
                          def go():
                              result = bq(sess).all()
                              eq_(assert_result[1:2], result)
                          self.assert_sql_count(testing.db, go, 2)
                  else:
                      if cond1:
                          def go():
                              result = bq(sess).all()
                              eq_(assert_result[0:1], result)
                          self.assert_sql_count(testing.db, go, 2)
                      else:
                          def go():
                              result = bq(sess).all()
                              eq_(assert_result[1:3], result)
                          self.assert_sql_count(testing.db, go, 3)
  
                  sess.close()
  
      def test_baked_lazy_loading_m2o(self):
          User, Address = self._m2o_fixture()
  
          base_bq = self.bakery(
              lambda s: s.query(Address))
  
          base_bq += lambda q: q.options(lazyload(Address.user))
          base_bq += lambda q: q.order_by(Address.id)
  
          assert_result = self.static.address_user_result
  
          for i in range(4):
              for cond1 in (False, True):
                  bq = base_bq._clone()
  
                  sess = Session()
  
                  if cond1:
                      bq += lambda q: q.filter(
                          Address.email_address == 'jack@bean.com')
                  else:
                      bq += lambda q: q.filter(
                          Address.email_address.like('ed@%'))
  
                  if cond1:
                      def go():
                          result = bq(sess).all()
                          eq_(assert_result[0:1], result)
                      self.assert_sql_count(testing.db, go, 2)
                  else:
                      def go():
                          result = bq(sess).all()
                          eq_(assert_result[1:4], result)
                      self.assert_sql_count(testing.db, go, 2)
  
                  sess.close()
  
      def test_useget_cancels_eager(self):
          """test that a one to many lazyload cancels the unnecessary
          eager many-to-one join on the other side."""
  
          User = self.classes.User
          Address = self.classes.Address
  
          mapper(User, self.tables.users)
          mapper(Address, self.tables.addresses, properties={
              'user': relationship(
                  User, lazy='joined',
                  backref=backref('addresses', lazy='baked_select')
              )
          })
  
          sess = Session()
          u1 = sess.query(User).filter(User.id == 8).one()
  
          def go():
              eq_(u1.addresses[0].user, u1)
          self.assert_sql_execution(
              testing.db, go,
              CompiledSQL(
                  "SELECT addresses.id AS addresses_id, addresses.user_id AS "
                  "addresses_user_id, addresses.email_address AS "
                  "addresses_email_address FROM addresses WHERE :param_1 = "
                  "addresses.user_id",
                  {'param_1': 8})
              )
  
      def test_useget_cancels_eager_propagated_present(self):
          """test that a one to many lazyload cancels the unnecessary
          eager many-to-one join on the other side, even when a propagated
          option is present."""
  
          User = self.classes.User
          Address = self.classes.Address
  
          mapper(User, self.tables.users)
          mapper(Address, self.tables.addresses, properties={
              'user': relationship(
                  User, lazy='joined',
                  backref=backref('addresses', lazy='baked_select')
              )
          })
  
          from sqlalchemy.orm.interfaces import MapperOption
  
          class MyBogusOption(MapperOption):
              propagate_to_loaders = True
  
          sess = Session()
          u1 = sess.query(User).options(MyBogusOption()).filter(User.id == 8) \
              .one()
  
          def go():
              eq_(u1.addresses[0].user, u1)
          self.assert_sql_execution(
              testing.db, go,
              CompiledSQL(
                  "SELECT addresses.id AS addresses_id, addresses.user_id AS "
                  "addresses_user_id, addresses.email_address AS "
                  "addresses_email_address FROM addresses WHERE :param_1 = "
                  "addresses.user_id",
                  {'param_1': 8})
              )
  
      # additional tests:
      # 1. m2m w lazyload
      # 2. o2m lazyload where m2o backrefs have an eager load, test
      # that eager load is canceled out
      # 3. uselist = False, uselist=False assertion
  