summaryrefslogtreecommitdiff
path: root/test/ext/test_associationproxy.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2010-07-11 13:41:38 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2010-07-11 13:41:38 -0400
commitf2b43da1a8e3fa8f2afc49e04ec16479a6e7da64 (patch)
tree90f12adf613f4a97c8ad5861560f86abd8263ccb /test/ext/test_associationproxy.py
parenta6b62cc3fed5f06d3428b1f6ee13756175ded61b (diff)
downloadsqlalchemy-f2b43da1a8e3fa8f2afc49e04ec16479a6e7da64.tar.gz
tidy test/base, test/ex, test/ext
Diffstat (limited to 'test/ext/test_associationproxy.py')
-rw-r--r--test/ext/test_associationproxy.py263
1 files changed, 137 insertions, 126 deletions
diff --git a/test/ext/test_associationproxy.py b/test/ext/test_associationproxy.py
index df51c322b..d22e45796 100644
--- a/test/ext/test_associationproxy.py
+++ b/test/ext/test_associationproxy.py
@@ -610,8 +610,9 @@ class CustomObjectTest(_CollectionOperations):
p = self.roundtrip(p)
self.assert_(len(list(p.children)) == 1)
- # We didn't provide an alternate _AssociationList implementation for
- # our ObjectCollection, so indexing will fail.
+ # We didn't provide an alternate _AssociationList implementation
+ # for our ObjectCollection, so indexing will fail.
+
try:
v = p.children[1]
self.fail()
@@ -635,10 +636,22 @@ class ProxyFactoryTest(ListTest):
Column('name', String(128)))
class CustomProxy(_AssociationList):
- def __init__(self, lazy_collection, creator, value_attr, parent):
+ def __init__(
+ self,
+ lazy_collection,
+ creator,
+ value_attr,
+ parent,
+ ):
getter, setter = parent._default_getset(lazy_collection)
- _AssociationList.__init__(self, lazy_collection, creator, getter, setter, parent)
-
+ _AssociationList.__init__(
+ self,
+ lazy_collection,
+ creator,
+ getter,
+ setter,
+ parent,
+ )
class Parent(object):
children = association_proxy('_children', 'name',
@@ -897,102 +910,94 @@ class LazyLoadTest(TestBase):
class ReconstitutionTest(TestBase):
+
def setup(self):
metadata = MetaData(testing.db)
- parents = Table('parents', metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('name', String(30)))
- children = Table('children', metadata,
- Column('id', Integer, primary_key=True,
- test_needs_autoincrement=True),
- Column('parent_id', Integer, ForeignKey('parents.id')),
- Column('name', String(30)))
+ parents = Table('parents', metadata, Column('id', Integer,
+ primary_key=True,
+ test_needs_autoincrement=True), Column('name',
+ String(30)))
+ children = Table('children', metadata, Column('id', Integer,
+ primary_key=True,
+ test_needs_autoincrement=True),
+ Column('parent_id', Integer,
+ ForeignKey('parents.id')), Column('name',
+ String(30)))
metadata.create_all()
parents.insert().execute(name='p1')
-
-
self.metadata = metadata
self.parents = parents
self.children = children
-
+
def teardown(self):
self.metadata.drop_all()
clear_mappers()
def test_weak_identity_map(self):
- mapper(Parent, self.parents, properties=dict(children=relationship(Child)))
+ mapper(Parent, self.parents,
+ properties=dict(children=relationship(Child)))
mapper(Child, self.children)
-
session = create_session(weak_identity_map=True)
def add_child(parent_name, child_name):
- parent = (session.query(Parent).
- filter_by(name=parent_name)).one()
+ parent = \
+ session.query(Parent).filter_by(name=parent_name).one()
parent.kids.append(child_name)
-
add_child('p1', 'c1')
gc_collect()
add_child('p1', 'c2')
-
session.flush()
p = session.query(Parent).filter_by(name='p1').one()
assert set(p.kids) == set(['c1', 'c2']), p.kids
def test_copy(self):
- mapper(Parent, self.parents, properties=dict(children=relationship(Child)))
+ mapper(Parent, self.parents,
+ properties=dict(children=relationship(Child)))
mapper(Child, self.children)
-
p = Parent('p1')
p.kids.extend(['c1', 'c2'])
p_copy = copy.copy(p)
del p
gc_collect()
-
assert set(p_copy.kids) == set(['c1', 'c2']), p.kids
def test_pickle_list(self):
- mapper(Parent, self.parents, properties=dict(children=relationship(Child)))
+ mapper(Parent, self.parents,
+ properties=dict(children=relationship(Child)))
mapper(Child, self.children)
-
p = Parent('p1')
p.kids.extend(['c1', 'c2'])
-
r1 = pickle.loads(pickle.dumps(p))
assert r1.kids == ['c1', 'c2']
-
r2 = pickle.loads(pickle.dumps(p.kids))
assert r2 == ['c1', 'c2']
def test_pickle_set(self):
- mapper(Parent, self.parents, properties=dict(children=relationship(Child, collection_class=set)))
+ mapper(Parent, self.parents,
+ properties=dict(children=relationship(Child,
+ collection_class=set)))
mapper(Child, self.children)
-
p = Parent('p1')
p.kids.update(['c1', 'c2'])
-
r1 = pickle.loads(pickle.dumps(p))
assert r1.kids == set(['c1', 'c2'])
-
r2 = pickle.loads(pickle.dumps(p.kids))
assert r2 == set(['c1', 'c2'])
def test_pickle_dict(self):
- mapper(Parent, self.parents, properties=dict(
- children=relationship(KVChild, collection_class=collections.mapped_collection(PickleKeyFunc('name')))
- ))
+ mapper(Parent, self.parents,
+ properties=dict(children=relationship(KVChild,
+ collection_class=
+ collections.mapped_collection(PickleKeyFunc('name')))))
mapper(KVChild, self.children)
-
p = Parent('p1')
- p.kids.update({'c1':'v1', 'c2':'v2'})
- assert p.kids == {'c1':'c1', 'c2':'c2'}
-
+ p.kids.update({'c1': 'v1', 'c2': 'v2'})
+ assert p.kids == {'c1': 'c1', 'c2': 'c2'}
r1 = pickle.loads(pickle.dumps(p))
- assert r1.kids == {'c1':'c1', 'c2':'c2'}
-
+ assert r1.kids == {'c1': 'c1', 'c2': 'c2'}
r2 = pickle.loads(pickle.dumps(p.kids))
- assert r2 == {'c1':'c1', 'c2':'c2'}
+ assert r2 == {'c1': 'c1', 'c2': 'c2'}
class PickleKeyFunc(object):
def __init__(self, name):
@@ -1002,40 +1007,37 @@ class PickleKeyFunc(object):
return getattr(obj, self.name)
class ComparatorTest(_base.MappedTest):
+
run_inserts = 'once'
run_deletes = None
run_setup_mappers = 'once'
+ run_setup_classes = 'once'
@classmethod
def define_tables(cls, metadata):
-
- Table(
- 'userkeywords', metadata,
- Column('keyword_id', Integer, ForeignKey('keywords.id'),
- primary_key=True),
- Column('user_id', Integer, ForeignKey('users.id')))
-
- Table(
- 'users', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('name', String(64)))
-
- Table(
- 'keywords', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('keyword', String(64)))
+ Table('userkeywords', metadata, Column('keyword_id', Integer,
+ ForeignKey('keywords.id'), primary_key=True),
+ Column('user_id', Integer, ForeignKey('users.id')))
+ Table('users', metadata, Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(64)))
+ Table('keywords', metadata, Column('id', Integer,
+ primary_key=True, test_needs_autoincrement=True),
+ Column('keyword', String(64)))
@classmethod
def setup_classes(cls):
class User(_base.ComparableEntity):
def __init__(self, name):
self.name = name
+
keywords = association_proxy('user_keywords', 'keyword',
- creator=lambda k: UserKeyword(keyword=k))
+ creator=lambda k: UserKeyword(keyword=k))
class Keyword(_base.ComparableEntity):
def __init__(self, keyword):
self.keyword = keyword
+
user = association_proxy('user_keyword', 'user')
class UserKeyword(_base.ComparableEntity):
@@ -1046,32 +1048,35 @@ class ComparatorTest(_base.MappedTest):
@classmethod
@testing.resolve_artifact_names
def setup_mappers(cls):
-
mapper(User, users)
- mapper(Keyword, keywords, properties={
- 'user_keyword': relationship(UserKeyword, uselist=False)
- })
- mapper(UserKeyword, userkeywords, properties={
- 'user': relationship(User, backref='user_keywords'),
- 'keyword': relationship(Keyword),
- })
+ mapper(Keyword, keywords, properties={'user_keyword'
+ : relationship(UserKeyword, uselist=False)})
+ mapper(UserKeyword, userkeywords, properties={'user'
+ : relationship(User, backref='user_keywords'), 'keyword'
+ : relationship(Keyword)})
@classmethod
@testing.resolve_artifact_names
def insert_data(cls):
session = sessionmaker()()
- words = ('quick', 'brown', 'fox', 'jumped', 'over', 'the', 'lazy')
+ words = (
+ 'quick',
+ 'brown',
+ 'fox',
+ 'jumped',
+ 'over',
+ 'the',
+ 'lazy',
+ )
for ii in range(4):
user = User('user%d' % ii)
session.add(user)
- for jj in words[ii:ii+3]:
+ for jj in words[ii:ii + 3]:
user.keywords.append(Keyword(jj))
-
orphan = Keyword('orphan')
orphan.user_keyword = UserKeyword(keyword=orphan, user=None)
session.add(orphan)
session.commit()
-
cls.u = user
cls.kw = user.keywords[0]
cls.session = session
@@ -1081,88 +1086,94 @@ class ComparatorTest(_base.MappedTest):
@testing.resolve_artifact_names
def test_filter_any_kwarg(self):
- self._equivalent(
- self.session.query(User).\
- filter(User.keywords.any(keyword='jumped')),
- self.session.query(User).\
- filter(User.user_keywords.any(
- UserKeyword.keyword.has(keyword='jumped'))))
+ self._equivalent(self.session.query(User).
+ filter(User.keywords.any(keyword='jumped'
+ )),
+ self.session.query(User).filter(
+ User.user_keywords.any(
+ UserKeyword.keyword.has(keyword='jumped'
+ ))))
@testing.resolve_artifact_names
def test_filter_has_kwarg(self):
- self._equivalent(
- self.session.query(Keyword).\
- filter(Keyword.user.has(name='user2')),
- self.session.query(Keyword).\
- filter(Keyword.user_keyword.has(
- UserKeyword.user.has(name='user2'))))
+ self._equivalent(self.session.query(Keyword).
+ filter(Keyword.user.has(name='user2'
+ )),
+ self.session.query(Keyword).
+ filter(Keyword.user_keyword.has(
+ UserKeyword.user.has(name='user2'
+ ))))
@testing.resolve_artifact_names
def test_filter_any_criterion(self):
- self._equivalent(
- self.session.query(User).\
- filter(User.keywords.any(Keyword.keyword == 'jumped')),
- self.session.query(User).\
- filter(User.user_keywords.any(
- UserKeyword.keyword.has(Keyword.keyword == 'jumped'))))
+ self._equivalent(self.session.query(User).
+ filter(User.keywords.any(Keyword.keyword
+ == 'jumped')),
+ self.session.query(User).
+ filter(User.user_keywords.any(
+ UserKeyword.keyword.has(Keyword.keyword
+ == 'jumped'))))
@testing.resolve_artifact_names
def test_filter_has_criterion(self):
- self._equivalent(
- self.session.query(Keyword).\
- filter(Keyword.user.has(User.name == 'user2')),
- self.session.query(Keyword).\
- filter(Keyword.user_keyword.has(
- UserKeyword.user.has(User.name == 'user2'))))
-
+ self._equivalent(self.session.query(Keyword).
+ filter(Keyword.user.has(User.name
+ == 'user2')),
+ self.session.query(Keyword).
+ filter(Keyword.user_keyword.has(
+ UserKeyword.user.has(User.name
+ == 'user2'))))
+
@testing.resolve_artifact_names
def test_filter_contains(self):
- self._equivalent(
- self.session.query(User).\
- filter(User.keywords.contains(self.kw)),
- self.session.query(User).\
- filter(User.user_keywords.any(keyword=self.kw)))
-
+ self._equivalent(self.session.query(User).
+ filter(User.keywords.contains(self.kw)),
+ self.session.query(User).
+ filter(User.user_keywords.any(keyword=self.kw)))
+
@testing.resolve_artifact_names
def test_filter_eq(self):
- self._equivalent(
- self.session.query(Keyword).\
- filter(Keyword.user == self.u),
- self.session.query(Keyword).\
- filter(Keyword.user_keyword.has(user=self.u)))
-
+ self._equivalent(self.session.query(Keyword).filter(Keyword.user
+ == self.u),
+ self.session.query(Keyword).
+ filter(Keyword.user_keyword.has(user=self.u)))
+
@testing.resolve_artifact_names
def test_filter_ne(self):
- self._equivalent(
- self.session.query(Keyword).\
- filter(Keyword.user != self.u),
- self.session.query(Keyword).\
- filter(not_(Keyword.user_keyword.has(user=self.u))))
+ self._equivalent(self.session.query(Keyword).filter(Keyword.user
+ != self.u),
+ self.session.query(Keyword).
+ filter(not_(Keyword.user_keyword.has(user=self.u))))
@testing.resolve_artifact_names
def test_filter_eq_null(self):
- self._equivalent(
- self.session.query(Keyword).\
- filter(Keyword.user == None),
- self.session.query(Keyword).\
- filter(Keyword.user_keyword.has(UserKeyword.user == None)))
+ self._equivalent(self.session.query(Keyword).filter(Keyword.user
+ == None),
+ self.session.query(Keyword).
+ filter(Keyword.user_keyword.has(UserKeyword.user
+ == None)))
@testing.resolve_artifact_names
def test_filter_scalar_contains_fails(self):
- assert_raises(exceptions.InvalidRequestError, lambda: Keyword.user.contains(self.u))
-
+ assert_raises(exceptions.InvalidRequestError, lambda : \
+ Keyword.user.contains(self.u))
+
@testing.resolve_artifact_names
def test_filter_scalar_any_fails(self):
- assert_raises(exceptions.InvalidRequestError, lambda: Keyword.user.any(name='user2'))
+ assert_raises(exceptions.InvalidRequestError, lambda : \
+ Keyword.user.any(name='user2'))
@testing.resolve_artifact_names
def test_filter_collection_has_fails(self):
- assert_raises(exceptions.InvalidRequestError, lambda: User.keywords.has(keyword='quick'))
+ assert_raises(exceptions.InvalidRequestError, lambda : \
+ User.keywords.has(keyword='quick'))
@testing.resolve_artifact_names
def test_filter_collection_eq_fails(self):
- assert_raises(exceptions.InvalidRequestError, lambda: User.keywords == self.kw)
+ assert_raises(exceptions.InvalidRequestError, lambda : \
+ User.keywords == self.kw)
@testing.resolve_artifact_names
def test_filter_collection_ne_fails(self):
- assert_raises(exceptions.InvalidRequestError, lambda: User.keywords != self.kw)
+ assert_raises(exceptions.InvalidRequestError, lambda : \
+ User.keywords != self.kw)