text: from sqlalchemy.testing import assert_raises, assert_raises_message
  import datetime
  import sqlalchemy as sa
  from sqlalchemy import testing
  from sqlalchemy import Integer, String, ForeignKey, MetaData, and_, \
      select, func
  from sqlalchemy.testing.schema import Table, Column
  from sqlalchemy.orm import mapper, relationship, relation, \
      backref, create_session, configure_mappers, \
      clear_mappers, sessionmaker, attributes,\
      Session, composite, column_property, foreign,\
      remote, synonym, joinedload, subqueryload
  from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE
  from sqlalchemy.testing import eq_, startswith_, AssertsCompiledSQL, is_, in_
  from sqlalchemy.testing import fixtures
  from test.orm import _fixtures
  from sqlalchemy import exc
  from sqlalchemy import inspect
  
  
  class _RelationshipErrors(object):
  
      def _assert_raises_no_relevant_fks(self, fn, expr, relname,
                                         primary, *arg, **kw):
          assert_raises_message(
              sa.exc.ArgumentError,
              "Could not locate any relevant foreign key columns "
              "for %s join condition '%s' on relationship %s.  "
              "Ensure that referencing columns are associated with "
              "a ForeignKey or ForeignKeyConstraint, or are annotated "
              r"in the join condition with the foreign\(\) annotation."
              % (
                  primary, expr, relname
              ),
              fn, *arg, **kw
          )
  
      def _assert_raises_no_equality(self, fn, expr, relname,
                                     primary, *arg, **kw):
          assert_raises_message(
              sa.exc.ArgumentError,
              "Could not locate any simple equality expressions "
              "involving locally mapped foreign key columns for %s join "
              "condition '%s' on relationship %s.  "
              "Ensure that referencing columns are associated with a "
              "ForeignKey or ForeignKeyConstraint, or are annotated in "
              r"the join condition with the foreign\(\) annotation. "
              "To allow comparison operators other than '==', "
              "the relationship can be marked as viewonly=True." % (
                  primary, expr, relname
              ),
              fn, *arg, **kw
          )
  
      def _assert_raises_ambig_join(self, fn, relname, secondary_arg,
                                    *arg, **kw):
          if secondary_arg is not None:
              assert_raises_message(
                  exc.ArgumentError,
                  "Could not determine join condition between "
                  "parent/child tables on relationship %s - "
                  "there are multiple foreign key paths linking the "
                  "tables via secondary table '%s'.  "
                  "Specify the 'foreign_keys' argument, providing a list "
                  "of those columns which should be counted as "
                  "containing a foreign key reference from the "
                  "secondary table to each of the parent and child tables."
                  % (relname, secondary_arg),
                  fn, *arg, **kw)
          else:
              assert_raises_message(
                  exc.ArgumentError,
                  "Could not determine join "
                  "condition between parent/child tables on "
                  "relationship %s - there are multiple foreign key "
                  "paths linking the tables.  Specify the "
                  "'foreign_keys' argument, providing a list of those "
                  "columns which should be counted as containing a "
                  "foreign key reference to the parent table."
                  % (relname,),
                  fn, *arg, **kw)
  
      def _assert_raises_no_join(self, fn, relname, secondary_arg,
                                 *arg, **kw):
          if secondary_arg is not None:
              assert_raises_message(
                  exc.NoForeignKeysError,
                  "Could not determine join condition between "
                  "parent/child tables on relationship %s - "
                  "there are no foreign keys linking these tables "
                  "via secondary table '%s'.  "
                  "Ensure that referencing columns are associated with a "
                  "ForeignKey "
                  "or ForeignKeyConstraint, or specify 'primaryjoin' and "
                  "'secondaryjoin' expressions"
                  % (relname, secondary_arg),
                  fn, *arg, **kw)
          else:
              assert_raises_message(
                  exc.NoForeignKeysError,
                  "Could not determine join condition between "
                  "parent/child tables on relationship %s - "
                  "there are no foreign keys linking these tables.  "
                  "Ensure that referencing columns are associated with a "
                  "ForeignKey "
                  "or ForeignKeyConstraint, or specify a 'primaryjoin' "
                  "expression."
                  % (relname,),
                  fn, *arg, **kw)
  
      def _assert_raises_ambiguous_direction(self, fn, relname, *arg, **kw):
          assert_raises_message(
              sa.exc.ArgumentError,
              "Can't determine relationship"
              " direction for relationship '%s' - foreign "
              "key columns within the join condition are present "
              "in both the parent and the child's mapped tables.  "
              "Ensure that only those columns referring to a parent column "
              r"are marked as foreign, either via the foreign\(\) annotation or "
              "via the foreign_keys argument."
              % relname,
              fn, *arg, **kw
          )
  
      def _assert_raises_no_local_remote(self, fn, relname, *arg, **kw):
          assert_raises_message(
              sa.exc.ArgumentError,
              "Relationship %s could not determine "
              "any unambiguous local/remote column "
              "pairs based on join condition and remote_side arguments.  "
              r"Consider using the remote\(\) annotation to "
              "accurately mark those elements of the join "
              "condition that are on the remote side of the relationship." % (
                  relname
              ),
  
              fn, *arg, **kw
          )
  
  
  class DependencyTwoParentTest(fixtures.MappedTest):
  
      """Test flush() when a mapper is dependent on multiple relationships"""
  
      run_setup_mappers = 'once'
      run_inserts = 'once'
      run_deletes = None
  
      @classmethod
      def define_tables(cls, metadata):
          Table("tbl_a", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("name", String(128)))
          Table("tbl_b", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("name", String(128)))
          Table("tbl_c", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("tbl_a_id", Integer, ForeignKey("tbl_a.id"),
                       nullable=False),
                Column("name", String(128)))
          Table("tbl_d", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("tbl_c_id", Integer, ForeignKey("tbl_c.id"),
                       nullable=False),
                Column("tbl_b_id", Integer, ForeignKey("tbl_b.id")),
                Column("name", String(128)))
  
      @classmethod
      def setup_classes(cls):
          class A(cls.Basic):
              pass
  
          class B(cls.Basic):
              pass
  
          class C(cls.Basic):
              pass
  
          class D(cls.Basic):
              pass
  
      @classmethod
      def setup_mappers(cls):
          A, C, B, D, tbl_b, tbl_c, tbl_a, tbl_d = (cls.classes.A,
                                                    cls.classes.C,
                                                    cls.classes.B,
                                                    cls.classes.D,
                                                    cls.tables.tbl_b,
                                                    cls.tables.tbl_c,
                                                    cls.tables.tbl_a,
                                                    cls.tables.tbl_d)
  
          mapper(A, tbl_a, properties=dict(
              c_rows=relationship(C, cascade="all, delete-orphan",
                                  backref="a_row")))
          mapper(B, tbl_b)
          mapper(C, tbl_c, properties=dict(
              d_rows=relationship(D, cascade="all, delete-orphan",
                                  backref="c_row")))
          mapper(D, tbl_d, properties=dict(
              b_row=relationship(B)))
  
      @classmethod
      def insert_data(cls):
          A, C, B, D = (cls.classes.A,
                        cls.classes.C,
                        cls.classes.B,
                        cls.classes.D)
  
          session = create_session()
          a = A(name='a1')
          b = B(name='b1')
          c = C(name='c1', a_row=a)
  
          d1 = D(name='d1', b_row=b, c_row=c)  # noqa
          d2 = D(name='d2', b_row=b, c_row=c)  # noqa
          d3 = D(name='d3', b_row=b, c_row=c)  # noqa
          session.add(a)
          session.add(b)
          session.flush()
  
      def test_DeleteRootTable(self):
          A = self.classes.A
  
          session = create_session()
          a = session.query(A).filter_by(name='a1').one()
  
          session.delete(a)
          session.flush()
  
      def test_DeleteMiddleTable(self):
          C = self.classes.C
  
          session = create_session()
          c = session.query(C).filter_by(name='c1').one()
  
          session.delete(c)
          session.flush()
  
  
  class M2ODontOverwriteFKTest(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table(
              'a', metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', ForeignKey('b.id'))
          )
          Table(
              'b', metadata,
              Column('id', Integer, primary_key=True),
          )
  
      def _fixture(self, uselist=False):
          a, b = self.tables.a, self.tables.b
  
          class A(fixtures.BasicEntity):
              pass
  
          class B(fixtures.BasicEntity):
              pass
  
          mapper(A, a, properties={
              'b': relationship(B, uselist=uselist)
          })
          mapper(B, b)
          return A, B
  
      def test_joinedload_doesnt_produce_bogus_event(self):
          A, B = self._fixture()
          sess = Session()
  
          b1 = B()
          sess.add(b1)
          sess.flush()
  
          a1 = A()
          sess.add(a1)
          sess.commit()
  
          # test that was broken by #3060
          a1 = sess.query(A).options(joinedload("b")).first()
          a1.bid = b1.id
          sess.flush()
  
          eq_(a1.bid, b1.id)
  
      def test_init_doesnt_produce_scalar_event(self):
          A, B = self._fixture()
          sess = Session()
  
          b1 = B()
          sess.add(b1)
          sess.flush()
  
          a1 = A()
          assert a1.b is None
          a1.bid = b1.id
          sess.add(a1)
          sess.flush()
          assert a1.bid is not None
  
      def test_init_doesnt_produce_collection_event(self):
          A, B = self._fixture(uselist=True)
          sess = Session()
  
          b1 = B()
          sess.add(b1)
          sess.flush()
  
          a1 = A()
          assert a1.b == []
          a1.bid = b1.id
          sess.add(a1)
          sess.flush()
          assert a1.bid is not None
  
      def test_scalar_relationship_overrides_fk(self):
          A, B = self._fixture()
          sess = Session()
  
          b1 = B()
          sess.add(b1)
          sess.flush()
  
          a1 = A()
          a1.bid = b1.id
          a1.b = None
          sess.add(a1)
          sess.flush()
          assert a1.bid is None
  
      def test_collection_relationship_overrides_fk(self):
          A, B = self._fixture(uselist=True)
          sess = Session()
  
          b1 = B()
          sess.add(b1)
          sess.flush()
  
          a1 = A()
          a1.bid = b1.id
          a1.b = []
          sess.add(a1)
          sess.flush()
          # this is weird
          assert a1.bid is not None
  
  
  class DirectSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
  
      """Tests the ultimate join condition, a single column
      that points to itself, e.g. within a SQL function or similar.
      The test is against a materialized path setup.
  
      this is an **extremely** unusual case::
  
      Entity
      ------
       path -------+
         ^         |
         +---------+
  
      In this case, one-to-many and many-to-one are no longer accurate.
      Both relationships return collections.   I'm not sure if this is a good
      idea.
  
      """
  
      __dialect__ = 'default'
  
      @classmethod
      def define_tables(cls, metadata):
          Table('entity', metadata,
                Column('path', String(100), primary_key=True)
                )
  
      @classmethod
      def setup_classes(cls):
          class Entity(cls.Basic):
  
              def __init__(self, path):
                  self.path = path
  
      def _descendants_fixture(self, data=True):
          Entity = self.classes.Entity
          entity = self.tables.entity
  
          m = mapper(Entity, entity, properties={
              "descendants": relationship(
                  Entity,
                  primaryjoin=remote(foreign(entity.c.path)).like(
                      entity.c.path.concat('/%')),
                  viewonly=True,
                  order_by=entity.c.path)
          })
          configure_mappers()
          assert m.get_property("descendants").direction is ONETOMANY
          if data:
              return self._fixture()
  
      def _anscestors_fixture(self, data=True):
          Entity = self.classes.Entity
          entity = self.tables.entity
  
          m = mapper(Entity, entity, properties={
              "anscestors": relationship(
                  Entity,
                  primaryjoin=entity.c.path.like(
                      remote(foreign(entity.c.path)).concat('/%')),
                  viewonly=True,
                  order_by=entity.c.path)
          })
          configure_mappers()
          assert m.get_property("anscestors").direction is ONETOMANY
          if data:
              return self._fixture()
  
      def _fixture(self):
          Entity = self.classes.Entity
          sess = Session()
          sess.add_all([
              Entity("/foo"),
              Entity("/foo/bar1"),
              Entity("/foo/bar2"),
              Entity("/foo/bar2/bat1"),
              Entity("/foo/bar2/bat2"),
              Entity("/foo/bar3"),
              Entity("/bar"),
              Entity("/bar/bat1")
          ])
          return sess
  
      def test_descendants_lazyload_clause(self):
          self._descendants_fixture(data=False)
          Entity = self.classes.Entity
          self.assert_compile(
              Entity.descendants.property.strategy._lazywhere,
              "entity.path LIKE (:param_1 || :path_1)"
          )
  
          self.assert_compile(
              Entity.descendants.property.strategy._rev_lazywhere,
              ":param_1 LIKE (entity.path || :path_1)"
          )
  
      def test_ancestors_lazyload_clause(self):
          self._anscestors_fixture(data=False)
          Entity = self.classes.Entity
          # :param_1 LIKE (:param_1 || :path_1)
          self.assert_compile(
              Entity.anscestors.property.strategy._lazywhere,
              ":param_1 LIKE (entity.path || :path_1)"
          )
  
          self.assert_compile(
              Entity.anscestors.property.strategy._rev_lazywhere,
              "entity.path LIKE (:param_1 || :path_1)"
          )
  
      def test_descendants_lazyload(self):
          sess = self._descendants_fixture()
          Entity = self.classes.Entity
          e1 = sess.query(Entity).filter_by(path="/foo").first()
          eq_(
              [e.path for e in e1.descendants],
              ["/foo/bar1", "/foo/bar2", "/foo/bar2/bat1",
                  "/foo/bar2/bat2", "/foo/bar3"]
          )
  
      def test_anscestors_lazyload(self):
          sess = self._anscestors_fixture()
          Entity = self.classes.Entity
          e1 = sess.query(Entity).filter_by(path="/foo/bar2/bat1").first()
          eq_(
              [e.path for e in e1.anscestors],
              ["/foo", "/foo/bar2"]
          )
  
      def test_descendants_joinedload(self):
          sess = self._descendants_fixture()
          Entity = self.classes.Entity
          e1 = sess.query(Entity).filter_by(path="/foo").\
              options(joinedload(Entity.descendants)).first()
  
          eq_(
              [e.path for e in e1.descendants],
              ["/foo/bar1", "/foo/bar2", "/foo/bar2/bat1",
                  "/foo/bar2/bat2", "/foo/bar3"]
          )
  
      def test_descendants_subqueryload(self):
          sess = self._descendants_fixture()
          Entity = self.classes.Entity
          e1 = sess.query(Entity).filter_by(path="/foo").\
              options(subqueryload(Entity.descendants)).first()
  
          eq_(
              [e.path for e in e1.descendants],
              ["/foo/bar1", "/foo/bar2", "/foo/bar2/bat1",
                  "/foo/bar2/bat2", "/foo/bar3"]
          )
  
      def test_anscestors_joinedload(self):
          sess = self._anscestors_fixture()
          Entity = self.classes.Entity
          e1 = sess.query(Entity).filter_by(path="/foo/bar2/bat1").\
              options(joinedload(Entity.anscestors)).first()
          eq_(
              [e.path for e in e1.anscestors],
              ["/foo", "/foo/bar2"]
          )
  
      def test_plain_join_descendants(self):
          self._descendants_fixture(data=False)
          Entity = self.classes.Entity
          sess = Session()
          self.assert_compile(
              sess.query(Entity).join(Entity.descendants, aliased=True),
              "SELECT entity.path AS entity_path FROM entity JOIN entity AS "
              "entity_1 ON entity_1.path LIKE (entity.path || :path_1)"
          )
  
  
  class CompositeSelfRefFKTest(fixtures.MappedTest, AssertsCompiledSQL):
  
      """Tests a composite FK where, in
      the relationship(), one col points
      to itself in the same table.
  
      this is a very unusual case::
  
      company         employee
      ----------      ----------
      company_id <--- company_id ------+
      name                ^            |
                          +------------+
  
                      emp_id <---------+
                      name             |
                      reports_to_id ---+
  
      employee joins to its sub-employees
      both on reports_to_id, *and on company_id to itself*.
  
      """
  
      __dialect__ = 'default'
  
      @classmethod
      def define_tables(cls, metadata):
          Table('company_t', metadata,
                Column('company_id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('name', String(30)))
  
          Table('employee_t', metadata,
                Column('company_id', Integer, primary_key=True),
                Column('emp_id', Integer, primary_key=True),
                Column('name', String(30)),
                Column('reports_to_id', Integer),
                sa.ForeignKeyConstraint(
                    ['company_id'],
                    ['company_t.company_id']),
                sa.ForeignKeyConstraint(
                    ['company_id', 'reports_to_id'],
                    ['employee_t.company_id', 'employee_t.emp_id']))
  
      @classmethod
      def setup_classes(cls):
          class Company(cls.Basic):
  
              def __init__(self, name):
                  self.name = name
  
          class Employee(cls.Basic):
  
              def __init__(self, name, company, emp_id, reports_to=None):
                  self.name = name
                  self.company = company
                  self.emp_id = emp_id
                  self.reports_to = reports_to
  
      def test_explicit(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company,
                                      primaryjoin=employee_t.c.company_id ==
                                      company_t.c.company_id,
                                      backref='employees'),
              'reports_to': relationship(Employee, primaryjoin=sa.and_(
                  employee_t.c.emp_id == employee_t.c.reports_to_id,
                  employee_t.c.company_id == employee_t.c.company_id
              ),
                  remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
                  foreign_keys=[
                      employee_t.c.reports_to_id, employee_t.c.company_id],
                  backref=backref('employees',
                                  foreign_keys=[employee_t.c.reports_to_id,
                                                employee_t.c.company_id]))
          })
  
          self._test()
  
      def test_implicit(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company, backref='employees'),
              'reports_to': relationship(
                  Employee,
                  remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
                  foreign_keys=[employee_t.c.reports_to_id,
                                employee_t.c.company_id],
                  backref=backref(
                      'employees',
                      foreign_keys=[
                          employee_t.c.reports_to_id, employee_t.c.company_id])
              )
          })
  
          self._test()
  
      def test_very_implicit(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company, backref='employees'),
              'reports_to': relationship(
                  Employee,
                  remote_side=[employee_t.c.emp_id, employee_t.c.company_id],
                  backref='employees'
              )
          })
  
          self._test()
  
      def test_very_explicit(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company, backref='employees'),
              'reports_to': relationship(
                  Employee,
                  _local_remote_pairs=[
                      (employee_t.c.reports_to_id, employee_t.c.emp_id),
                      (employee_t.c.company_id, employee_t.c.company_id)
                  ],
                  foreign_keys=[
                      employee_t.c.reports_to_id,
                      employee_t.c.company_id],
                  backref=backref(
                      'employees',
                      foreign_keys=[
                          employee_t.c.reports_to_id, employee_t.c.company_id])
              )
          })
  
          self._test()
  
      def test_annotated(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company, backref='employees'),
              'reports_to': relationship(
                  Employee,
                  primaryjoin=sa.and_(
                      remote(employee_t.c.emp_id) == employee_t.c.reports_to_id,
                      remote(employee_t.c.company_id) == employee_t.c.company_id
                  ),
                  backref=backref('employees')
              )
          })
  
          self._assert_lazy_clauses()
          self._test()
  
      def test_overlapping_warning(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company, backref='employees'),
              'reports_to': relationship(
                  Employee,
                  primaryjoin=sa.and_(
                      remote(employee_t.c.emp_id) == employee_t.c.reports_to_id,
                      remote(employee_t.c.company_id) == employee_t.c.company_id
                  ),
                  backref=backref('employees')
              )
          })
  
          assert_raises_message(
              exc.SAWarning,
              r"relationship .* will copy column .* to column "
              r"employee_t.company_id, which conflicts with relationship\(s\)",
              configure_mappers
          )
  
      def test_annotated_no_overwriting(self):
          Employee, Company, employee_t, company_t = (self.classes.Employee,
                                                      self.classes.Company,
                                                      self.tables.employee_t,
                                                      self.tables.company_t)
  
          mapper(Company, company_t)
          mapper(Employee, employee_t, properties={
              'company': relationship(Company, backref='employees'),
              'reports_to': relationship(
                  Employee,
                  primaryjoin=sa.and_(
                      remote(employee_t.c.emp_id) ==
                      foreign(employee_t.c.reports_to_id),
                      remote(employee_t.c.company_id) == employee_t.c.company_id
                  ),
                  backref=backref('employees')
              )
          })
  
          self._assert_lazy_clauses()
          self._test_no_warning()
  
      def _test_no_overwrite(self, sess, expect_failure):
          # test [ticket:3230]
  
          Employee, Company = self.classes.Employee, self.classes.Company
  
          c1 = sess.query(Company).filter_by(name='c1').one()
          e3 = sess.query(Employee).filter_by(name='emp3').one()
          e3.reports_to = None
  
          if expect_failure:
              # if foreign() isn't applied specifically to
              # employee_t.c.reports_to_id only, then
              # employee_t.c.company_id goes foreign as well and then
              # this happens
              assert_raises_message(
                  AssertionError,
                  "Dependency rule tried to blank-out primary key column "
                  "'employee_t.company_id'",
                  sess.flush
              )
          else:
              sess.flush()
              eq_(e3.company, c1)
  
      @testing.emits_warning("relationship .* will copy column ")
      def _test(self):
          self._test_no_warning(overwrites=True)
  
      def _test_no_warning(self, overwrites=False):
          configure_mappers()
          self._test_relationships()
          sess = Session()
          self._setup_data(sess)
          self._test_lazy_relations(sess)
          self._test_join_aliasing(sess)
          self._test_no_overwrite(sess, expect_failure=overwrites)
  
      @testing.emits_warning("relationship .* will copy column ")
      def _assert_lazy_clauses(self):
          configure_mappers()
          Employee = self.classes.Employee
          self.assert_compile(
              Employee.employees.property.strategy._lazywhere,
              ":param_1 = employee_t.reports_to_id AND "
              ":param_2 = employee_t.company_id"
          )
  
          self.assert_compile(
              Employee.employees.property.strategy._rev_lazywhere,
              "employee_t.emp_id = :param_1 AND "
              "employee_t.company_id = :param_2"
          )
  
      def _test_relationships(self):
          Employee = self.classes.Employee
          employee_t = self.tables.employee_t
          eq_(
              set(Employee.employees.property.local_remote_pairs),
              set([
                  (employee_t.c.company_id, employee_t.c.company_id),
                  (employee_t.c.emp_id, employee_t.c.reports_to_id),
              ])
          )
          eq_(
              Employee.employees.property.remote_side,
              set([employee_t.c.company_id, employee_t.c.reports_to_id])
          )
          eq_(
              set(Employee.reports_to.property.local_remote_pairs),
              set([
                  (employee_t.c.company_id, employee_t.c.company_id),
                  (employee_t.c.reports_to_id, employee_t.c.emp_id),
              ])
          )
  
      def _setup_data(self, sess):
          Employee, Company = self.classes.Employee, self.classes.Company
  
          c1 = Company('c1')
          c2 = Company('c2')
  
          e1 = Employee('emp1', c1, 1)
          e2 = Employee('emp2', c1, 2, e1)  # noqa
          e3 = Employee('emp3', c1, 3, e1)
          e4 = Employee('emp4', c1, 4, e3)  # noqa
          e5 = Employee('emp5', c2, 1)
          e6 = Employee('emp6', c2, 2, e5)  # noqa
          e7 = Employee('emp7', c2, 3, e5)  # noqa
  
          sess.add_all((c1, c2))
          sess.commit()
          sess.close()
  
      def _test_lazy_relations(self, sess):
          Employee, Company = self.classes.Employee, self.classes.Company
  
          c1 = sess.query(Company).filter_by(name='c1').one()
          c2 = sess.query(Company).filter_by(name='c2').one()
          e1 = sess.query(Employee).filter_by(name='emp1').one()
          e5 = sess.query(Employee).filter_by(name='emp5').one()
  
          test_e1 = sess.query(Employee).get([c1.company_id, e1.emp_id])
          assert test_e1.name == 'emp1', test_e1.name
          test_e5 = sess.query(Employee).get([c2.company_id, e5.emp_id])
          assert test_e5.name == 'emp5', test_e5.name
          assert [x.name for x in test_e1.employees] == ['emp2', 'emp3']
          assert sess.query(Employee).\
              get([c1.company_id, 3]).reports_to.name == 'emp1'
          assert sess.query(Employee).\
              get([c2.company_id, 3]).reports_to.name == 'emp5'
  
      def _test_join_aliasing(self, sess):
          Employee, Company = self.classes.Employee, self.classes.Company
          eq_(
              [n for n, in sess.query(Employee.name).
               join(Employee.reports_to, aliased=True).
               filter_by(name='emp5').
               reset_joinpoint().
               order_by(Employee.name)],
              ['emp6', 'emp7']
          )
  
  
  class CompositeJoinPartialFK(fixtures.MappedTest, AssertsCompiledSQL):
      __dialect__ = 'default'
  
      @classmethod
      def define_tables(cls, metadata):
          Table("parent", metadata,
                Column('x', Integer, primary_key=True),
                Column('y', Integer, primary_key=True),
                Column('z', Integer),
                )
          Table("child", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('x', Integer),
                Column('y', Integer),
                Column('z', Integer),
                # note 'z' is not here
                sa.ForeignKeyConstraint(
                    ["x", "y"],
                    ["parent.x", "parent.y"]
                )
                )
  
      @classmethod
      def setup_mappers(cls):
          parent, child = cls.tables.parent, cls.tables.child
  
          class Parent(cls.Comparable):
              pass
  
          class Child(cls.Comparable):
              pass
          mapper(Parent, parent, properties={
              'children': relationship(Child, primaryjoin=and_(
                  parent.c.x == child.c.x,
                  parent.c.y == child.c.y,
                  parent.c.z == child.c.z,
              ))
          })
          mapper(Child, child)
  
      def test_joins_fully(self):
          Parent, Child = self.classes.Parent, self.classes.Child
  
          self.assert_compile(
              Parent.children.property.strategy._lazywhere,
              ":param_1 = child.x AND :param_2 = child.y AND :param_3 = child.z"
          )
  
  
  class SynonymsAsFKsTest(fixtures.MappedTest):
  
      """Syncrules on foreign keys that are also primary"""
  
      @classmethod
      def define_tables(cls, metadata):
          Table("tableA", metadata,
                Column("id", Integer, primary_key=True),
                Column("foo", Integer,),
                test_needs_fk=True)
  
          Table("tableB", metadata,
                Column("id", Integer, primary_key=True),
                Column("_a_id", Integer, key='a_id', primary_key=True),
                test_needs_fk=True)
  
      @classmethod
      def setup_classes(cls):
          class A(cls.Basic):
              pass
  
          class B(cls.Basic):
  
              @property
              def a_id(self):
                  return self._a_id
  
      def test_synonym_fk(self):
          """test that active history is enabled on a
          one-to-many/one that has use_get==True"""
  
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          mapper(B, tableB, properties={
              'a_id': synonym('_a_id', map_column=True)})
          mapper(A, tableA, properties={
              'b': relationship(B, primaryjoin=(tableA.c.id == foreign(B.a_id)),
                                uselist=False)})
  
          sess = create_session()
  
          b = B(id=0)
          a = A(id=0, b=b)
          sess.add(a)
          sess.add(b)
          sess.flush()
          sess.expunge_all()
  
          assert a.b == b
          assert a.id == b.a_id
          assert a.id == b._a_id
  
  
  class FKsAsPksTest(fixtures.MappedTest):
  
      """Syncrules on foreign keys that are also primary"""
  
      @classmethod
      def define_tables(cls, metadata):
          Table("tableA", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("foo", Integer,),
                test_needs_fk=True)
  
          Table("tableB", metadata,
                Column("id", Integer, ForeignKey("tableA.id"), primary_key=True),
                test_needs_fk=True)
  
      @classmethod
      def setup_classes(cls):
          class A(cls.Basic):
              pass
  
          class B(cls.Basic):
              pass
  
      def test_onetoone_switch(self):
          """test that active history is enabled on a
          one-to-many/one that has use_get==True"""
  
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          mapper(A, tableA, properties={
              'b': relationship(B, cascade="all,delete-orphan", uselist=False)})
          mapper(B, tableB)
  
          configure_mappers()
          assert A.b.property.strategy.use_get
  
          sess = create_session()
  
          a1 = A()
          sess.add(a1)
          sess.flush()
          sess.close()
          a1 = sess.query(A).first()
          a1.b = B()
          sess.flush()
  
      def test_no_delete_PK_AtoB(self):
          """A cant be deleted without B because B would have no PK value."""
  
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          mapper(A, tableA, properties={
              'bs': relationship(B, cascade="save-update")})
          mapper(B, tableB)
  
          a1 = A()
          a1.bs.append(B())
          sess = create_session()
          sess.add(a1)
          sess.flush()
  
          sess.delete(a1)
          try:
              sess.flush()
              assert False
          except AssertionError as e:
              startswith_(str(e),
                          "Dependency rule tried to blank-out "
                          "primary key column 'tableB.id' on instance ")
  
      def test_no_delete_PK_BtoA(self):
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          mapper(B, tableB, properties={
              'a': relationship(A, cascade="save-update")})
          mapper(A, tableA)
  
          b1 = B()
          a1 = A()
          b1.a = a1
          sess = create_session()
          sess.add(b1)
          sess.flush()
          b1.a = None
          try:
              sess.flush()
              assert False
          except AssertionError as e:
              startswith_(str(e),
                          "Dependency rule tried to blank-out "
                          "primary key column 'tableB.id' on instance ")
  
      @testing.fails_on_everything_except(
          'sqlite', testing.requires.mysql_non_strict)
      def test_nullPKsOK_BtoA(self):
          A, tableA = self.classes.A, self.tables.tableA
  
          # postgresql cant handle a nullable PK column...?
          tableC = Table(
              'tablec', tableA.metadata,
              Column('id', Integer, primary_key=True),
              Column('a_id', Integer, ForeignKey('tableA.id'),
                     primary_key=True, nullable=True))
          tableC.create()
  
          class C(fixtures.BasicEntity):
              pass
          mapper(C, tableC, properties={
              'a': relationship(A, cascade="save-update")
          })
          mapper(A, tableA)
  
          c1 = C()
          c1.id = 5
          c1.a = None
          sess = create_session()
          sess.add(c1)
          # test that no error is raised.
          sess.flush()
  
      def test_delete_cascade_BtoA(self):
          """No 'blank the PK' error when the child is to
          be deleted as part of a cascade"""
  
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          for cascade in ("save-update, delete",
                          # "save-update, delete-orphan",
                          "save-update, delete, delete-orphan"):
              mapper(B, tableB, properties={
                  'a': relationship(A, cascade=cascade, single_parent=True)
              })
              mapper(A, tableA)
  
              b1 = B()
              a1 = A()
              b1.a = a1
              sess = create_session()
              sess.add(b1)
              sess.flush()
              sess.delete(b1)
              sess.flush()
              assert a1 not in sess
              assert b1 not in sess
              sess.expunge_all()
              sa.orm.clear_mappers()
  
      def test_delete_cascade_AtoB(self):
          """No 'blank the PK' error when the child is to
          be deleted as part of a cascade"""
  
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          for cascade in ("save-update, delete",
                          # "save-update, delete-orphan",
                          "save-update, delete, delete-orphan"):
              mapper(A, tableA, properties={
                  'bs': relationship(B, cascade=cascade)
              })
              mapper(B, tableB)
  
              a1 = A()
              b1 = B()
              a1.bs.append(b1)
              sess = create_session()
              sess.add(a1)
              sess.flush()
  
              sess.delete(a1)
              sess.flush()
              assert a1 not in sess
              assert b1 not in sess
              sess.expunge_all()
              sa.orm.clear_mappers()
  
      def test_delete_manual_AtoB(self):
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          mapper(A, tableA, properties={
              'bs': relationship(B, cascade="none")})
          mapper(B, tableB)
  
          a1 = A()
          b1 = B()
          a1.bs.append(b1)
          sess = create_session()
          sess.add(a1)
          sess.add(b1)
          sess.flush()
  
          sess.delete(a1)
          sess.delete(b1)
          sess.flush()
          assert a1 not in sess
          assert b1 not in sess
          sess.expunge_all()
  
      def test_delete_manual_BtoA(self):
          tableB, A, B, tableA = (self.tables.tableB,
                                  self.classes.A,
                                  self.classes.B,
                                  self.tables.tableA)
  
          mapper(B, tableB, properties={
              'a': relationship(A, cascade="none")})
          mapper(A, tableA)
  
          b1 = B()
          a1 = A()
          b1.a = a1
          sess = create_session()
          sess.add(b1)
          sess.add(a1)
          sess.flush()
          sess.delete(b1)
          sess.delete(a1)
          sess.flush()
          assert a1 not in sess
          assert b1 not in sess
  
  
  class UniqueColReferenceSwitchTest(fixtures.MappedTest):
  
      """test a relationship based on a primary
      join against a unique non-pk column"""
  
      @classmethod
      def define_tables(cls, metadata):
          Table("table_a", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("ident", String(10), nullable=False,
                       unique=True),
                )
  
          Table("table_b", metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("a_ident", String(10),
                       ForeignKey('table_a.ident'),
                       nullable=False),
                )
  
      @classmethod
      def setup_classes(cls):
          class A(cls.Comparable):
              pass
  
          class B(cls.Comparable):
              pass
  
      def test_switch_parent(self):
          A, B, table_b, table_a = (self.classes.A,
                                    self.classes.B,
                                    self.tables.table_b,
                                    self.tables.table_a)
  
          mapper(A, table_a)
          mapper(B, table_b, properties={"a": relationship(A, backref="bs")})
  
          session = create_session()
          a1, a2 = A(ident="uuid1"), A(ident="uuid2")
          session.add_all([a1, a2])
          a1.bs = [
              B(), B()
          ]
          session.flush()
          session.expire_all()
          a1, a2 = session.query(A).all()
  
          for b in list(a1.bs):
              b.a = a2
          session.delete(a1)
          session.flush()
  
  
  class RelationshipToSelectableTest(fixtures.MappedTest):
  
      """Test a map to a select that relates to a map to the table."""
  
      @classmethod
      def define_tables(cls, metadata):
          Table('items', metadata,
                Column('item_policy_num', String(10), primary_key=True,
                       key='policyNum'),
                Column('item_policy_eff_date', sa.Date, primary_key=True,
                       key='policyEffDate'),
                Column('item_type', String(20), primary_key=True,
                       key='type'),
                Column('item_id', Integer, primary_key=True,
                       key='id', autoincrement=False))
  
      def test_basic(self):
          items = self.tables.items
  
          class Container(fixtures.BasicEntity):
              pass
  
          class LineItem(fixtures.BasicEntity):
              pass
  
          container_select = sa.select(
              [items.c.policyNum, items.c.policyEffDate, items.c.type],
              distinct=True,
          ).alias('container_select')
  
          mapper(LineItem, items)
  
          mapper(
              Container,
              container_select,
              properties=dict(
                  lineItems=relationship(
                      LineItem,
                      lazy='select',
                      cascade='all, delete-orphan',
                      order_by=sa.asc(items.c.id),
                      primaryjoin=sa.and_(
                          container_select.c.policyNum == items.c.policyNum,
                          container_select.c.policyEffDate ==
                          items.c.policyEffDate,
                          container_select.c.type == items.c.type),
                      foreign_keys=[
                          items.c.policyNum,
                          items.c.policyEffDate,
                          items.c.type
                      ]
                  )
              )
          )
  
          session = create_session()
          con = Container()
          con.policyNum = "99"
          con.policyEffDate = datetime.date.today()
          con.type = "TESTER"
          session.add(con)
          for i in range(0, 10):
              li = LineItem()
              li.id = i
              con.lineItems.append(li)
              session.add(li)
          session.flush()
          session.expunge_all()
          newcon = session.query(Container).\
              order_by(container_select.c.type).first()
          assert con.policyNum == newcon.policyNum
          assert len(newcon.lineItems) == 10
          for old, new in zip(con.lineItems, newcon.lineItems):
              eq_(old.id, new.id)
  
  
  class FKEquatedToConstantTest(fixtures.MappedTest):
  
      """test a relationship with a non-column entity in the primary join,
      is not viewonly, and also has the non-column's clause mentioned in the
      foreign keys list.
  
      """
  
      @classmethod
      def define_tables(cls, metadata):
          Table('tags', metadata, Column("id", Integer, primary_key=True,
                                         test_needs_autoincrement=True),
                Column("data", String(50)),
                )
  
          Table('tag_foo', metadata,
                Column("id", Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('tagid', Integer),
                Column("data", String(50)),
                )
  
      def test_basic(self):
          tag_foo, tags = self.tables.tag_foo, self.tables.tags
  
          class Tag(fixtures.ComparableEntity):
              pass
  
          class TagInstance(fixtures.ComparableEntity):
              pass
  
          mapper(Tag, tags, properties={
              'foo': relationship(
                  TagInstance,
                  primaryjoin=sa.and_(tag_foo.c.data == 'iplc_case',
                                      tag_foo.c.tagid == tags.c.id),
                  foreign_keys=[tag_foo.c.tagid, tag_foo.c.data]),
          })
  
          mapper(TagInstance, tag_foo)
  
          sess = create_session()
          t1 = Tag(data='some tag')
          t1.foo.append(TagInstance(data='iplc_case'))
          t1.foo.append(TagInstance(data='not_iplc_case'))
          sess.add(t1)
          sess.flush()
          sess.expunge_all()
  
          # relationship works
          eq_(
              sess.query(Tag).all(),
              [Tag(data='some tag', foo=[TagInstance(data='iplc_case')])]
          )
  
          # both TagInstances were persisted
          eq_(
              sess.query(TagInstance).order_by(TagInstance.data).all(),
              [TagInstance(data='iplc_case'), TagInstance(data='not_iplc_case')]
          )
  
  
  class BackrefPropagatesForwardsArgs(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table('users', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('name', String(50))
                )
          Table('addresses', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('user_id', Integer),
                Column('email', String(50))
                )
  
      @classmethod
      def setup_classes(cls):
          class User(cls.Comparable):
              pass
  
          class Address(cls.Comparable):
              pass
  
      def test_backref(self):
          User, Address, users, addresses = (self.classes.User,
                                             self.classes.Address,
                                             self.tables.users,
                                             self.tables.addresses)
  
          mapper(User, users, properties={
              'addresses': relationship(
                  Address,
                  primaryjoin=addresses.c.user_id == users.c.id,
                  foreign_keys=addresses.c.user_id,
                  backref='user')
          })
          mapper(Address, addresses)
  
          sess = sessionmaker()()
          u1 = User(name='u1', addresses=[Address(email='a1')])
          sess.add(u1)
          sess.commit()
          eq_(sess.query(Address).all(), [
              Address(email='a1', user=User(name='u1'))
          ])
  
  
  class AmbiguousJoinInterpretedAsSelfRef(fixtures.MappedTest):
  
      """test ambiguous joins due to FKs on both sides treated as
      self-referential.
  
      this mapping is very similar to that of
      test/orm/inheritance/query.py
      SelfReferentialTestJoinedToBase , except that inheritance is
      not used here.
  
      """
  
      @classmethod
      def define_tables(cls, metadata):
          Table(
              'subscriber', metadata,
              Column(
                  'id', Integer, primary_key=True,
                  test_needs_autoincrement=True))
  
          Table(
              'address', metadata,
              Column(
                  'subscriber_id', Integer,
                  ForeignKey('subscriber.id'), primary_key=True),
              Column('type', String(1), primary_key=True),
          )
  
      @classmethod
      def setup_mappers(cls):
          subscriber, address = cls.tables.subscriber, cls.tables.address
  
          subscriber_and_address = subscriber.join(
              address,
              and_(address.c.subscriber_id == subscriber.c.id,
                   address.c.type.in_(['A', 'B', 'C'])))
  
          class Address(cls.Comparable):
              pass
  
          class Subscriber(cls.Comparable):
              pass
  
          mapper(Address, address)
  
          mapper(Subscriber, subscriber_and_address, properties={
              'id': [subscriber.c.id, address.c.subscriber_id],
              'addresses': relationship(Address,
                                        backref=backref("customer"))
          })
  
      def test_mapping(self):
          Subscriber, Address = self.classes.Subscriber, self.classes.Address
  
          sess = create_session()
          assert Subscriber.addresses.property.direction is ONETOMANY
          assert Address.customer.property.direction is MANYTOONE
  
          s1 = Subscriber(type='A',
                          addresses=[
                              Address(type='D'),
                              Address(type='E'),
                          ]
                          )
          a1 = Address(type='B', customer=Subscriber(type='C'))
  
          assert s1.addresses[0].customer is s1
          assert a1.customer.addresses[0] is a1
  
          sess.add_all([s1, a1])
  
          sess.flush()
          sess.expunge_all()
  
          eq_(
              sess.query(Subscriber).order_by(Subscriber.type).all(),
              [
                  Subscriber(id=1, type='A'),
                  Subscriber(id=2, type='B'),
                  Subscriber(id=2, type='C')
              ]
          )
  
  
  class ManualBackrefTest(_fixtures.FixtureTest):
  
      """Test explicit relationships that are backrefs to each other."""
  
      run_inserts = None
  
      def test_o2m(self):
          users, Address, addresses, User = (self.tables.users,
                                             self.classes.Address,
                                             self.tables.addresses,
                                             self.classes.User)
  
          mapper(User, users, properties={
              'addresses': relationship(Address, back_populates='user')
          })
  
          mapper(Address, addresses, properties={
              'user': relationship(User, back_populates='addresses')
          })
  
          sess = create_session()
  
          u1 = User(name='u1')
          a1 = Address(email_address='foo')
          u1.addresses.append(a1)
          assert a1.user is u1
  
          sess.add(u1)
          sess.flush()
          sess.expire_all()
          assert sess.query(Address).one() is a1
          assert a1.user is u1
          assert a1 in u1.addresses
  
      def test_invalid_key(self):
          users, Address, addresses, User = (self.tables.users,
                                             self.classes.Address,
                                             self.tables.addresses,
                                             self.classes.User)
  
          mapper(User, users, properties={
              'addresses': relationship(Address, back_populates='userr')
          })
  
          mapper(Address, addresses, properties={
              'user': relationship(User, back_populates='addresses')
          })
  
          assert_raises(sa.exc.InvalidRequestError, configure_mappers)
  
      def test_invalid_target(self):
          addresses, Dingaling, User, dingalings, Address, users = (
              self.tables.addresses,
              self.classes.Dingaling,
              self.classes.User,
              self.tables.dingalings,
              self.classes.Address,
              self.tables.users)
  
          mapper(User, users, properties={
              'addresses': relationship(Address, back_populates='dingaling'),
          })
  
          mapper(Dingaling, dingalings)
          mapper(Address, addresses, properties={
              'dingaling': relationship(Dingaling)
          })
  
          assert_raises_message(sa.exc.ArgumentError,
                                r"reverse_property 'dingaling' on relationship "
                                r"User.addresses references "
                                r"relationship Address.dingaling, "
                                r"which does not "
                                r"reference mapper Mapper\|User\|users",
                                configure_mappers)
  
  
  class NoLoadBackPopulates(_fixtures.FixtureTest):
  
      """test the noload stratgegy which unlike others doesn't use
      lazyloader to set up instrumentation"""
  
      def test_o2m(self):
          users, Address, addresses, User = (self.tables.users,
                                             self.classes.Address,
                                             self.tables.addresses,
                                             self.classes.User)
  
          mapper(User, users, properties={
              'addresses': relationship(
                  Address, back_populates='user', lazy="noload")
          })
  
          mapper(Address, addresses, properties={
              'user': relationship(User)
          })
  
          u1 = User()
          a1 = Address()
          u1.addresses.append(a1)
          is_(a1.user, u1)
  
      def test_m2o(self):
          users, Address, addresses, User = (self.tables.users,
                                             self.classes.Address,
                                             self.tables.addresses,
                                             self.classes.User)
  
          mapper(User, users, properties={
              'addresses': relationship(
                  Address)
          })
  
          mapper(Address, addresses, properties={
              'user': relationship(
                  User, back_populates='addresses', lazy="noload")
          })
  
          u1 = User()
          a1 = Address()
          a1.user = u1
          in_(a1, u1.addresses)
  
  
  class JoinConditionErrorTest(fixtures.TestBase):
  
      def test_clauseelement_pj(self):
          from sqlalchemy.ext.declarative import declarative_base
          Base = declarative_base()
  
          class C1(Base):
              __tablename__ = 'c1'
              id = Column('id', Integer, primary_key=True)
  
          class C2(Base):
              __tablename__ = 'c2'
              id = Column('id', Integer, primary_key=True)
              c1id = Column('c1id', Integer, ForeignKey('c1.id'))
              c2 = relationship(C1, primaryjoin=C1.id)
  
          assert_raises(sa.exc.ArgumentError, configure_mappers)
  
      def test_clauseelement_pj_false(self):
          from sqlalchemy.ext.declarative import declarative_base
          Base = declarative_base()
  
          class C1(Base):
              __tablename__ = 'c1'
              id = Column('id', Integer, primary_key=True)
  
          class C2(Base):
              __tablename__ = 'c2'
              id = Column('id', Integer, primary_key=True)
              c1id = Column('c1id', Integer, ForeignKey('c1.id'))
              c2 = relationship(C1, primaryjoin="x" == "y")
  
          assert_raises(sa.exc.ArgumentError, configure_mappers)
  
      def test_only_column_elements(self):
          m = MetaData()
          t1 = Table('t1', m,
                     Column('id', Integer, primary_key=True),
                     Column('foo_id', Integer, ForeignKey('t2.id')),
                     )
          t2 = Table('t2', m,
                     Column('id', Integer, primary_key=True),
                     )
  
          class C1(object):
              pass
  
          class C2(object):
              pass
  
          mapper(C1, t1, properties={
              'c2': relationship(C2, primaryjoin=t1.join(t2))})
          mapper(C2, t2)
          assert_raises(sa.exc.ArgumentError, configure_mappers)
  
      def test_invalid_string_args(self):
          from sqlalchemy.ext.declarative import declarative_base
  
          for argname, arg in [
              ('remote_side', ['c1.id']),
              ('remote_side', ['id']),
              ('foreign_keys', ['c1id']),
              ('foreign_keys', ['C2.c1id']),
              ('order_by', ['id']),
          ]:
              clear_mappers()
              kw = {argname: arg}
              Base = declarative_base()
  
              class C1(Base):
                  __tablename__ = 'c1'
                  id = Column('id', Integer, primary_key=True)
  
              class C2(Base):
                  __tablename__ = 'c2'
                  id_ = Column('id', Integer, primary_key=True)
                  c1id = Column('c1id', Integer, ForeignKey('c1.id'))
                  c2 = relationship(C1, **kw)
  
              assert_raises_message(
                  sa.exc.ArgumentError,
                  "Column-based expression object expected "
                  "for argument '%s'; got: '%s', type %r" %
                  (argname, arg[0], type(arg[0])),
                  configure_mappers)
  
      def test_fk_error_not_raised_unrelated(self):
          m = MetaData()
          t1 = Table('t1', m,
                     Column('id', Integer, primary_key=True),
                     Column('foo_id', Integer, ForeignKey('t2.nonexistent_id')),
                     )
          t2 = Table('t2', m,  # noqa
                     Column('id', Integer, primary_key=True),
                     )
  
          t3 = Table('t3', m,
                     Column('id', Integer, primary_key=True),
                     Column('t1id', Integer, ForeignKey('t1.id'))
                     )
  
          class C1(object):
              pass
  
          class C2(object):
              pass
  
          mapper(C1, t1, properties={'c2': relationship(C2)})
          mapper(C2, t3)
          assert C1.c2.property.primaryjoin.compare(t1.c.id == t3.c.t1id)
  
      def test_join_error_raised(self):
          m = MetaData()
          t1 = Table('t1', m,
                     Column('id', Integer, primary_key=True),
                     )
          t2 = Table('t2', m,  # noqa
                     Column('id', Integer, primary_key=True),
                     )
  
          t3 = Table('t3', m,
                     Column('id', Integer, primary_key=True),
                     Column('t1id', Integer)
                     )
  
          class C1(object):
              pass
  
          class C2(object):
              pass
  
          mapper(C1, t1, properties={'c2': relationship(C2)})
          mapper(C2, t3)
  
          assert_raises(sa.exc.ArgumentError, configure_mappers)
  
      def teardown(self):
          clear_mappers()
  
  
  class TypeMatchTest(fixtures.MappedTest):
  
      """test errors raised when trying to add items
          whose type is not handled by a relationship"""
  
      @classmethod
      def define_tables(cls, metadata):
          Table("a", metadata,
                Column('aid', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('adata', String(30)))
          Table("b", metadata,
                Column('bid', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("a_id", Integer, ForeignKey("a.aid")),
                Column('bdata', String(30)))
          Table("c", metadata,
                Column('cid', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("b_id", Integer, ForeignKey("b.bid")),
                Column('cdata', String(30)))
          Table("d", metadata,
                Column('did', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column("a_id", Integer, ForeignKey("a.aid")),
                Column('ddata', String(30)))
  
      def test_o2m_oncascade(self):
          a, c, b = (self.tables.a,
                     self.tables.c,
                     self.tables.b)
  
          class A(fixtures.BasicEntity):
              pass
  
          class B(fixtures.BasicEntity):
              pass
  
          class C(fixtures.BasicEntity):
              pass
          mapper(A, a, properties={'bs': relationship(B)})
          mapper(B, b)
          mapper(C, c)
  
          a1 = A()
          b1 = B()
          c1 = C()
          a1.bs.append(b1)
          a1.bs.append(c1)
          sess = create_session()
          try:
              sess.add(a1)
              assert False
          except AssertionError as err:
              eq_(str(err),
                  "Attribute 'bs' on class '%s' doesn't handle "
                  "objects of type '%s'" % (A, C))
  
      def test_o2m_onflush(self):
          a, c, b = (self.tables.a,
                     self.tables.c,
                     self.tables.b)
  
          class A(fixtures.BasicEntity):
              pass
  
          class B(fixtures.BasicEntity):
              pass
  
          class C(fixtures.BasicEntity):
              pass
          mapper(A, a, properties={'bs': relationship(B, cascade="none")})
          mapper(B, b)
          mapper(C, c)
  
          a1 = A()
          b1 = B()
          c1 = C()
          a1.bs.append(b1)
          a1.bs.append(c1)
          sess = create_session()
          sess.add(a1)
          sess.add(b1)
          sess.add(c1)
          assert_raises_message(sa.orm.exc.FlushError,
                                "Attempting to flush an item",
                                sess.flush)
  
      def test_o2m_nopoly_onflush(self):
          a, c, b = (self.tables.a,
                     self.tables.c,
                     self.tables.b)
  
          class A(fixtures.BasicEntity):
              pass
  
          class B(fixtures.BasicEntity):
              pass
  
          class C(B):
              pass
          mapper(A, a, properties={'bs': relationship(B, cascade="none")})
          mapper(B, b)
          mapper(C, c, inherits=B)
  
          a1 = A()
          b1 = B()
          c1 = C()
          a1.bs.append(b1)
          a1.bs.append(c1)
          sess = create_session()
          sess.add(a1)
          sess.add(b1)
          sess.add(c1)
          assert_raises_message(sa.orm.exc.FlushError,
                                "Attempting to flush an item",
                                sess.flush)
  
      def test_m2o_nopoly_onflush(self):
          a, b, d = (self.tables.a,
                     self.tables.b,
                     self.tables.d)
  
          class A(fixtures.BasicEntity):
              pass
  
          class B(A):
              pass
  
          class D(fixtures.BasicEntity):
              pass
          mapper(A, a)
          mapper(B, b, inherits=A)
          mapper(D, d, properties={"a": relationship(A, cascade="none")})
          b1 = B()
          d1 = D()
          d1.a = b1
          sess = create_session()
          sess.add(b1)
          sess.add(d1)
          assert_raises_message(sa.orm.exc.FlushError,
                                "Attempting to flush an item",
                                sess.flush)
  
      def test_m2o_oncascade(self):
          a, b, d = (self.tables.a,
                     self.tables.b,
                     self.tables.d)
  
          class A(fixtures.BasicEntity):
              pass
  
          class B(fixtures.BasicEntity):
              pass
  
          class D(fixtures.BasicEntity):
              pass
          mapper(A, a)
          mapper(B, b)
          mapper(D, d, properties={"a": relationship(A)})
          b1 = B()
          d1 = D()
          d1.a = b1
          sess = create_session()
          assert_raises_message(AssertionError,
                                "doesn't handle objects of type",
                                sess.add, d1)
  
  
  class TypedAssociationTable(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          class MySpecialType(sa.types.TypeDecorator):
              impl = String
  
              def process_bind_param(self, value, dialect):
                  return "lala" + value
  
              def process_result_value(self, value, dialect):
                  return value[4:]
  
          Table('t1', metadata,
                Column('col1', MySpecialType(30), primary_key=True),
                Column('col2', String(30)))
          Table('t2', metadata,
                Column('col1', MySpecialType(30), primary_key=True),
                Column('col2', String(30)))
          Table('t3', metadata,
                Column('t1c1', MySpecialType(30), ForeignKey('t1.col1')),
                Column('t2c1', MySpecialType(30), ForeignKey('t2.col1')))
  
      def test_m2m(self):
          """Many-to-many tables with special types for candidate keys."""
  
          t2, t3, t1 = (self.tables.t2,
                        self.tables.t3,
                        self.tables.t1)
  
          class T1(fixtures.BasicEntity):
              pass
  
          class T2(fixtures.BasicEntity):
              pass
          mapper(T2, t2)
          mapper(T1, t1, properties={
              't2s': relationship(T2, secondary=t3, backref='t1s')})
  
          a = T1()
          a.col1 = "aid"
          b = T2()
          b.col1 = "bid"
          c = T2()
          c.col1 = "cid"
          a.t2s.append(b)
          a.t2s.append(c)
          sess = create_session()
          sess.add(a)
          sess.flush()
  
          eq_(select([func.count('*')]).select_from(t3).scalar(), 2)
  
          a.t2s.remove(c)
          sess.flush()
  
          eq_(select([func.count('*')]).select_from(t3).scalar(), 1)
  
  
  class CustomOperatorTest(fixtures.MappedTest, AssertsCompiledSQL):
  
      """test op() in conjunction with join conditions"""
  
      run_create_tables = run_deletes = None
  
      __dialect__ = 'default'
  
      @classmethod
      def define_tables(cls, metadata):
          Table('a', metadata,
                Column('id', Integer, primary_key=True),
                Column('foo', String(50))
                )
          Table('b', metadata,
                Column('id', Integer, primary_key=True),
                Column('foo', String(50))
                )
  
      def test_join_on_custom_op(self):
          class A(fixtures.BasicEntity):
              pass
  
          class B(fixtures.BasicEntity):
              pass
  
          mapper(A, self.tables.a, properties={
              'bs': relationship(B,
                                 primaryjoin=self.tables.a.c.foo.op(
                                     '&*', is_comparison=True
                                 )(foreign(self.tables.b.c.foo)),
                                 viewonly=True
                                 )
          })
          mapper(B, self.tables.b)
          self.assert_compile(
              Session().query(A).join(A.bs),
              "SELECT a.id AS a_id, a.foo AS a_foo "
              "FROM a JOIN b ON a.foo &* b.foo"
          )
  
  
  class ViewOnlyHistoryTest(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table("t1", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)))
          Table("t2", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)),
                Column('t1id', Integer, ForeignKey('t1.id')))
  
      def _assert_fk(self, a1, b1, is_set):
          s = Session(testing.db)
          s.add_all([a1, b1])
          s.flush()
  
          if is_set:
              eq_(b1.t1id, a1.id)
          else:
              eq_(b1.t1id, None)
  
          return s
  
      def test_o2m_viewonly_oneside(self):
          class A(fixtures.ComparableEntity):
              pass
  
          class B(fixtures.ComparableEntity):
              pass
  
          mapper(A, self.tables.t1, properties={
              "bs": relationship(B, viewonly=True,
                                 backref=backref("a", viewonly=False))
          })
          mapper(B, self.tables.t2)
  
          a1 = A()
          b1 = B()
          a1.bs.append(b1)
          assert b1.a is a1
          assert not inspect(a1).attrs.bs.history.has_changes()
          assert inspect(b1).attrs.a.history.has_changes()
  
          sess = self._assert_fk(a1, b1, True)
  
          a1.bs.remove(b1)
          assert a1 not in sess.dirty
          assert b1 in sess.dirty
  
      def test_m2o_viewonly_oneside(self):
          class A(fixtures.ComparableEntity):
              pass
  
          class B(fixtures.ComparableEntity):
              pass
  
          mapper(A, self.tables.t1, properties={
              "bs": relationship(B, viewonly=False,
                                 backref=backref("a", viewonly=True))
          })
          mapper(B, self.tables.t2)
  
          a1 = A()
          b1 = B()
          b1.a = a1
          assert b1 in a1.bs
          assert inspect(a1).attrs.bs.history.has_changes()
          assert not inspect(b1).attrs.a.history.has_changes()
  
          sess = self._assert_fk(a1, b1, True)
  
          a1.bs.remove(b1)
          assert a1 in sess.dirty
          assert b1 not in sess.dirty
  
      def test_o2m_viewonly_only(self):
          class A(fixtures.ComparableEntity):
              pass
  
          class B(fixtures.ComparableEntity):
              pass
  
          mapper(A, self.tables.t1, properties={
              "bs": relationship(B, viewonly=True)
          })
          mapper(B, self.tables.t2)
  
          a1 = A()
          b1 = B()
          a1.bs.append(b1)
          assert not inspect(a1).attrs.bs.history.has_changes()
  
          self._assert_fk(a1, b1, False)
  
      def test_m2o_viewonly_only(self):
          class A(fixtures.ComparableEntity):
              pass
  
          class B(fixtures.ComparableEntity):
              pass
  
          mapper(A, self.tables.t1)
          mapper(B, self.tables.t2, properties={
              'a': relationship(A, viewonly=True)
          })
  
          a1 = A()
          b1 = B()
          b1.a = a1
          assert not inspect(b1).attrs.a.history.has_changes()
  
          self._assert_fk(a1, b1, False)
  
  
  class ViewOnlyM2MBackrefTest(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table("t1", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)))
          Table("t2", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)),
                )
          Table("t1t2", metadata,
                Column('t1id', Integer, ForeignKey('t1.id'), primary_key=True),
                Column('t2id', Integer, ForeignKey('t2.id'), primary_key=True),
                )
  
      def test_viewonly(self):
          t1t2, t2, t1 = (self.tables.t1t2,
                          self.tables.t2,
                          self.tables.t1)
  
          class A(fixtures.ComparableEntity):
              pass
  
          class B(fixtures.ComparableEntity):
              pass
  
          mapper(A, t1, properties={
              'bs': relationship(B, secondary=t1t2,
                                 backref=backref('as_', viewonly=True))
          })
          mapper(B, t2)
  
          sess = create_session()
          a1 = A()
          b1 = B(as_=[a1])
  
          assert not inspect(b1).attrs.as_.history.has_changes()
  
          sess.add(a1)
          sess.flush()
          eq_(
              sess.query(A).first(), A(bs=[B(id=b1.id)])
          )
          eq_(
              sess.query(B).first(), B(as_=[A(id=a1.id)])
          )
  
  
  class ViewOnlyOverlappingNames(fixtures.MappedTest):
  
      """'viewonly' mappings with overlapping PK column names."""
  
      @classmethod
      def define_tables(cls, metadata):
          Table("t1", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)))
          Table("t2", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)),
                Column('t1id', Integer, ForeignKey('t1.id')))
          Table("t3", metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)),
                Column('t2id', Integer, ForeignKey('t2.id')))
  
      def test_three_table_view(self):
          """A three table join with overlapping PK names.
  
          A third table is pulled into the primary join condition using
          overlapping PK column names and should not produce 'conflicting column'
          error.
  
          """
  
          t2, t3, t1 = (self.tables.t2,
                        self.tables.t3,
                        self.tables.t1)
  
          class C1(fixtures.BasicEntity):
              pass
  
          class C2(fixtures.BasicEntity):
              pass
  
          class C3(fixtures.BasicEntity):
              pass
  
          mapper(C1, t1, properties={
              't2s': relationship(C2),
              't2_view': relationship(
                  C2,
                  viewonly=True,
                  primaryjoin=sa.and_(t1.c.id == t2.c.t1id,
                                      t3.c.t2id == t2.c.id,
                                      t3.c.data == t1.c.data))})
          mapper(C2, t2)
          mapper(C3, t3, properties={
              't2': relationship(C2)})
  
          c1 = C1()
          c1.data = 'c1data'
          c2a = C2()
          c1.t2s.append(c2a)
          c2b = C2()
          c1.t2s.append(c2b)
          c3 = C3()
          c3.data = 'c1data'
          c3.t2 = c2b
          sess = create_session()
          sess.add(c1)
          sess.add(c3)
          sess.flush()
          sess.expunge_all()
  
          c1 = sess.query(C1).get(c1.id)
          assert set([x.id for x in c1.t2s]) == set([c2a.id, c2b.id])
          assert set([x.id for x in c1.t2_view]) == set([c2b.id])
  
  
  class ViewOnlyUniqueNames(fixtures.MappedTest):
  
      """'viewonly' mappings with unique PK column names."""
  
      @classmethod
      def define_tables(cls, metadata):
          Table("t1", metadata,
                Column('t1id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)))
          Table("t2", metadata,
                Column('t2id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)),
                Column('t1id_ref', Integer, ForeignKey('t1.t1id')))
          Table("t3", metadata,
                Column('t3id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(40)),
                Column('t2id_ref', Integer, ForeignKey('t2.t2id')))
  
      def test_three_table_view(self):
          """A three table join with overlapping PK names.
  
          A third table is pulled into the primary join condition using unique
          PK column names and should not produce 'mapper has no columnX' error.
  
          """
  
          t2, t3, t1 = (self.tables.t2,
                        self.tables.t3,
                        self.tables.t1)
  
          class C1(fixtures.BasicEntity):
              pass
  
          class C2(fixtures.BasicEntity):
              pass
  
          class C3(fixtures.BasicEntity):
              pass
  
          mapper(C1, t1, properties={
              't2s': relationship(C2),
              't2_view': relationship(
                  C2,
                  viewonly=True,
                  primaryjoin=sa.and_(t1.c.t1id == t2.c.t1id_ref,
                                      t3.c.t2id_ref == t2.c.t2id,
                                      t3.c.data == t1.c.data))})
          mapper(C2, t2)
          mapper(C3, t3, properties={
              't2': relationship(C2)})
  
          c1 = C1()
          c1.data = 'c1data'
          c2a = C2()
          c1.t2s.append(c2a)
          c2b = C2()
          c1.t2s.append(c2b)
          c3 = C3()
          c3.data = 'c1data'
          c3.t2 = c2b
          sess = create_session()
  
          sess.add_all((c1, c3))
          sess.flush()
          sess.expunge_all()
  
          c1 = sess.query(C1).get(c1.t1id)
          assert set([x.t2id for x in c1.t2s]) == set([c2a.t2id, c2b.t2id])
          assert set([x.t2id for x in c1.t2_view]) == set([c2b.t2id])
  
  
  class ViewOnlyLocalRemoteM2M(fixtures.TestBase):
  
      """test that local-remote is correctly determined for m2m"""
  
      def test_local_remote(self):
          meta = MetaData()
  
          t1 = Table('t1', meta,
                     Column('id', Integer, primary_key=True),
                     )
          t2 = Table('t2', meta,
                     Column('id', Integer, primary_key=True),
                     )
          t12 = Table('tab', meta,
                      Column('t1_id', Integer, ForeignKey('t1.id',)),
                      Column('t2_id', Integer, ForeignKey('t2.id',)),
                      )
  
          class A(object):
              pass
  
          class B(object):
              pass
          mapper(B, t2, )
          m = mapper(A, t1, properties=dict(
              b_view=relationship(B, secondary=t12, viewonly=True),
              b_plain=relationship(B, secondary=t12),
          )
          )
          configure_mappers()
          assert m.get_property('b_view').local_remote_pairs == \
              m.get_property('b_plain').local_remote_pairs == \
              [(t1.c.id, t12.c.t1_id), (t2.c.id, t12.c.t2_id)]
  
  
  class ViewOnlyNonEquijoin(fixtures.MappedTest):
  
      """'viewonly' mappings based on non-equijoins."""
  
      @classmethod
      def define_tables(cls, metadata):
          Table('foos', metadata,
                Column('id', Integer, primary_key=True))
          Table('bars', metadata,
                Column('id', Integer, primary_key=True),
                Column('fid', Integer))
  
      def test_viewonly_join(self):
          bars, foos = self.tables.bars, self.tables.foos
  
          class Foo(fixtures.ComparableEntity):
              pass
  
          class Bar(fixtures.ComparableEntity):
              pass
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id > bars.c.fid,
                                   foreign_keys=[bars.c.fid],
                                   viewonly=True)})
  
          mapper(Bar, bars)
  
          sess = create_session()
          sess.add_all((Foo(id=4),
                        Foo(id=9),
                        Bar(id=1, fid=2),
                        Bar(id=2, fid=3),
                        Bar(id=3, fid=6),
                        Bar(id=4, fid=7)))
          sess.flush()
  
          sess = create_session()
          eq_(sess.query(Foo).filter_by(id=4).one(),
              Foo(id=4, bars=[Bar(fid=2), Bar(fid=3)]))
          eq_(sess.query(Foo).filter_by(id=9).one(),
              Foo(id=9, bars=[Bar(fid=2), Bar(fid=3), Bar(fid=6), Bar(fid=7)]))
  
  
  class ViewOnlyRepeatedRemoteColumn(fixtures.MappedTest):
  
      """'viewonly' mappings that contain the same 'remote' column twice"""
  
      @classmethod
      def define_tables(cls, metadata):
          Table('foos', metadata,
                Column(
                    'id', Integer, primary_key=True,
                    test_needs_autoincrement=True),
                Column('bid1', Integer, ForeignKey('bars.id')),
                Column('bid2', Integer, ForeignKey('bars.id')))
  
          Table('bars', metadata,
                Column(
                    'id', Integer, primary_key=True,
                    test_needs_autoincrement=True),
                Column('data', String(50)))
  
      def test_relationship_on_or(self):
          bars, foos = self.tables.bars, self.tables.foos
  
          class Foo(fixtures.ComparableEntity):
              pass
  
          class Bar(fixtures.ComparableEntity):
              pass
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=sa.or_(bars.c.id == foos.c.bid1,
                                                      bars.c.id == foos.c.bid2),
                                   uselist=True,
                                   viewonly=True)})
          mapper(Bar, bars)
  
          sess = create_session()
          b1 = Bar(id=1, data='b1')
          b2 = Bar(id=2, data='b2')
          b3 = Bar(id=3, data='b3')
          f1 = Foo(bid1=1, bid2=2)
          f2 = Foo(bid1=3, bid2=None)
  
          sess.add_all((b1, b2, b3))
          sess.flush()
  
          sess.add_all((f1, f2))
          sess.flush()
  
          sess.expunge_all()
          eq_(sess.query(Foo).filter_by(id=f1.id).one(),
              Foo(bars=[Bar(data='b1'), Bar(data='b2')]))
          eq_(sess.query(Foo).filter_by(id=f2.id).one(),
              Foo(bars=[Bar(data='b3')]))
  
  
  class ViewOnlyRepeatedLocalColumn(fixtures.MappedTest):
  
      """'viewonly' mappings that contain the same 'local' column twice"""
  
      @classmethod
      def define_tables(cls, metadata):
          Table('foos', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(50)))
  
          Table('bars', metadata, Column('id', Integer, primary_key=True,
                                         test_needs_autoincrement=True),
                Column('fid1', Integer, ForeignKey('foos.id')),
                Column('fid2', Integer, ForeignKey('foos.id')),
                Column('data', String(50)))
  
      def test_relationship_on_or(self):
          bars, foos = self.tables.bars, self.tables.foos
  
          class Foo(fixtures.ComparableEntity):
              pass
  
          class Bar(fixtures.ComparableEntity):
              pass
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=sa.or_(bars.c.fid1 == foos.c.id,
                                                      bars.c.fid2 == foos.c.id),
                                   viewonly=True)})
          mapper(Bar, bars)
  
          sess = create_session()
          f1 = Foo(id=1, data='f1')
          f2 = Foo(id=2, data='f2')
          b1 = Bar(fid1=1, data='b1')
          b2 = Bar(fid2=1, data='b2')
          b3 = Bar(fid1=2, data='b3')
          b4 = Bar(fid1=1, fid2=2, data='b4')
  
          sess.add_all((f1, f2))
          sess.flush()
  
          sess.add_all((b1, b2, b3, b4))
          sess.flush()
  
          sess.expunge_all()
          eq_(sess.query(Foo).filter_by(id=f1.id).one(),
              Foo(bars=[Bar(data='b1'), Bar(data='b2'), Bar(data='b4')]))
          eq_(sess.query(Foo).filter_by(id=f2.id).one(),
              Foo(bars=[Bar(data='b3'), Bar(data='b4')]))
  
  
  class ViewOnlyComplexJoin(_RelationshipErrors, fixtures.MappedTest):
  
      """'viewonly' mappings with a complex join condition."""
  
      @classmethod
      def define_tables(cls, metadata):
          Table('t1', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(50)))
          Table('t2', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(50)),
                Column('t1id', Integer, ForeignKey('t1.id')))
          Table('t3', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(50)))
          Table('t2tot3', metadata,
                Column('t2id', Integer, ForeignKey('t2.id')),
                Column('t3id', Integer, ForeignKey('t3.id')))
  
      @classmethod
      def setup_classes(cls):
          class T1(cls.Comparable):
              pass
  
          class T2(cls.Comparable):
              pass
  
          class T3(cls.Comparable):
              pass
  
      def test_basic(self):
          T1, t2, T2, T3, t3, t2tot3, t1 = (self.classes.T1,
                                            self.tables.t2,
                                            self.classes.T2,
                                            self.classes.T3,
                                            self.tables.t3,
                                            self.tables.t2tot3,
                                            self.tables.t1)
  
          mapper(T1, t1, properties={
              't3s': relationship(T3, primaryjoin=sa.and_(
                  t1.c.id == t2.c.t1id,
                  t2.c.id == t2tot3.c.t2id,
                  t3.c.id == t2tot3.c.t3id),
                  viewonly=True,
                  foreign_keys=t3.c.id, remote_side=t2.c.t1id)
          })
          mapper(T2, t2, properties={
              't1': relationship(T1),
              't3s': relationship(T3, secondary=t2tot3)
          })
          mapper(T3, t3)
  
          sess = create_session()
          sess.add(T2(data='t2', t1=T1(data='t1'), t3s=[T3(data='t3')]))
          sess.flush()
          sess.expunge_all()
  
          a = sess.query(T1).first()
          eq_(a.t3s, [T3(data='t3')])
  
      def test_remote_side_escalation(self):
          T1, t2, T2, T3, t3, t2tot3, t1 = (self.classes.T1,
                                            self.tables.t2,
                                            self.classes.T2,
                                            self.classes.T3,
                                            self.tables.t3,
                                            self.tables.t2tot3,
                                            self.tables.t1)
  
          mapper(T1, t1, properties={
              't3s': relationship(T3,
                                  primaryjoin=sa.and_(t1.c.id == t2.c.t1id,
                                                      t2.c.id == t2tot3.c.t2id,
                                                      t3.c.id == t2tot3.c.t3id
                                                      ),
                                  viewonly=True,
                                  foreign_keys=t3.c.id)})
          mapper(T2, t2, properties={
              't1': relationship(T1),
              't3s': relationship(T3, secondary=t2tot3)})
          mapper(T3, t3)
          self._assert_raises_no_local_remote(configure_mappers, "T1.t3s")
  
  
  class RemoteForeignBetweenColsTest(fixtures.DeclarativeMappedTest):
  
      """test a complex annotation using between().
  
      Using declarative here as an integration test for the local()
      and remote() annotations in conjunction with already annotated
      instrumented attributes, etc.
  
      """
      @classmethod
      def setup_classes(cls):
          Base = cls.DeclarativeBasic
  
          class Network(fixtures.ComparableEntity, Base):
              __tablename__ = "network"
  
              id = Column(sa.Integer, primary_key=True,
                          test_needs_autoincrement=True)
              ip_net_addr = Column(Integer)
              ip_broadcast_addr = Column(Integer)
  
              addresses = relationship(
                  "Address",
                  primaryjoin="remote(foreign(Address.ip_addr)).between("
                  "Network.ip_net_addr,"
                  "Network.ip_broadcast_addr)",
                  viewonly=True
              )
  
          class Address(fixtures.ComparableEntity, Base):
              __tablename__ = "address"
  
              ip_addr = Column(Integer, primary_key=True)
  
      @classmethod
      def insert_data(cls):
          Network, Address = cls.classes.Network, cls.classes.Address
          s = Session(testing.db)
  
          s.add_all([
              Network(ip_net_addr=5, ip_broadcast_addr=10),
              Network(ip_net_addr=15, ip_broadcast_addr=25),
              Network(ip_net_addr=30, ip_broadcast_addr=35),
              Address(ip_addr=17), Address(ip_addr=18), Address(ip_addr=9),
              Address(ip_addr=27)
          ])
          s.commit()
  
      def test_col_query(self):
          Network, Address = self.classes.Network, self.classes.Address
  
          session = Session(testing.db)
          eq_(
              session.query(Address.ip_addr).
              select_from(Network).
              join(Network.addresses).
              filter(Network.ip_net_addr == 15).
              all(),
              [(17, ), (18, )]
          )
  
      def test_lazyload(self):
          Network, Address = self.classes.Network, self.classes.Address
  
          session = Session(testing.db)
  
          n3 = session.query(Network).filter(Network.ip_net_addr == 5).one()
          eq_([a.ip_addr for a in n3.addresses], [9])
  
  
  class ExplicitLocalRemoteTest(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table('t1', metadata,
                Column('id', String(50), primary_key=True),
                Column('data', String(50)))
          Table('t2', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('data', String(50)),
                Column('t1id', String(50)))
  
      @classmethod
      def setup_classes(cls):
          class T1(cls.Comparable):
              pass
  
          class T2(cls.Comparable):
              pass
  
      def test_onetomany_funcfk_oldstyle(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          # old _local_remote_pairs
          mapper(T1, t1, properties={
              't2s': relationship(
                  T2,
                  primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
                  _local_remote_pairs=[(t1.c.id, t2.c.t1id)],
                  foreign_keys=[t2.c.t1id]
              )
          })
          mapper(T2, t2)
          self._test_onetomany()
  
      def test_onetomany_funcfk_annotated(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          # use annotation
          mapper(T1, t1, properties={
              't2s': relationship(T2,
                                  primaryjoin=t1.c.id ==
                                  foreign(sa.func.lower(t2.c.t1id)),
                                  )})
          mapper(T2, t2)
          self._test_onetomany()
  
      def _test_onetomany(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
          is_(T1.t2s.property.direction, ONETOMANY)
          eq_(T1.t2s.property.local_remote_pairs, [(t1.c.id, t2.c.t1id)])
          sess = create_session()
          a1 = T1(id='number1', data='a1')
          a2 = T1(id='number2', data='a2')
          b1 = T2(data='b1', t1id='NuMbEr1')
          b2 = T2(data='b2', t1id='Number1')
          b3 = T2(data='b3', t1id='Number2')
          sess.add_all((a1, a2, b1, b2, b3))
          sess.flush()
          sess.expunge_all()
  
          eq_(sess.query(T1).first(),
              T1(id='number1', data='a1', t2s=[
                 T2(data='b1', t1id='NuMbEr1'),
                 T2(data='b2', t1id='Number1')]))
  
      def test_manytoone_funcfk(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          mapper(T1, t1)
          mapper(T2, t2, properties={
              't1': relationship(T1,
                                 primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
                                 _local_remote_pairs=[(t2.c.t1id, t1.c.id)],
                                 foreign_keys=[t2.c.t1id],
                                 uselist=True)})
  
          sess = create_session()
          a1 = T1(id='number1', data='a1')
          a2 = T1(id='number2', data='a2')
          b1 = T2(data='b1', t1id='NuMbEr1')
          b2 = T2(data='b2', t1id='Number1')
          b3 = T2(data='b3', t1id='Number2')
          sess.add_all((a1, a2, b1, b2, b3))
          sess.flush()
          sess.expunge_all()
  
          eq_(sess.query(T2).filter(T2.data.in_(['b1', 'b2'])).all(),
              [T2(data='b1', t1=[T1(id='number1', data='a1')]),
               T2(data='b2', t1=[T1(id='number1', data='a1')])])
  
      def test_onetomany_func_referent(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          mapper(T1, t1, properties={
              't2s': relationship(
                  T2,
                  primaryjoin=sa.func.lower(t1.c.id) == t2.c.t1id,
                  _local_remote_pairs=[(t1.c.id, t2.c.t1id)],
                  foreign_keys=[t2.c.t1id])})
          mapper(T2, t2)
  
          sess = create_session()
          a1 = T1(id='NuMbeR1', data='a1')
          a2 = T1(id='NuMbeR2', data='a2')
          b1 = T2(data='b1', t1id='number1')
          b2 = T2(data='b2', t1id='number1')
          b3 = T2(data='b2', t1id='number2')
          sess.add_all((a1, a2, b1, b2, b3))
          sess.flush()
          sess.expunge_all()
  
          eq_(sess.query(T1).first(),
              T1(id='NuMbeR1', data='a1', t2s=[
                  T2(data='b1', t1id='number1'),
                  T2(data='b2', t1id='number1')]))
  
      def test_manytoone_func_referent(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          mapper(T1, t1)
          mapper(T2, t2, properties={
              't1': relationship(T1,
                                 primaryjoin=sa.func.lower(t1.c.id) == t2.c.t1id,
                                 _local_remote_pairs=[(t2.c.t1id, t1.c.id)],
                                 foreign_keys=[t2.c.t1id], uselist=True)})
  
          sess = create_session()
          a1 = T1(id='NuMbeR1', data='a1')
          a2 = T1(id='NuMbeR2', data='a2')
          b1 = T2(data='b1', t1id='number1')
          b2 = T2(data='b2', t1id='number1')
          b3 = T2(data='b3', t1id='number2')
          sess.add_all((a1, a2, b1, b2, b3))
          sess.flush()
          sess.expunge_all()
  
          eq_(sess.query(T2).filter(T2.data.in_(['b1', 'b2'])).all(),
              [T2(data='b1', t1=[T1(id='NuMbeR1', data='a1')]),
               T2(data='b2', t1=[T1(id='NuMbeR1', data='a1')])])
  
      def test_escalation_1(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          mapper(T1, t1, properties={
              't2s': relationship(
                  T2,
                  primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
                  _local_remote_pairs=[(t1.c.id, t2.c.t1id)],
                  foreign_keys=[t2.c.t1id],
                  remote_side=[t2.c.t1id])})
          mapper(T2, t2)
          assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
  
      def test_escalation_2(self):
          T2, T1, t2, t1 = (self.classes.T2,
                            self.classes.T1,
                            self.tables.t2,
                            self.tables.t1)
  
          mapper(T1, t1, properties={
              't2s': relationship(
                  T2,
                  primaryjoin=t1.c.id == sa.func.lower(t2.c.t1id),
                  _local_remote_pairs=[(t1.c.id, t2.c.t1id)])})
          mapper(T2, t2)
          assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
  
  
  class InvalidRemoteSideTest(fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table('t1', metadata,
                Column('id', Integer, primary_key=True),
                Column('data', String(50)),
                Column('t_id', Integer, ForeignKey('t1.id'))
                )
  
      @classmethod
      def setup_classes(cls):
          class T1(cls.Comparable):
              pass
  
      def test_o2m_backref(self):
          T1, t1 = self.classes.T1, self.tables.t1
  
          mapper(T1, t1, properties={
              't1s': relationship(T1, backref='parent')
          })
  
          assert_raises_message(
              sa.exc.ArgumentError,
              "T1.t1s and back-reference T1.parent are "
              r"both of the same direction symbol\('ONETOMANY'\).  Did you "
              "mean to set remote_side on the many-to-one side ?",
              configure_mappers)
  
      def test_m2o_backref(self):
          T1, t1 = self.classes.T1, self.tables.t1
  
          mapper(T1, t1, properties={
              't1s': relationship(T1,
                                  backref=backref('parent', remote_side=t1.c.id),
                                  remote_side=t1.c.id)
          })
  
          assert_raises_message(
              sa.exc.ArgumentError,
              "T1.t1s and back-reference T1.parent are "
              r"both of the same direction symbol\('MANYTOONE'\).  Did you "
              "mean to set remote_side on the many-to-one side ?",
              configure_mappers)
  
      def test_o2m_explicit(self):
          T1, t1 = self.classes.T1, self.tables.t1
  
          mapper(T1, t1, properties={
              't1s': relationship(T1, back_populates='parent'),
              'parent': relationship(T1, back_populates='t1s'),
          })
  
          # can't be sure of ordering here
          assert_raises_message(
              sa.exc.ArgumentError,
              r"both of the same direction symbol\('ONETOMANY'\).  Did you "
              "mean to set remote_side on the many-to-one side ?",
              configure_mappers)
  
      def test_m2o_explicit(self):
          T1, t1 = self.classes.T1, self.tables.t1
  
          mapper(T1, t1, properties={
              't1s': relationship(T1, back_populates='parent',
                                  remote_side=t1.c.id),
              'parent': relationship(T1, back_populates='t1s',
                                     remote_side=t1.c.id)
          })
  
          # can't be sure of ordering here
          assert_raises_message(
              sa.exc.ArgumentError,
              r"both of the same direction symbol\('MANYTOONE'\).  Did you "
              "mean to set remote_side on the many-to-one side ?",
              configure_mappers)
  
  
  class AmbiguousFKResolutionTest(_RelationshipErrors, fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table("a", metadata,
                Column('id', Integer, primary_key=True)
                )
          Table("b", metadata,
                Column('id', Integer, primary_key=True),
                Column('aid_1', Integer, ForeignKey('a.id')),
                Column('aid_2', Integer, ForeignKey('a.id')),
                )
          Table("atob", metadata,
                Column('aid', Integer),
                Column('bid', Integer),
                )
          Table("atob_ambiguous", metadata,
                Column('aid1', Integer, ForeignKey('a.id')),
                Column('bid1', Integer, ForeignKey('b.id')),
                Column('aid2', Integer, ForeignKey('a.id')),
                Column('bid2', Integer, ForeignKey('b.id')),
                )
  
      @classmethod
      def setup_classes(cls):
          class A(cls.Basic):
              pass
  
          class B(cls.Basic):
              pass
  
      def test_ambiguous_fks_o2m(self):
          A, B = self.classes.A, self.classes.B
          a, b = self.tables.a, self.tables.b
          mapper(A, a, properties={
              'bs': relationship(B)
          })
          mapper(B, b)
          self._assert_raises_ambig_join(
              configure_mappers,
              "A.bs",
              None
          )
  
      def test_with_fks_o2m(self):
          A, B = self.classes.A, self.classes.B
          a, b = self.tables.a, self.tables.b
          mapper(A, a, properties={
              'bs': relationship(B, foreign_keys=b.c.aid_1)
          })
          mapper(B, b)
          sa.orm.configure_mappers()
          assert A.bs.property.primaryjoin.compare(
              a.c.id == b.c.aid_1
          )
          eq_(
              A.bs.property._calculated_foreign_keys,
              set([b.c.aid_1])
          )
  
      def test_with_pj_o2m(self):
          A, B = self.classes.A, self.classes.B
          a, b = self.tables.a, self.tables.b
          mapper(A, a, properties={
              'bs': relationship(B, primaryjoin=a.c.id == b.c.aid_1)
          })
          mapper(B, b)
          sa.orm.configure_mappers()
          assert A.bs.property.primaryjoin.compare(
              a.c.id == b.c.aid_1
          )
          eq_(
              A.bs.property._calculated_foreign_keys,
              set([b.c.aid_1])
          )
  
      def test_with_annotated_pj_o2m(self):
          A, B = self.classes.A, self.classes.B
          a, b = self.tables.a, self.tables.b
          mapper(A, a, properties={
              'bs': relationship(B, primaryjoin=a.c.id == foreign(b.c.aid_1))
          })
          mapper(B, b)
          sa.orm.configure_mappers()
          assert A.bs.property.primaryjoin.compare(
              a.c.id == b.c.aid_1
          )
          eq_(
              A.bs.property._calculated_foreign_keys,
              set([b.c.aid_1])
          )
  
      def test_no_fks_m2m(self):
          A, B = self.classes.A, self.classes.B
          a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob
          mapper(A, a, properties={
              'bs': relationship(B, secondary=a_to_b)
          })
          mapper(B, b)
          self._assert_raises_no_join(
              sa.orm.configure_mappers,
              "A.bs", a_to_b,
          )
  
      def test_ambiguous_fks_m2m(self):
          A, B = self.classes.A, self.classes.B
          a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob_ambiguous
          mapper(A, a, properties={
              'bs': relationship(B, secondary=a_to_b)
          })
          mapper(B, b)
  
          self._assert_raises_ambig_join(
              configure_mappers,
              "A.bs",
              "atob_ambiguous"
          )
  
      def test_with_fks_m2m(self):
          A, B = self.classes.A, self.classes.B
          a, b, a_to_b = self.tables.a, self.tables.b, self.tables.atob_ambiguous
          mapper(A, a, properties={
              'bs': relationship(B, secondary=a_to_b,
                                 foreign_keys=[a_to_b.c.aid1, a_to_b.c.bid1])
          })
          mapper(B, b)
          sa.orm.configure_mappers()
  
  
  class SecondaryNestedJoinTest(fixtures.MappedTest, AssertsCompiledSQL,
                                testing.AssertsExecutionResults):
  
      """test support for a relationship where the 'secondary' table is a
      compound join().
  
      join() and joinedload() should use a "flat" alias, lazyloading needs
      to ensure the join renders.
  
      """
      run_setup_mappers = 'once'
      run_inserts = 'once'
      run_deletes = None
  
      @classmethod
      def define_tables(cls, metadata):
          Table(
              'a', metadata,
              Column(
                  'id', Integer, primary_key=True,
                  test_needs_autoincrement=True),
              Column('name', String(30)),
              Column('b_id', ForeignKey('b.id'))
          )
          Table('b', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('name', String(30)),
                Column('d_id', ForeignKey('d.id'))
                )
          Table('c', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('name', String(30)),
                Column('a_id', ForeignKey('a.id')),
                Column('d_id', ForeignKey('d.id'))
                )
          Table('d', metadata,
                Column('id', Integer, primary_key=True,
                       test_needs_autoincrement=True),
                Column('name', String(30)),
                )
  
      @classmethod
      def setup_classes(cls):
          class A(cls.Comparable):
              pass
  
          class B(cls.Comparable):
              pass
  
          class C(cls.Comparable):
              pass
  
          class D(cls.Comparable):
              pass
  
      @classmethod
      def setup_mappers(cls):
          A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
          a, b, c, d = cls.tables.a, cls.tables.b, cls.tables.c, cls.tables.d
          j = sa.join(b, d, b.c.d_id == d.c.id).join(c, c.c.d_id == d.c.id)
          # j = join(b, d, b.c.d_id == d.c.id).join(c, c.c.d_id == d.c.id) \
          # .alias()
          mapper(A, a, properties={
              "b": relationship(B),
              "d": relationship(
                  D, secondary=j,
                  primaryjoin=and_(a.c.b_id == b.c.id, a.c.id == c.c.a_id),
                  secondaryjoin=d.c.id == b.c.d_id,
                  # primaryjoin=and_(a.c.b_id == j.c.b_id, a.c.id == j.c.c_a_id),
                  # secondaryjoin=d.c.id == j.c.b_d_id,
                  uselist=False,
                  viewonly=True
              )
          })
          mapper(B, b, properties={
              "d": relationship(D)
          })
          mapper(C, c, properties={
              "a": relationship(A),
              "d": relationship(D)
          })
          mapper(D, d)
  
      @classmethod
      def insert_data(cls):
          A, B, C, D = cls.classes.A, cls.classes.B, cls.classes.C, cls.classes.D
          sess = Session()
          a1, a2, a3, a4 = A(name='a1'), A(name='a2'), A(name='a3'), A(name='a4')
          b1, b2, b3, b4 = B(name='b1'), B(name='b2'), B(name='b3'), B(name='b4')
          c1, c2, c3, c4 = C(name='c1'), C(name='c2'), C(name='c3'), C(name='c4')
          d1, d2 = D(name='d1'), D(name='d2')
  
          a1.b = b1
          a2.b = b2
          a3.b = b3
          a4.b = b4
  
          c1.a = a1
          c2.a = a2
          c3.a = a2
          c4.a = a4
  
          c1.d = d1
          c2.d = d2
          c3.d = d1
          c4.d = d2
  
          b1.d = d1
          b2.d = d1
          b3.d = d2
          b4.d = d2
  
          sess.add_all([a1, a2, a3, a4, b1, b2, b3, b4, c1, c2, c4, c4, d1, d2])
          sess.commit()
  
      def test_render_join(self):
          A, D = self.classes.A, self.classes.D
          sess = Session()
          self.assert_compile(
              sess.query(A).join(A.d),
              "SELECT a.id AS a_id, a.name AS a_name, a.b_id AS a_b_id "
              "FROM a JOIN (b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id "
              "JOIN c AS c_1 ON c_1.d_id = d_1.id) ON a.b_id = b_1.id "
              "AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id",
              dialect="postgresql"
          )
  
      def test_render_joinedload(self):
          A, D = self.classes.A, self.classes.D
          sess = Session()
          self.assert_compile(
              sess.query(A).options(joinedload(A.d)),
              "SELECT a.id AS a_id, a.name AS a_name, a.b_id AS a_b_id, "
              "d_1.id AS d_1_id, d_1.name AS d_1_name FROM a LEFT OUTER JOIN "
              "(b AS b_1 JOIN d AS d_2 ON b_1.d_id = d_2.id JOIN c AS c_1 "
              "ON c_1.d_id = d_2.id JOIN d AS d_1 ON d_1.id = b_1.d_id) "
              "ON a.b_id = b_1.id AND a.id = c_1.a_id",
              dialect="postgresql"
          )
  
      def test_render_lazyload(self):
          from sqlalchemy.testing.assertsql import CompiledSQL
  
          A, D = self.classes.A, self.classes.D
          sess = Session()
          a1 = sess.query(A).filter(A.name == 'a1').first()
  
          def go():
              a1.d
  
          # here, the "lazy" strategy has to ensure the "secondary"
          # table is part of the "select_from()", since it's a join().
          # referring to just the columns wont actually render all those
          # join conditions.
          self.assert_sql_execution(
              testing.db,
              go,
              CompiledSQL(
                  "SELECT d.id AS d_id, d.name AS d_name FROM b "
                  "JOIN d ON b.d_id = d.id JOIN c ON c.d_id = d.id "
                  "WHERE :param_1 = b.id AND :param_2 = c.a_id "
                  "AND d.id = b.d_id",
                  {'param_1': a1.id, 'param_2': a1.id}
              )
          )
  
      mapping = {
          "a1": "d1",
          "a2": None,
          "a3": None,
          "a4": "d2"
      }
  
      def test_join(self):
          A, D = self.classes.A, self.classes.D
          sess = Session()
  
          for a, d in sess.query(A, D).outerjoin(A.d):
              eq_(self.mapping[a.name], d.name if d is not None else None)
  
      def test_joinedload(self):
          A, D = self.classes.A, self.classes.D
          sess = Session()
  
          for a in sess.query(A).options(joinedload(A.d)):
              d = a.d
              eq_(self.mapping[a.name], d.name if d is not None else None)
  
      def test_lazyload(self):
          A, D = self.classes.A, self.classes.D
          sess = Session()
  
          for a in sess.query(A):
              d = a.d
              eq_(self.mapping[a.name], d.name if d is not None else None)
  
  
  class InvalidRelationshipEscalationTest(
          _RelationshipErrors, fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table('foos', metadata,
                Column('id', Integer, primary_key=True),
                Column('fid', Integer))
          Table('bars', metadata,
                Column('id', Integer, primary_key=True),
                Column('fid', Integer))
  
          Table('foos_with_fks', metadata,
                Column('id', Integer, primary_key=True),
                Column('fid', Integer, ForeignKey('foos_with_fks.id')))
          Table('bars_with_fks', metadata,
                Column('id', Integer, primary_key=True),
                Column('fid', Integer, ForeignKey('foos_with_fks.id')))
  
      @classmethod
      def setup_classes(cls):
          class Foo(cls.Basic):
              pass
  
          class Bar(cls.Basic):
              pass
  
      def test_no_join(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar)})
          mapper(Bar, bars)
  
          self._assert_raises_no_join(sa.orm.configure_mappers,
                                      "Foo.bars", None
                                      )
  
      def test_no_join_self_ref(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo)})
          mapper(Bar, bars)
  
          self._assert_raises_no_join(
              configure_mappers,
              "Foo.foos",
              None
          )
  
      def test_no_equated(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id > bars.c.fid)})
          mapper(Bar, bars)
  
          self._assert_raises_no_relevant_fks(
              configure_mappers,
              "foos.id > bars.fid", "Foo.bars", "primary"
          )
  
      def test_no_equated_fks(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id > bars.c.fid,
                                   foreign_keys=bars.c.fid)})
          mapper(Bar, bars)
          self._assert_raises_no_equality(
              sa.orm.configure_mappers,
              "foos.id > bars.fid", "Foo.bars", "primary"
          )
  
      def test_no_equated_wo_fks_works_on_relaxed(self):
          foos_with_fks, Foo, Bar, bars_with_fks, foos = (
              self.tables.foos_with_fks,
              self.classes.Foo,
              self.classes.Bar,
              self.tables.bars_with_fks,
              self.tables.foos)
  
          # very unique - the join between parent/child
          # has no fks, but there is an fk join between two other
          # tables in the join condition, for those users that try creating
          # these big-long-string-of-joining-many-tables primaryjoins.
          # in this case we don't get eq_pairs, but we hit the
          # "works if viewonly" rule.  so here we add another clause regarding
          # "try foreign keys".
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=and_(
                                       bars_with_fks.c.fid == foos_with_fks.c.id,
                                       foos_with_fks.c.id == foos.c.id,
                                   )
                                   )})
          mapper(Bar, bars_with_fks)
  
          self._assert_raises_no_equality(
              sa.orm.configure_mappers,
              "bars_with_fks.fid = foos_with_fks.id "
              "AND foos_with_fks.id = foos.id",
              "Foo.bars", "primary"
          )
  
      def test_ambiguous_fks(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id == bars.c.fid,
                                   foreign_keys=[foos.c.id, bars.c.fid])})
          mapper(Bar, bars)
  
          self._assert_raises_ambiguous_direction(
              sa.orm.configure_mappers,
              "Foo.bars"
          )
  
      def test_ambiguous_remoteside_o2m(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id == bars.c.fid,
                                   foreign_keys=[bars.c.fid],
                                   remote_side=[foos.c.id, bars.c.fid],
                                   viewonly=True
                                   )})
          mapper(Bar, bars)
  
          self._assert_raises_no_local_remote(
              configure_mappers,
              "Foo.bars",
          )
  
      def test_ambiguous_remoteside_m2o(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id == bars.c.fid,
                                   foreign_keys=[foos.c.id],
                                   remote_side=[foos.c.id, bars.c.fid],
                                   viewonly=True
                                   )})
          mapper(Bar, bars)
  
          self._assert_raises_no_local_remote(
              configure_mappers,
              "Foo.bars",
          )
  
      def test_no_equated_self_ref_no_fks(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo,
                                   primaryjoin=foos.c.id > foos.c.fid)})
          mapper(Bar, bars)
  
          self._assert_raises_no_relevant_fks(
              configure_mappers,
              "foos.id > foos.fid", "Foo.foos", "primary"
          )
  
      def test_no_equated_self_ref_no_equality(self):
          bars, Foo, Bar, foos = (self.tables.bars,
                                  self.classes.Foo,
                                  self.classes.Bar,
                                  self.tables.foos)
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo,
                                   primaryjoin=foos.c.id > foos.c.fid,
                                   foreign_keys=[foos.c.fid])})
          mapper(Bar, bars)
  
          self._assert_raises_no_equality(configure_mappers,
                                          "foos.id > foos.fid", "Foo.foos",
                                          "primary")
  
      def test_no_equated_viewonly(self):
          bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (
              self.tables.bars,
              self.classes.Bar,
              self.tables.bars_with_fks,
              self.tables.foos_with_fks,
              self.classes.Foo,
              self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id > bars.c.fid,
                                   viewonly=True)})
          mapper(Bar, bars)
  
          self._assert_raises_no_relevant_fks(
              sa.orm.configure_mappers,
              "foos.id > bars.fid", "Foo.bars", "primary"
          )
  
          sa.orm.clear_mappers()
          mapper(Foo, foos_with_fks, properties={
              'bars': relationship(
                  Bar,
                  primaryjoin=foos_with_fks.c.id > bars_with_fks.c.fid,
                  viewonly=True)})
          mapper(Bar, bars_with_fks)
          sa.orm.configure_mappers()
  
      def test_no_equated_self_ref_viewonly(self):
          bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (
              self.tables.bars,
              self.classes.Bar,
              self.tables.bars_with_fks,
              self.tables.foos_with_fks,
              self.classes.Foo,
              self.tables.foos)
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo,
                                   primaryjoin=foos.c.id > foos.c.fid,
                                   viewonly=True)})
          mapper(Bar, bars)
  
          self._assert_raises_no_relevant_fks(
              sa.orm.configure_mappers,
              "foos.id > foos.fid", "Foo.foos", "primary"
          )
  
          sa.orm.clear_mappers()
          mapper(Foo, foos_with_fks, properties={
              'foos': relationship(
                  Foo,
                  primaryjoin=foos_with_fks.c.id > foos_with_fks.c.fid,
                  viewonly=True)})
          mapper(Bar, bars_with_fks)
          sa.orm.configure_mappers()
  
      def test_no_equated_self_ref_viewonly_fks(self):
          Foo, foos = self.classes.Foo, self.tables.foos
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo,
                                   primaryjoin=foos.c.id > foos.c.fid,
                                   viewonly=True,
                                   foreign_keys=[foos.c.fid])})
  
          sa.orm.configure_mappers()
          eq_(Foo.foos.property.local_remote_pairs, [(foos.c.id, foos.c.fid)])
  
      def test_equated(self):
          bars, Bar, bars_with_fks, foos_with_fks, Foo, foos = (
              self.tables.bars,
              self.classes.Bar,
              self.tables.bars_with_fks,
              self.tables.foos_with_fks,
              self.classes.Foo,
              self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   primaryjoin=foos.c.id == bars.c.fid)})
          mapper(Bar, bars)
  
          self._assert_raises_no_relevant_fks(
              configure_mappers,
              "foos.id = bars.fid", "Foo.bars", "primary"
          )
  
          sa.orm.clear_mappers()
          mapper(Foo, foos_with_fks, properties={
              'bars': relationship(
                  Bar,
                  primaryjoin=foos_with_fks.c.id == bars_with_fks.c.fid)})
          mapper(Bar, bars_with_fks)
          sa.orm.configure_mappers()
  
      def test_equated_self_ref(self):
          Foo, foos = self.classes.Foo, self.tables.foos
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo,
                                   primaryjoin=foos.c.id == foos.c.fid)})
  
          self._assert_raises_no_relevant_fks(
              configure_mappers,
              "foos.id = foos.fid", "Foo.foos", "primary"
          )
  
      def test_equated_self_ref_wrong_fks(self):
          bars, Foo, foos = (self.tables.bars,
                             self.classes.Foo,
                             self.tables.foos)
  
          mapper(Foo, foos, properties={
              'foos': relationship(Foo,
                                   primaryjoin=foos.c.id == foos.c.fid,
                                   foreign_keys=[bars.c.id])})
  
          self._assert_raises_no_relevant_fks(
              configure_mappers,
              "foos.id = foos.fid", "Foo.foos", "primary"
          )
  
  
  class InvalidRelationshipEscalationTestM2M(
          _RelationshipErrors, fixtures.MappedTest):
  
      @classmethod
      def define_tables(cls, metadata):
          Table('foos', metadata,
                Column('id', Integer, primary_key=True))
          Table('foobars', metadata,
                Column('fid', Integer), Column('bid', Integer))
          Table('bars', metadata,
                Column('id', Integer, primary_key=True))
  
          Table('foobars_with_fks', metadata,
                Column('fid', Integer, ForeignKey('foos.id')),
                Column('bid', Integer, ForeignKey('bars.id'))
                )
  
          Table('foobars_with_many_columns', metadata,
                Column('fid', Integer),
                Column('bid', Integer),
                Column('fid1', Integer),
                Column('bid1', Integer),
                Column('fid2', Integer),
                Column('bid2', Integer),
                )
  
      @classmethod
      def setup_classes(cls):
          class Foo(cls.Basic):
              pass
  
          class Bar(cls.Basic):
              pass
  
      def test_no_join(self):
          foobars, bars, Foo, Bar, foos = (self.tables.foobars,
                                           self.tables.bars,
                                           self.classes.Foo,
                                           self.classes.Bar,
                                           self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar, secondary=foobars)})
          mapper(Bar, bars)
  
          self._assert_raises_no_join(
              configure_mappers,
              "Foo.bars",
              "foobars"
          )
  
      def test_no_secondaryjoin(self):
          foobars, bars, Foo, Bar, foos = (self.tables.foobars,
                                           self.tables.bars,
                                           self.classes.Foo,
                                           self.classes.Bar,
                                           self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   secondary=foobars,
                                   primaryjoin=foos.c.id > foobars.c.fid)})
          mapper(Bar, bars)
  
          self._assert_raises_no_join(
              configure_mappers,
              "Foo.bars",
              "foobars"
          )
  
      def test_no_fks(self):
          foobars_with_many_columns, bars, Bar, foobars, Foo, foos = (
              self.tables.foobars_with_many_columns,
              self.tables.bars,
              self.classes.Bar,
              self.tables.foobars,
              self.classes.Foo,
              self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar, secondary=foobars,
                                   primaryjoin=foos.c.id == foobars.c.fid,
                                   secondaryjoin=foobars.c.bid == bars.c.id)})
          mapper(Bar, bars)
          sa.orm.configure_mappers()
          eq_(
              Foo.bars.property.synchronize_pairs,
              [(foos.c.id, foobars.c.fid)]
          )
          eq_(
              Foo.bars.property.secondary_synchronize_pairs,
              [(bars.c.id, foobars.c.bid)]
          )
  
          sa.orm.clear_mappers()
          mapper(Foo, foos, properties={
              'bars': relationship(
                  Bar,
                  secondary=foobars_with_many_columns,
                  primaryjoin=foos.c.id ==
                  foobars_with_many_columns.c.fid,
                  secondaryjoin=foobars_with_many_columns.c.bid ==
                  bars.c.id)})
          mapper(Bar, bars)
          sa.orm.configure_mappers()
          eq_(
              Foo.bars.property.synchronize_pairs,
              [(foos.c.id, foobars_with_many_columns.c.fid)]
          )
          eq_(
              Foo.bars.property.secondary_synchronize_pairs,
              [(bars.c.id, foobars_with_many_columns.c.bid)]
          )
  
      def test_local_col_setup(self):
          foobars_with_fks, bars, Bar, Foo, foos = (
              self.tables.foobars_with_fks,
              self.tables.bars,
              self.classes.Bar,
              self.classes.Foo,
              self.tables.foos)
  
          # ensure m2m backref is set up with correct annotations
          # [ticket:2578]
          mapper(Foo, foos, properties={
              'bars': relationship(Bar, secondary=foobars_with_fks,
                                   backref="foos")
          })
          mapper(Bar, bars)
          sa.orm.configure_mappers()
          eq_(
              Foo.bars.property._join_condition.local_columns,
              set([foos.c.id])
          )
          eq_(
              Bar.foos.property._join_condition.local_columns,
              set([bars.c.id])
          )
  
      def test_bad_primaryjoin(self):
          foobars_with_fks, bars, Bar, foobars, Foo, foos = (
              self.tables.foobars_with_fks,
              self.tables.bars,
              self.classes.Bar,
              self.tables.foobars,
              self.classes.Foo,
              self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   secondary=foobars,
                                   primaryjoin=foos.c.id > foobars.c.fid,
                                   secondaryjoin=foobars.c.bid <= bars.c.id)})
          mapper(Bar, bars)
  
          self._assert_raises_no_equality(
              configure_mappers,
              'foos.id > foobars.fid',
              "Foo.bars",
              "primary")
  
          sa.orm.clear_mappers()
          mapper(Foo, foos, properties={
              'bars': relationship(
                  Bar,
                  secondary=foobars_with_fks,
                  primaryjoin=foos.c.id > foobars_with_fks.c.fid,
                  secondaryjoin=foobars_with_fks.c.bid <= bars.c.id)})
          mapper(Bar, bars)
          self._assert_raises_no_equality(
              configure_mappers,
              'foos.id > foobars_with_fks.fid',
              "Foo.bars",
              "primary")
  
          sa.orm.clear_mappers()
          mapper(Foo, foos, properties={
              'bars': relationship(
                  Bar,
                  secondary=foobars_with_fks,
                  primaryjoin=foos.c.id > foobars_with_fks.c.fid,
                  secondaryjoin=foobars_with_fks.c.bid <= bars.c.id,
                  viewonly=True)})
          mapper(Bar, bars)
          sa.orm.configure_mappers()
  
      def test_bad_secondaryjoin(self):
          foobars, bars, Foo, Bar, foos = (self.tables.foobars,
                                           self.tables.bars,
                                           self.classes.Foo,
                                           self.classes.Bar,
                                           self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   secondary=foobars,
                                   primaryjoin=foos.c.id == foobars.c.fid,
                                   secondaryjoin=foobars.c.bid <= bars.c.id,
                                   foreign_keys=[foobars.c.fid])})
          mapper(Bar, bars)
          self._assert_raises_no_relevant_fks(
              configure_mappers,
              "foobars.bid <= bars.id",
              "Foo.bars",
              "secondary"
          )
  
      def test_no_equated_secondaryjoin(self):
          foobars, bars, Foo, Bar, foos = (self.tables.foobars,
                                           self.tables.bars,
                                           self.classes.Foo,
                                           self.classes.Bar,
                                           self.tables.foos)
  
          mapper(Foo, foos, properties={
              'bars': relationship(Bar,
                                   secondary=foobars,
                                   primaryjoin=foos.c.id == foobars.c.fid,
                                   secondaryjoin=foobars.c.bid <= bars.c.id,
                                   foreign_keys=[foobars.c.fid, foobars.c.bid])})
          mapper(Bar, bars)
  
          self._assert_raises_no_equality(
              configure_mappers,
              "foobars.bid <= bars.id",
              "Foo.bars",
              "secondary"
          )
  
  
  class ActiveHistoryFlagTest(_fixtures.FixtureTest):
      run_inserts = None
      run_deletes = None
  
      def _test_attribute(self, obj, attrname, newvalue):
          sess = Session()
          sess.add(obj)
          oldvalue = getattr(obj, attrname)
          sess.commit()
  
          # expired
          assert attrname not in obj.__dict__
  
          setattr(obj, attrname, newvalue)
          eq_(
              attributes.get_history(obj, attrname),
              ([newvalue, ], (), [oldvalue, ])
          )
  
      def test_column_property_flag(self):
          User, users = self.classes.User, self.tables.users
  
          mapper(User, users, properties={
              'name': column_property(users.c.name,
                                      active_history=True)
          })
          u1 = User(name='jack')
          self._test_attribute(u1, 'name', 'ed')
  
      def test_relationship_property_flag(self):
          Address, addresses, users, User = (self.classes.Address,
                                             self.tables.addresses,
                                             self.tables.users,
                                             self.classes.User)
  
          mapper(Address, addresses, properties={
              'user': relationship(User, active_history=True)
          })
          mapper(User, users)
          u1 = User(name='jack')
          u2 = User(name='ed')
          a1 = Address(email_address='a1', user=u1)
          self._test_attribute(a1, 'user', u2)
  
      def test_composite_property_flag(self):
          Order, orders = self.classes.Order, self.tables.orders
  
          class MyComposite(object):
  
              def __init__(self, description, isopen):
                  self.description = description
                  self.isopen = isopen
  
              def __composite_values__(self):
                  return [self.description, self.isopen]
  
              def __eq__(self, other):
                  return isinstance(other, MyComposite) and \
                      other.description == self.description
          mapper(Order, orders, properties={
              'composite': composite(
                  MyComposite,
                  orders.c.description,
                  orders.c.isopen,
                  active_history=True)
          })
          o1 = Order(composite=MyComposite('foo', 1))
          self._test_attribute(o1, "composite", MyComposite('bar', 1))
  
  
  class RelationDeprecationTest(fixtures.MappedTest):
  
      """test usage of the old 'relation' function."""
  
      run_inserts = 'once'
      run_deletes = None
  
      @classmethod
      def define_tables(cls, metadata):
          Table('users_table', metadata,
                Column('id', Integer, primary_key=True),
                Column('name', String(64)))
  
          Table('addresses_table', metadata,
                Column('id', Integer, primary_key=True),
                Column('user_id', Integer, ForeignKey('users_table.id')),
                Column('email_address', String(128)),
                Column('purpose', String(16)),
                Column('bounces', Integer, default=0))
  
      @classmethod
      def setup_classes(cls):
          class User(cls.Basic):
              pass
  
          class Address(cls.Basic):
              pass
  
      @classmethod
      def fixtures(cls):
          return dict(
              users_table=(
                  ('id', 'name'),
                  (1, 'jack'),
                  (2, 'ed'),
                  (3, 'fred'),
                  (4, 'chuck')),
  
              addresses_table=(
                  ('id', 'user_id', 'email_address', 'purpose', 'bounces'),
                  (1, 1, 'jack@jack.home', 'Personal', 0),
                  (2, 1, 'jack@jack.bizz', 'Work', 1),
                  (3, 2, 'ed@foo.bar', 'Personal', 0),
                  (4, 3, 'fred@the.fred', 'Personal', 10)))
  
      def test_relation(self):
          addresses_table, User, users_table, Address = (
              self.tables.addresses_table,
              self.classes.User,
              self.tables.users_table,
              self.classes.Address)
  
          mapper(User, users_table, properties=dict(
              addresses=relation(Address, backref='user'),
          ))
          mapper(Address, addresses_table)
  
          session = create_session()
  
          session.query(User).filter(User.addresses.any(
              Address.email_address == 'ed@foo.bar')).one()
  