- Added new parameter :paramref:`.orm.mapper.passive_deletes` to

available mapper options.   This allows a DELETE to proceed
for a joined-table inheritance mapping against the base table only,
while allowing for ON DELETE CASCADE to handle deleting the row
from the subclass tables.
fixes #2349

1 files changed, 236 insertions, 0 deletions

diff --git a/test/orm/inheritance/test_basic.py b/test/orm/inheritance/test_basic.py
index 911d4bc5c..341e142de 100644
--- a/test/orm/inheritance/test_basic.py
+++ b/test/orm/inheritance/test_basic.py
@@ -1149,6 +1149,242 @@ class FlushTest(fixtures.MappedTest):
         assert user_roles.count().scalar() == 1
 
 
+class PassiveDeletesTest(fixtures.MappedTest):
+    __requires__ = ('foreign_keys',)
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "a", metadata,
+            Column('id', Integer, primary_key=True),
+            Column('type', String(30))
+        )
+        Table(
+            "b", metadata,
+            Column(
+                'id', Integer, ForeignKey('a.id', ondelete="CASCADE"),
+                primary_key=True),
+            Column('data', String(10))
+        )
+
+        Table(
+            "c", metadata,
+            Column('cid', Integer, primary_key=True),
+            Column('bid', ForeignKey('b.id', ondelete="CASCADE"))
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class A(cls.Basic):
+            pass
+
+        class B(A):
+            pass
+
+        class C(B):
+            pass
+
+    def _fixture(self, a_p=False, b_p=False, c_p=False):
+        A, B, C = self.classes("A", "B", "C")
+        a, b, c = self.tables("a", "b", "c")
+
+        mapper(
+            A, a, passive_deletes=a_p,
+            polymorphic_on=a.c.type, polymorphic_identity='a')
+        mapper(
+            B, b, inherits=A, passive_deletes=b_p, polymorphic_identity='b')
+        mapper(
+            C, c, inherits=B, passive_deletes=c_p, polymorphic_identity='c')
+
+    def test_none(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture()
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        # want to see if the 'C' table loads even though
+        # a and b are loaded
+        c1 = s.query(A).filter_by(id=3).first()
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT c.bid AS c_bid, b.data AS b_data, c.cid AS c_cid "
+                "FROM c, b WHERE :param_1 = b.id AND b.id = c.bid",
+                [{'param_1': 3}]
+            ),
+            CompiledSQL(
+                "DELETE FROM c WHERE c.cid = :cid",
+                [{'cid': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id",
+                [{'id': 3}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+    def test_c_only(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture(c_p=True)
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        s.delete(a1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.type AS a_type "
+                "FROM a WHERE a.id = :param_1",
+                [{'param_1': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 1}]
+            )
+        )
+
+        b1.id
+        s.delete(b1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id",
+                [{'id': 2}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 2}]
+            )
+        )
+
+        # want to see if the 'C' table loads even though
+        # a and b are loaded
+        c1 = s.query(A).filter_by(id=3).first()
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM b WHERE b.id = :id",
+                [{'id': 3}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+    def test_b_only(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture(b_p=True)
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        s.delete(a1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.type AS a_type "
+                "FROM a WHERE a.id = :param_1",
+                [{'param_1': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 1}]
+            )
+        )
+
+        b1.id
+        s.delete(b1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 2}]
+            )
+        )
+
+        c1.id
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+    def test_a_only(self):
+        A, B, C = self.classes("A", "B", "C")
+        self._fixture(a_p=True)
+
+        s = Session()
+        a1, b1, c1 = A(id=1), B(id=2), C(cid=1, id=3)
+        s.add_all([a1, b1, c1])
+        s.commit()
+
+        s.delete(a1)
+
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT a.id AS a_id, a.type AS a_type "
+                "FROM a WHERE a.id = :param_1",
+                [{'param_1': 1}]
+            ),
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 1}]
+            )
+        )
+
+        b1.id
+        s.delete(b1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 2}]
+            )
+        )
+
+        # want to see if the 'C' table loads even though
+        # a and b are loaded
+        c1 = s.query(A).filter_by(id=3).first()
+        s.delete(c1)
+        with self.sql_execution_asserter(testing.db) as asserter:
+            s.flush()
+        asserter.assert_(
+            CompiledSQL(
+                "DELETE FROM a WHERE a.id = :id",
+                [{'id': 3}]
+            )
+        )
+
+
 class OptimizedGetOnDeferredTest(fixtures.MappedTest):
     """test that the 'optimized get' path accommodates deferred columns."""