- A warning is emitted in the case of multiple relationships thatticket_3230

ultimately will populate a foreign key column in conflict with
another, where the relationships are attempting to copy values
from different source columns.  This occurs in the case where
composite foreign keys with overlapping columns are mapped to
relationships that each refer to a different referenced column.
A new documentation section illustrates the example as well as how
to overcome the issue by specifying "foreign" columns specifically
on a per-relationship basis.
fixes #3230

3 files changed, 102 insertions, 21 deletions

diff --git a/test/orm/test_assorted_eager.py b/test/orm/test_assorted_eager.py
index 2bee3cbd6..48faa172f 100644
--- a/test/orm/test_assorted_eager.py
+++ b/test/orm/test_assorted_eager.py
@@ -82,8 +82,8 @@ class EagerTest(fixtures.MappedTest):
         mapper(Category, categories)
 
         mapper(Option, options, properties=dict(
-            owner=relationship(Owner),
-            test=relationship(Thing)))
+            owner=relationship(Owner, viewonly=True),
+            test=relationship(Thing, viewonly=True)))
 
         mapper(Thing, tests, properties=dict(
             owner=relationship(Owner, backref='tests'),
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index 40bc01b5d..eba47dbec 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -361,6 +361,27 @@ class InheritedJoinTest(fixtures.MappedTest, AssertsCompiledSQL):
         )
 
 
+class JoinOnSynonymTest(_fixtures.FixtureTest, AssertsCompiledSQL):
+    @classmethod
+    def setup_mappers(cls):
+        User = cls.classes.User
+        Address = cls.classes.Address
+        users, addresses = (cls.tables.users, cls.tables.addresses)
+        mapper(User, users, properties={
+            'addresses': relationship(Address),
+            'ad_syn': synonym("addresses")
+        })
+        mapper(Address, addresses)
+
+    def test_join_on_synonym(self):
+        User = self.classes.User
+        self.assert_compile(
+            Session().query(User).join(User.ad_syn),
+            "SELECT users.id AS users_id, users.name AS users_name "
+            "FROM users JOIN addresses ON users.id = addresses.user_id"
+        )
+
+
 class JoinTest(QueryTest, AssertsCompiledSQL):
     __dialect__ = 'default'
 
@@ -409,24 +430,6 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
             sess.query(literal_column('x'), User).join, Address
         )
 
-    def test_join_on_synonym(self):
-
-        class User(object):
-            pass
-        class Address(object):
-            pass
-        users, addresses = (self.tables.users, self.tables.addresses)
-        mapper(User, users, properties={
-            'addresses':relationship(Address),
-            'ad_syn':synonym("addresses")
-        })
-        mapper(Address, addresses)
-        self.assert_compile(
-            Session().query(User).join(User.ad_syn),
-            "SELECT users.id AS users_id, users.name AS users_name "
-            "FROM users JOIN addresses ON users.id = addresses.user_id"
-        )
-
     def test_multi_tuple_form(self):
         """test the 'tuple' form of join, now superseded
         by the two-element join() form.
diff --git a/test/orm/test_relationships.py b/test/orm/test_relationships.py
index 4c5a5abee..2a15ce666 100644
--- a/test/orm/test_relationships.py
+++ b/test/orm/test_relationships.py
@@ -672,12 +672,89 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
 
         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 "
+            "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._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):
         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)
 
     def _test_relationships(self):
         configure_mappers()
@@ -3044,7 +3121,8 @@ class SecondaryNestedJoinTest(fixtures.MappedTest, AssertsCompiledSQL,
                 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
+                uselist=False,
+                viewonly=True
             )
         })
         mapper(B, b, properties={