diff options
Diffstat (limited to 'test/orm/inheritance/test_basic.py')
| -rw-r--r-- | test/orm/inheritance/test_basic.py | 236 |
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.""" |