+++ /dev/null
-"""copy of ComparableEntity and eq_() from test.lib.
-
-This is just to support running the example outside of
-the SQLA testing environment which is no longer part of
-SQLAlchemy as of 0.7.
-
-"""
-
-import sqlalchemy as sa
-from sqlalchemy import exc as sa_exc
-
-
-def eq_(a, b, msg=None):
- """Assert a == b, with repr messaging on failure."""
- assert a == b, msg or "%r != %r" % (a, b)
-
-_repr_stack = set()
-class BasicEntity(object):
- def __init__(self, **kw):
- for key, value in kw.items():
- setattr(self, key, value)
-
- def __repr__(self):
- if id(self) in _repr_stack:
- return object.__repr__(self)
- _repr_stack.add(id(self))
- try:
- return "%s(%s)" % (
- (self.__class__.__name__),
- ', '.join(["%s=%r" % (key, getattr(self, key))
- for key in sorted(self.__dict__.keys())
- if not key.startswith('_')]))
- finally:
- _repr_stack.remove(id(self))
-
-_recursion_stack = set()
-class ComparableEntity(BasicEntity):
- def __hash__(self):
- return hash(self.__class__)
-
- def __ne__(self, other):
- return not self.__eq__(other)
-
- def __eq__(self, other):
- """'Deep, sparse compare.
-
- Deeply compare two entities, following the non-None attributes of the
- non-persisted object, if possible.
-
- """
- if other is self:
- return True
- elif not self.__class__ == other.__class__:
- return False
-
- if id(self) in _recursion_stack:
- return True
- _recursion_stack.add(id(self))
-
- try:
- # pick the entity thats not SA persisted as the source
- try:
- self_key = sa.orm.attributes.instance_state(self).key
- except sa.orm.exc.NO_STATE:
- self_key = None
-
- if other is None:
- a = self
- b = other
- elif self_key is not None:
- a = other
- b = self
- else:
- a = self
- b = other
-
- for attr in a.__dict__.keys():
- if attr.startswith('_'):
- continue
- value = getattr(a, attr)
-
- try:
- # handle lazy loader errors
- battr = getattr(b, attr)
- except (AttributeError, sa_exc.UnboundExecutionError):
- return False
-
- if hasattr(value, '__iter__'):
- if list(value) != list(battr):
- return False
- else:
- if value is not None and value != battr:
- return False
- return True
- finally:
- _recursion_stack.remove(id(self))
from .history_meta import Versioned, versioned_session
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import clear_mappers, Session, deferred, relationship
-from ._lib import ComparableEntity, eq_
+from sqlalchemy.testing import AssertsCompiledSQL, eq_
+from sqlalchemy.testing.entities import BasicEntity, ComparableEntity
engine = None
global engine
engine = create_engine('sqlite://', echo=True)
-class TestVersioning(TestCase):
+class TestVersioning(TestCase, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
def setUp(self):
self.session = Session(engine)
self.Base = declarative_base()
]
)
+ def test_joined_inheritance_multilevel(self):
+ class BaseClass(Versioned, self.Base, ComparableEntity):
+ __tablename__ = 'basetable'
+
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+ type = Column(String(20))
+
+ __mapper_args__ = {'polymorphic_on': type,
+ 'polymorphic_identity': 'base'}
+
+ class SubClass(BaseClass):
+ __tablename__ = 'subtable'
+
+ id = Column(Integer, primary_key=True)
+ base_id = Column(Integer, ForeignKey('basetable.id'))
+ subdata1 = Column(String(50))
+
+ __mapper_args__ = {'polymorphic_identity': 'sub'}
+
+ class SubSubClass(SubClass):
+ __tablename__ = 'subsubtable'
+
+ id = Column(Integer, ForeignKey('subtable.id'), primary_key=True)
+ subdata2 = Column(String(50))
+
+ __mapper_args__ = {'polymorphic_identity': 'subsub'}
+
+ self.create_tables()
+
+ SubSubHistory = SubSubClass.__history_mapper__.class_
+ sess = self.session
+ q = sess.query(SubSubHistory)
+ self.assert_compile(
+ q,
+ "SELECT subsubtable_history.id AS subsubtable_history_id, "
+ "subtable_history.id AS subtable_history_id, "
+ "basetable_history.id AS basetable_history_id, "
+ "basetable_history.name AS basetable_history_name, "
+ "basetable_history.type AS basetable_history_type, "
+ "subsubtable_history.version AS subsubtable_history_version, "
+ "subtable_history.version AS subtable_history_version, "
+ "basetable_history.version AS basetable_history_version, "
+ "subtable_history.base_id AS subtable_history_base_id, "
+ "subtable_history.subdata1 AS subtable_history_subdata1, "
+ "subsubtable_history.subdata2 AS subsubtable_history_subdata2 "
+ "FROM basetable_history "
+ "JOIN subtable_history "
+ "ON basetable_history.id = subtable_history.base_id "
+ "AND basetable_history.version = subtable_history.version "
+ "JOIN subsubtable_history ON subtable_history.id = "
+ "subsubtable_history.id AND subtable_history.version = subsubtable_history.version"
+ )
+
+ ssc = SubSubClass(name='ss1', subdata1='sd1', subdata2='sd2')
+ sess.add(ssc)
+ sess.commit()
+ eq_(
+ sess.query(SubSubHistory).all(),
+ []
+ )
+ ssc.subdata1 = 'sd11'
+ ssc.subdata2 = 'sd22'
+ sess.commit()
+ eq_(
+ sess.query(SubSubHistory).all(),
+ [SubSubHistory(name='ss1', subdata1='sd1',
+ subdata2='sd2', type='subsub', version=1)]
+ )
+ eq_(ssc, SubSubClass(name='ss1', subdata1='sd11',
+ subdata2='sd22', version=2))
+
+
+
def test_single_inheritance(self):
class BaseClass(Versioned, self.Base, ComparableEntity):
__tablename__ = 'basetable'
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
