from sqlalchemy import Boolean

from sqlalchemy import case

from sqlalchemy import column

from sqlalchemy import event

from sqlalchemy import exc as sa_exc

from sqlalchemy import ForeignKey

from sqlalchemy import func

from sqlalchemy import inspect

from sqlalchemy import Integer

from sqlalchemy import MetaData

from sqlalchemy import select

from sqlalchemy import String

from sqlalchemy import table

from sqlalchemy import testing

from sqlalchemy import util

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import attributes

from sqlalchemy.orm import class_mapper

from sqlalchemy.orm import clear_mappers

from sqlalchemy.orm import column_property

from sqlalchemy.orm import composite

from sqlalchemy.orm import create_session

from sqlalchemy.orm import deferred

from sqlalchemy.orm import exc as orm_exc

from sqlalchemy.orm import joinedload

from sqlalchemy.orm import mapper

from sqlalchemy.orm import object_mapper

from sqlalchemy.orm import polymorphic_union

from sqlalchemy.orm import relationship

from sqlalchemy.orm import Session

from sqlalchemy.orm import sessionmaker

from sqlalchemy.orm import synonym

from sqlalchemy.orm.util import instance_str

from sqlalchemy.testing import assert_raises

from sqlalchemy.testing import assert_raises_message

from sqlalchemy.testing import engines

from sqlalchemy.testing import eq_

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import is_

from sqlalchemy.testing import mock

from sqlalchemy.testing.assertsql import AllOf

from sqlalchemy.testing.assertsql import CompiledSQL

from sqlalchemy.testing.assertsql import Or

from sqlalchemy.testing.assertsql import RegexSQL

from sqlalchemy.testing.schema import Column

from sqlalchemy.testing.schema import Table

class O2MTest(fixtures.MappedTest):

"""deals with inheritance and one-to-many relationships"""

@classmethod

def define_tables(cls, metadata):

global foo, bar, blub

foo = Table(

"foo",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("data", String(20)),

)

bar = Table(

"bar",

metadata,

Column("id", Integer, ForeignKey("foo.id"), primary_key=True),

Column("bar_data", String(20)),

)

blub = Table(

"blub",

metadata,

Column("id", Integer, ForeignKey("bar.id"), primary_key=True),

Column("foo_id", Integer, ForeignKey("foo.id"), nullable=False),

Column("blub_data", String(20)),

)

def test_basic(self):

class Foo(object):

def __init__(self, data=None):

self.data = data

def __repr__(self):

return "Foo id %d, data %s" % (self.id, self.data)

mapper(Foo, foo)

class Bar(Foo):

def __repr__(self):

return "Bar id %d, data %s" % (self.id, self.data)

mapper(Bar, bar, inherits=Foo)

class Blub(Bar):

def __repr__(self):

return "Blub id %d, data %s" % (self.id, self.data)

mapper(

Blub,

blub,

inherits=Bar,

properties={"parent_foo": relationship(Foo)},

)

sess = create_session()

b1 = Blub("blub #1")

b2 = Blub("blub #2")

f = Foo("foo #1")

sess.add(b1)

sess.add(b2)

sess.add(f)

b1.parent_foo = f

b2.parent_foo = f

sess.flush()

compare = ",".join(

[repr(b1), repr(b2), repr(b1.parent_foo), repr(b2.parent_foo)]

)

sess.expunge_all()

result = sess.query(Blub).all()

result_str = ",".join(

[

repr(result[0]),

repr(result[1]),

repr(result[0].parent_foo),

repr(result[1].parent_foo),

]

)

eq_(compare, result_str)

eq_(result[0].parent_foo.data, "foo #1")

eq_(result[1].parent_foo.data, "foo #1")

class ColExpressionsTest(fixtures.DeclarativeMappedTest):

__backend__ = True

@classmethod

def setup_classes(cls):

Base = cls.DeclarativeBasic

class A(Base):

__tablename__ = "a"

id = Column(

Integer, primary_key=True, test_needs_autoincrement=True

)

type = Column(String(10))

__mapper_args__ = {

"polymorphic_on": type,

"polymorphic_identity": "a",

}

class B(A):

__tablename__ = "b"

id = Column(ForeignKey("a.id"), primary_key=True)

data = Column(Integer)

__mapper_args__ = {"polymorphic_identity": "b"}

@classmethod

def insert_data(cls, connection):

A, B = cls.classes("A", "B")

s = Session(connection)

s.add_all([B(data=5), B(data=7)])

s.commit()

def test_group_by(self):

B = self.classes.B

s = Session()

rows = (

s.query(B.id.expressions[0], B.id.expressions[1], func.sum(B.data))

.group_by(*B.id.expressions)

.all()

)

eq_(rows, [(1, 1, 5), (2, 2, 7)])

class PolyExpressionEagerLoad(fixtures.DeclarativeMappedTest):

run_setup_mappers = "once"

__dialect__ = "default"

@classmethod

def setup_classes(cls):

Base = cls.DeclarativeBasic

class A(fixtures.ComparableEntity, Base):

__tablename__ = "a"

id = Column(

Integer, primary_key=True, test_needs_autoincrement=True

)

discriminator = Column(String(50), nullable=False)

child_id = Column(Integer, ForeignKey("a.id"))

child = relationship("A")

p_a = case([(discriminator == "a", "a")], else_="b")

__mapper_args__ = {

"polymorphic_identity": "a",

"polymorphic_on": p_a,

}

class B(A):

__mapper_args__ = {"polymorphic_identity": "b"}

@classmethod

def insert_data(cls, connection):

A = cls.classes.A

session = Session(connection)

session.add_all(

[

A(id=1, discriminator="a"),

A(id=2, discriminator="b", child_id=1),

A(id=3, discriminator="c", child_id=1),

]

)

session.commit()

def test_joinedload(self):

A = self.classes.A

B = self.classes.B

session = Session(testing.db)

result = (

session.query(A)

.filter_by(child_id=None)

.options(joinedload("child"))

.one()

)

eq_(result, A(id=1, discriminator="a", child=[B(id=2), B(id=3)]))

class PolymorphicResolutionMultiLevel(

fixtures.DeclarativeMappedTest, testing.AssertsCompiledSQL

):

run_setup_mappers = "once"

__dialect__ = "default"

@classmethod

def setup_classes(cls):

Base = cls.DeclarativeBasic

class A(Base):

__tablename__ = "a"

id = Column(Integer, primary_key=True)

class B(A):

__tablename__ = "b"

id = Column(Integer, ForeignKey("a.id"), primary_key=True)

class C(A):

__tablename__ = "c"

id = Column(Integer, ForeignKey("a.id"), primary_key=True)

class D(B):

__tablename__ = "d"

id = Column(Integer, ForeignKey("b.id"), primary_key=True)

def test_ordered_b_d(self):

a_mapper = inspect(self.classes.A)

eq_(

a_mapper._mappers_from_spec(

[self.classes.B, self.classes.D], None

),

[a_mapper, inspect(self.classes.B), inspect(self.classes.D)],

)

def test_a(self):

a_mapper = inspect(self.classes.A)

eq_(a_mapper._mappers_from_spec([self.classes.A], None), [a_mapper])

def test_b_d_selectable(self):

a_mapper = inspect(self.classes.A)

spec = [self.classes.D, self.classes.B]

eq_(

a_mapper._mappers_from_spec(

spec, self.classes.B.__table__.join(self.classes.D.__table__)

),

[inspect(self.classes.B), inspect(self.classes.D)],

)

def test_d_selectable(self):

a_mapper = inspect(self.classes.A)

spec = [self.classes.D]

eq_(

a_mapper._mappers_from_spec(

spec, self.classes.B.__table__.join(self.classes.D.__table__)

),

[inspect(self.classes.D)],

)

def test_reverse_d_b(self):

a_mapper = inspect(self.classes.A)

spec = [self.classes.D, self.classes.B]

eq_(

a_mapper._mappers_from_spec(spec, None),

[a_mapper, inspect(self.classes.B), inspect(self.classes.D)],

)

mappers, selectable = a_mapper._with_polymorphic_args(spec=spec)

self.assert_compile(

selectable,

"a LEFT OUTER JOIN b ON a.id = b.id "

"LEFT OUTER JOIN d ON b.id = d.id",

)

def test_d_b_missing(self):

a_mapper = inspect(self.classes.A)

spec = [self.classes.D]

eq_(

a_mapper._mappers_from_spec(spec, None),

[a_mapper, inspect(self.classes.B), inspect(self.classes.D)],

)

mappers, selectable = a_mapper._with_polymorphic_args(spec=spec)

self.assert_compile(

selectable,

"a LEFT OUTER JOIN b ON a.id = b.id "

"LEFT OUTER JOIN d ON b.id = d.id",

)

def test_d_c_b(self):

a_mapper = inspect(self.classes.A)

spec = [self.classes.D, self.classes.C, self.classes.B]

ms = a_mapper._mappers_from_spec(spec, None)

eq_(ms[-1], inspect(self.classes.D))

eq_(ms[0], a_mapper)

eq_(set(ms[1:3]), set(a_mapper._inheriting_mappers))

class PolymorphicOnNotLocalTest(fixtures.MappedTest):

@classmethod

def define_tables(cls, metadata):

Table(

"t1",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("x", String(10)),

Column("q", String(10)),

)

Table(

"t2",

metadata,

Column(

"t2id",

Integer,

primary_key=True,

test_needs_autoincrement=True,

),

Column("y", String(10)),

Column("xid", ForeignKey("t1.id")),

)

@classmethod

def setup_classes(cls):

class Parent(cls.Comparable):

pass

class Child(Parent):

pass

def test_non_col_polymorphic_on(self):

Parent = self.classes.Parent

t2 = self.tables.t2

assert_raises_message(

sa_exc.ArgumentError,

"Can't determine polymorphic_on "

"value 'im not a column' - no "

"attribute is mapped to this name.",

mapper,

Parent,

t2,

polymorphic_on="im not a column",

)

def test_polymorphic_on_non_expr_prop(self):

t2 = self.tables.t2

Parent = self.classes.Parent

assert_raises_message(

sa_exc.ArgumentError,

r"Column expression or string key expected for argument "

r"'polymorphic_on'; got .*function",

mapper,

Parent,

t2,

polymorphic_on=lambda: "hi",

polymorphic_identity=0,

)

def test_polymorphic_on_not_present_col(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent = self.classes.Parent

t1t2_join = select([t1.c.x], from_obj=[t1.join(t2)]).alias()

def go():

t1t2_join_2 = select([t1.c.q], from_obj=[t1.join(t2)]).alias()

mapper(

Parent,

t2,

polymorphic_on=t1t2_join.c.x,

with_polymorphic=("*", t1t2_join_2),

polymorphic_identity=0,

)

assert_raises_message(

sa_exc.InvalidRequestError,

"Could not map polymorphic_on column 'x' to the mapped table - "

"polymorphic loads will not function properly",

go,

)

def test_polymorphic_on_only_in_with_poly(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent = self.classes.Parent

t1t2_join = select([t1.c.x], from_obj=[t1.join(t2)]).alias()

# if its in the with_polymorphic, then its OK

mapper(

Parent,

t2,

polymorphic_on=t1t2_join.c.x,

with_polymorphic=("*", t1t2_join),

polymorphic_identity=0,

)

def test_polymorphic_on_not_in_with_poly(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent = self.classes.Parent

t1t2_join = select([t1.c.x], from_obj=[t1.join(t2)]).alias()

# if with_polymorphic, but its not present, not OK

def go():

t1t2_join_2 = select([t1.c.q], from_obj=[t1.join(t2)]).alias()

mapper(

Parent,

t2,

polymorphic_on=t1t2_join.c.x,

with_polymorphic=("*", t1t2_join_2),

polymorphic_identity=0,

)

assert_raises_message(

sa_exc.InvalidRequestError,

"Could not map polymorphic_on column 'x' "

"to the mapped table - "

"polymorphic loads will not function properly",

go,

)

def test_polymorphic_on_expr_explicit_map(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")])

mapper(

Parent,

t1,

properties={"discriminator": column_property(expr)},

polymorphic_identity="parent",

polymorphic_on=expr,

)

mapper(Child, t2, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_expr_implicit_map_no_label_joined(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")])

mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr)

mapper(Child, t2, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_expr_implicit_map_w_label_joined(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case(

[(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]

).label(None)

mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr)

mapper(Child, t2, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_expr_implicit_map_no_label_single(self):

"""test that single_table_criterion is propagated

with a standalone expr"""

t1 = self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")])

mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr)

mapper(Child, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_expr_implicit_map_w_label_single(self):

"""test that single_table_criterion is propagated

with a standalone expr"""

t1 = self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case(

[(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")]

).label(None)

mapper(Parent, t1, polymorphic_identity="parent", polymorphic_on=expr)

mapper(Child, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_column_prop(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")])

cprop = column_property(expr)

mapper(

Parent,

t1,

properties={"discriminator": cprop},

polymorphic_identity="parent",

polymorphic_on=cprop,

)

mapper(Child, t2, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_column_str_prop(self):

t2, t1 = self.tables.t2, self.tables.t1

Parent, Child = self.classes.Parent, self.classes.Child

expr = case([(t1.c.x == "p", "parent"), (t1.c.x == "c", "child")])

cprop = column_property(expr)

mapper(

Parent,

t1,

properties={"discriminator": cprop},

polymorphic_identity="parent",

polymorphic_on="discriminator",

)

mapper(Child, t2, inherits=Parent, polymorphic_identity="child")

self._roundtrip(parent_ident="p", child_ident="c")

def test_polymorphic_on_synonym(self):

t1 = self.tables.t1

Parent = self.classes.Parent

cprop = column_property(t1.c.x)

assert_raises_message(

sa_exc.ArgumentError,

"Only direct column-mapped property or "

"SQL expression can be passed for polymorphic_on",

mapper,

Parent,

t1,

properties={"discriminator": cprop, "discrim_syn": synonym(cprop)},

polymorphic_identity="parent",

polymorphic_on="discrim_syn",

)

def _roundtrip(

self, set_event=True, parent_ident="parent", child_ident="child"

):

Parent, Child = self.classes.Parent, self.classes.Child

# locate the "polymorphic_on" ColumnProperty. This isn't

# "officially" stored at the moment so do some heuristics to find it.

parent_mapper = inspect(Parent)

for prop in parent_mapper.column_attrs:

if not prop.instrument:

break

else:

prop = parent_mapper._columntoproperty[

parent_mapper.polymorphic_on

]

# then make sure the column we will query on matches.

is_(parent_mapper.polymorphic_on, prop.columns[0])

if set_event:

@event.listens_for(Parent, "init", propagate=True)

def set_identity(instance, *arg, **kw):

ident = object_mapper(instance).polymorphic_identity

if ident == "parent":

instance.x = parent_ident

elif ident == "child":

instance.x = child_ident

else:

assert False, "Got unexpected identity %r" % ident

s = Session(testing.db)

s.add_all([Parent(q="p1"), Child(q="c1", y="c1"), Parent(q="p2")])

s.commit()

s.close()

eq_(

[type(t) for t in s.query(Parent).order_by(Parent.id)],

[Parent, Child, Parent],

)

eq_([type(t) for t in s.query(Child).all()], [Child])

class SortOnlyOnImportantFKsTest(fixtures.MappedTest):

@classmethod

def define_tables(cls, metadata):

Table(

"a",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column(

"b_id",

Integer,

ForeignKey("b.id", use_alter=True, name="b_fk"),

),

)

Table(

"b",

metadata,

Column("id", Integer, ForeignKey("a.id"), primary_key=True),

)

@classmethod

def setup_classes(cls):

Base = declarative_base()

class A(Base):

__tablename__ = "a"

id = Column(

Integer, primary_key=True, test_needs_autoincrement=True

)

b_id = Column(Integer, ForeignKey("b.id"))

class B(A):

__tablename__ = "b"

id = Column(Integer, ForeignKey("a.id"), primary_key=True)

__mapper_args__ = {"inherit_condition": id == A.id}

cls.classes.A = A

cls.classes.B = B

def test_flush(self):

s = Session(testing.db)

s.add(self.classes.B())

s.flush()

class FalseDiscriminatorTest(fixtures.MappedTest):

@classmethod

def define_tables(cls, metadata):

global t1

t1 = Table(

"t1",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("type", Boolean, nullable=False),

)

def test_false_on_sub(self):

class Foo(object):

pass

class Bar(Foo):

pass

mapper(Foo, t1, polymorphic_on=t1.c.type, polymorphic_identity=True)

mapper(Bar, inherits=Foo, polymorphic_identity=False)

sess = create_session()

b1 = Bar()

sess.add(b1)

sess.flush()

assert b1.type is False

sess.expunge_all()

assert isinstance(sess.query(Foo).one(), Bar)

def test_false_on_base(self):

class Ding(object):

pass

class Bat(Ding):

pass

mapper(Ding, t1, polymorphic_on=t1.c.type, polymorphic_identity=False)

mapper(Bat, inherits=Ding, polymorphic_identity=True)

sess = create_session()

d1 = Ding()

sess.add(d1)

sess.flush()

assert d1.type is False

sess.expunge_all()

assert sess.query(Ding).one() is not None

class PolymorphicSynonymTest(fixtures.MappedTest):

@classmethod

def define_tables(cls, metadata):

global t1, t2

t1 = Table(

"t1",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("type", String(10), nullable=False),

Column("info", String(255)),

)

t2 = Table(

"t2",

metadata,

Column("id", Integer, ForeignKey("t1.id"), primary_key=True),

Column("data", String(10), nullable=False),

)

def test_polymorphic_synonym(self):

class T1(fixtures.ComparableEntity):

def info(self):

return "THE INFO IS:" + self._info

def _set_info(self, x):

self._info = x

info = property(info, _set_info)

class T2(T1):

pass

mapper(

T1,

t1,

polymorphic_on=t1.c.type,

polymorphic_identity="t1",

properties={"info": synonym("_info", map_column=True)},

)

mapper(T2, t2, inherits=T1, polymorphic_identity="t2")

sess = create_session()

at1 = T1(info="at1")

at2 = T2(info="at2", data="t2 data")

sess.add(at1)

sess.add(at2)

sess.flush()

sess.expunge_all()

eq_(sess.query(T2).filter(T2.info == "at2").one(), at2)

eq_(at2.info, "THE INFO IS:at2")

class PolymorphicAttributeManagementTest(fixtures.MappedTest):

"""Test polymorphic_on can be assigned, can be mirrored, etc."""

run_setup_mappers = "once"

@classmethod

def define_tables(cls, metadata):

Table(

"table_a",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("class_name", String(50)),

)

Table(

"table_b",

metadata,

Column("id", Integer, ForeignKey("table_a.id"), primary_key=True),

Column("class_name", String(50)),

)

Table(

"table_c",

metadata,

Column("id", Integer, ForeignKey("table_b.id"), primary_key=True),

Column("data", String(10)),

)

@classmethod

def setup_classes(cls):

table_b, table_c, table_a = (

cls.tables.table_b,

cls.tables.table_c,

cls.tables.table_a,

)

class A(cls.Basic):

pass

class B(A):

pass

class C(B):

pass

class D(B):

pass

mapper(

A,

table_a,

polymorphic_on=table_a.c.class_name,

polymorphic_identity="a",

)

mapper(

B,

table_b,

inherits=A,

polymorphic_on=table_b.c.class_name,

polymorphic_identity="b",

properties=dict(

class_name=[table_a.c.class_name, table_b.c.class_name]

),

)

mapper(C, table_c, inherits=B, polymorphic_identity="c")

mapper(D, inherits=B, polymorphic_identity="d")

def test_poly_configured_immediate(self):

A, C, B = (self.classes.A, self.classes.C, self.classes.B)

a = A()

b = B()

c = C()

eq_(a.class_name, "a")

eq_(b.class_name, "b")

eq_(c.class_name, "c")

def test_base_class(self):

A, C, B = (self.classes.A, self.classes.C, self.classes.B)

sess = Session()

c1 = C()

sess.add(c1)

sess.commit()

assert isinstance(sess.query(B).first(), C)

sess.close()

assert isinstance(sess.query(A).first(), C)

def test_valid_assignment_upwards(self):

"""test that we can assign 'd' to a B, since B/D

both involve the same set of tables.

"""

D, B = self.classes.D, self.classes.B

sess = Session()

b1 = B()

b1.class_name = "d"

sess.add(b1)

sess.commit()

sess.close()

assert isinstance(sess.query(B).first(), D)

def test_invalid_assignment_downwards(self):

"""test that we warn on assign of 'b' to a C, since this adds

a row to the C table we'd never load.

"""

C = self.classes.C

sess = Session()

c1 = C()

c1.class_name = "b"

sess.add(c1)

assert_raises_message(

sa_exc.SAWarning,

"Flushing object %s with incompatible "

"polymorphic identity 'b'; the object may not "

"refresh and/or load correctly" % instance_str(c1),

sess.flush,

)

def test_invalid_assignment_upwards(self):

"""test that we warn on assign of 'c' to a B, since we will have a

"C" row that has no joined row, which will cause object

deleted errors.

"""

B = self.classes.B

sess = Session()

b1 = B()

b1.class_name = "c"

sess.add(b1)

assert_raises_message(

sa_exc.SAWarning,

"Flushing object %s with incompatible "

"polymorphic identity 'c'; the object may not "

"refresh and/or load correctly" % instance_str(b1),

sess.flush,

)

def test_entirely_oob_assignment(self):

"""test warn on an unknown polymorphic identity.

"""

B = self.classes.B

sess = Session()

b1 = B()

b1.class_name = "xyz"

sess.add(b1)

assert_raises_message(

sa_exc.SAWarning,

"Flushing object %s with incompatible "

"polymorphic identity 'xyz'; the object may not "

"refresh and/or load correctly" % instance_str(b1),

sess.flush,

)

def test_not_set_on_upate(self):

C = self.classes.C

sess = Session()

c1 = C()

sess.add(c1)

sess.commit()

sess.expire(c1)

c1.data = "foo"

sess.flush()

def test_validate_on_upate(self):

C = self.classes.C

sess = Session()

c1 = C()

sess.add(c1)

sess.commit()

sess.expire(c1)

c1.class_name = "b"

assert_raises_message(

sa_exc.SAWarning,

"Flushing object %s with incompatible "

"polymorphic identity 'b'; the object may not "

"refresh and/or load correctly" % instance_str(c1),

sess.flush,

)

class CascadeTest(fixtures.MappedTest):

"""that cascades on polymorphic relationships continue

cascading along the path of the instance's mapper, not

the base mapper."""

@classmethod

def define_tables(cls, metadata):

global t1, t2, t3, t4

t1 = Table(

"t1",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("data", String(30)),

)

t2 = Table(

"t2",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("t1id", Integer, ForeignKey("t1.id")),

Column("type", String(30)),

Column("data", String(30)),

)

t3 = Table(

"t3",

metadata,

Column("id", Integer, ForeignKey("t2.id"), primary_key=True),

Column("moredata", String(30)),

)

t4 = Table(

"t4",

metadata,

Column(

"id", Integer, primary_key=True, test_needs_autoincrement=True

),

Column("t3id", Integer, ForeignKey("t3.id")),

Column("data", String(30)),

)

def test_cascade(self):

class T1(fixtures.BasicEntity):

pass

class T2(fixtures.BasicEntity):

pass

class T3(T2):

pass

class T4(fixtures.BasicEntity):

pass