summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJason Kirtland <jek@discorporate.us>2007-11-05 22:11:30 +0000
committerJason Kirtland <jek@discorporate.us>2007-11-05 22:11:30 +0000
commitbcb586a93c68b4f58ded49b253448f4fe32c558a (patch)
tree91fa5e449e44ef4549c76fa49510cf2e724e921d
parentd36a3940a0958f444ad9ca9a2e6b5eb54fab9b80 (diff)
downloadsqlalchemy-bcb586a93c68b4f58ded49b253448f4fe32c558a.tar.gz
Migrated maxdb behavioral assumptions from unsupported to fails_on
-rw-r--r--test/orm/assorted_eager.py95
-rw-r--r--test/orm/cascade.py86
-rw-r--r--test/orm/dynamic.py34
-rw-r--r--test/orm/eager_relations.py130
-rw-r--r--test/orm/generative.py33
-rw-r--r--test/orm/inheritance/basic.py67
-rw-r--r--test/orm/inheritance/manytomany.py37
-rw-r--r--test/orm/inheritance/poly_linked_list.py44
-rw-r--r--test/orm/mapper.py226
-rw-r--r--test/orm/query.py139
-rw-r--r--test/sql/case_statement.py16
-rw-r--r--test/sql/query.py182
12 files changed, 558 insertions, 531 deletions
diff --git a/test/orm/assorted_eager.py b/test/orm/assorted_eager.py
index 25a848976..194a836c0 100644
--- a/test/orm/assorted_eager.py
+++ b/test/orm/assorted_eager.py
@@ -12,7 +12,7 @@ class EagerTest(AssertMixin):
def setUpAll(self):
global dbmeta, owners, categories, tests, options, Owner, Category, Test, Option, false
dbmeta = MetaData(testbase.db)
-
+
# determine a literal value for "false" based on the dialect
# FIXME: this PassiveDefault setup is bogus.
bp = Boolean().dialect_impl(testbase.db.dialect).bind_processor(testbase.db.dialect)
@@ -22,7 +22,7 @@ class EagerTest(AssertMixin):
false = text('FALSE')
else:
false = str(False)
-
+
owners = Table ( 'owners', dbmeta ,
Column ( 'id', Integer, primary_key=True, nullable=False ),
Column('data', String(30)) )
@@ -56,7 +56,7 @@ class EagerTest(AssertMixin):
'category':relation(Category),
'owner_option': relation(Option,primaryjoin=and_(tests.c.id==options.c.test_id,tests.c.owner_id==options.c.owner_id),
foreignkey=[options.c.test_id, options.c.owner_id],
- uselist=False )
+ uselist=False )
})
s=create_session()
@@ -92,7 +92,7 @@ class EagerTest(AssertMixin):
def test_noorm(self):
"""test the control case"""
- # I want to display a list of tests owned by owner 1
+ # I want to display a list of tests owned by owner 1
# if someoption is false or he hasn't specified it yet (null)
# but not if he set it to true (example someoption is for hiding)
@@ -109,7 +109,7 @@ class EagerTest(AssertMixin):
from_obj=[tests.join(categories).outerjoin(options,and_(tests.c.id==options.c.test_id,tests.c.owner_id==options.c.owner_id))] ).execute().fetchall()
print result
assert result == [(1, u'Some Category'), (3, u'Some Category')]
-
+
def test_withouteagerload(self):
s = create_session()
l=s.query(Test).select ( and_(tests.c.owner_id==1,or_(options.c.someoption==None,options.c.someoption==False)),
@@ -119,7 +119,7 @@ class EagerTest(AssertMixin):
assert result == [u'1 Some Category', u'3 Some Category']
def test_witheagerload(self):
- """test that an eagerload locates the correct "from" clause with
+ """test that an eagerload locates the correct "from" clause with
which to attach to, when presented with a query that already has a complicated from clause."""
s = create_session()
q=s.query(Test).options(eagerload('category'))
@@ -133,10 +133,10 @@ class EagerTest(AssertMixin):
"""test the same as witheagerload except using generative"""
s = create_session()
q=s.query(Test).options(eagerload('category'))
- l=q.filter (
+ l=q.filter (
and_(tests.c.owner_id==1,or_(options.c.someoption==None,options.c.someoption==False))
).outerjoin('owner_option')
-
+
result = ["%d %s" % ( t.id,t.category.name ) for t in l]
print result
assert result == [u'1 Some Category', u'3 Some Category']
@@ -146,7 +146,7 @@ class EagerTest(AssertMixin):
s = create_session()
q=s.query(Test).options(eagerload('category'))
l=q.select( (tests.c.owner_id==1) & ('options.someoption is null or options.someoption=%s' % false) & q.join_to('owner_option') )
- result = ["%d %s" % ( t.id,t.category.name ) for t in l]
+ result = ["%d %s" % ( t.id,t.category.name ) for t in l]
print result
assert result == [u'3 Some Category']
@@ -154,7 +154,7 @@ class EagerTest(AssertMixin):
s = create_session()
q=s.query(Test).options(eagerload('category'))
l=q.select( (tests.c.owner_id==1) & ((options.c.someoption==None) | (options.c.someoption==False)) & q.join_to('owner_option') )
- result = ["%d %s" % ( t.id,t.category.name ) for t in l]
+ result = ["%d %s" % ( t.id,t.category.name ) for t in l]
print result
assert result == [u'3 Some Category']
@@ -182,10 +182,11 @@ class EagerTest2(AssertMixin):
def tearDown(self):
for t in metadata.table_iterator(reverse=True):
t.delete().execute()
-
+
+ @testing.fails_on('maxdb')
def testeagerterminate(self):
"""test that eager query generation does not include the same mapper's table twice.
-
+
or, that bi-directional eager loads dont include each other in eager query generation."""
class Middle(object):
def __init__(self, data): self.data = data
@@ -228,7 +229,8 @@ class EagerTest3(ORMTest):
Column ( 'id', Integer, primary_key=True, nullable=False ),
Column ( 'data_id', Integer, ForeignKey('datas.id')),
Column ( 'somedata', Integer, nullable=False ))
-
+
+ @testing.fails_on('maxdb')
def test_nesting_with_functions(self):
class Data(object): pass
class Foo(object):pass
@@ -245,7 +247,7 @@ class EagerTest3(ORMTest):
d.a=x
s.save(d)
data.append(d)
-
+
for x in range(10):
rid=random.randint(0,len(data) - 1)
somedata=random.randint(1,50000)
@@ -260,22 +262,22 @@ class EagerTest3(ORMTest):
[stats.c.data_id,func.max(stats.c.somedata).label('max')],
stats.c.data_id<=25,
group_by=[stats.c.data_id]).alias('arb')
-
+
arb_result = arb_data.execute().fetchall()
# order the result list descending based on 'max'
arb_result.sort(lambda a, b:cmp(b['max'],a['max']))
# extract just the "data_id" from it
arb_result = [row['data_id'] for row in arb_result]
-
- # now query for Data objects using that above select, adding the
+
+ # now query for Data objects using that above select, adding the
# "order by max desc" separately
q=s.query(Data).options(eagerload('foo')).select(
from_obj=[datas.join(arb_data,arb_data.c.data_id==datas.c.id)],
order_by=[desc(arb_data.c.max)],limit=10)
-
+
# extract "data_id" from the list of result objects
verify_result = [d.id for d in q]
-
+
# assert equality including ordering (may break if the DB "ORDER BY" and python's sort() used differing
# algorithms and there are repeated 'somedata' values in the list)
assert verify_result == arb_result
@@ -287,12 +289,13 @@ class EagerTest4(ORMTest):
Column('department_id', Integer, primary_key=True),
Column('name', String(50)))
- employees = Table('employees', metadata,
+ employees = Table('employees', metadata,
Column('person_id', Integer, primary_key=True),
Column('name', String(50)),
Column('department_id', Integer,
ForeignKey('departments.department_id')))
+ @testing.fails_on('maxdb')
def test_basic(self):
class Department(object):
def __init__(self, **kwargs):
@@ -333,13 +336,13 @@ class EagerTest4(ORMTest):
assert q[0] is d2
class EagerTest5(ORMTest):
- """test the construction of AliasedClauses for the same eager load property but different
+ """test the construction of AliasedClauses for the same eager load property but different
parent mappers, due to inheritance"""
def define_tables(self, metadata):
global base, derived, derivedII, comments
base = Table(
'base', metadata,
- Column('uid', String(30), primary_key=True),
+ Column('uid', String(30), primary_key=True),
Column('x', String(30))
)
@@ -426,15 +429,15 @@ class EagerTest5(ORMTest):
class EagerTest6(ORMTest):
def define_tables(self, metadata):
global designType, design, part, inheritedPart
- designType = Table('design_types', metadata,
+ designType = Table('design_types', metadata,
Column('design_type_id', Integer, primary_key=True),
)
- design =Table('design', metadata,
+ design =Table('design', metadata,
Column('design_id', Integer, primary_key=True),
Column('design_type_id', Integer, ForeignKey('design_types.design_type_id')))
- part = Table('parts', metadata,
+ part = Table('parts', metadata,
Column('part_id', Integer, primary_key=True),
Column('design_id', Integer, ForeignKey('design.design_id')),
Column('design_type_id', Integer, ForeignKey('design_types.design_type_id')))
@@ -466,9 +469,9 @@ class EagerTest6(ORMTest):
))
class_mapper(Design).add_property("type", relation(DesignType, lazy=False, backref="designs"))
-
+
class_mapper(Part).add_property("design", relation(Design, lazy=False, backref=backref("parts", cascade="all, delete-orphan")))
-
+
#Part.mapper.add_property("designType", relation(DesignType))
d = Design()
@@ -508,7 +511,7 @@ class EagerTest7(ORMTest):
invoice_table = Table('invoices', metadata,
Column('invoice_id', Integer, Sequence('invoice_id_seq', optional=True), primary_key = True),
Column('company_id', Integer, ForeignKey("companies.company_id")),
- Column('date', DateTime),
+ Column('date', DateTime),
)
items_table = Table('items', metadata,
@@ -543,7 +546,7 @@ class EagerTest7(ORMTest):
return "Item: " + repr(getattr(self, 'item_id', None)) + " " + repr(getattr(self, 'invoice_id', None)) + " " + repr(self.code) + " " + repr(self.qty)
def testone(self):
- """tests eager load of a many-to-one attached to a one-to-many. this testcase illustrated
+ """tests eager load of a many-to-one attached to a one-to-many. this testcase illustrated
the bug, which is that when the single Company is loaded, no further processing of the rows
occurred in order to load the Company's second Address object."""
@@ -729,6 +732,7 @@ class EagerTest8(ORMTest):
testbase.db.execute(task_type_t.insert(), {'id':1})
testbase.db.execute(task_t.insert(), {'title':'task 1', 'task_type_id':1, 'status_id':1, 'prj_id':1})
+ @testing.fails_on('maxdb')
def test_nested_joins(self):
# this is testing some subtle column resolution stuff,
# concerning corresponding_column() being extremely accurate
@@ -741,7 +745,7 @@ class EagerTest8(ORMTest):
tsk_cnt_join = outerjoin(project_t, task_t, task_t.c.prj_id==project_t.c.id)
- ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')],
+ ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')],
from_obj=[tsk_cnt_join], group_by=[project_t.c.id]).alias('prj_tsk_cnt_s')
j = join(project_t, ss, project_t.c.id == ss.c.prj_id)
@@ -753,12 +757,12 @@ class EagerTest8(ORMTest):
mapper(Message_Type, message_type_t)
- mapper(Message, message_t,
+ mapper(Message, message_t,
properties=dict(type=relation(Message_Type, lazy=False, uselist=False),
))
tsk_cnt_join = outerjoin(project_t, task_t, task_t.c.prj_id==project_t.c.id)
- ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')],
+ ss = select([project_t.c.id.label('prj_id'), func.count(task_t.c.id).label('tasks_number')],
from_obj=[tsk_cnt_join], group_by=[project_t.c.id]).alias('prj_tsk_cnt_s')
j = join(project_t, ss, project_t.c.id == ss.c.prj_id)
@@ -777,16 +781,16 @@ class EagerTest8(ORMTest):
session = create_session()
for t in session.query(cls.mapper).limit(10).offset(0).list():
- print t.id, t.title, t.props_cnt
+ print t.id, t.title, t.props_cnt
class EagerTest9(ORMTest):
- """test the usage of query options to eagerly load specific paths.
-
- this relies upon the 'path' construct used by PropertyOption to relate
- LoaderStrategies to specific paths, as well as the path state maintained
+ """test the usage of query options to eagerly load specific paths.
+
+ this relies upon the 'path' construct used by PropertyOption to relate
+ LoaderStrategies to specific paths, as well as the path state maintained
throughout the query setup/mapper instances process.
"""
-
+
def define_tables(self, metadata):
global accounts_table, transactions_table, entries_table
accounts_table = Table('accounts', metadata,
@@ -804,6 +808,7 @@ class EagerTest9(ORMTest):
Column('transaction_id', Integer, ForeignKey(transactions_table.c.transaction_id)),
)
+ @testing.fails_on('maxdb')
def test_eagerload_on_path(self):
class Account(fixtures.Base):
pass
@@ -840,10 +845,10 @@ class EagerTest9(ORMTest):
def go():
# load just the first Account. eager loading will actually load all objects saved thus far,
- # but will not eagerly load the "accounts" off the immediate "entries"; only the
+ # but will not eagerly load the "accounts" off the immediate "entries"; only the
# "accounts" off the entries->transaction->entries
acc = session.query(Account).options(eagerload_all('entries.transaction.entries.account')).first()
-
+
# no sql occurs
assert acc.name == 'acc1'
assert acc.entries[0].transaction.entries[0].account.name == 'acc1'
@@ -852,10 +857,10 @@ class EagerTest9(ORMTest):
# lazyload triggers but no sql occurs because many-to-one uses cached query.get()
for e in acc.entries:
assert e.account is acc
-
+
self.assert_sql_count(testbase.db, go, 1)
-
-
-
-if __name__ == "__main__":
+
+
+
+if __name__ == "__main__":
testbase.main()
diff --git a/test/orm/cascade.py b/test/orm/cascade.py
index adbb95265..7583507b2 100644
--- a/test/orm/cascade.py
+++ b/test/orm/cascade.py
@@ -23,34 +23,34 @@ class O2MCascadeTest(AssertMixin):
orders = relation(
mapper(tables.Order, tables.orders, properties = dict (
items = relation(mapper(tables.Item, tables.orderitems), lazy=True, uselist =True, cascade="all, delete-orphan")
- )),
+ )),
lazy = True, uselist = True, cascade="all, delete-orphan")
))
def setUp(self):
global data
data = [tables.User,
- {'user_name' : 'ed',
+ {'user_name' : 'ed',
'address' : (tables.Address, {'email_address' : 'foo@bar.com'}),
'orders' : (tables.Order, [
- {'description' : 'eds 1st order', 'items' : (tables.Item, [{'item_name' : 'eds o1 item'}, {'item_name' : 'eds other o1 item'}])},
+ {'description' : 'eds 1st order', 'items' : (tables.Item, [{'item_name' : 'eds o1 item'}, {'item_name' : 'eds other o1 item'}])},
{'description' : 'eds 2nd order', 'items' : (tables.Item, [{'item_name' : 'eds o2 item'}, {'item_name' : 'eds other o2 item'}])}
])
},
- {'user_name' : 'jack',
+ {'user_name' : 'jack',
'address' : (tables.Address, {'email_address' : 'jack@jack.com'}),
'orders' : (tables.Order, [
{'description' : 'jacks 1st order', 'items' : (tables.Item, [{'item_name' : 'im a lumberjack'}, {'item_name' : 'and im ok'}])}
])
},
- {'user_name' : 'foo',
+ {'user_name' : 'foo',
'address' : (tables.Address, {'email_address': 'hi@lala.com'}),
'orders' : (tables.Order, [
- {'description' : 'foo order', 'items' : (tables.Item, [])},
- {'description' : 'foo order 2', 'items' : (tables.Item, [{'item_name' : 'hi'}])},
+ {'description' : 'foo order', 'items' : (tables.Item, [])},
+ {'description' : 'foo order 2', 'items' : (tables.Item, [{'item_name' : 'hi'}])},
{'description' : 'foo order three', 'items' : (tables.Item, [{'item_name' : 'there'}])}
])
- }
+ }
]
sess = create_session()
@@ -90,7 +90,7 @@ class O2MCascadeTest(AssertMixin):
assert o2 in sess
sess.flush()
sess.clear()
-
+
u = sess.query(tables.User).get(u.user_id)
o3 = tables.Order()
o3.description='order3'
@@ -100,7 +100,7 @@ class O2MCascadeTest(AssertMixin):
assert o3 in sess
assert o4 in sess
sess.flush()
-
+
o5 = tables.Order()
o5.description='order5'
sess.save(o5)
@@ -109,8 +109,8 @@ class O2MCascadeTest(AssertMixin):
assert False
except exceptions.FlushError, e:
assert "is an orphan" in str(e)
-
-
+
+
def testdelete(self):
sess = create_session()
l = sess.query(tables.User).all()
@@ -130,7 +130,7 @@ class O2MCascadeTest(AssertMixin):
def testdelete2(self):
"""test that unloaded collections are still included in a delete-cascade by default."""
-
+
sess = create_session()
u = sess.query(tables.User).get_by(user_name='ed')
# assert 'addresses' collection not loaded
@@ -144,7 +144,7 @@ class O2MCascadeTest(AssertMixin):
"""test that cascade only reaches instances that are still part of the collection,
not those that have been removed"""
sess = create_session()
-
+
u = tables.User()
u.user_name = 'newuser'
o = tables.Order()
@@ -152,12 +152,12 @@ class O2MCascadeTest(AssertMixin):
u.orders.append(o)
sess.save(u)
sess.flush()
-
+
u.orders.remove(o)
sess.delete(u)
assert u in sess.deleted
assert o not in sess.deleted
-
+
def testorphan(self):
sess = create_session()
@@ -180,23 +180,23 @@ class M2OCascadeTest(AssertMixin):
ctx.current.clear()
for t in metadata.table_iterator(reverse=True):
t.delete().execute()
-
+
def tearDownAll(self):
clear_mappers()
metadata.drop_all()
-
+
def setUpAll(self):
global ctx, data, metadata, User, Pref, Extra
ctx = SessionContext(create_session)
metadata = MetaData(testbase.db)
- extra = Table("extra", metadata,
+ extra = Table("extra", metadata,
Column("extra_id", Integer, Sequence("extra_id_seq", optional=True), primary_key=True),
Column("prefs_id", Integer, ForeignKey("prefs.prefs_id"))
)
- prefs = Table('prefs', metadata,
+ prefs = Table('prefs', metadata,
Column('prefs_id', Integer, Sequence('prefs_id_seq', optional=True), primary_key=True),
Column('prefs_data', String(40)))
-
+
users = Table('users', metadata,
Column('user_id', Integer, Sequence('user_id_seq', optional=True), primary_key = True),
Column('user_name', String(40)),
@@ -236,6 +236,7 @@ class M2OCascadeTest(AssertMixin):
ctx.current.flush()
ctx.current.clear()
+ @testing.fails_on('maxdb')
def testorphan(self):
jack = ctx.current.query(User).get_by(user_name='jack')
p = jack.pref
@@ -245,6 +246,7 @@ class M2OCascadeTest(AssertMixin):
assert p not in ctx.current
assert e not in ctx.current
+ @testing.fails_on('maxdb')
def testorphan2(self):
jack = ctx.current.query(User).get_by(user_name='jack')
p = jack.pref
@@ -260,28 +262,28 @@ class M2OCascadeTest(AssertMixin):
ctx.current.flush()
assert p not in ctx.current
assert e not in ctx.current
-
-
+
+
class M2MCascadeTest(AssertMixin):
def setUpAll(self):
global metadata, a, b, atob
metadata = MetaData(testbase.db)
- a = Table('a', metadata,
+ a = Table('a', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30))
)
- b = Table('b', metadata,
+ b = Table('b', metadata,
Column('id', Integer, primary_key=True),
Column('data', String(30))
)
- atob = Table('atob', metadata,
+ atob = Table('atob', metadata,
Column('aid', Integer, ForeignKey('a.id')),
Column('bid', Integer, ForeignKey('b.id'))
-
+
)
metadata.create_all()
-
+
def tearDownAll(self):
metadata.drop_all()
@@ -292,26 +294,26 @@ class M2MCascadeTest(AssertMixin):
class B(object):
def __init__(self, data):
self.data = data
-
+
mapper(A, a, properties={
# if no backref here, delete-orphan failed until [ticket:427] was fixed
'bs':relation(B, secondary=atob, cascade="all, delete-orphan")
})
mapper(B, b)
-
+
sess = create_session()
a1 = A('a1')
b1 = B('b1')
a1.bs.append(b1)
sess.save(a1)
sess.flush()
-
+
a1.bs.remove(b1)
sess.flush()
assert atob.count().scalar() ==0
assert b.count().scalar() == 0
assert a.count().scalar() == 1
-
+
def testcascadedelete(self):
class A(object):
def __init__(self, data):
@@ -319,19 +321,19 @@ class M2MCascadeTest(AssertMixin):
class B(object):
def __init__(self, data):
self.data = data
-
+
mapper(A, a, properties={
'bs':relation(B, secondary=atob, cascade="all, delete-orphan")
})
mapper(B, b)
-
+
sess = create_session()
a1 = A('a1')
b1 = B('b1')
a1.bs.append(b1)
sess.save(a1)
sess.flush()
-
+
sess.delete(a1)
sess.flush()
assert atob.count().scalar() ==0
@@ -340,7 +342,7 @@ class M2MCascadeTest(AssertMixin):
class UnsavedOrphansTest(ORMTest):
"""tests regarding pending entities that are orphans"""
-
+
def define_tables(self, metadata):
global users, addresses, User, Address
users = Table('users', metadata,
@@ -355,7 +357,7 @@ class UnsavedOrphansTest(ORMTest):
)
class User(object):pass
class Address(object):pass
-
+
def test_pending_orphan(self):
"""test that an entity that never had a parent on a delete-orphan cascade cant be saved."""
@@ -373,7 +375,7 @@ class UnsavedOrphansTest(ORMTest):
assert a.address_id is None, "Error: address should not be persistent"
def test_delete_new_object(self):
- """test that an entity which is attached then detached from its
+ """test that an entity which is attached then detached from its
parent with a delete-orphan cascade gets counted as an orphan"""
mapper(Address, addresses)
@@ -422,7 +424,7 @@ class UnsavedOrphansTest2(ORMTest):
)
def testdeletechildwithchild(self):
- """test that an entity which is attached then detached from its
+ """test that an entity which is attached then detached from its
parent with a delete-orphan cascade gets counted as an orphan, as well
as its own child instances"""
@@ -460,7 +462,7 @@ class UnsavedOrphansTest2(ORMTest):
class DoubleParentOrphanTest(AssertMixin):
"""test orphan detection for an entity with two parent relations"""
-
+
def setUpAll(self):
global metadata, address_table, businesses, homes
metadata = MetaData(testbase.db)
@@ -487,7 +489,7 @@ class DoubleParentOrphanTest(AssertMixin):
metadata.drop_all()
def test_non_orphan(self):
"""test that an entity can have two parent delete-orphan cascades, and persists normally."""
-
+
class Address(object):pass
class Home(object):pass
class Business(object):pass
@@ -504,7 +506,7 @@ class DoubleParentOrphanTest(AssertMixin):
b1.address = a2
[session.save(x) for x in [h1,b1]]
session.flush()
-
+
def test_orphan(self):
"""test that an entity can have two parent delete-orphan cascades, and is detected as an orphan
when saved without a parent."""
diff --git a/test/orm/dynamic.py b/test/orm/dynamic.py
index 7d0d748e3..80e3982da 100644
--- a/test/orm/dynamic.py
+++ b/test/orm/dynamic.py
@@ -9,7 +9,7 @@ from query import QueryTest
class DynamicTest(QueryTest):
keep_mappers = False
-
+
def setup_mappers(self):
pass
@@ -38,12 +38,12 @@ class DynamicTest(QueryTest):
def go():
assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(User.id==7).all()
self.assert_sql_count(testbase.db, go, 2)
-
+
class FlushTest(FixtureTest):
def test_basic(self):
class Fixture(Base):
pass
-
+
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
})
@@ -55,18 +55,19 @@ class FlushTest(FixtureTest):
sess.save(u1)
sess.save(u2)
sess.flush()
-
+
sess.clear()
# test the test fixture a little bit
assert User(name='jack', addresses=[Address(email_address='wrong')]) != sess.query(User).first()
assert User(name='jack', addresses=[Address(email_address='lala@hoho.com')]) == sess.query(User).first()
-
+
assert [
User(name='jack', addresses=[Address(email_address='lala@hoho.com')]),
User(name='ed', addresses=[Address(email_address='foo@bar.com')])
] == sess.query(User).all()
+ @testing.fails_on('maxdb')
def test_delete(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), backref='user')
@@ -80,16 +81,17 @@ class FlushTest(FixtureTest):
u.addresses.append(Address(email_address='e'))
u.addresses.append(Address(email_address='f'))
sess.save(u)
-
+
assert Address(email_address='c') == u.addresses[2]
sess.delete(u.addresses[2])
sess.delete(u.addresses[4])
sess.delete(u.addresses[3])
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)
-
+
sess.delete(u)
sess.close()
+ @testing.fails_on('maxdb')
def test_remove_orphans(self):
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses), cascade="all, delete-orphan", backref='user')
@@ -104,20 +106,20 @@ class FlushTest(FixtureTest):
u.addresses.append(Address(email_address='f'))
sess.save(u)
- assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='c'),
+ assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='c'),
Address(email_address='d'), Address(email_address='e'), Address(email_address='f')] == sess.query(Address).all()
assert Address(email_address='c') == u.addresses[2]
-
+
try:
del u.addresses[3]
assert False
except TypeError, e:
assert "doesn't support item deletion" in str(e), str(e)
-
+
for a in u.addresses.filter(Address.email_address.in_(['c', 'e', 'f'])):
u.addresses.remove(a)
-
+
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)
assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == sess.query(Address).all()
@@ -144,10 +146,10 @@ def create_backref_test(autoflush, saveuser):
if not autoflush:
sess.flush()
-
+
assert u in sess
assert a in sess
-
+
self.assert_(list(u.addresses) == [a])
a.user = None
@@ -165,9 +167,9 @@ def create_backref_test(autoflush, saveuser):
setattr(FlushTest, test_backref.__name__, test_backref)
for autoflush in (False, True):
- for saveuser in (False, True):
- create_backref_test(autoflush, saveuser)
+ for saveuser in (False, True):
+ create_backref_test(autoflush, saveuser)
if __name__ == '__main__':
testbase.main()
-
+
diff --git a/test/orm/eager_relations.py b/test/orm/eager_relations.py
index 61c41dd7e..cd971a7f0 100644
--- a/test/orm/eager_relations.py
+++ b/test/orm/eager_relations.py
@@ -44,15 +44,15 @@ class EagerTest(QueryTest):
assert [
User(id=7, addresses=[
Address(id=1)
- ]),
+ ]),
User(id=8, addresses=[
Address(id=3, email_address='ed@bettyboop.com'),
Address(id=4, email_address='ed@lala.com'),
Address(id=2, email_address='ed@wood.com')
- ]),
+ ]),
User(id=9, addresses=[
Address(id=5)
- ]),
+ ]),
User(id=10, addresses=[])
] == q.all()
@@ -64,22 +64,22 @@ class EagerTest(QueryTest):
mapper(User, users, properties = dict(
addresses = relation(Address, lazy=False),
))
-
+
q = create_session().query(User)
l = q.filter(User.id==Address.user_id).order_by(Address.email_address).all()
-
+
assert [
User(id=8, addresses=[
Address(id=2, email_address='ed@wood.com'),
Address(id=3, email_address='ed@bettyboop.com'),
Address(id=4, email_address='ed@lala.com'),
- ]),
+ ]),
User(id=9, addresses=[
Address(id=5)
- ]),
+ ]),
User(id=7, addresses=[
Address(id=1)
- ]),
+ ]),
] == l
def test_orderby_desc(self):
@@ -92,15 +92,15 @@ class EagerTest(QueryTest):
assert [
User(id=7, addresses=[
Address(id=1)
- ]),
+ ]),
User(id=8, addresses=[
Address(id=2, email_address='ed@wood.com'),
Address(id=4, email_address='ed@lala.com'),
Address(id=3, email_address='ed@bettyboop.com'),
- ]),
+ ]),
User(id=9, addresses=[
Address(id=5)
- ]),
+ ]),
User(id=10, addresses=[])
] == sess.query(User).all()
@@ -115,7 +115,7 @@ class EagerTest(QueryTest):
def go():
assert fixtures.item_keyword_result == q.all()
self.assert_sql_count(testbase.db, go, 1)
-
+
def go():
assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
self.assert_sql_count(testbase.db, go, 1)
@@ -131,22 +131,22 @@ class EagerTest(QueryTest):
def go():
assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all()
-
+
self.assert_sql_count(testbase.db, go, 1)
def test_cyclical(self):
"""test that a circular eager relationship breaks the cycle with a lazy loader"""
-
+
mapper(Address, addresses)
mapper(User, users, properties = dict(
addresses = relation(Address, lazy=False, backref=backref('user', lazy=False))
))
assert class_mapper(User).get_property('addresses').lazy is False
assert class_mapper(Address).get_property('user').lazy is False
-
+
sess = create_session()
assert fixtures.user_address_result == sess.query(User).all()
-
+
def test_double(self):
"""tests eager loading with two relations simulatneously, from the same table, using aliases. """
openorders = alias(orders, 'openorders')
@@ -228,23 +228,24 @@ class EagerTest(QueryTest):
def test_no_false_hits(self):
"""test that eager loaders don't interpret main table columns as part of their eager load."""
-
+
mapper(User, users, properties={
'addresses':relation(Address, lazy=False),
'orders':relation(Order, lazy=False)
})
mapper(Address, addresses)
mapper(Order, orders)
-
+
allusers = create_session().query(User).all()
-
+
# using a textual select, the columns will be 'id' and 'name'.
- # the eager loaders have aliases which should not hit on those columns, they should
+ # the eager loaders have aliases which should not hit on those columns, they should
# be required to locate only their aliased/fully table qualified column name.
noeagers = create_session().query(User).from_statement("select * from users").all()
assert 'orders' not in noeagers[0].__dict__
assert 'addresses' not in noeagers[0].__dict__
+ @testing.fails_on('maxdb')
def test_limit(self):
"""test limit operations combined with lazy-load relationships."""
@@ -263,12 +264,12 @@ class EagerTest(QueryTest):
if testing.against('mysql'):
l = q.limit(2).all()
assert fixtures.user_all_result[:2] == l
- else:
+ else:
l = q.limit(2).offset(1).order_by(User.id).all()
print fixtures.user_all_result[1:3]
print l
assert fixtures.user_all_result[1:3] == l
-
+
def test_distinct(self):
# this is an involved 3x union of the users table to get a lot of rows.
# then see if the "distinct" works its way out. you actually get the same
@@ -287,24 +288,26 @@ class EagerTest(QueryTest):
l = q.filter(s.c.u2_id==User.c.id).distinct().all()
assert fixtures.user_address_result == l
self.assert_sql_count(testbase.db, go, 1)
-
+
+ @testing.fails_on('maxdb')
def test_limit_2(self):
mapper(Keyword, keywords)
mapper(Item, items, properties = dict(
keywords = relation(Keyword, secondary=item_keywords, lazy=False, order_by=[keywords.c.id]),
))
-
+
sess = create_session()
q = sess.query(Item)
l = q.filter((Item.c.description=='item 2') | (Item.c.description=='item 5') | (Item.c.description=='item 3')).\
order_by(Item.id).limit(2).all()
assert fixtures.item_keyword_result[1:3] == l
-
+
+ @testing.fails_on('maxdb')
def test_limit_3(self):
- """test that the ORDER BY is propigated from the inner select to the outer select, when using the
+ """test that the ORDER BY is propigated from the inner select to the outer select, when using the
'wrapped' select statement resulting from the combination of eager loading and limit/offset clauses."""
-
+
mapper(Item, items)
mapper(Order, orders, properties = dict(
items = relation(Item, secondary=order_items, lazy=False)
@@ -316,17 +319,17 @@ class EagerTest(QueryTest):
orders = relation(Order, lazy=False),
))
sess = create_session()
-
+
q = sess.query(User)
if not testing.against('maxdb', 'mssql'):
l = q.join('orders').order_by(Order.user_id.desc()).limit(2).offset(1)
assert [
- User(id=9,
+ User(id=9,
orders=[Order(id=2), Order(id=4)],
addresses=[Address(id=5)]
),
- User(id=7,
+ User(id=7,
orders=[Order(id=1), Order(id=3), Order(id=5)],
addresses=[Address(id=1)]
)
@@ -334,7 +337,7 @@ class EagerTest(QueryTest):
l = q.join('addresses').order_by(Address.email_address.desc()).limit(1).offset(0)
assert [
- User(id=7,
+ User(id=7,
orders=[Order(id=1), Order(id=3), Order(id=5)],
addresses=[Address(id=1)]
)
@@ -345,43 +348,44 @@ class EagerTest(QueryTest):
address = relation(mapper(Address, addresses), lazy=False, uselist=False)
))
q = create_session().query(User)
-
+
def go():
l = q.filter(users.c.id == 7).all()
assert [User(id=7, address=Address(id=1))] == l
self.assert_sql_count(testbase.db, go, 1)
+ @testing.fails_on('maxdb')
def test_many_to_one(self):
mapper(Address, addresses, properties = dict(
user = relation(mapper(User, users), lazy=False)
))
sess = create_session()
q = sess.query(Address)
-
+
def go():
a = q.filter(addresses.c.id==1).one()
assert a.user is not None
u1 = sess.query(User).get(7)
assert a.user is u1
self.assert_sql_count(testbase.db, go, 1)
-
+
def test_one_and_many(self):
- """tests eager load for a parent object with a child object that
+ """tests eager load for a parent object with a child object that
contains a many-to-many relationship to a third object."""
-
+
mapper(User, users, properties={
'orders':relation(Order, lazy=False)
})
- mapper(Item, items)
+ mapper(Item, items)
mapper(Order, orders, properties = dict(
items = relation(Item, secondary=order_items, lazy=False, order_by=items.c.id)
))
-
+
q = create_session().query(User)
-
+
l = q.filter("users.id in (7, 8, 9)")
-
+
def go():
assert fixtures.user_order_result[0:3] == l.all()
self.assert_sql_count(testbase.db, go, 1)
@@ -389,27 +393,27 @@ class EagerTest(QueryTest):
def test_double_with_aggregate(self):
max_orders_by_user = select([func.max(orders.c.id).label('order_id')], group_by=[orders.c.user_id]).alias('max_orders_by_user')
-
+
max_orders = orders.select(orders.c.id==max_orders_by_user.c.order_id).alias('max_orders')
-
+
mapper(Order, orders)
mapper(User, users, properties={
'orders':relation(Order, backref='user', lazy=False),
'max_order':relation(mapper(Order, max_orders, non_primary=True), lazy=False, uselist=False)
})
q = create_session().query(User)
-
+
def go():
assert [
User(id=7, orders=[
Order(id=1),
Order(id=3),
Order(id=5),
- ],
+ ],
max_order=Order(id=5)
),
User(id=8, orders=[]),
- User(id=9, orders=[Order(id=2),Order(id=4)],
+ User(id=9, orders=[Order(id=2),Order(id=4)],
max_order=Order(id=4)
),
User(id=10),
@@ -431,7 +435,7 @@ class EagerTest(QueryTest):
"""test eager loading of a mapper which is against a select"""
s = select([orders], orders.c.isopen==1).alias('openorders')
-
+
mapper(Order, s, properties={
'user':relation(User, lazy=False)
})
@@ -443,7 +447,7 @@ class EagerTest(QueryTest):
Order(id=3, user=User(id=7)),
Order(id=4, user=User(id=9))
] == q.all()
-
+
q = q.select_from(s.join(order_items).join(items)).filter(~Item.id.in_([1, 2, 5]))
assert [
Order(id=3, user=User(id=7)),
@@ -451,7 +455,7 @@ class EagerTest(QueryTest):
def test_aliasing(self):
"""test that eager loading uses aliases to insulate the eager load from regular criterion against those tables."""
-
+
mapper(User, users, properties = dict(
addresses = relation(mapper(Address, addresses), lazy=False)
))
@@ -466,12 +470,13 @@ class SelfReferentialEagerTest(ORMTest):
Column('id', Integer, Sequence('node_id_seq', optional=True), primary_key=True),
Column('parent_id', Integer, ForeignKey('nodes.id')),
Column('data', String(30)))
-
+
+ @testing.fails_on('maxdb')
def test_basic(self):
class Node(Base):
def append(self, node):
self.children.append(node)
-
+
mapper(Node, nodes, properties={
'children':relation(Node, lazy=False, join_depth=3)
})
@@ -545,6 +550,7 @@ class SelfReferentialEagerTest(ORMTest):
),
])
+ @testing.fails_on('maxdb')
def test_no_depth(self):
class Node(Base):
def append(self, node):
@@ -580,7 +586,7 @@ class SelfReferentialEagerTest(ORMTest):
class SelfReferentialM2MEagerTest(ORMTest):
def define_tables(self, metadata):
global widget, widget_rel
-
+
widget = Table('widget', metadata,
Column('id', Integer, primary_key=True),
Column('name', Unicode(40), nullable=False, unique=True),
@@ -610,49 +616,49 @@ class SelfReferentialM2MEagerTest(ORMTest):
sess.save(w1)
sess.flush()
sess.clear()
-
+
# l = sess.query(Widget).filter(Widget.name=='w1').all()
# print l
assert [Widget(name='w1', children=[Widget(name='w2')])] == sess.query(Widget).filter(Widget.name=='w1').all()
-
+
class CyclicalInheritingEagerTest(ORMTest):
def define_tables(self, metadata):
global t1, t2
- t1 = Table('t1', metadata,
+ t1 = Table('t1', metadata,
Column('c1', Integer, primary_key=True),
Column('c2', String(30)),
Column('type', String(30))
)
- t2 = Table('t2', metadata,
+ t2 = Table('t2', metadata,
Column('c1', Integer, primary_key=True),
Column('c2', String(30)),
Column('type', String(30)),
Column('t1.id', Integer, ForeignKey('t1.c1')))
-
+
def test_basic(self):
class T(object):
pass
-
+
class SubT(T):
pass
-
+
class T2(object):
pass
class SubT2(T2):
pass
-
+
mapper(T, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1')
mapper(SubT, None, inherits=T, polymorphic_identity='subt1', properties={
't2s':relation(SubT2, lazy=False, backref=backref('subt', lazy=False))
})
mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity='t2')
mapper(SubT2, None, inherits=T2, polymorphic_identity='subt2')
-
+
# testing a particular endless loop condition in eager join setup
create_session().query(SubT).all()
-
-
+
+
if __name__ == '__main__':
testbase.main()
diff --git a/test/orm/generative.py b/test/orm/generative.py
index c850069ab..d8fb7a5a6 100644
--- a/test/orm/generative.py
+++ b/test/orm/generative.py
@@ -20,25 +20,26 @@ class GenerativeQueryTest(PersistTest):
Column('id', Integer, Sequence('foo_id_seq'), primary_key=True),
Column('bar', Integer),
Column('range', Integer))
-
+
mapper(Foo, foo)
metadata.create_all()
-
+
sess = create_session(bind=testbase.db)
for i in range(100):
sess.save(Foo(bar=i, range=i%10))
sess.flush()
-
+
def tearDownAll(self):
metadata.drop_all()
clear_mappers()
-
+
def test_selectby(self):
res = create_session(bind=testbase.db).query(Foo).filter_by(range=5)
assert res.order_by([Foo.c.bar])[0].bar == 5
assert res.order_by([desc(Foo.c.bar)])[0].bar == 95
-
+
@testing.unsupported('mssql')
+ @testing.fails_on('maxdb')
def test_slice(self):
sess = create_session(bind=testbase.db)
query = sess.query(Foo)
@@ -90,14 +91,14 @@ class GenerativeQueryTest(PersistTest):
query = create_session(bind=testbase.db).query(Foo)
assert query.filter(foo.c.bar<30).apply_avg(foo.c.bar).first() == 14.5
assert query.filter(foo.c.bar<30).apply_avg(foo.c.bar).one() == 14.5
-
+
def test_filter(self):
query = create_session(bind=testbase.db).query(Foo)
assert query.count() == 100
assert query.filter(Foo.c.bar < 30).count() == 30
res2 = query.filter(Foo.c.bar < 30).filter(Foo.c.bar > 10)
assert res2.count() == 19
-
+
def test_options(self):
query = create_session(bind=testbase.db).query(Foo)
class ext1(MapperExtension):
@@ -105,7 +106,7 @@ class GenerativeQueryTest(PersistTest):
instance.TEST = "hello world"
return EXT_CONTINUE
assert query.options(extension(ext1()))[0].TEST == "hello world"
-
+
def test_order_by(self):
query = create_session(bind=testbase.db).query(Foo)
assert query.order_by([Foo.c.bar])[0].bar == 0
@@ -114,7 +115,7 @@ class GenerativeQueryTest(PersistTest):
def test_offset(self):
query = create_session(bind=testbase.db).query(Foo)
assert list(query.order_by([Foo.c.bar]).offset(10))[0].bar == 10
-
+
def test_offset(self):
query = create_session(bind=testbase.db).query(Foo)
assert len(list(query.limit(10))) == 10
@@ -209,7 +210,7 @@ class RelationsTest(AssertMixin):
filter(or_(tables.Order.c.order_id==None,tables.Item.c.item_id==2))
print x.compile()
self.assert_result(list(x), tables.User, *tables.user_result[1:3])
-
+
class CaseSensitiveTest(PersistTest):
def setUpAll(self):
@@ -232,7 +233,7 @@ class CaseSensitiveTest(PersistTest):
def tearDownAll(self):
metadata.drop_all()
clear_mappers()
-
+
def test_distinctcount(self):
q = create_session(bind=testbase.db).query(Obj1)
assert q.count() == 4
@@ -244,7 +245,7 @@ class CaseSensitiveTest(PersistTest):
class SelfRefTest(ORMTest):
def define_tables(self, metadata):
global t1
- t1 = Table('t1', metadata,
+ t1 = Table('t1', metadata,
Column('id', Integer, primary_key=True),
Column('parent_id', Integer, ForeignKey('t1.id'))
)
@@ -263,8 +264,8 @@ class SelfRefTest(ORMTest):
assert False
except exceptions.InvalidRequestError, e:
assert str(e) == "Self-referential query on 'T.children (T)' property requires create_aliases=True argument.", str(e)
-
-
-
+
+
+
if __name__ == "__main__":
- testbase.main()
+ testbase.main()
diff --git a/test/orm/inheritance/basic.py b/test/orm/inheritance/basic.py
index 000fddc45..9fa7fffba 100644
--- a/test/orm/inheritance/basic.py
+++ b/test/orm/inheritance/basic.py
@@ -37,7 +37,7 @@ class O2MTest(ORMTest):
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)
@@ -65,7 +65,7 @@ class O2MTest(ORMTest):
self.assert_(l[0].parent_foo.data == 'foo #1' and l[1].parent_foo.data == 'foo #1')
-class GetTest(ORMTest):
+class GetTest(ORMTest):
def define_tables(self, metadata):
global foo, bar, blub
foo = Table('foo', metadata,
@@ -83,7 +83,7 @@ class GetTest(ORMTest):
Column('foo_id', Integer, ForeignKey('foo.id')),
Column('bar_id', Integer, ForeignKey('bar.id')),
Column('data', String(20)))
-
+
def create_test(polymorphic):
def test_get(self):
class Foo(object):
@@ -91,7 +91,7 @@ class GetTest(ORMTest):
class Bar(Foo):
pass
-
+
class Blub(Bar):
pass
@@ -103,7 +103,7 @@ class GetTest(ORMTest):
mapper(Foo, foo)
mapper(Bar, bar, inherits=Foo)
mapper(Blub, blub, inherits=Bar)
-
+
sess = create_session()
f = Foo()
b = Bar()
@@ -112,7 +112,7 @@ class GetTest(ORMTest):
sess.save(b)
sess.save(bl)
sess.flush()
-
+
if polymorphic:
def go():
assert sess.query(Foo).get(f.id) == f
@@ -121,33 +121,33 @@ class GetTest(ORMTest):
assert sess.query(Bar).get(b.id) == b
assert sess.query(Bar).get(bl.id) == bl
assert sess.query(Blub).get(bl.id) == bl
-
+
self.assert_sql_count(testbase.db, go, 0)
else:
- # this is testing the 'wrong' behavior of using get()
+ # this is testing the 'wrong' behavior of using get()
# polymorphically with mappers that are not configured to be
# polymorphic. the important part being that get() always
# returns an instance of the query's type.
def go():
assert sess.query(Foo).get(f.id) == f
-
+
bb = sess.query(Foo).get(b.id)
assert isinstance(b, Foo) and bb.id==b.id
-
+
bll = sess.query(Foo).get(bl.id)
assert isinstance(bll, Foo) and bll.id==bl.id
-
+
assert sess.query(Bar).get(b.id) == b
-
+
bll = sess.query(Bar).get(bl.id)
assert isinstance(bll, Bar) and bll.id == bl.id
-
+
assert sess.query(Blub).get(bl.id) == bl
-
+
self.assert_sql_count(testbase.db, go, 3)
-
+
return test_get
-
+
test_get_polymorphic = create_test(True)
test_get_nonpolymorphic = create_test(False)
@@ -155,7 +155,7 @@ class GetTest(ORMTest):
class ConstructionTest(ORMTest):
def define_tables(self, metadata):
global content_type, content, product
- content_type = Table('content_type', metadata,
+ content_type = Table('content_type', metadata,
Column('id', Integer, primary_key=True)
)
content = Table('content', metadata,
@@ -163,7 +163,7 @@ class ConstructionTest(ORMTest):
Column('content_type_id', Integer, ForeignKey('content_type.id')),
Column('type', String(30))
)
- product = Table('product', metadata,
+ product = Table('product', metadata,
Column('id', Integer, ForeignKey('content.id'), primary_key=True)
)
@@ -195,7 +195,7 @@ class ConstructionTest(ORMTest):
p = Product()
p.contenttype = ContentType()
# TODO: assertion ??
-
+
class EagerLazyTest(ORMTest):
"""tests eager load/lazy load of child items off inheritance mappers, tests that
LazyLoader constructs the right query condition."""
@@ -214,6 +214,7 @@ class EagerLazyTest(ORMTest):
Column('foo_id', Integer, ForeignKey('foo.id'))
)
+ @testing.fails_on('maxdb')
def testbasic(self):
class Foo(object): pass
class Bar(Foo): pass
@@ -234,7 +235,7 @@ class EagerLazyTest(ORMTest):
bar_foo.insert().execute(bar_id=1, foo_id=3)
bar_foo.insert().execute(bar_id=2, foo_id=4)
-
+
sess = create_session()
q = sess.query(Bar)
self.assert_(len(q.selectfirst().lazy) == 1)
@@ -253,7 +254,7 @@ class FlushTest(ORMTest):
roles = Table('role', metadata,
Column('id', Integer, primary_key=True),
- Column('description', String(32))
+ Column('description', String(32))
)
user_roles = Table('user_role', metadata,
@@ -265,14 +266,14 @@ class FlushTest(ORMTest):
Column('admin_id', Integer, primary_key=True),
Column('user_id', Integer, ForeignKey('users.id'))
)
-
+
def testone(self):
class User(object):pass
class Role(object):pass
class Admin(User):pass
role_mapper = mapper(Role, roles)
user_mapper = mapper(User, users, properties = {
- 'roles' : relation(Role, secondary=user_roles, lazy=False, private=False)
+ 'roles' : relation(Role, secondary=user_roles, lazy=False, private=False)
}
)
admin_mapper = mapper(Admin, admins, inherits=user_mapper)
@@ -290,7 +291,7 @@ class FlushTest(ORMTest):
a.password = 'admin'
sess.save(a)
sess.flush()
-
+
assert user_roles.count().scalar() == 1
def testtwo(self):
@@ -311,7 +312,7 @@ class FlushTest(ORMTest):
}
)
- admin_mapper = mapper(Admin, admins, inherits=user_mapper)
+ admin_mapper = mapper(Admin, admins, inherits=user_mapper)
# create roles
adminrole = Role('admin')
@@ -411,7 +412,7 @@ class SyncCompileTest(ORMTest):
"""test that syncrules compile properly on custom inherit conds"""
def define_tables(self, metadata):
global _a_table, _b_table, _c_table
-
+
_a_table = Table('a', metadata,
Column('id', Integer, primary_key=True),
Column('data1', String)
@@ -427,14 +428,14 @@ class SyncCompileTest(ORMTest):
Column('b_a_id', Integer, ForeignKey('b.a_id'), primary_key=True),
Column('data3', String)
)
-
+
def test_joins(self):
for j1 in (None, _b_table.c.a_id==_a_table.c.id, _a_table.c.id==_b_table.c.a_id):
for j2 in (None, _b_table.c.a_id==_c_table.c.b_a_id, _c_table.c.b_a_id==_b_table.c.a_id):
self._do_test(j1, j2)
for t in _a_table.metadata.table_iterator(reverse=True):
t.delete().execute().close()
-
+
def _do_test(self, j1, j2):
class A(object):
def __init__(self, **kwargs):
@@ -446,12 +447,12 @@ class SyncCompileTest(ORMTest):
class C(B):
pass
-
+
mapper(A, _a_table)
- mapper(B, _b_table, inherits=A,
+ mapper(B, _b_table, inherits=A,
inherit_condition=j1
)
- mapper(C, _c_table, inherits=B,
+ mapper(C, _c_table, inherits=B,
inherit_condition=j2
)
@@ -473,7 +474,7 @@ class SyncCompileTest(ORMTest):
assert len(session.query(B).all()) == 2
assert len(session.query(C).all()) == 1
-
-if __name__ == "__main__":
+
+if __name__ == "__main__":
testbase.main()
diff --git a/test/orm/inheritance/manytomany.py b/test/orm/inheritance/manytomany.py
index 343345aa3..bed719d3f 100644
--- a/test/orm/inheritance/manytomany.py
+++ b/test/orm/inheritance/manytomany.py
@@ -18,7 +18,7 @@ class InheritTest(ORMTest):
primary_key=True),
Column('name', String(50), nullable=False))
- users = Table('prin_users', metadata,
+ users = Table('prin_users', metadata,
Column('principal_id', Integer,
ForeignKey('principals.principal_id'), primary_key=True),
Column('password', String(50), nullable=False),
@@ -53,7 +53,7 @@ class InheritTest(ORMTest):
mapper(Group, groups, inherits=Principal, properties={
'users': relation(User, secondary=user_group_map,
- lazy=True, backref="groups")
+ lazy=True, backref="groups")
})
g = Group(name="group1")
@@ -62,7 +62,7 @@ class InheritTest(ORMTest):
sess.save(g)
sess.flush()
# TODO: put an assertion
-
+
class InheritTest2(ORMTest):
"""deals with inheritance and many-to-many relationships"""
def define_tables(self, metadata):
@@ -87,7 +87,7 @@ class InheritTest2(ORMTest):
def __init__(self, data=None):
self.data = data
class Bar(Foo):pass
-
+
mapper(Foo, foo)
mapper(Bar, bar, inherits=Foo)
print foo.join(bar).primary_key
@@ -97,13 +97,13 @@ class InheritTest2(ORMTest):
sess.save(b)
sess.flush()
sess.clear()
-
+
# test that "bar.bid" does not need to be referenced in a get
# (ticket 185)
assert sess.query(Bar).get(b.id).id == b.id
-
+
def testbasic(self):
- class Foo(object):
+ class Foo(object):
def __init__(self, data=None):
self.data = data
@@ -114,7 +114,7 @@ class InheritTest2(ORMTest):
mapper(Bar, bar, inherits=Foo, properties={
'foos': relation(Foo, secondary=foo_bar, lazy=False)
})
-
+
sess = create_session()
b = Bar('barfoo')
sess.save(b)
@@ -155,10 +155,10 @@ class InheritTest3(ORMTest):
Column('id', Integer, ForeignKey('bar.id'), primary_key=True),
Column('data', String(20)))
- bar_foo = Table('bar_foo', metadata,
+ bar_foo = Table('bar_foo', metadata,
Column('bar_id', Integer, ForeignKey('bar.id')),
Column('foo_id', Integer, ForeignKey('foo.id')))
-
+
blub_bar = Table('bar_blub', metadata,
Column('blub_id', Integer, ForeignKey('blub.id')),
Column('bar_id', Integer, ForeignKey('bar.id')))
@@ -166,7 +166,7 @@ class InheritTest3(ORMTest):
blub_foo = Table('blub_foo', metadata,
Column('blub_id', Integer, ForeignKey('blub.id')),
Column('foo_id', Integer, ForeignKey('foo.id')))
-
+
def testbasic(self):
class Foo(object):
def __init__(self, data=None):
@@ -178,7 +178,7 @@ class InheritTest3(ORMTest):
class Bar(Foo):
def __repr__(self):
return "Bar id %d, data %s" % (self.id, self.data)
-
+
mapper(Bar, bar, inherits=Foo, properties={
'foos' :relation(Foo, secondary=bar_foo, lazy=True)
})
@@ -194,8 +194,9 @@ class InheritTest3(ORMTest):
l = sess.query(Bar).select()
print repr(l[0]) + repr(l[0].foos)
self.assert_(repr(l[0]) + repr(l[0].foos) == compare)
-
- def testadvanced(self):
+
+ @testing.fails_on('maxdb')
+ def testadvanced(self):
class Foo(object):
def __init__(self, data=None):
self.data = data
@@ -211,7 +212,7 @@ class InheritTest3(ORMTest):
class Blub(Bar):
def __repr__(self):
return "Blub id %d, data %s, bars %s, foos %s" % (self.id, self.data, repr([b for b in self.bars]), repr([f for f in self.foos]))
-
+
mapper(Blub, blub, inherits=Bar, properties={
'bars':relation(Bar, secondary=blub_bar, lazy=False),
'foos':relation(Foo, secondary=blub_foo, lazy=False),
@@ -238,7 +239,7 @@ class InheritTest3(ORMTest):
x = sess.query(Blub).get_by(id=blubid)
print x
self.assert_(repr(x) == compare)
-
-
-if __name__ == "__main__":
+
+
+if __name__ == "__main__":
testbase.main()
diff --git a/test/orm/inheritance/poly_linked_list.py b/test/orm/inheritance/poly_linked_list.py
index a7f15d3bc..1b14b4426 100644
--- a/test/orm/inheritance/poly_linked_list.py
+++ b/test/orm/inheritance/poly_linked_list.py
@@ -23,22 +23,22 @@ class PolymorphicCircularTest(ORMTest):
Column('id', Integer, ForeignKey('table1.id'), primary_key=True),
)
- data = Table('data', metadata,
+ data = Table('data', metadata,
Column('id', Integer, primary_key=True),
Column('node_id', Integer, ForeignKey('table1.id')),
Column('data', String(30))
)
-
+
#join = polymorphic_union(
# {
# 'table3' : table1.join(table3),
# 'table2' : table1.join(table2),
# 'table1' : table1.select(table1.c.type.in_(['table1', 'table1b'])),
# }, None, 'pjoin')
-
+
join = table1.outerjoin(table2).outerjoin(table3).alias('pjoin')
#join = None
-
+
class Table1(object):
def __init__(self, name, data=None):
self.name = name
@@ -49,19 +49,19 @@ class PolymorphicCircularTest(ORMTest):
class Table1B(Table1):
pass
-
+
class Table2(Table1):
pass
class Table3(Table1):
pass
-
+
class Data(object):
def __init__(self, data):
self.data = data
def __repr__(self):
return "%s(%d, %s)" % (self.__class__.__name__, self.id, repr(str(self.data)))
-
+
try:
# this is how the mapping used to work. ensure that this raises an error now
table1_mapper = mapper(Table1, table1,
@@ -69,8 +69,8 @@ class PolymorphicCircularTest(ORMTest):
polymorphic_on=table1.c.type,
polymorphic_identity='table1',
properties={
- 'next': relation(Table1,
- backref=backref('prev', primaryjoin=join.c.id==join.c.related_id, foreignkey=join.c.id, uselist=False),
+ 'next': relation(Table1,
+ backref=backref('prev', primaryjoin=join.c.id==join.c.related_id, foreignkey=join.c.id, uselist=False),
uselist=False, primaryjoin=join.c.id==join.c.related_id),
'data':relation(mapper(Data, data))
}
@@ -80,10 +80,10 @@ class PolymorphicCircularTest(ORMTest):
except:
assert True
clear_mappers()
-
+
# currently, the "eager" relationships degrade to lazy relationships
# due to the polymorphic load.
- # the "next" relation used to have a "lazy=False" on it, but the EagerLoader raises the "self-referential"
+ # the "next" relation used to have a "lazy=False" on it, but the EagerLoader raises the "self-referential"
# exception now. since eager loading would never work for that relation anyway, its better that the user
# gets an exception instead of it silently not eager loading.
table1_mapper = mapper(Table1, table1,
@@ -91,8 +91,8 @@ class PolymorphicCircularTest(ORMTest):
polymorphic_on=table1.c.type,
polymorphic_identity='table1',
properties={
- 'next': relation(Table1,
- backref=backref('prev', primaryjoin=table1.c.id==table1.c.related_id, remote_side=table1.c.id, uselist=False),
+ 'next': relation(Table1,
+ backref=backref('prev', primaryjoin=table1.c.id==table1.c.related_id, remote_side=table1.c.id, uselist=False),
uselist=False, primaryjoin=table1.c.id==table1.c.related_id),
'data':relation(mapper(Data, data), lazy=False)
}
@@ -105,27 +105,31 @@ class PolymorphicCircularTest(ORMTest):
polymorphic_identity='table2')
table3_mapper = mapper(Table3, table3, inherits=table1_mapper, polymorphic_identity='table3')
-
+
table1_mapper.compile()
assert table1_mapper.primary_key == [table1.c.id], table1_mapper.primary_key
-
+
+ @testing.fails_on('maxdb')
def testone(self):
self.do_testlist([Table1, Table2, Table1, Table2])
+ @testing.fails_on('maxdb')
def testtwo(self):
self.do_testlist([Table3])
-
+
+ @testing.fails_on('maxdb')
def testthree(self):
self.do_testlist([Table2, Table1, Table1B, Table3, Table3, Table1B, Table1B, Table2, Table1])
+ @testing.fails_on('maxdb')
def testfour(self):
self.do_testlist([
- Table2('t2', [Data('data1'), Data('data2')]),
+ Table2('t2', [Data('data1'), Data('data2')]),
Table1('t1', []),
Table3('t3', [Data('data3')]),
Table1B('t1b', [Data('data4'), Data('data5')])
])
-
+
def do_testlist(self, classes):
sess = create_session( )
@@ -147,7 +151,7 @@ class PolymorphicCircularTest(ORMTest):
# save to DB
sess.save(t)
sess.flush()
-
+
# string version of the saved list
assertlist = []
node = t
@@ -183,7 +187,7 @@ class PolymorphicCircularTest(ORMTest):
assert n.next is node
node = n
backwards = repr(assertlist)
-
+
# everything should match !
print "ORIGNAL", original
print "BACKWARDS",backwards
diff --git a/test/orm/mapper.py b/test/orm/mapper.py
index 4402fe48f..0a993dd10 100644
--- a/test/orm/mapper.py
+++ b/test/orm/mapper.py
@@ -20,7 +20,7 @@ class MapperSuperTest(AssertMixin):
clear_mappers()
def setUp(self):
pass
-
+
class MapperTest(MapperSuperTest):
def test_propconflict(self):
@@ -37,7 +37,7 @@ class MapperTest(MapperSuperTest):
assert False
except exceptions.ArgumentError:
pass
-
+
def test_prop_accessor(self):
mapper(User, users)
try:
@@ -45,7 +45,7 @@ class MapperTest(MapperSuperTest):
assert False
except NotImplementedError, uoe:
assert str(uoe) == "Public collection of MapperProperty objects is provided by the get_property() and iterate_properties accessors."
-
+
def test_badcascade(self):
mapper(Address, addresses)
try:
@@ -53,10 +53,10 @@ class MapperTest(MapperSuperTest):
assert False
except exceptions.ArgumentError, e:
assert str(e) == "Invalid cascade option 'fake'"
-
+
def test_columnprefix(self):
- mapper(User, users, column_prefix='_', properties={
- 'user_name':synonym('_user_name')
+ mapper(User, users, column_prefix='_', properties={
+ 'user_name':synonym('_user_name')
})
s = create_session()
@@ -64,9 +64,9 @@ class MapperTest(MapperSuperTest):
assert u._user_name=='jack'
assert u._user_id ==7
assert not hasattr(u, 'user_name')
- u2 = s.query(User).filter_by(user_name='jack').one()
+ u2 = s.query(User).filter_by(user_name='jack').one()
assert u is u2
-
+
def test_refresh(self):
mapper(User, users, properties={'addresses':relation(mapper(Address, addresses), backref='user')})
s = create_session()
@@ -79,15 +79,15 @@ class MapperTest(MapperSuperTest):
self.assert_(a in u.addresses)
s.refresh(u)
-
+
# its refreshed, so not dirty
self.assert_(u not in s.dirty)
-
+
# username is back to the DB
self.assert_(u.user_name == 'jack')
-
+
self.assert_(a not in u.addresses)
-
+
u.user_name = 'foo'
u.addresses.append(a)
# now its dirty
@@ -103,7 +103,7 @@ class MapperTest(MapperSuperTest):
def test_compileonsession(self):
m = mapper(User, users)
session = create_session()
- session.connection(m)
+ session.connection(m)
def test_expirecascade(self):
mapper(User, users, properties={'addresses':relation(mapper(Address, addresses), cascade="all, refresh-expire")})
@@ -112,7 +112,7 @@ class MapperTest(MapperSuperTest):
u.addresses[0].email_address = 'someotheraddress'
s.expire(u)
assert u.addresses[0].email_address == 'ed@wood.com'
-
+
def test_refreshwitheager(self):
"""test that a refresh/expire operation loads rows properly and sends correct "isnew" state to eager loaders"""
mapper(User, users, properties={'addresses':relation(mapper(Address, addresses), lazy=False)})
@@ -127,7 +127,7 @@ class MapperTest(MapperSuperTest):
assert len(u.addresses) == 3
s.expire(u)
assert len(u.addresses) == 3
-
+
def test_incompletecolumns(self):
"""test loading from a select which does not contain all columns"""
mapper(Address, addresses)
@@ -136,7 +136,7 @@ class MapperTest(MapperSuperTest):
assert a.user_id == 7
assert a.address_id == 1
assert a.email_address is None
-
+
def test_badconstructor(self):
"""test that if the construction of a mapped class fails, the instnace does not get placed in the session"""
class Foo(object):
@@ -156,7 +156,7 @@ class MapperTest(MapperSuperTest):
pass
def test_constructorexceptions(self):
- """test that exceptions raised in the mapped class are not masked by sa decorations"""
+ """test that exceptions raised in the mapped class are not masked by sa decorations"""
ex = AssertionError('oops')
sess = create_session()
@@ -185,7 +185,7 @@ class MapperTest(MapperSuperTest):
assert False
except Exception, e:
assert e is ex
-
+
def test_refresh_lazy(self):
"""test that when a lazy loader is set as a trigger on an object's attribute (at the attribute level, not the class level), a refresh() operation doesnt fire the lazy loader or create any problems"""
s = create_session()
@@ -208,7 +208,7 @@ class MapperTest(MapperSuperTest):
# test plain expire
self.assert_(u.user_name =='jack')
self.assert_(len(u.addresses) == 1)
-
+
# we're changing the database here, so if this test fails in the middle,
# it'll screw up the other tests which are hardcoded to 7/'jack'
u.user_name = 'foo'
@@ -222,14 +222,15 @@ class MapperTest(MapperSuperTest):
s.query(User).select()
# test that it refreshed
self.assert_(u.__dict__['user_name'] == 'jack')
-
- # object should be back to normal now,
+
+ # object should be back to normal now,
# this should *not* produce a SELECT statement (not tested here though....)
self.assert_(u.user_name =='jack')
-
+
+ @testing.fails_on('maxdb')
def test_refresh2(self):
"""test a hang condition that was occuring on expire/refresh"""
-
+
s = create_session()
m1 = mapper(Address, addresses)
@@ -249,13 +250,13 @@ class MapperTest(MapperSuperTest):
print u.user_name #this line runs
s.refresh(u) #hangs
-
+
def test_props(self):
m = mapper(User, users, properties = {
'addresses' : relation(mapper(Address, addresses))
}).compile()
self.assert_(User.addresses.property is m.get_property('addresses'))
-
+
def test_compileonprop(self):
mapper(User, users, properties = {
'addresses' : relation(mapper(Address, addresses))
@@ -274,31 +275,31 @@ class MapperTest(MapperSuperTest):
mapper(User, users)
mapper(Foo, addresses, inherits=User)
assert getattr(Foo().__class__, 'user_name').impl is not None
-
+
def test_addproperty(self):
m = mapper(User, users)
mapper(Address, addresses)
m.add_property('user_name', deferred(users.c.user_name))
m.add_property('name', synonym('user_name'))
m.add_property('addresses', relation(Address))
-
+
sess = create_session(transactional=True)
assert sess.query(User).get(7)
u = sess.query(User).filter_by(name='jack').one()
-
+
def go():
self.assert_result([u], User, user_address_result[0])
assert u.user_name == 'jack'
-
+
self.assert_sql_count(testbase.db, go, 2)
-
+
u3 = User()
u3.user_name = 'some user'
sess.save(u3)
sess.flush()
sess.rollback()
-
+
def test_propfilters(self):
t = Table('person', MetaData(),
Column('id', Integer, primary_key=True),
@@ -334,7 +335,7 @@ class MapperTest(MapperSuperTest):
p_m.compile()
#compile_mappers()
-
+
def assert_props(cls, want):
have = set([n for n in dir(cls) if not n.startswith('_')])
want = set(want)
@@ -366,7 +367,7 @@ class MapperTest(MapperSuperTest):
q = create_session().query(m)
l = q.select()
self.assert_result(l, User, *user_result[0:2])
-
+
def test_mappingtojoinnopk(self):
metadata = MetaData()
account_ids_table = Table('account_ids', metadata,
@@ -390,7 +391,7 @@ class MapperTest(MapperSuperTest):
assert testbase.db.execute(account_stuff_table.count()).scalar() == 0
finally:
metadata.drop_all(testbase.db)
-
+
def test_mappingtoouterjoin(self):
"""test mapping to an outer join, with a composite primary key that allows nulls"""
result = [
@@ -400,14 +401,14 @@ class MapperTest(MapperSuperTest):
{'user_id' : 8, 'address_id' : 4},
{'user_id' : 9, 'address_id':None}
]
-
+
j = join(users, addresses, isouter=True)
m = mapper(User, j, allow_null_pks=True, primary_key=[users.c.user_id, addresses.c.address_id])
q = create_session().query(m)
l = q.select()
self.assert_result(l, User, *result)
-
+
def test_customjoin(self):
"""test that the from_obj parameter to query.select() can be used
to totally replace the FROM parameters of the generated query."""
@@ -421,21 +422,21 @@ class MapperTest(MapperSuperTest):
q = create_session().query(m)
l = q.select((orderitems.c.item_name=='item 4'), from_obj=[users.join(orders).join(orderitems)])
self.assert_result(l, User, user_result[0])
-
+
def test_orderby(self):
"""test ordering at the mapper and query level"""
# TODO: make a unit test out of these various combinations
# m = mapper(User, users, order_by=desc(users.c.user_name))
mapper(User, users, order_by=None)
# mapper(User, users)
-
+
# l = create_session().query(User).select(order_by=[desc(users.c.user_name), asc(users.c.user_id)])
l = create_session().query(User).select()
# l = create_session().query(User).select(order_by=[])
# l = create_session().query(User).select(order_by=None)
-
-
- @testing.unsupported('firebird')
+
+
+ @testing.unsupported('firebird')
def test_function(self):
"""test mapping to a SELECT statement that has functions in it."""
s = select([users, (users.c.user_id * 2).label('concat'), func.count(addresses.c.address_id).label('count')],
@@ -448,10 +449,10 @@ class MapperTest(MapperSuperTest):
assert l[0].concat == l[0].user_id * 2 == 14
assert l[1].concat == l[1].user_id * 2 == 16
- @testing.unsupported('firebird')
+ @testing.unsupported('firebird')
def test_count(self):
"""test the count function on Query.
-
+
(why doesnt this work on firebird?)"""
mapper(User, users)
q = create_session().query(User)
@@ -475,13 +476,13 @@ class MapperTest(MapperSuperTest):
self.assert_(False, "should have raised ArgumentError")
except exceptions.ArgumentError, e:
self.assert_(True)
-
+
clear_mappers()
# assert that allow_column_override cancels the error
m = mapper(User, users, properties = {
'user_name' : relation(mapper(Address, addresses))
}, allow_column_override=True)
-
+
clear_mappers()
# assert that the column being named else where also cancels the error
m = mapper(User, users, properties = {
@@ -497,39 +498,40 @@ class MapperTest(MapperSuperTest):
adlist = synonym('addresses', proxy=True),
adname = synonym('addresses')
))
-
+
assert hasattr(User, 'adlist')
assert not hasattr(User, 'adname')
-
+
u = sess.query(User).get_by(uname='jack')
self.assert_result(u.adlist, Address, *(user_address_result[0]['addresses'][1]))
assert hasattr(u, 'adlist')
assert not hasattr(u, 'adname')
-
+
addr = sess.query(Address).get_by(address_id=user_address_result[0]['addresses'][1][0]['address_id'])
u = sess.query(User).get_by(adname=addr)
u2 = sess.query(User).get_by(adlist=addr)
assert u is u2
-
+
assert u not in sess.dirty
u.uname = "some user name"
assert u.uname == "some user name"
assert u.user_name == "some user name"
assert u in sess.dirty
+ @testing.fails_on('maxdb')
def test_synonymoptions(self):
sess = create_session()
mapper(User, users, properties = dict(
addresses = relation(mapper(Address, addresses), lazy = True),
adlist = synonym('addresses', proxy=True)
))
-
+
def go():
u = sess.query(User).options(eagerload('adlist')).get_by(user_name='jack')
self.assert_result(u.adlist, Address, *(user_address_result[0]['addresses'][1]))
self.assert_sql_count(testbase.db, go, 1)
-
+
def test_extensionoptions(self):
sess = create_session()
class ext1(MapperExtension):
@@ -556,7 +558,7 @@ class MapperTest(MapperSuperTest):
assert l.TEST_2 == "also hello world"
assert not hasattr(l.addresses[0], 'TEST')
assert not hasattr(l.addresses[0], 'TEST2')
-
+
def test_eageroptions(self):
"""tests that a lazy relation can be upgraded to an eager relation via the options method"""
sess = create_session()
@@ -569,6 +571,7 @@ class MapperTest(MapperSuperTest):
self.assert_result(l, User, *user_address_result)
self.assert_sql_count(testbase.db, go, 0)
+ @testing.fails_on('maxdb')
def test_eageroptionswithlimit(self):
sess = create_session()
mapper(User, users, properties = dict(
@@ -582,14 +585,15 @@ class MapperTest(MapperSuperTest):
self.assert_sql_count(testbase.db, go, 0)
sess.clear()
-
+
# test that eager loading doesnt modify parent mapper
def go():
u = sess.query(User).get_by(user_id=8)
assert u.user_id == 8
assert len(u.addresses) == 3
assert "tbl_row_count" not in self.capture_sql(testbase.db, go)
-
+
+ @testing.fails_on('maxdb')
def test_lazyoptionswithlimit(self):
sess = create_session()
mapper(User, users, properties = dict(
@@ -616,7 +620,7 @@ class MapperTest(MapperSuperTest):
self.assert_sql_count(testbase.db, go, 1)
sess.clear()
-
+
# then select just from users. run it into instances.
# then assert the data, which will launch 3 more lazy loads
# (previous users in session fell out of scope and were removed from session's identity map)
@@ -625,14 +629,14 @@ class MapperTest(MapperSuperTest):
l = usermapper.instances(r, sess)
self.assert_result(l, User, *user_address_result)
self.assert_sql_count(testbase.db, go, 4)
-
+
clear_mappers()
sess.clear()
-
+
# test with a deeper set of eager loads. when we first load the three
# users, they will have no addresses or orders. the number of lazy loads when
- # traversing the whole thing will be three for the addresses and three for the
+ # traversing the whole thing will be three for the addresses and three for the
# orders.
# (previous users in session fell out of scope and were removed from session's identity map)
usermapper = mapper(User, users,
@@ -654,7 +658,7 @@ class MapperTest(MapperSuperTest):
self.assert_sql_count(testbase.db, go, 1)
sess.clear()
-
+
# then select just from users. run it into instances.
# then assert the data, which will launch 6 more lazy loads
def go():
@@ -662,8 +666,8 @@ class MapperTest(MapperSuperTest):
l = usermapper.instances(r, sess)
self.assert_result(l, User, *user_all_result)
self.assert_sql_count(testbase.db, go, 7)
-
-
+
+
def test_lazyoptions(self):
"""tests that an eager relation can be upgraded to a lazy relation via the options method"""
sess = create_session()
@@ -677,12 +681,12 @@ class MapperTest(MapperSuperTest):
def test_latecompile(self):
"""tests mappers compiling late in the game"""
-
+
mapper(User, users, properties = {'orders': relation(Order)})
mapper(Item, orderitems, properties={'keywords':relation(Keyword, secondary=itemkeywords)})
mapper(Keyword, keywords)
mapper(Order, orders, properties={'items':relation(Item)})
-
+
sess = create_session()
u = sess.query(User).select()
def go():
@@ -698,17 +702,17 @@ class MapperTest(MapperSuperTest):
}))
}))
})
-
+
sess = create_session()
-
+
# eagerload nothing.
u = sess.query(User).select()
def go():
print u[0].orders[1].items[0].keywords[1]
self.assert_sql_count(testbase.db, go, 3)
sess.clear()
-
-
+
+
print "-------MARK----------"
# eagerload orders.items.keywords; eagerload_all() implies eager load of orders, orders.items
q2 = sess.query(User).options(eagerload_all('orders.items.keywords'))
@@ -730,24 +734,24 @@ class MapperTest(MapperSuperTest):
print "-------MARK4----------"
sess.clear()
-
+
# eagerload "keywords" on items. it will lazy load "orders", then lazy load
# the "items" on the order, but on "items" it will eager load the "keywords"
print "-------MARK5----------"
q3 = sess.query(User).options(eagerload('orders.items.keywords'))
u = q3.select()
self.assert_sql_count(testbase.db, go, 2)
-
-
+
+
class DeferredTest(MapperSuperTest):
def test_basic(self):
"""tests a basic "deferred" load"""
-
+
m = mapper(Order, orders, properties={
'description':deferred(orders.c.description)
})
-
+
o = Order()
self.assert_(o.description is None)
@@ -768,7 +772,7 @@ class DeferredTest(MapperSuperTest):
m = mapper(Order, orders, properties={
'description':deferred(orders.c.description)
})
-
+
sess = create_session()
o = Order()
sess.save(o)
@@ -791,19 +795,19 @@ class DeferredTest(MapperSuperTest):
def go():
o.description = "some description"
self.assert_sql_count(testbase.db, go, 0)
-
+
def test_save(self):
m = mapper(Order, orders, properties={
'description':deferred(orders.c.description)
})
-
+
sess = create_session()
q = sess.query(m)
l = q.select()
o2 = l[2]
o2.isopen = 1
sess.flush()
-
+
def test_group(self):
"""tests deferred load with a group"""
m = mapper(Order, orders, properties = {
@@ -825,7 +829,7 @@ class DeferredTest(MapperSuperTest):
("SELECT orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}),
("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders WHERE orders.order_id = :param_1", {'param_1':3})
])
-
+
o2 = q.select()[2]
# assert o2.opened == 1
assert o2.description == 'order 3'
@@ -834,11 +838,11 @@ class DeferredTest(MapperSuperTest):
def go():
sess.flush()
self.assert_sql_count(testbase.db, go, 0)
-
+
def test_commitsstate(self):
"""test that when deferred elements are loaded via a group, they get the proper CommittedState
and dont result in changes being committed"""
-
+
m = mapper(Order, orders, properties = {
'userident':deferred(orders.c.user_id, group='primary'),
'description':deferred(orders.c.description, group='primary'),
@@ -856,7 +860,7 @@ class DeferredTest(MapperSuperTest):
# therefore the flush() shouldnt actually issue any SQL
sess.flush()
self.assert_sql_count(testbase.db, go, 0)
-
+
def test_options(self):
"""tests using options on a mapper to create deferred and undeferred columns"""
m = mapper(Order, orders)
@@ -866,7 +870,7 @@ class DeferredTest(MapperSuperTest):
def go():
l = q2.select()
print l[2].user_id
-
+
orderby = str(orders.default_order_by()[0].compile(testbase.db))
self.assert_sql(testbase.db, go, [
("SELECT orders.order_id AS orders_order_id, orders.description AS orders_description, orders.isopen AS orders_isopen FROM orders ORDER BY %s" % orderby, {}),
@@ -902,7 +906,7 @@ class DeferredTest(MapperSuperTest):
("SELECT orders.user_id AS orders_user_id, orders.description AS orders_description, orders.isopen AS orders_isopen, orders.order_id AS orders_order_id FROM orders ORDER BY %s" % orderby, {}),
])
-
+
def test_deepoptions(self):
m = mapper(User, users, properties={
'orders':relation(mapper(Order, orders, properties={
@@ -935,8 +939,8 @@ class CompositeTypesTest(ORMTest):
Column('id', Integer, primary_key=True),
Column('version_id', Integer, primary_key=True),
Column('name', String(30)))
-
- edges = Table('edges', metadata,
+
+ edges = Table('edges', metadata,
Column('id', Integer, primary_key=True),
Column('graph_id', Integer, nullable=False),
Column('graph_version_id', Integer, nullable=False),
@@ -953,7 +957,7 @@ class CompositeTypesTest(ORMTest):
self.x = x
self.y = y
def __composite_values__(self):
- return [self.x, self.y]
+ return [self.x, self.y]
def __eq__(self, other):
return other.x == self.x and other.y == self.y
def __ne__(self, other):
@@ -965,7 +969,7 @@ class CompositeTypesTest(ORMTest):
def __init__(self, start, end):
self.start = start
self.end = end
-
+
mapper(Graph, graphs, properties={
'edges':relation(Edge)
})
@@ -973,7 +977,7 @@ class CompositeTypesTest(ORMTest):
'start':composite(Point, edges.c.x1, edges.c.y1),
'end':composite(Point, edges.c.x2, edges.c.y2)
})
-
+
sess = create_session()
g = Graph()
g.id = 1
@@ -982,13 +986,13 @@ class CompositeTypesTest(ORMTest):
g.edges.append(Edge(Point(14, 5), Point(2, 7)))
sess.save(g)
sess.flush()
-
+
sess.clear()
g2 = sess.query(Graph).get([g.id, g.version_id])
for e1, e2 in zip(g.edges, g2.edges):
assert e1.start == e2.start
assert e1.end == e2.end
-
+
g2.edges[1].end = Point(18, 4)
sess.flush()
sess.clear()
@@ -1002,7 +1006,7 @@ class CompositeTypesTest(ORMTest):
assert sess.query(Edge).get(g2.edges[1].id).end == Point(19, 5)
g.edges[1].end = Point(19, 5)
-
+
sess.clear()
def go():
g2 = sess.query(Graph).options(eagerload('edges')).get([g.id, g.version_id])
@@ -1010,19 +1014,19 @@ class CompositeTypesTest(ORMTest):
assert e1.start == e2.start
assert e1.end == e2.end
self.assert_sql_count(testbase.db, go, 1)
-
+
# test comparison of CompositeProperties to their object instances
g = sess.query(Graph).get([1, 1])
assert sess.query(Edge).filter(Edge.start==Point(3, 4)).one() is g.edges[0]
-
+
assert sess.query(Edge).filter(Edge.start!=Point(3, 4)).first() is g.edges[1]
assert sess.query(Edge).filter(Edge.start==None).all() == []
-
-
+
+
def test_pk(self):
"""test using a composite type as a primary key"""
-
+
class Version(object):
def __init__(self, id, version):
self.id = id
@@ -1034,30 +1038,30 @@ class CompositeTypesTest(ORMTest):
return other.id == self.id and other.version == self.version
def __ne__(self, other):
return not self.__eq__(other)
-
+
class Graph(object):
def __init__(self, version):
self.version = version
-
+
mapper(Graph, graphs, properties={
'version':composite(Version, graphs.c.id, graphs.c.version_id)
})
-
+
sess = create_session()
g = Graph(Version(1, 1))
sess.save(g)
sess.flush()
-
+
sess.clear()
g2 = sess.query(Graph).get([1, 1])
assert g.version == g2.version
sess.clear()
-
+
g2 = sess.query(Graph).get(Version(1, 1))
assert g.version == g2.version
-
-
-
+
+
+
class NoLoadTest(MapperSuperTest):
def test_basic(self):
"""tests a basic one-to-many lazy load"""
@@ -1071,7 +1075,7 @@ class NoLoadTest(MapperSuperTest):
x[0].addresses
l[0] = x
self.assert_sql_count(testbase.db, go, 1)
-
+
self.assert_result(l[0], User,
{'user_id' : 7, 'addresses' : (Address, [])},
)
@@ -1086,7 +1090,7 @@ class NoLoadTest(MapperSuperTest):
x[0].addresses
l[0] = x
self.assert_sql_count(testbase.db, go, 2)
-
+
self.assert_result(l[0], User,
{'user_id' : 7, 'addresses' : (Address, [{'address_id' : 1}])},
)
@@ -1110,14 +1114,14 @@ class MapperExtensionTest(MapperSuperTest):
m = mapper(User, users, extension=Ext(), properties = dict(
addresses = relation(Address, lazy=True),
))
-
+
q = create_session().query(m)
l = q.select();
self.assert_result(l, User, *user_address_result)
-
+
def test_methods(self):
"""test that common user-defined methods get called."""
-
+
methods = set()
class Ext(MapperExtension):
def load(self, query, *args, **kwargs):
@@ -1167,7 +1171,7 @@ class MapperExtensionTest(MapperSuperTest):
def after_delete(self, mapper, connection, instance):
methods.add('after_delete')
return EXT_CONTINUE
-
+
mapper(User, users, extension=Ext())
sess = create_session()
u = User()
@@ -1180,9 +1184,9 @@ class MapperExtensionTest(MapperSuperTest):
sess.flush()
sess.delete(u)
sess.flush()
- assert methods == set(['load', 'append_result', 'before_delete', 'create_instance', 'translate_row', 'get',
+ assert methods == set(['load', 'append_result', 'before_delete', 'create_instance', 'translate_row', 'get',
'after_delete', 'after_insert', 'before_update', 'before_insert', 'after_update', 'populate_instance'])
-
+
class RequirementsTest(AssertMixin):
"""Tests the contract for user classes."""
diff --git a/test/orm/query.py b/test/orm/query.py
index e82041ff9..d879f7192 100644
--- a/test/orm/query.py
+++ b/test/orm/query.py
@@ -12,16 +12,16 @@ from testlib.fixtures import *
class QueryTest(FixtureTest):
keep_mappers = True
keep_data = True
-
+
def setUpAll(self):
super(QueryTest, self).setUpAll()
install_fixture_data()
self.setup_mappers()
-
+
def tearDownAll(self):
clear_mappers()
super(QueryTest, self).tearDownAll()
-
+
def setup_mappers(self):
mapper(User, users, properties={
'addresses':relation(Address, backref='user'),
@@ -39,10 +39,10 @@ class QueryTest(FixtureTest):
class UnicodeSchemaTest(QueryTest):
keep_mappers = False
-
+
def setup_mappers(self):
pass
-
+
def define_tables(self, metadata):
super(UnicodeSchemaTest, self).define_tables(metadata)
global uni_meta, uni_users
@@ -50,11 +50,11 @@ class UnicodeSchemaTest(QueryTest):
uni_users = Table(u'users', uni_meta,
Column(u'id', Integer, primary_key=True),
Column(u'name', String(30), nullable=False))
-
+
def test_get(self):
mapper(User, uni_users)
assert User(id=7) == create_session(bind=testbase.db).query(User).get(7)
-
+
class GetTest(QueryTest):
def test_get(self):
s = create_session()
@@ -74,45 +74,45 @@ class GetTest(QueryTest):
print s.primary_key
print m.primary_key
assert s.primary_key == m.primary_key
-
+
row = s.select(use_labels=True).execute().fetchone()
print row[s.primary_key[0]]
-
+
sess = create_session()
assert sess.query(SomeUser).get(7).name == 'jack'
def test_load(self):
s = create_session()
-
+
try:
assert s.query(User).load(19) is None
assert False
except exceptions.InvalidRequestError:
assert True
-
+
u = s.query(User).load(7)
u2 = s.query(User).load(7)
assert u is u2
s.clear()
u2 = s.query(User).load(7)
assert u is not u2
-
+
u2.name = 'some name'
a = Address(email_address='some other name')
u2.addresses.append(a)
assert u2 in s.dirty
assert a in u2.addresses
-
+
s.query(User).load(7)
assert u2 not in s.dirty
assert u2.name =='jack'
assert a not in u2.addresses
-
+
@testing.exclude('mysql', '<', (4, 1))
def test_unicode(self):
- """test that Query.get properly sets up the type for the bind parameter. using unicode would normally fail
+ """test that Query.get properly sets up the type for the bind parameter. using unicode would normally fail
on postgres, mysql and oracle unless it is converted to an encoded string"""
-
+
metadata = MetaData(engines.utf8_engine())
table = Table('unicode_data', metadata,
Column('id', Unicode(40), primary_key=True),
@@ -155,19 +155,19 @@ class GetTest(QueryTest):
s.query(User).populate_existing().all()
assert u.addresses[0].email_address == 'lala'
assert u.orders[1].items[2].description == 'item 12'
-
+
# eager load does
s.query(User).options(eagerload('addresses'), eagerload_all('orders.items')).populate_existing().all()
assert u.addresses[0].email_address == 'jack@bean.com'
assert u.orders[1].items[2].description == 'item 5'
-
+
class OperatorTest(QueryTest):
"""test sql.Comparator implementation for MapperProperties"""
-
+
def _test(self, clause, expected):
c = str(clause.compile(dialect = default.DefaultDialect()))
assert c == expected, "%s != %s" % (c, expected)
-
+
def test_arithmetic(self):
create_session().query(User)
for (py_op, sql_op) in ((operator.add, '+'), (operator.mul, '*'),
@@ -213,7 +213,7 @@ class OperatorTest(QueryTest):
self.assert_(compiled == fwd_sql or compiled == rev_sql,
"\n'" + compiled + "'\n does not match\n'" +
fwd_sql + "'\n or\n'" + rev_sql + "'")
-
+
def test_in(self):
self._test(User.id.in_(['a', 'b']),
"users.id IN (:users_id, :users_id_1)")
@@ -232,24 +232,24 @@ class OperatorTest(QueryTest):
):
c = expr.compile(dialect=default.DefaultDialect())
assert str(c) == compare, "%s != %s" % (str(c), compare)
-
-
+
+
class CompileTest(QueryTest):
def test_deferred(self):
session = create_session()
s = session.query(User).filter(and_(addresses.c.email_address == bindparam('emailad'), Address.user_id==User.id)).compile()
-
+
l = session.query(User).instances(s.execute(emailad = 'jack@bean.com'))
assert [User(id=7)] == l
-
+
class SliceTest(QueryTest):
def test_first(self):
assert User(id=7) == create_session().query(User).first()
-
+
assert create_session().query(User).filter(User.id==27).first() is None
-
+
# more slice tests are available in test/orm/generative.py
-
+
class TextTest(QueryTest):
def test_fulltext(self):
assert [User(id=7), User(id=8), User(id=9),User(id=10)] == create_session().query(User).from_statement("select * from users").all()
@@ -263,25 +263,25 @@ class TextTest(QueryTest):
def test_binds(self):
assert [User(id=8), User(id=9)] == create_session().query(User).filter("id in (:id1, :id2)").params(id1=8, id2=9).all()
-
+
class FilterTest(QueryTest):
def test_basic(self):
assert [User(id=7), User(id=8), User(id=9),User(id=10)] == create_session().query(User).all()
- @testing.unsupported('maxdb')
+ @testing.fails_on('maxdb')
def test_limit(self):
assert [User(id=8), User(id=9)] == create_session().query(User).limit(2).offset(1).all()
assert [User(id=8), User(id=9)] == list(create_session().query(User)[1:3])
assert User(id=8) == create_session().query(User)[1]
-
+
def test_onefilter(self):
assert [User(id=8), User(id=9)] == create_session().query(User).filter(User.name.endswith('ed')).all()
def test_contains(self):
"""test comparing a collection to an object instance."""
-
+
sess = create_session()
address = sess.query(Address).get(3)
assert [User(id=8)] == sess.query(User).filter(User.addresses.contains(address)).all()
@@ -299,9 +299,9 @@ class FilterTest(QueryTest):
assert False
except exceptions.InvalidRequestError:
assert True
-
+
#assert [User(id=7), User(id=9), User(id=10)] == sess.query(User).filter(User.addresses!=address).all()
-
+
def test_any(self):
sess = create_session()
@@ -314,16 +314,15 @@ class FilterTest(QueryTest):
assert [User(id=9)] == sess.query(User).filter(User.addresses.any(email_address='fred@fred.com')).all()
- # THIS ONE
- @testing.unsupported('maxdb')
+ @testing.unsupported('maxdb') # can core
def test_has(self):
sess = create_session()
assert [Address(id=5)] == sess.query(Address).filter(Address.user.has(name='fred')).all()
-
+
assert [Address(id=2), Address(id=3), Address(id=4), Address(id=5)] == sess.query(Address).filter(Address.user.has(User.name.like('%ed%'))).all()
-
+
assert [Address(id=2), Address(id=3), Address(id=4)] == sess.query(Address).filter(Address.user.has(User.name.like('%ed%'), id=8)).all()
-
+
def test_contains_m2m(self):
sess = create_session()
item = sess.query(Item).get(3)
@@ -333,7 +332,7 @@ class FilterTest(QueryTest):
def test_comparison(self):
"""test scalar comparison to an object instance"""
-
+
sess = create_session()
user = sess.query(User).get(8)
assert [Address(id=2), Address(id=3), Address(id=4)] == sess.query(Address).filter(Address.user==user).all()
@@ -349,8 +348,8 @@ class AggregateTest(QueryTest):
def test_apply(self):
sess = create_session()
assert sess.query(Order).apply_sum(Order.user_id * Order.address_id).filter(Order.id.in_([2, 3, 4])).one() == 79
-
-
+
+
class CountTest(QueryTest):
def test_basic(self):
assert 4 == create_session().query(User).count()
@@ -370,13 +369,13 @@ class TextTest(QueryTest):
def test_binds(self):
assert [User(id=8), User(id=9)] == create_session().query(User).filter("id in (:id1, :id2)").params(id1=8, id2=9).all()
-
-
+
+
class ParentTest(QueryTest):
def test_o2m(self):
sess = create_session()
q = sess.query(User)
-
+
u1 = q.filter_by(name='jack').one()
# test auto-lookup of property
@@ -398,7 +397,7 @@ class ParentTest(QueryTest):
def test_noparent(self):
sess = create_session()
q = sess.query(User)
-
+
u1 = q.filter_by(name='jack').one()
try:
@@ -412,8 +411,8 @@ class ParentTest(QueryTest):
i1 = sess.query(Item).filter_by(id=2).one()
k = sess.query(Keyword).with_parent(i1).all()
assert [Keyword(name='red'), Keyword(name='small'), Keyword(name='square')] == k
-
-
+
+
class JoinTest(QueryTest):
def test_overlapping_paths(self):
@@ -434,13 +433,13 @@ class JoinTest(QueryTest):
result = create_session().query(User).outerjoin(['orders', 'items'], aliased=aliased).filter_by(id=3).reset_joinpoint().outerjoin(['orders','address'], aliased=aliased).filter_by(id=1).all()
assert [User(id=7, name='jack')] == result
-
+
def test_overlap_with_aliases(self):
oalias = orders.alias('oalias')
result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_(["order 1", "order 2", "order 3"])).join(['orders', 'items']).all()
assert [User(id=7, name='jack'), User(id=9, name='fred')] == result
-
+
result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_(["order 1", "order 2", "order 3"])).join(['orders', 'items']).filter_by(id=4).all()
assert [User(id=7, name='jack')] == result
@@ -472,7 +471,7 @@ class JoinTest(QueryTest):
q = sess.query(User).join('orders').filter(Order.description=="order 3").join(['orders', 'items']).filter(Order.description=="item 1")
assert [] == q.all()
assert q.count() == 0
-
+
q = sess.query(User).join('orders', aliased=True).filter(Order.items.any(Item.description=='item 4'))
assert [User(id=7)] == q.all()
@@ -544,8 +543,8 @@ class MultiplePathTest(ORMTest):
create_session().query(T1).join('t2s_1', aliased=True).filter(t2.c.id==5).reset_joinpoint().join('t2s_2').all()
create_session().query(T1).join('t2s_1').filter(t2.c.id==5).reset_joinpoint().join('t2s_2', aliased=True).all()
-
-
+
+
class SynonymTest(QueryTest):
keep_mappers = True
@@ -588,12 +587,12 @@ class SynonymTest(QueryTest):
):
sess = create_session()
q = sess.query(User)
-
+
u1 = q.filter_by(**{nameprop:'jack'}).one()
o = sess.query(Order).with_parent(u1, property=orderprop).all()
assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
-
+
class InstancesTest(QueryTest):
def test_from_alias(self):
@@ -669,7 +668,7 @@ class InstancesTest(QueryTest):
(user8, address4),
(user9, address5),
(user10, None)]
-
+
selectquery = users.outerjoin(addresses).select(use_labels=True, order_by=[users.c.id, addresses.c.id])
q = sess.query(User)
l = q.instances(selectquery.execute(), Address)
@@ -684,7 +683,7 @@ class InstancesTest(QueryTest):
q = sess.query(User).add_entity(Address)
l = q.join('addresses', aliased=aliased).filter_by(email_address='ed@bettyboop.com').all()
assert l == [(user8, address3)]
-
+
q = sess.query(User, Address).join('addresses', aliased=aliased).filter_by(email_address='ed@bettyboop.com')
assert q.all() == [(user8, address3)]
@@ -704,7 +703,7 @@ class InstancesTest(QueryTest):
(user8, address4),
(user9, address5),
(user10, None)]
-
+
q = sess.query(User)
adalias = addresses.alias('adalias')
q = q.add_entity(Address, alias=adalias).select_from(users.outerjoin(adalias))
@@ -714,7 +713,7 @@ class InstancesTest(QueryTest):
q = sess.query(User).add_entity(Address, alias=adalias)
l = q.select_from(users.outerjoin(adalias)).filter(adalias.c.email_address=='ed@bettyboop.com').all()
assert l == [(user8, address3)]
-
+
def test_multi_columns(self):
"""test aliased/nonalised joins with the usage of add_column()"""
sess = create_session()
@@ -724,7 +723,7 @@ class InstancesTest(QueryTest):
(user9, 1),
(user10, 0)
]
-
+
for aliased in (False, True):
q = sess.query(User)
q = q.group_by([c for c in users.c]).order_by(User.id).outerjoin('addresses', aliased=aliased).add_column(func.count(Address.id).label('count'))
@@ -750,12 +749,12 @@ class InstancesTest(QueryTest):
q = create_session().query(User)
l = q.add_column("count").add_column("concat").from_statement(s).all()
assert l == expected
-
+
# test with select_from()
q = create_session().query(User).add_column(func.count(addresses.c.id))\
.add_column(("Name:" + users.c.name)).select_from(users.outerjoin(addresses))\
.group_by([c for c in users.c]).order_by(users.c.id)
-
+
assert q.all() == expected
# test with outerjoin() both aliased and non
@@ -763,7 +762,7 @@ class InstancesTest(QueryTest):
q = create_session().query(User).add_column(func.count(addresses.c.id))\
.add_column(("Name:" + users.c.name)).outerjoin('addresses', aliased=aliased)\
.group_by([c for c in users.c]).order_by(users.c.id)
-
+
assert q.all() == expected
class CustomJoinTest(QueryTest):
@@ -802,7 +801,7 @@ class SelfReferentialJoinTest(ORMTest):
self.children.append(node)
mapper(Node, nodes, properties={
- 'children':relation(Node, lazy=True, join_depth=3,
+ 'children':relation(Node, lazy=True, join_depth=3,
backref=backref('parent', remote_side=[nodes.c.id])
)
})
@@ -817,7 +816,7 @@ class SelfReferentialJoinTest(ORMTest):
sess.save(n1)
sess.flush()
sess.clear()
-
+
# TODO: the aliasing of the join in query._join_to has to limit the aliasing
# among local_side / remote_side (add local_side as an attribute on PropertyLoader)
# also implement this idea in EagerLoader
@@ -826,7 +825,7 @@ class SelfReferentialJoinTest(ORMTest):
node = sess.query(Node).join(['children', 'children'], aliased=True).filter_by(data='n122').first()
assert node.data=='n1'
-
+
node = sess.query(Node).filter_by(data='n122').join('parent', aliased=True).filter_by(data='n12').\
join('parent', aliased=True, from_joinpoint=True).filter_by(data='n1').first()
assert node.data == 'n122'
@@ -868,7 +867,7 @@ class ExternalColumnsTest(QueryTest):
mapper(Address, addresses, properties={
'user':relation(User, lazy=True)
- })
+ })
sess = create_session()
l = sess.query(User).all()
@@ -886,9 +885,9 @@ class ExternalColumnsTest(QueryTest):
Address(id=4, user=User(id=8, concat=16, count=3)),
Address(id=5, user=User(id=9, concat=18, count=1))
]
-
+
assert address_result == sess.query(Address).all()
-
+
# run the eager version twice to test caching of aliased clauses
for x in range(2):
sess.clear()
@@ -897,7 +896,7 @@ class ExternalColumnsTest(QueryTest):
self.assert_sql_count(testbase.db, go, 1)
tuple_address_result = [(address, address.user) for address in address_result]
-
+
tuple_address_result == sess.query(Address).join('user').add_entity(User).all()
assert tuple_address_result == sess.query(Address).join('user', aliased=True, id='ualias').add_entity(User, id='ualias').all()
diff --git a/test/sql/case_statement.py b/test/sql/case_statement.py
index 7856d758b..bddf70b99 100644
--- a/test/sql/case_statement.py
+++ b/test/sql/case_statement.py
@@ -24,15 +24,15 @@ class CaseTest(PersistTest):
{'pk':6, 'info':'pk_6_data'})
def tearDownAll(self):
info_table.drop()
-
- @testing.unsupported('maxdb')
+
+ @testing.fails_on('maxdb')
def testcase(self):
inner = select([case([
- [info_table.c.pk < 3,
+ [info_table.c.pk < 3,
literal('lessthan3', type_=String)],
- [and_(info_table.c.pk >= 3, info_table.c.pk < 7),
+ [and_(info_table.c.pk >= 3, info_table.c.pk < 7),
literal('gt3', type_=String)]]).label('x'),
- info_table.c.pk, info_table.c.info],
+ info_table.c.pk, info_table.c.info],
from_obj=[info_table]).alias('q_inner')
inner_result = inner.execute().fetchall()
@@ -67,12 +67,12 @@ class CaseTest(PersistTest):
]
w_else = select([case([
- [info_table.c.pk < 3,
+ [info_table.c.pk < 3,
literal(3, type_=Integer)],
- [and_(info_table.c.pk >= 3, info_table.c.pk < 6),
+ [and_(info_table.c.pk >= 3, info_table.c.pk < 6),
literal(6, type_=Integer)]],
else_ = 0).label('x'),
- info_table.c.pk, info_table.c.info],
+ info_table.c.pk, info_table.c.info],
from_obj=[info_table]).alias('q_inner')
else_result = w_else.execute().fetchall()
diff --git a/test/sql/query.py b/test/sql/query.py
index c4a3e9e0d..9a4524fdb 100644
--- a/test/sql/query.py
+++ b/test/sql/query.py
@@ -7,7 +7,7 @@ from testlib import *
class QueryTest(PersistTest):
-
+
def setUpAll(self):
global users, addresses, metadata
metadata = MetaData(testbase.db)
@@ -15,23 +15,23 @@ class QueryTest(PersistTest):
Column('user_id', INT, primary_key = True),
Column('user_name', VARCHAR(20)),
)
- addresses = Table('query_addresses', metadata,
+ addresses = Table('query_addresses', metadata,
Column('address_id', Integer, primary_key=True),
Column('user_id', Integer, ForeignKey('query_users.user_id')),
Column('address', String(30)))
metadata.create_all()
-
+
def tearDown(self):
addresses.delete().execute()
users.delete().execute()
-
+
def tearDownAll(self):
metadata.drop_all()
-
+
def testinsert(self):
users.insert().execute(user_id = 7, user_name = 'jack')
assert users.count().scalar() == 1
-
+
def test_insert_heterogeneous_params(self):
users.insert().execute(
{'user_id':7, 'user_name':'jack'},
@@ -39,7 +39,7 @@ class QueryTest(PersistTest):
{'user_id':9}
)
assert users.select().execute().fetchall() == [(7, 'jack'), (8, 'ed'), (9, None)]
-
+
def testupdate(self):
users.insert().execute(user_id = 7, user_name = 'jack')
assert users.count().scalar() == 1
@@ -56,7 +56,7 @@ class QueryTest(PersistTest):
INSERTed including defaults that fired off on the DB side and
detects rows that had defaults and post-fetches.
"""
-
+
result = table.insert().execute(**values)
ret = values.copy()
@@ -73,7 +73,7 @@ class QueryTest(PersistTest):
for supported, table, values, assertvalues in [
(
{'unsupported':['sqlite']},
- Table("t1", metadata,
+ Table("t1", metadata,
Column('id', Integer, Sequence('t1_id_seq', optional=True), primary_key=True),
Column('foo', String(30), primary_key=True)),
{'foo':'hi'},
@@ -81,7 +81,7 @@ class QueryTest(PersistTest):
),
(
{'unsupported':['sqlite']},
- Table("t2", metadata,
+ Table("t2", metadata,
Column('id', Integer, Sequence('t2_id_seq', optional=True), primary_key=True),
Column('foo', String(30), primary_key=True),
Column('bar', String(30), PassiveDefault('hi'))
@@ -91,7 +91,7 @@ class QueryTest(PersistTest):
),
(
{'unsupported':[]},
- Table("t3", metadata,
+ Table("t3", metadata,
Column("id", String(40), primary_key=True),
Column('foo', String(30), primary_key=True),
Column("bar", String(30))
@@ -101,7 +101,7 @@ class QueryTest(PersistTest):
),
(
{'unsupported':[]},
- Table("t4", metadata,
+ Table("t4", metadata,
Column('id', Integer, Sequence('t4_id_seq', optional=True), primary_key=True),
Column('foo', String(30), primary_key=True),
Column('bar', String(30), PassiveDefault('hi'))
@@ -111,7 +111,7 @@ class QueryTest(PersistTest):
),
(
{'unsupported':[]},
- Table("t5", metadata,
+ Table("t5", metadata,
Column('id', String(10), primary_key=True),
Column('bar', String(30), PassiveDefault('hi'))
),
@@ -139,30 +139,30 @@ class QueryTest(PersistTest):
self.assert_(len(l) == 3)
def test_fetchmany(self):
- users.insert().execute(user_id = 7, user_name = 'jack')
- users.insert().execute(user_id = 8, user_name = 'ed')
- users.insert().execute(user_id = 9, user_name = 'fred')
- r = users.select().execute()
- l = []
- for row in r.fetchmany(size=2):
- l.append(row)
+ users.insert().execute(user_id = 7, user_name = 'jack')
+ users.insert().execute(user_id = 8, user_name = 'ed')
+ users.insert().execute(user_id = 9, user_name = 'fred')
+ r = users.select().execute()
+ l = []
+ for row in r.fetchmany(size=2):
+ l.append(row)
self.assert_(len(l) == 2, "fetchmany(size=2) got %s rows" % len(l))
-
+
def test_compiled_execute(self):
- users.insert().execute(user_id = 7, user_name = 'jack')
+ users.insert().execute(user_id = 7, user_name = 'jack')
s = select([users], users.c.user_id==bindparam('id')).compile()
c = testbase.db.connect()
assert c.execute(s, id=7).fetchall()[0]['user_id'] == 7
-
+
def test_compiled_insert_execute(self):
- users.insert().compile().execute(user_id = 7, user_name = 'jack')
+ users.insert().compile().execute(user_id = 7, user_name = 'jack')
s = select([users], users.c.user_id==bindparam('id')).compile()
c = testbase.db.connect()
assert c.execute(s, id=7).fetchall()[0]['user_id'] == 7
def test_repeated_bindparams(self):
"""Tests that a BindParam can be used more than once.
-
+
This should be run for DB-APIs with both positional and named
paramstyles.
"""
@@ -173,7 +173,7 @@ class QueryTest(PersistTest):
s = users.select(or_(users.c.user_name==u, users.c.user_name==u))
r = s.execute(userid='fred').fetchall()
assert len(r) == 1
-
+
def test_bindparam_shortname(self):
"""test the 'shortname' field on BindParamClause."""
users.insert().execute(user_id = 7, user_name = 'jack')
@@ -213,14 +213,14 @@ class QueryTest(PersistTest):
a_eq(prep(":this \:that"), "? :that")
a_eq(prep(r"(\:that$other)"), "(:that$other)")
a_eq(prep(r".\:that$ :other."), ".:that$ ?.")
-
+
def testdelete(self):
users.insert().execute(user_id = 7, user_name = 'jack')
users.insert().execute(user_id = 8, user_name = 'fred')
print repr(users.select().execute().fetchall())
users.delete(users.c.user_name == 'fred').execute()
-
+
print repr(users.select().execute().fetchall())
def testselectlimit(self):
@@ -233,8 +233,9 @@ class QueryTest(PersistTest):
users.insert().execute(user_id=7, user_name='fido')
r = users.select(limit=3, order_by=[users.c.user_id]).execute().fetchall()
self.assert_(r == [(1, 'john'), (2, 'jack'), (3, 'ed')], repr(r))
-
- @testing.unsupported('mssql', 'maxdb')
+
+ @testing.unsupported('mssql')
+ @testing.fails_on('maxdb')
def testselectlimitoffset(self):
users.insert().execute(user_id=1, user_name='john')
users.insert().execute(user_id=2, user_name='jack')
@@ -247,8 +248,9 @@ class QueryTest(PersistTest):
self.assert_(r==[(3, 'ed'), (4, 'wendy'), (5, 'laura')])
r = users.select(offset=5, order_by=[users.c.user_id]).execute().fetchall()
self.assert_(r==[(6, 'ralph'), (7, 'fido')])
-
- @testing.supported('mssql', 'maxdb')
+
+ @testing.supported('mssql')
+ @testing.fails_on('maxdb')
def test_select_limit_nooffset(self):
try:
r = users.select(limit=3, offset=2, order_by=[users.c.user_id]).execute().fetchall()
@@ -256,11 +258,11 @@ class QueryTest(PersistTest):
except exceptions.InvalidRequestError:
pass
- @testing.unsupported('mysql')
+ @testing.unsupported('mysql')
def test_scalar_select(self):
"""test that scalar subqueries with labels get their type propigated to the result set."""
# mysql and/or mysqldb has a bug here, type isnt propigated for scalar subquery.
- datetable = Table('datetable', metadata,
+ datetable = Table('datetable', metadata,
Column('id', Integer, primary_key=True),
Column('today', DateTime))
datetable.create()
@@ -278,7 +280,7 @@ class QueryTest(PersistTest):
Tests simple, compound, aliased and DESC clauses.
"""
-
+
users.insert().execute(user_id=1, user_name='c')
users.insert().execute(user_id=2, user_name='b')
users.insert().execute(user_id=3, user_name='a')
@@ -291,7 +293,7 @@ class QueryTest(PersistTest):
a_eq(users.select(order_by=[users.c.user_id],
use_labels=labels),
[(1, 'c'), (2, 'b'), (3, 'a')])
-
+
a_eq(users.select(order_by=[users.c.user_name, users.c.user_id],
use_labels=labels),
[(3, 'a'), (2, 'b'), (1, 'c')])
@@ -300,7 +302,7 @@ class QueryTest(PersistTest):
use_labels=labels,
order_by=[users.c.user_id]),
[(1,), (2,), (3,)])
-
+
a_eq(select([users.c.user_id.label('foo'), users.c.user_name],
use_labels=labels,
order_by=[users.c.user_name, users.c.user_id]),
@@ -339,7 +341,7 @@ class QueryTest(PersistTest):
users.insert().execute(user_id=1, user_name='john')
users.insert().execute(user_id=2, user_name='jack')
addresses.insert().execute(address_id=1, user_id=2, address='foo@bar.com')
-
+
r = users.select(users.c.user_id==2).execute().fetchone()
self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2)
self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack')
@@ -347,13 +349,13 @@ class QueryTest(PersistTest):
r = text("select * from query_users where user_id=2", bind=testbase.db).execute().fetchone()
self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2)
self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack')
-
+
# test slices
r = text("select * from query_addresses", bind=testbase.db).execute().fetchone()
self.assert_(r[0:1] == (1,))
self.assert_(r[1:] == (2, 'foo@bar.com'))
self.assert_(r[:-1] == (1, 2))
-
+
def test_ambiguous_column(self):
users.insert().execute(user_id=1, user_name='john')
r = users.outerjoin(addresses).select().execute().fetchone()
@@ -363,7 +365,7 @@ class QueryTest(PersistTest):
except exceptions.InvalidRequestError, e:
assert str(e) == "Ambiguous column name 'user_id' in result set! try 'use_labels' option on select statement." or \
str(e) == "Ambiguous column name 'USER_ID' in result set! try 'use_labels' option on select statement."
-
+
def test_column_label_targeting(self):
users.insert().execute(user_id=7, user_name='ed')
@@ -374,7 +376,7 @@ class QueryTest(PersistTest):
row = s.select(use_labels=True).execute().fetchone()
assert row[s.c.user_id] == 7
assert row[s.c.user_name] == 'ed'
-
+
def test_keys(self):
users.insert().execute(user_id=1, user_name='foo')
r = users.select().execute().fetchone()
@@ -396,19 +398,19 @@ class QueryTest(PersistTest):
r = testbase.db.execute('select user_name from query_users', {}).fetchone()
self.assertEqual(len(r), 1)
r.close()
-
+
def test_cant_execute_join(self):
try:
users.join(addresses).execute()
except exceptions.ArgumentError, e:
assert str(e).startswith('Not an executable clause: ')
-
+
def test_functions(self):
x = testbase.db.func.current_date().execute().scalar()
y = testbase.db.func.current_date().select().execute().scalar()
z = testbase.db.func.current_date().scalar()
assert (x == y == z) is True
-
+
x = testbase.db.func.current_date(type_=Date)
assert isinstance(x.type, Date)
assert isinstance(x.execute().scalar(), datetime.date)
@@ -422,7 +424,7 @@ class QueryTest(PersistTest):
finally:
conn.close()
assert (x == y == z) is True
-
+
def test_update_functions(self):
"""
Tests sending functions and SQL expressions to the VALUES and SET
@@ -459,25 +461,25 @@ class QueryTest(PersistTest):
res = exec_sorted(select([t2.c.value, t2.c.stuff]))
self.assertEquals(res, [(-14, 'hi'), (3, None), (7, None)])
-
+
t2.update(values=dict(value=func.length("asdsafasd"))).execute(stuff="some stuff")
assert select([t2.c.value, t2.c.stuff]).execute().fetchall() == [(9,"some stuff"), (9,"some stuff"), (9,"some stuff")]
-
+
t2.delete().execute()
-
+
t2.insert(values=dict(value=func.length("one") + 8)).execute()
assert t2.select().execute().fetchone()['value'] == 11
-
+
t2.update(values=dict(value=func.length("asfda"))).execute()
assert select([t2.c.value, t2.c.stuff]).execute().fetchone() == (5, "thisisstuff")
t2.update(values={t2.c.value:func.length("asfdaasdf"), t2.c.stuff:"foo"}).execute()
print "HI", select([t2.c.value, t2.c.stuff]).execute().fetchone()
assert select([t2.c.value, t2.c.stuff]).execute().fetchone() == (9, "foo")
-
+
finally:
meta.drop_all()
-
+
@testing.supported('postgres')
def test_functions_with_cols(self):
# TODO: shouldnt this work on oracle too ?
@@ -485,14 +487,14 @@ class QueryTest(PersistTest):
y = testbase.db.func.current_date().select().execute().scalar()
z = testbase.db.func.current_date().scalar()
w = select(['*'], from_obj=[testbase.db.func.current_date()]).scalar()
-
+
# construct a column-based FROM object out of a function, like in [ticket:172]
s = select([sql.column('date', type_=DateTime)], from_obj=[testbase.db.func.current_date()])
q = s.execute().fetchone()[s.c.date]
r = s.alias('datequery').select().scalar()
-
+
assert x == y == z == w == q == r
-
+
def test_column_order_with_simple_query(self):
# should return values in column definition order
users.insert().execute(user_id=1, user_name='foo')
@@ -501,7 +503,7 @@ class QueryTest(PersistTest):
self.assertEqual(r[1], 'foo')
self.assertEqual([x.lower() for x in r.keys()], ['user_id', 'user_name'])
self.assertEqual(r.values(), [1, 'foo'])
-
+
def test_column_order_with_text_query(self):
# should return values in query order
users.insert().execute(user_id=1, user_name='foo')
@@ -510,7 +512,7 @@ class QueryTest(PersistTest):
self.assertEqual(r[1], 1)
self.assertEqual([x.lower() for x in r.keys()], ['user_name', 'user_id'])
self.assertEqual(r.values(), ['foo', 1])
-
+
@testing.unsupported('oracle', 'firebird', 'maxdb')
def test_column_accessor_shadow(self):
meta = MetaData(testbase.db)
@@ -540,7 +542,7 @@ class QueryTest(PersistTest):
r.close()
finally:
shadowed.drop(checkfirst=True)
-
+
@testing.supported('mssql')
def test_fetchid_trigger(self):
meta = MetaData(testbase.db)
@@ -566,16 +568,16 @@ class QueryTest(PersistTest):
tr.commit()
con.execute("""drop trigger paj""")
meta.drop_all()
-
+
@testing.supported('mssql')
def test_insertid_schema(self):
meta = MetaData(testbase.db)
con = testbase.db.connect()
con.execute('create schema paj')
tbl = Table('test', meta, Column('id', Integer, primary_key=True), schema='paj')
- tbl.create()
+ tbl.create()
try:
- tbl.insert().execute({'id':1})
+ tbl.insert().execute({'id':1})
finally:
tbl.drop()
con.execute('drop schema paj')
@@ -584,58 +586,58 @@ class QueryTest(PersistTest):
def test_insertid_reserved(self):
meta = MetaData(testbase.db)
table = Table(
- 'select', meta,
+ 'select', meta,
Column('col', Integer, primary_key=True)
)
table.create()
-
+
meta2 = MetaData(testbase.db)
try:
table.insert().execute(col=7)
finally:
table.drop()
- @testing.unsupported('maxdb')
+ @testing.fails_on('maxdb')
def test_in_filtering(self):
"""test the behavior of the in_() function."""
-
+
users.insert().execute(user_id = 7, user_name = 'jack')
users.insert().execute(user_id = 8, user_name = 'fred')
users.insert().execute(user_id = 9, user_name = None)
-
+
s = users.select(users.c.user_name.in_([]))
r = s.execute().fetchall()
# No username is in empty set
assert len(r) == 0
-
+
s = users.select(not_(users.c.user_name.in_([])))
r = s.execute().fetchall()
# All usernames with a value are outside an empty set
assert len(r) == 2
-
+
s = users.select(users.c.user_name.in_(['jack','fred']))
r = s.execute().fetchall()
assert len(r) == 2
-
+
s = users.select(not_(users.c.user_name.in_(['jack','fred'])))
r = s.execute().fetchall()
# Null values are not outside any set
assert len(r) == 0
-
+
u = bindparam('search_key')
-
+
s = users.select(u.in_([]))
r = s.execute(search_key='john').fetchall()
assert len(r) == 0
r = s.execute(search_key=None).fetchall()
assert len(r) == 0
-
+
s = users.select(not_(u.in_([])))
r = s.execute(search_key='john').fetchall()
assert len(r) == 3
r = s.execute(search_key=None).fetchall()
assert len(r) == 0
-
+
s = users.select(users.c.user_name.in_([]) == True)
r = s.execute().fetchall()
assert len(r) == 0
@@ -653,24 +655,24 @@ class CompoundTest(PersistTest):
def setUpAll(self):
global metadata, t1, t2, t3
metadata = MetaData(testbase.db)
- t1 = Table('t1', metadata,
+ t1 = Table('t1', metadata,
Column('col1', Integer, Sequence('t1pkseq'), primary_key=True),
Column('col2', String(30)),
Column('col3', String(40)),
Column('col4', String(30))
)
- t2 = Table('t2', metadata,
+ t2 = Table('t2', metadata,
Column('col1', Integer, Sequence('t2pkseq'), primary_key=True),
Column('col2', String(30)),
Column('col3', String(40)),
Column('col4', String(30)))
- t3 = Table('t3', metadata,
+ t3 = Table('t3', metadata,
Column('col1', Integer, Sequence('t3pkseq'), primary_key=True),
Column('col2', String(30)),
Column('col3', String(40)),
Column('col4', String(30)))
metadata.create_all()
-
+
t1.insert().execute([
dict(col2="t1col2r1", col3="aaa", col4="aaa"),
dict(col2="t1col2r2", col3="bbb", col4="bbb"),
@@ -686,7 +688,7 @@ class CompoundTest(PersistTest):
dict(col2="t3col2r2", col3="bbb", col4="aaa"),
dict(col2="t3col2r3", col3="ccc", col4="bbb"),
])
-
+
def tearDownAll(self):
metadata.drop_all()
@@ -699,7 +701,7 @@ class CompoundTest(PersistTest):
t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
- )
+ )
u = union(s1, s2)
wanted = [('aaa', 'aaa'), ('bbb', 'bbb'), ('bbb', 'ccc'),
@@ -709,28 +711,28 @@ class CompoundTest(PersistTest):
found2 = self._fetchall_sorted(u.alias('bar').select().execute())
self.assertEquals(found2, wanted)
-
+
def test_union_ordered(self):
(s1, s2) = (
select([t1.c.col3.label('col3'), t1.c.col4.label('col4')],
t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
- )
+ )
u = union(s1, s2, order_by=['col3', 'col4'])
wanted = [('aaa', 'aaa'), ('bbb', 'bbb'), ('bbb', 'ccc'),
('ccc', 'aaa')]
self.assertEquals(u.execute().fetchall(), wanted)
- @testing.unsupported('maxdb')
+ @testing.fails_on('maxdb')
def test_union_ordered_alias(self):
(s1, s2) = (
select([t1.c.col3.label('col3'), t1.c.col4.label('col4')],
t1.c.col2.in_(["t1col2r1", "t1col2r2"])),
select([t2.c.col3.label('col3'), t2.c.col4.label('col4')],
t2.c.col2.in_(["t2col2r2", "t2col2r3"]))
- )
+ )
u = union(s1, s2, order_by=['col3', 'col4'])
wanted = [('aaa', 'aaa'), ('bbb', 'bbb'), ('bbb', 'ccc'),
@@ -826,7 +828,7 @@ class CompoundTest(PersistTest):
)
wanted = [('aaa', 'bbb'), ('bbb', 'ccc'), ('ccc', 'aaa')]
found = self._fetchall_sorted(u.execute())
-
+
self.assertEquals(found, wanted)
@testing.unsupported('mysql')
@@ -889,7 +891,7 @@ class JoinTest(PersistTest):
def assertRows(self, statement, expected):
"""Execute a statement and assert that rows returned equal expected."""
-
+
found = exec_sorted(statement)
self.assertEquals(found, sorted(expected))
@@ -1069,7 +1071,7 @@ class JoinTest(PersistTest):
from_obj=[(t1.join(t2).outerjoin(t3, criteria))])
print expr
self.assertRows(expr, [(10, 20, 30), (11, 21, None)])
-
+
def test_mixed_where(self):
"""Joins t1->t2, outer t2->t3, plus a where on each table in turn."""
@@ -1117,13 +1119,13 @@ class OperatorTest(PersistTest):
def setUpAll(self):
global metadata, flds
metadata = MetaData(testbase.db)
- flds = Table('flds', metadata,
+ flds = Table('flds', metadata,
Column('idcol', Integer, Sequence('t1pkseq'), primary_key=True),
Column('intcol', Integer),
Column('strcol', String(50)),
)
metadata.create_all()
-
+
flds.insert().execute([
dict(intcol=5, strcol='foo'),
dict(intcol=13, strcol='bar')
@@ -1132,7 +1134,7 @@ class OperatorTest(PersistTest):
def tearDownAll(self):
metadata.drop_all()
- @testing.unsupported('maxdb')
+ @testing.fails_on('maxdb')
def test_modulo(self):
self.assertEquals(
select([flds.c.intcol % 3],
@@ -1149,4 +1151,4 @@ def exec_sorted(statement, *args, **kw):
if __name__ == "__main__":
- testbase.main()
+ testbase.main()