eq_(sess.query(T3).all(), [])
class M2MCascadeTest(_base.MappedTest):
+ """delete-orphan cascade is deprecated on many-to-many."""
+
def define_tables(self, metadata):
Table('a', metadata,
Column('id', Integer, primary_key=True),
- Column('data', String(30)))
+ Column('data', String(30)),
+ test_needs_fk=True
+ )
Table('b', metadata,
Column('id', Integer, primary_key=True),
- Column('data', String(30)))
+ Column('data', String(30)),
+ test_needs_fk=True
+
+ )
Table('atob', metadata,
Column('aid', Integer, ForeignKey('a.id')),
- Column('bid', Integer, ForeignKey('b.id')))
+ Column('bid', Integer, ForeignKey('b.id')),
+ test_needs_fk=True
+
+ )
Table('c', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30)),
- Column('bid', Integer, ForeignKey('b.id')))
+ Column('bid', Integer, ForeignKey('b.id')),
+ test_needs_fk=True
+
+ )
def setup_classes(self):
class A(_fixtures.Base):
class C(_fixtures.Base):
pass
+ @testing.emits_warning(".*not supported on a many-to-many")
@testing.resolve_artifact_names
def test_delete_orphan(self):
mapper(A, a, properties={
assert b.count().scalar() == 0
assert a.count().scalar() == 1
+ @testing.emits_warning(".*not supported on a many-to-many")
@testing.resolve_artifact_names
def test_delete_orphan_cascades(self):
mapper(A, a, properties={
assert a.count().scalar() == 1
assert c.count().scalar() == 0
+ @testing.emits_warning(".*not supported on a many-to-many")
@testing.resolve_artifact_names
def test_cascade_delete(self):
mapper(A, a, properties={
assert b.count().scalar() == 0
assert a.count().scalar() == 0
+ @testing.emits_warning(".*not supported on a many-to-many")
+ @testing.fails_on_everything_except('sqlite')
+ @testing.resolve_artifact_names
+ def test_this_doesnt_work(self):
+ """illustrates why cascade with m2m should not be supported
+ (i.e. many parents...)
+
+ """
+ mapper(A, a, properties={
+ 'bs':relation(B, secondary=atob, cascade="all, delete-orphan")
+ })
+ mapper(B, b)
+
+ sess = create_session()
+ b1 =B(data='b1')
+ a1 = A(data='a1', bs=[b1])
+ a2 = A(data='a2', bs=[b1])
+ sess.add(a1)
+ sess.add(a2)
+ sess.flush()
+
+ sess.delete(a1)
+
+ # this raises an integrity error on DBs that support FKs
+ sess.flush()
+
+ # still a row present !
+ assert atob.count().scalar() ==1
+
+ # but no bs !
+ assert b.count().scalar() == 0
+ assert a.count().scalar() == 1
+
class UnsavedOrphansTest(_base.MappedTest):
"""Pending entities that are orphans"""