add "merge" to viewonly cascades; propagate NO_RAISE when merging

Fixed bug where :meth:`_orm.Session.merge` would fail to preserve the
current loaded contents of relationship attributes that were indicated with
the :paramref:`_orm.relationship.viewonly` parameter, thus defeating
strategies that use :meth:`_orm.Session.merge` to pull fully loaded objects
from caches and other similar techniques. In a related change, fixed issue
where an object that contains a loaded relationship that was nonetheless
configured as ``lazy='raise'`` on the mapping would fail when passed to
:meth:`_orm.Session.merge`; checks for "raise" are now suspended within
the merge process assuming the :paramref:`_orm.Session.merge.load`
parameter remains at its default of ``True``.

Overall, this is a behavioral adjustment to a change introduced in the 1.4
series as of :ticket:`4994`, which took "merge" out of the set of cascades
applied by default to "viewonly" relationships. As "viewonly" relationships
aren't persisted under any circumstances, allowing their contents to
transfer during "merge" does not impact the persistence behavior of the
target object. This allows :meth:`_orm.Session.merge` to correctly suit one
of its use cases, that of adding objects to a :class:`.Session` that were
loaded elsewhere, often for the purposes of restoring from a cache.

Fixes: #8862
Change-Id: I8731c7810460e6a71f8bf5e8ded59142b9b02956

2 files changed, 151 insertions, 6 deletions

diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index e5710e90e..0e9e63c26 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -4253,7 +4253,7 @@ class ViewonlyCascadeUpdate(fixtures.MappedTest):
         ({"delete"}, {"none"}),
         (
             {"all, delete-orphan"},
-            {"refresh-expire", "expunge"},
+            {"refresh-expire", "expunge", "merge"},
         ),
         ({"save-update, expunge"}, {"expunge"}),
     )
@@ -4357,7 +4357,10 @@ class ViewonlyCascadeUpdate(fixtures.MappedTest):
         not_in(o1, sess)
         not_in(o2, sess)
 
-    def test_default_merge_cascade(self):
+    @testing.combinations(
+        "persistent", "pending", argnames="collection_status"
+    )
+    def test_default_merge_cascade(self, collection_status):
         User, Order, orders, users = (
             self.classes.User,
             self.classes.Order,
@@ -4389,12 +4392,31 @@ class ViewonlyCascadeUpdate(fixtures.MappedTest):
             Order(id=2, user_id=1, description="someotherorder"),
         )
 
-        u1.orders.append(o1)
-        u1.orders.append(o2)
+        if collection_status == "pending":
+            # technically this is pointless, one should not be appending
+            # to this collection
+            u1.orders.append(o1)
+            u1.orders.append(o2)
+        elif collection_status == "persistent":
+            sess.add(u1)
+            sess.flush()
+            sess.add_all([o1, o2])
+            sess.flush()
+            u1.orders
+        else:
+            assert False
 
         u1 = sess.merge(u1)
 
-        assert not u1.orders
+        # in 1.4, as of #4993 this was asserting that u1.orders would
+        # not be present in the new object.  However, as observed during
+        # #8862, this defeats schemes that seek to restore fully loaded
+        # objects from caches which may even have lazy="raise", but
+        # in any case would want to not emit new SQL on those collections.
+        # so we assert here that u1.orders is in fact present
+        assert "orders" in u1.__dict__
+        assert u1.__dict__["orders"]
+        assert u1.orders
 
     def test_default_cascade(self):
         User, Order, orders, users = (
@@ -4420,7 +4442,7 @@ class ViewonlyCascadeUpdate(fixtures.MappedTest):
             },
         )
 
-        eq_(umapper.attrs["orders"].cascade, set())
+        eq_(umapper.attrs["orders"].cascade, {"merge"})
 
 
 class CollectionCascadesNoBackrefTest(fixtures.TestBase):
diff --git a/test/orm/test_merge.py b/test/orm/test_merge.py
index ef0db6d05..7ea2fa62b 100644
--- a/test/orm/test_merge.py
+++ b/test/orm/test_merge.py
@@ -7,6 +7,7 @@ from sqlalchemy import event
 from sqlalchemy import ForeignKey
 from sqlalchemy import Integer
 from sqlalchemy import PickleType
+from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import Text
@@ -16,6 +17,7 @@ from sqlalchemy.orm import configure_mappers
 from sqlalchemy.orm import defer
 from sqlalchemy.orm import deferred
 from sqlalchemy.orm import foreign
+from sqlalchemy.orm import joinedload
 from sqlalchemy.orm import relationship
 from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
@@ -28,6 +30,7 @@ from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import in_
 from sqlalchemy.testing import not_in
+from sqlalchemy.testing.assertsql import CountStatements
 from sqlalchemy.testing.fixtures import fixture_session
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
@@ -1374,6 +1377,126 @@ class MergeTest(_fixtures.FixtureTest):
         except sa.exc.InvalidRequestError as e:
             assert "load=False option does not support" in str(e)
 
+    @testing.combinations("viewonly", "normal", argnames="viewonly")
+    @testing.combinations("load", "noload", argnames="load")
+    @testing.combinations("select", "raise", "raise_on_sql", argnames="lazy")
+    @testing.combinations(
+        "merge_persistent", "merge_detached", argnames="merge_persistent"
+    )
+    @testing.combinations("detached", "persistent", argnames="detach_original")
+    @testing.combinations("o2m", "m2o", argnames="direction")
+    def test_relationship_population_maintained(
+        self,
+        viewonly,
+        load,
+        lazy,
+        merge_persistent,
+        direction,
+        detach_original,
+    ):
+        """test #8862"""
+
+        User, Address = self.classes("User", "Address")
+        users, addresses = self.tables("users", "addresses")
+
+        self.mapper_registry.map_imperatively(
+            User,
+            users,
+            properties={
+                "addresses": relationship(
+                    Address,
+                    viewonly=viewonly == "viewonly",
+                    lazy=lazy,
+                    back_populates="user",
+                )
+            },
+        )
+
+        self.mapper_registry.map_imperatively(
+            Address,
+            addresses,
+            properties={
+                "user": relationship(
+                    User,
+                    viewonly=viewonly == "viewonly",
+                    lazy=lazy,
+                    back_populates="addresses",
+                )
+            },
+        )
+
+        s = fixture_session()
+
+        u1 = User(id=1, name="u1")
+        s.add(u1)
+        s.flush()
+        s.add_all(
+            [Address(user_id=1, email_address="e%d" % i) for i in range(1, 4)]
+        )
+        s.commit()
+
+        if direction == "o2m":
+            cls_to_merge = User
+            obj_to_merge = (
+                s.scalars(select(User).options(joinedload(User.addresses)))
+                .unique()
+                .one()
+            )
+            attrname = "addresses"
+
+        elif direction == "m2o":
+            cls_to_merge = Address
+            obj_to_merge = (
+                s.scalars(
+                    select(Address)
+                    .filter_by(email_address="e1")
+                    .options(joinedload(Address.user))
+                )
+                .unique()
+                .one()
+            )
+            attrname = "user"
+        else:
+            assert False
+
+        assert attrname in obj_to_merge.__dict__
+
+        s2 = Session(testing.db)
+
+        if merge_persistent == "merge_persistent":
+            target_persistent = s2.get(cls_to_merge, obj_to_merge.id)  # noqa
+
+        if detach_original == "detach":
+            s.expunge(obj_to_merge)
+
+        with self.sql_execution_asserter(testing.db) as assert_:
+            merged_object = s2.merge(obj_to_merge, load=load == "load")
+
+        assert_.assert_(
+            CountStatements(
+                0
+                if load == "noload"
+                else 1
+                if merge_persistent == "merge_persistent"
+                else 2
+            )
+        )
+
+        assert attrname in merged_object.__dict__
+
+        with self.sql_execution_asserter(testing.db) as assert_:
+            if direction == "o2m":
+                eq_(
+                    merged_object.addresses,
+                    [
+                        Address(user_id=1, email_address="e%d" % i)
+                        for i in range(1, 4)
+                    ],
+                )
+            elif direction == "m2o":
+                eq_(merged_object.user, User(id=1, name="u1"))
+        assert_.assert_(CountStatements(0))
+
     def test_synonym(self):
         users = self.tables.users