collection based on the type of target given.
"""
+
def listen_one(x, y):
pass
PyPy) is not interpreted as an iterable.
"""
+
class IterString(str):
def __iter__(self):
return iter(self + '')
def test_default_flags(self):
"""test with no initialization or server version info"""
+
dialect = oracle.dialect(dbapi=testing.db.dialect.dbapi)
assert dialect._supports_char_length
assert dialect._supports_nchar
@testing.provide_metadata
def test_numerics_broken_inspection(self):
- metadata = self.metadata
"""Numeric scenarios where Oracle type info is 'broken',
returning us precision, scale of the form (0, 0) or (0, -127).
We convert to Decimal and let int()/float() processors take over.
"""
+ metadata = self.metadata
+
# this test requires cx_oracle 5
foo = Table('foo', metadata,
@testing.provide_metadata
def test_index_reflection(self):
- metadata = self.metadata
""" Reflecting partial & expression-based indexes should warn
"""
+ metadata = self.metadata
+
t1 = Table('party', metadata, Column('id', String(10),
nullable=False), Column('name', String(20),
index=True), Column('aname', String(20)))
'behavior of default displayed in pragma '
'table_info()')
def test_default_reflection_2(self):
+
db = testing.db
m = MetaData(db)
expected = ['my_default', '0']
"""CREATE TABLE r_defaults (
data VARCHAR(40) DEFAULT 'my_default',
val INTEGER NOT NULL DEFAULT 0
- )"""
+ )"""
try:
db.execute(table)
rt = Table('r_defaults', m, autoload=True)
@testing.provide_metadata
def test_no_rowcount_on_selects_inserts(self):
- metadata = self.metadata
"""assert that rowcount is only called on deletes and updates.
This because cursor.rowcount can be expensive on some dialects
"""
+ metadata = self.metadata
+
engine = engines.testing_engine()
metadata.bind = engine
@testing.provide_metadata
def test_override_composite_fk(self):
- metadata = self.metadata
"""Test double-remove of composite foreign key, when replaced."""
+ metadata = self.metadata
+
a = Table('a',
metadata,
Column('x', sa.Integer, primary_key=True),
column in its reflection.
"""
+
meta = MetaData(testing.db)
a1 = Table('a', meta,
Column('x', sa.Integer, primary_key=True),
def test_rollback(self):
"""test a basic rollback"""
+
connection = testing.db.connect()
transaction = connection.begin()
connection.execute(users.insert(), user_id=1, user_name='user1')
decorated class' compiler.
"""
+
t1 = table('t1', column('c1'), column('c2'))
dispatch = Select._compiler_dispatch
See [ticket:1892].
"""
+
class Booking(Base):
__tablename__ = 'booking'
id = Column(Integer, primary_key=True)
See [ticket:1892].
"""
+
class Booking(Base):
__tablename__ = 'booking'
id = Column(Integer, primary_key=True)
of the superclass checks that the superclass property is a column.
"""
+
class General(Base):
__tablename__ = 'test'
id = Column(Integer, primary_key=True)
of the superclass checks that the subclass property is a column.
"""
+
def go():
class General(Base):
__tablename__ = 'test'
)
def test_optimized_passes(self):
- base, sub = self.tables.base, self.tables.sub
-
""""test that the 'optimized load' routine doesn't crash when
a column in the join condition is not available."""
+ base, sub = self.tables.base, self.tables.sub
+
+
class Base(fixtures.ComparableEntity):
pass
class Sub(Base):
assert sess.query(B).filter(B.bname == 'b1').one() is b1
def test_polymorphic_backref(self):
+ """test multiple backrefs to the same polymorphically-loading
+ attribute."""
+
A, C, B, c_table, b_table, a_table, Dest, dest_table = (self.classes.A,
self.classes.C,
self.classes.B,
self.classes.Dest,
self.tables.dest_table)
- """test multiple backrefs to the same polymorphically-loading
- attribute."""
ajoin = polymorphic_union({'a': a_table, 'b': b_table, 'c':c_table},
'type','ajoin')
def test_insert_order(self):
"""test that classes of multiple types mix up mapper inserts
so that insert order of individual tables is maintained"""
+
person_join = polymorphic_union(
{
'engineer':people.join(engineers),
def test_eager_empty(self):
"""test parent object with child relationship to an inheriting mapper, using eager loads,
works when there are no child objects present"""
+
class Person(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
@testing.uses_deprecated("fold_equivalents is deprecated.")
def test_manytoone_lazyload(self):
"""test that lazy load clause to a polymorphic child mapper generates correctly [ticket:493]"""
+
class PersistentObject(object):
def __init__(self, **kwargs):
for key, value in kwargs.iteritems():
def test_pjoin_compile(self):
"""test that remote_side columns in the secondary join table arent attempted to be
matched to the target polymorphic selectable"""
+
class BaseItem(object): pass
class Item(BaseItem): pass
class Collection(object): pass
def testfour(self):
"""this tests the RasterDocument being attached to the Assembly, but *not* the Document. this means only
a "sub-class" task, i.e. corresponding to an inheriting mapper but not the base mapper, is created. """
+
product_mapper = mapper(Product, products_table,
polymorphic_on=products_table.c.product_type,
polymorphic_identity='product')
session.flush()
def test_noorm(self):
+ """test the control case"""
+
tests, options, categories = (self.tables.tests,
self.tables.options,
self.tables.categories)
- """test the control case"""
# 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)
eq_(result, [u'1 Some Category', u'3 Some Category'])
def test_withjoinedload(self):
- Thing, tests, options = (self.classes.Thing,
- self.tables.tests,
- self.tables.options)
-
"""
Test that an joinedload locates the correct "from" clause with which to
attach to, when presented with a query that already has a complicated
from clause.
"""
+
+ Thing, tests, options = (self.classes.Thing,
+ self.tables.tests,
+ self.tables.options)
+
s = create_session()
q=s.query(Thing).options(sa.orm.joinedload('category'))
eq_(result, [u'1 Some Category', u'3 Some Category'])
def test_dslish(self):
+ """test the same as withjoinedload except using generative"""
+
Thing, tests, options = (self.classes.Thing,
self.tables.tests,
self.tables.options)
- """test the same as withjoinedload except using generative"""
s = create_session()
q = s.query(Thing).options(sa.orm.joinedload('category'))
l = q.filter (
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_eager_terminate(self):
- Middle, Right, Left = (self.classes.Middle,
- self.classes.Right,
- self.classes.Left)
-
"""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.
"""
+
+ Middle, Right, Left = (self.classes.Middle,
+ self.classes.Right,
+ self.classes.Left)
+
p = Middle('m1')
p.left.append(Left('l1'))
p.right.append(Right('r1'))
pass
def test_load_m2o_attached_to_o2(self):
- addresses, invoices, Company, companies, Invoice, Address = (self.tables.addresses,
- self.tables.invoices,
- self.classes.Company,
- self.tables.companies,
- self.classes.Invoice,
- self.classes.Address)
-
"""
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
the Company's second Address object.
"""
+
+ addresses, invoices, Company, companies, Invoice, Address = (self.tables.addresses,
+ self.tables.invoices,
+ self.classes.Company,
+ self.tables.companies,
+ self.classes.Invoice,
+ self.classes.Address)
+
mapper(Address, addresses)
mapper(Company, companies, properties={
def test_inheritance(self):
"""tests that attributes are polymorphic"""
+
class Foo(object):pass
class Bar(Foo):pass
since we use distinct objects in an inheritance scenario.
"""
+
class Parent(object):
pass
class Child(object):
assert a1 in u2.addresses
def test_plain_load_passive(self):
+ """test that many-to-one set doesn't load the old value."""
+
User, Address = self.classes.User, self.classes.Address
- """test that many-to-one set doesn't load the old value."""
sess = sessionmaker()()
u1 = User(name='jack')
assert_raises(sa_exc.DBAPIError, sess.flush)
def test_save_update_sends_pending(self):
- Order, User = self.classes.Order, self.classes.User
-
"""test that newly added and deleted collection items are
cascaded on save-update"""
+ Order, User = self.classes.Order, self.classes.User
+
+
sess = sessionmaker(expire_on_commit=False)()
o1, o2, o3 = Order(description='o1'), Order(description='o2'), \
Order(description='o3')
assert orders.count().scalar() == 0
def test_delete_unloaded_collections(self):
+ """Unloaded collections are still included in a delete-cascade
+ by default."""
+
User, addresses, users, Address = (self.classes.User,
self.tables.addresses,
self.tables.users,
self.classes.Address)
- """Unloaded collections are still included in a delete-cascade
- by default."""
sess = create_session()
u = User(name='jack',
addresses=[Address(email_address="address1"),
assert users.count().scalar() == 0
def test_cascades_onlycollection(self):
+ """Cascade only reaches instances that are still part of the
+ collection, not those that have been removed"""
+
User, Order, users, orders = (self.classes.User,
self.classes.Order,
self.tables.users,
self.tables.orders)
- """Cascade only reaches instances that are still part of the
- collection, not those that have been removed"""
sess = create_session()
u = User(name='jack',
orders=[Order(description='someorder')])])
def test_cascade_nosideeffects(self):
+ """test that cascade leaves the state of unloaded
+ scalars/collections unchanged."""
+
Dingaling, User, Address = (self.classes.Dingaling,
self.classes.User,
self.classes.Address)
- """test that cascade leaves the state of unloaded
- scalars/collections unchanged."""
sess = create_session()
u = User(name='jack')
assert extra.count().scalar() == 2
def test_cascade_on_deleted(self):
+ """test a bug introduced by r6711"""
+
Foo, User = self.classes.Foo, self.classes.User
- """test a bug introduced by r6711"""
sess = sessionmaker(expire_on_commit=True)()
sess.commit()
def test_save_update_sends_pending(self):
- Pref, User = self.classes.Pref, self.classes.User
-
"""test that newly added and deleted scalar items are cascaded
on save-update"""
+ Pref, User = self.classes.Pref, self.classes.User
+
+
sess = sessionmaker(expire_on_commit=False)()
p1, p2 = Pref(data='p1'), Pref(data='p2')
[Pref(data="someotherpref")])
def test_double_assignment(self):
+ """Double assignment will not accidentally reset the 'parent' flag."""
+
Pref, User = self.classes.Pref, self.classes.User
- """Double assignment will not accidentally reset the 'parent' flag."""
sess = create_session()
jack = sess.query(User).filter_by(name="jack").one()
)
def test_single_parent_backref(self):
+ """test that setting m2m via a uselist=False backref bypasses the single_parent raise"""
+
a, A, B, b, atob = (self.tables.a,
self.classes.A,
self.classes.B,
self.tables.b,
self.tables.atob)
- """test that setting m2m via a uselist=False backref bypasses the single_parent raise"""
mapper(A, a, properties={
'bs':relationship(B,
pass
def test_pending_standalone_orphan(self):
- users, orders, User, Address, Order, addresses = (self.tables.users,
- self.tables.orders,
- self.classes.User,
- self.classes.Address,
- self.classes.Order,
- self.tables.addresses)
-
"""Standalone 'orphan' objects can now be persisted, if the underlying
constraints of the database allow it.
"""
+ users, orders, User, Address, Order, addresses = (self.tables.users,
+ self.tables.orders,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.addresses)
+
+
mapper(Order, orders)
mapper(Address, addresses)
mapper(User, users, properties=dict(
def test_pending_collection_expunge(self):
+ """Removing a pending item from a collection expunges it from
+ the session."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """Removing a pending item from a collection expunges it from
- the session."""
mapper(Address, addresses)
mapper(User, users, properties=dict(
ForeignKey('accounts.account_id')))
def test_double_parent_expunge_o2m(self):
+ """test the delete-orphan uow event for multiple delete-orphan
+ parent relationships."""
+
sales_reps, customers, accounts = (self.tables.sales_reps,
self.tables.customers,
self.tables.accounts)
- """test the delete-orphan uow event for multiple delete-orphan
- parent relationships."""
class Customer(fixtures.ComparableEntity):
pass
'Should expunge customer when both parents are gone'
def test_double_parent_expunge_o2o(self):
+ """test the delete-orphan uow event for multiple delete-orphan
+ parent relationships."""
+
sales_reps, customers, accounts = (self.tables.sales_reps,
self.tables.customers,
self.tables.accounts)
- """test the delete-orphan uow event for multiple delete-orphan
- parent relationships."""
class Customer(fixtures.ComparableEntity):
pass
)
def test_non_orphan(self):
+ """test that an entity can have two parent delete-orphan
+ cascades, and persists normally."""
+
homes, businesses, addresses = (self.tables.homes,
self.tables.businesses,
self.tables.addresses)
- """test that an entity can have two parent delete-orphan
- cascades, and persists normally."""
class Address(fixtures.ComparableEntity):
pass
address=Address(street='address2')))
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."""
+
homes, businesses, addresses = (self.tables.homes,
self.tables.businesses,
self.tables.addresses)
- """test that an entity can have two parent delete-orphan
- cascades, and is detected as an orphan when saved without a
- parent."""
class Address(fixtures.ComparableEntity):
pass
@testing.uses_deprecated()
def test_circular_sort(self):
+ """test ticket 1306"""
+
base, inh_child, parent = (self.tables.base,
self.tables.inh_child,
self.tables.parent)
- """test ticket 1306"""
class Base(fixtures.ComparableEntity):
pass
assert isinstance(f.bars, MyList)
def test_lazyload(self):
+ """test that a 'set' can be used as a collection and can lazyload."""
+
someothertable, sometable = self.tables.someothertable, self.tables.sometable
- """test that a 'set' can be used as a collection and can lazyload."""
class Foo(object):
pass
class Bar(object):
f.bars.clear()
def test_dict(self):
+ """test that a 'dict' can be used as a collection and can lazyload."""
+
someothertable, sometable = self.tables.someothertable, self.tables.sometable
- """test that a 'dict' can be used as a collection and can lazyload."""
class Foo(object):
pass
f.bars.clear()
def test_dict_wrapper(self):
+ """test that the supplied 'dict' wrapper can be used as a collection and can lazyload."""
+
someothertable, sometable = self.tables.someothertable, self.tables.sometable
- """test that the supplied 'dict' wrapper can be used as a collection and can lazyload."""
class Foo(object):
pass
def test_conflicting_backref_one(self):
"""test that conflicting backrefs raises an exception"""
+
metadata = MetaData(testing.db)
order = Table('orders', metadata,
)
def test_save_null(self):
- Graph, Edge = self.classes.Graph, self.classes.Edge
-
"""test saving a null composite value
See google groups thread for more context:
http://groups.google.com/group/sqlalchemy/browse_thread/thread/0c6580a1761b2c29
"""
+
+ Graph, Edge = self.classes.Graph, self.classes.Edge
+
sess = Session()
g = Graph(id=1)
e = Edge(None, None)
sess.flush()
def test_many_to_one_only(self):
- C1, t1 = self.classes.C1, self.tables.t1
-
"""
test that the circular dependency sort can assemble a many-to-one
actually in the list of modified objects.
"""
+
+ C1, t1 = self.classes.C1, self.tables.t1
+
mapper(C1, t1, properties={
'parent':relationship(C1,
primaryjoin=t1.c.parent_c1 == t1.c.c1,
primaryjoin=child2.c.child1_id == child1.c.id)))
def test_many_to_one_only(self):
+ """test similar to SelfReferentialTest.testmanytooneonly"""
+
Child1, Child2 = self.classes.Child1, self.classes.Child2
- """test similar to SelfReferentialTest.testmanytooneonly"""
session = create_session()
sess.flush()
def test_reflush_2(self):
+ """A variant on test_reflush()"""
+
T2, T3, T1 = (self.classes.T2,
self.classes.T3,
self.classes.T1)
- """A variant on test_reflush()"""
o1 = T1()
o1.t2 = T2()
sess = create_session()
pass
def test_cycle(self):
- person, ball, Ball, Person = (self.tables.person,
- self.tables.ball,
- self.classes.Ball,
- self.classes.Person)
-
"""
This test has a peculiar aspect in that it doesnt create as many
dependent relationships as the other tests, and revealed a small
glitch in the circular dependency sorting.
"""
+
+ person, ball, Ball, Person = (self.tables.person,
+ self.tables.ball,
+ self.classes.Ball,
+ self.classes.Person)
+
mapper(Ball, ball)
mapper(Person, person, properties=dict(
balls=relationship(Ball,
sess.flush()
def test_post_update_m2o(self):
+ """A cycle between two rows, with a post_update on the many-to-one"""
+
person, ball, Ball, Person = (self.tables.person,
self.tables.ball,
self.classes.Ball,
self.classes.Person)
- """A cycle between two rows, with a post_update on the many-to-one"""
mapper(Ball, ball)
mapper(Person, person, properties=dict(
balls=relationship(Ball,
)
def test_post_update_backref(self):
+ """test bidirectional post_update."""
+
person, ball, Ball, Person = (self.tables.person,
self.tables.ball,
self.classes.Ball,
self.classes.Person)
- """test bidirectional post_update."""
mapper(Ball, ball)
mapper(Person, person, properties=dict(
def test_post_update_o2m(self):
+ """A cycle between two rows, with a post_update on the one-to-many"""
+
person, ball, Ball, Person = (self.tables.person,
self.tables.ball,
self.classes.Ball,
self.classes.Person)
- """A cycle between two rows, with a post_update on the one-to-many"""
mapper(Ball, ball)
mapper(Person, person, properties=dict(
self.path = path
def test_one(self):
- node, Node = self.tables.node, self.classes.Node
-
"""Post_update only fires off when needed.
This test case used to produce many superfluous update statements,
"""
+ node, Node = self.tables.node, self.classes.Node
+
+
mapper(Node, node, properties={
'children': relationship(
Node,
pass
def test_one(self):
- A, a_table = self.classes.A, self.tables.a_table
-
"""
Test that post_update remembers to be involved in update operations as
well, since it replaces the normal dependency processing completely
"""
+ A, a_table = self.classes.A, self.tables.a_table
+
+
mapper(A, a_table, properties={
'foo': relationship(A,
remote_side=[a_table.c.id],
output = session.query(User).deprecatedmethod(inputs)
"""
+
# 0.4+
output = session.query(User).newway(inputs)
assert output is correct
######################################################################
def test_override_get(self):
- Address = self.classes.Address
-
"""MapperExtension.get()
x = session.query.get(5)
"""
+
+ Address = self.classes.Address
+
from sqlalchemy.orm.query import Query
cache = {}
class MyQuery(Query):
assert ad1 in cache.values()
def test_load(self):
- Address = self.classes.Address
-
"""x = session.query(Address).load(1)
x = session.load(Address, 1)
"""
+ Address = self.classes.Address
+
+
session = create_session()
ad1 = session.query(Address).populate_existing().get(1)
assert bool(ad1)
def test_apply_max(self):
- Address = self.classes.Address
-
"""Query.apply_max(col)
max = session.query(Address).apply_max(Address.bounces)
"""
+
+ Address = self.classes.Address
+
session = create_session()
# 0.5.0
assert max == 10
def test_apply_min(self):
- Address = self.classes.Address
-
"""Query.apply_min(col)
min = session.query(Address).apply_min(Address.bounces)
"""
+
+ Address = self.classes.Address
+
session = create_session()
# 0.5.0
assert min == 0
def test_apply_avg(self):
- Address = self.classes.Address
-
"""Query.apply_avg(col)
avg = session.query(Address).apply_avg(Address.bounces)
"""
+
+ Address = self.classes.Address
+
session = create_session()
avgs = list(session.query(Address).values(func.avg(Address.bounces)))
assert avg > 0 and avg < 10
def test_apply_sum(self):
- Address = self.classes.Address
-
"""Query.apply_sum(col)
avg = session.query(Address).apply_avg(Address.bounces)
"""
+
+ Address = self.classes.Address
+
session = create_session()
avgs = list(session.query(Address).values(func.sum(Address.bounces)))
assert avg == 11
def test_count_by(self):
- User, Address = self.classes.User, self.classes.Address
-
"""Query.count_by(*args, **params)
num = session.query(Address).count_by(purpose='Personal')
num = session.query(User).count_by(purpose='Personal')
"""
+
+ User, Address = self.classes.User, self.classes.Address
+
session = create_session()
num = session.query(Address).filter_by(purpose='Personal').count()
assert num == 3, num
def test_count_whereclause(self):
- Address = self.classes.Address
-
"""Query.count(whereclause=None, params=None, **kwargs)
num = session.query(Address).count(address_table.c.bounces > 1)
"""
+
+ Address = self.classes.Address
+
session = create_session()
num = session.query(Address).filter(Address.bounces > 1).count()
assert num == 1, num
def test_execute(self):
- User, users_table = self.classes.User, self.tables.users_table
-
"""Query.execute(clauseelement, params=None, *args, **kwargs)
users = session.query(User).execute(users_table.select())
"""
+
+ User, users_table = self.classes.User, self.tables.users_table
+
session = create_session()
users = session.query(User).from_statement(users_table.select()).all()
assert len(users) == 4
def test_get_by(self):
- User, Address = self.classes.User, self.classes.Address
-
"""Query.get_by(*args, **params)
user = session.query(User).get_by(name='ed')
user = session.query(User).get_by(email_addresss='fred@the.fred')
"""
+
+ User, Address = self.classes.User, self.classes.Address
+
session = create_session()
user = session.query(User).filter_by(name='ed').first()
assert user.name == 'fred'
def test_instances_entities(self):
- addresses_table, User, users_table, Address = (self.tables.addresses_table,
- self.classes.User,
- self.tables.users_table,
- self.classes.Address)
-
"""Query.instances(cursor, *mappers_or_columns, **kwargs)
sel = users_table.join(addresses_table).select(use_labels=True)
res = session.query(User).instances(sel.execute(), Address)
"""
+
+ addresses_table, User, users_table, Address = (self.tables.addresses_table,
+ self.classes.User,
+ self.tables.users_table,
+ self.classes.Address)
+
session = create_session()
sel = users_table.join(addresses_table).select(use_labels=True)
TODO
"""
+
session = create_session()
TODO
"""
+
session = create_session()
TODO
"""
+
session = create_session()
def test_list(self):
- User = self.classes.User
-
"""Query.list()
users = session.query(User).list()
"""
+
+ User = self.classes.User
+
session = create_session()
users = session.query(User).all()
assert len(users) == 4
def test_scalar(self):
- User = self.classes.User
-
"""Query.scalar()
user = session.query(User).filter(User.id==1).scalar()
"""
+
+ User = self.classes.User
+
session = create_session()
user = session.query(User).filter(User.id==1).first()
assert user.id==1
def test_select(self):
- User = self.classes.User
-
"""Query.select(arg=None, **kwargs)
users = session.query(User).select(users_table.c.name != None)
"""
+
+ User = self.classes.User
+
session = create_session()
users = session.query(User).filter(User.name != None).all()
assert len(users) == 4
def test_select_by(self):
- User, Address = self.classes.User, self.classes.Address
-
"""Query.select_by(*args, **params)
users = session.query(User).select_by(name='fred')
users = session.query(User).select_by(email_address='fred@the.fred')
"""
+
+ User, Address = self.classes.User, self.classes.Address
+
session = create_session()
users = session.query(User).filter_by(name='fred').all()
assert len(users) == 1
def test_selectfirst(self):
- Address = self.classes.Address
-
"""Query.selectfirst(arg=None, **kwargs)
bounced = session.query(Address).selectfirst(
addresses_table.c.bounces > 0)
"""
+
+ Address = self.classes.Address
+
session = create_session()
bounced = session.query(Address).filter(Address.bounces > 0).first()
assert bounced.bounces > 0
def test_selectfirst_by(self):
- User, Address = self.classes.User, self.classes.Address
-
"""Query.selectfirst_by(*args, **params)
onebounce = session.query(Address).selectfirst_by(bounces=1)
onebounce_user = session.query(User).selectfirst_by(bounces=1)
"""
+
+ User, Address = self.classes.User, self.classes.Address
+
session = create_session()
onebounce = session.query(Address).filter_by(bounces=1).first()
assert onebounce_user.name == 'jack'
def test_selectone(self):
- User = self.classes.User
-
"""Query.selectone(arg=None, **kwargs)
ed = session.query(User).selectone(users_table.c.name == 'ed')
"""
+
+ User = self.classes.User
+
session = create_session()
ed = session.query(User).filter(User.name == 'jack').one()
def test_selectone_by(self):
- User, Address = self.classes.User, self.classes.Address
-
"""Query.selectone_by
ed = session.query(User).selectone_by(name='ed')
ed = session.query(User).selectone_by(email_address='ed@foo.bar')
"""
+
+ User, Address = self.classes.User, self.classes.Address
+
session = create_session()
ed = session.query(User).filter_by(name='jack').one()
Address.email_address == 'ed@foo.bar')).one()
def test_select_statement(self):
- User, users_table = self.classes.User, self.tables.users_table
-
"""Query.select_statement(statement, **params)
users = session.query(User).select_statement(users_table.select())
"""
+
+ User, users_table = self.classes.User, self.tables.users_table
+
session = create_session()
users = session.query(User).from_statement(users_table.select()).all()
assert len(users) == 4
def test_select_text(self):
- User = self.classes.User
-
"""Query.select_text(text, **params)
users = session.query(User).select_text('SELECT * FROM users_table')
"""
+
+ User = self.classes.User
+
session = create_session()
users = (session.query(User).
assert len(users) == 4
def test_select_whereclause(self):
- User = self.classes.User
-
"""Query.select_whereclause(whereclause=None, params=None, **kwargs)
users = session.query(User).select_whereclause("name='ed'")
"""
+
+ User = self.classes.User
+
session = create_session()
users = session.query(User).filter(User.name=='ed').all()
eq_(self.static.user_address_result, q.all())
def test_statement(self):
+ """test that the .statement accessor returns the actual statement that
+ would render, without any _clones called."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test that the .statement accessor returns the actual statement that
- would render, without any _clones called."""
mapper(User, users, properties={
'addresses':dynamic_loader(mapper(Address, addresses))
def test_no_orphan(self):
+ """An eagerly loaded child object is not marked as an orphan"""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """An eagerly loaded child object is not marked as an orphan"""
mapper(User, users, properties={
'addresses':relationship(Address, cascade="all,delete-orphan", lazy='joined')
], q.order_by(User.id).all())
def test_orderby_related(self):
+ """A regular mapper select on a single table can
+ order by a relationship to a second table"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A regular mapper select on a single table can
- order by a relationship to a second table"""
mapper(Address, addresses)
mapper(User, users, properties = dict(
self.assert_sql_count(testing.db, go, count)
def test_disable_dynamic(self):
+ """test no joined option on a dynamic."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test no joined option on a dynamic."""
mapper(User, users, properties={
'addresses':relationship(Address, lazy="dynamic")
def test_cyclical(self):
+ """A circular eager relationship breaks the cycle with a lazy loader"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A circular eager relationship breaks the cycle with a lazy loader"""
mapper(Address, addresses)
mapper(User, users, properties = dict(
eq_(self.static.user_address_result, sess.query(User).order_by(User.id).all())
def test_double(self):
+ """Eager loading with two relationships simultaneously,
+ from the same table, using aliases."""
+
users, orders, User, Address, Order, addresses = (self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Order,
self.tables.addresses)
- """Eager loading with two relationships simultaneously,
- from the same table, using aliases."""
openorders = sa.alias(orders, 'openorders')
closedorders = sa.alias(orders, 'closedorders')
self.assert_sql_count(testing.db, go, 1)
def test_double_same_mappers(self):
+ """Eager loading with two relationships simulatneously,
+ from the same table, using aliases."""
+
addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.users)
- """Eager loading with two relationships simulatneously,
- from the same table, using aliases."""
mapper(Address, addresses)
mapper(Order, orders, properties={
self.assert_sql_count(testing.db, go, 1)
def test_no_false_hits(self):
+ """Eager loaders don't interpret main table columns as
+ part of their eager load."""
+
addresses, orders, User, Address, Order, users = (self.tables.addresses,
self.tables.orders,
self.classes.User,
self.classes.Order,
self.tables.users)
- """Eager loaders don't interpret main table columns as
- part of their eager load."""
mapper(User, users, properties={
'addresses':relationship(Address, lazy='joined'),
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_limit(self):
+ """Limit operations combined with lazy-load relationships."""
+
users, items, order_items, orders, Item, User, Address, Order, addresses = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.addresses)
- """Limit operations combined with lazy-load relationships."""
mapper(Item, items)
mapper(Order, orders, properties={
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_limit_3(self):
+ """test that the ORDER BY is propagated 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."""
+
addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.users)
- """test that the ORDER BY is propagated 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(
)
def test_useget_cancels_eager(self):
+ """test that a one to many lazyload cancels the unnecessary
+ eager many-to-one join on the other side."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test that a one to many lazyload cancels the unnecessary
- eager many-to-one join on the other side."""
mapper(User, users)
mapper(Address, addresses, properties={
def test_manytoone_limit(self):
+ """test that the subquery wrapping only occurs with
+ limit/offset and m2m or o2m joins present."""
+
users, items, order_items, Order, Item, User, Address, orders, addresses = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.tables.orders,
self.tables.addresses)
- """test that the subquery wrapping only occurs with
- limit/offset and m2m or o2m joins present."""
mapper(User, users, properties=odict(
orders=relationship(Order, backref='user')
self.assert_sql_count(testing.db, go, 1)
def test_many_to_one_null(self):
+ """test that a many-to-one eager load which loads None does
+ not later trigger a lazy load.
+
+ """
+
Order, Address, addresses, orders = (self.classes.Order,
self.classes.Address,
self.tables.addresses,
self.tables.orders)
- """test that a many-to-one eager load which loads None does
- not later trigger a lazy load.
-
- """
# use a primaryjoin intended to defeat SA's usage of
# query.get() for a many-to-one lazyload
self.assert_sql_count(testing.db, go, 1)
def test_one_and_many(self):
+ """tests eager load for a parent object with a child object that
+ contains a many-to-many relationship to a third object."""
+
users, items, order_items, orders, Item, User, Order = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.classes.User,
self.classes.Order)
- """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':relationship(Order, lazy='joined', order_by=orders.c.id)
self.assert_sql_count(testing.db, go, 1)
def test_uselist_false_warning(self):
+ """test that multiple rows received by a
+ uselist=False raises a warning."""
+
User, users, orders, Order = (self.classes.User,
self.tables.users,
self.tables.orders,
self.classes.Order)
- """test that multiple rows received by a
- uselist=False raises a warning."""
mapper(User, users, properties={
'order':relationship(Order, uselist=False)
eq_(self.static.user_all_result, q.order_by(User.id).all())
def test_against_select(self):
+ """test eager loading of a mapper which is against a select"""
+
users, items, order_items, orders, Item, User, Order = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.classes.User,
self.classes.Order)
- """test eager loading of a mapper which is against a select"""
s = sa.select([orders], orders.c.isopen==1).alias('openorders')
], q.all())
def test_aliasing(self):
+ """test that eager loading uses aliases to insulate the eager
+ load from regular criterion against those tables."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """test that eager loading uses aliases to insulate the eager
- load from regular criterion against those tables."""
mapper(User, users, properties = dict(
addresses = relationship(mapper(Address, addresses),
)
def test_label_anonymizing(self):
- tags_table, users_table = self.tables.tags_table, self.tables.users_table
-
"""Eager loading works with subqueries with labels,
Even if an explicit labelname which conflicts with a label on the
that type.
"""
+
+ tags_table, users_table = self.tables.tags_table, self.tables.users_table
+
class User(fixtures.ComparableEntity):
@property
def prop_score(self):
run_inserts = None
def test_instance_event_listen(self):
+ """test listen targets for instance events"""
+
users, addresses = self.tables.users, self.tables.addresses
- """test listen targets for instance events"""
canary = []
class A(object):
'after_delete'])
def test_before_after_only_collection(self):
+ """before_update is called on parent for collection modifications,
+ after_update is called even if no columns were updated.
+
+ """
+
keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
self.tables.items,
self.tables.item_keywords,
self.classes.Keyword,
self.classes.Item)
- """before_update is called on parent for collection modifications,
- after_update is called even if no columns were updated.
-
- """
mapper(Item, items, properties={
'keywords': relationship(Keyword, secondary=item_keywords)})
return Ext, methods
def test_basic(self):
+ """test that common user-defined methods get called."""
+
User, users = self.classes.User, self.tables.users
- """test that common user-defined methods get called."""
Ext, methods = self.extension()
mapper(User, users, extension=Ext())
'after_delete'])
def test_before_after_only_collection(self):
+ """before_update is called on parent for collection modifications,
+ after_update is called even if no columns were updated.
+
+ """
+
keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
self.tables.items,
self.tables.item_keywords,
self.classes.Keyword,
self.classes.Item)
- """before_update is called on parent for collection modifications,
- after_update is called even if no columns were updated.
-
- """
Ext1, methods1 = self.extension()
Ext2, methods2 = self.extension()
def test_inheritance_with_dupes(self):
+ """Inheritance with the same extension instance on both mappers."""
+
users, addresses, User = (self.tables.users,
self.tables.addresses,
self.classes.User)
- """Inheritance with the same extension instance on both mappers."""
Ext, methods = self.extension()
class AdminUser(User):
eq_(u.name, 'chuck')
def test_deferred(self):
+ """test that unloaded, deferred attributes aren't included in the expiry list."""
+
Order, orders = self.classes.Order, self.tables.orders
- """test that unloaded, deferred attributes aren't included in the expiry list."""
mapper(Order, orders, properties={'description':deferred(orders.c.description)})
])
def test_refresh_collection_exception(self):
+ """test graceful failure for currently unsupported
+ immediate refresh of a collection"""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test graceful failure for currently unsupported
- immediate refresh of a collection"""
mapper(User, users, properties={
'addresses':relationship(Address, order_by=addresses.c.email_address)
def test_expire_preserves_changes(self):
+ """test that the expire load operation doesn't revert post-expire changes"""
+
Order, orders = self.classes.Order, self.tables.orders
- """test that the expire load operation doesn't revert post-expire changes"""
mapper(Order, orders)
sess = create_session()
assert o.description == 'another new description'
def test_expire_committed(self):
+ """test that the committed state of the attribute receives the most recent DB data"""
+
orders, Order = self.tables.orders, self.classes.Order
- """test that the committed state of the attribute receives the most recent DB data"""
mapper(Order, orders)
sess = create_session()
assert len(list(sess)) == 9
def test_state_change_col_to_deferred(self):
+ """Behavioral test to verify the current activity of loader callables."""
+
users, User = self.tables.users, self.classes.User
- """Behavioral test to verify the current activity of loader callables."""
mapper(User, users)
assert 'name' in attributes.instance_state(u1).callables
def test_state_deferred_to_col(self):
+ """Behavioral test to verify the current activity of loader callables."""
+
users, User = self.tables.users, self.classes.User
- """Behavioral test to verify the current activity of loader callables."""
mapper(User, users, properties={'name':deferred(users.c.name)})
)
def test_state_noload_to_lazy(self):
+ """Behavioral test to verify the current activity of loader callables."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """Behavioral test to verify the current activity of loader callables."""
mapper(User, users, properties={'addresses':relationship(Address, lazy='noload')})
mapper(Address, addresses)
assert u.name == 'jack'
def test_refresh_with_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"""
+
User, Address, addresses, users = (self.classes.User,
self.classes.Address,
self.tables.addresses,
self.tables.users)
- """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()
mapper(User, users, properties={'addresses':relationship(mapper(Address, addresses))})
self.assert_sql_count(testing.db, go, 1)
def test_refresh_with_eager(self):
+ """test that a refresh/expire operation loads rows properly and sends correct "isnew" state to eager loaders"""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test that a refresh/expire operation loads rows properly and sends correct "isnew" state to eager loaders"""
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses), lazy='joined')
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_refresh2(self):
+ """test a hang condition that was occurring on expire/refresh"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """test a hang condition that was occurring on expire/refresh"""
s = create_session()
mapper(Address, addresses)
)
def test_no_joinedload(self):
+ """test that joinedloads are pushed outwards and not rendered in subqueries."""
+
User = self.classes.User
- """test that joinedloads are pushed outwards and not rendered in subqueries."""
s = create_session()
)
def test_aliases(self):
+ """test that aliased objects are accessible externally to a from_self() call."""
+
User, Address = self.classes.User, self.classes.Address
- """test that aliased objects are accessible externally to a from_self() call."""
s = create_session()
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_chaining(self):
+ """test that contains_eager() 'chains' by default."""
+
Dingaling, User, Address = (self.classes.Dingaling,
self.classes.User,
self.classes.Address)
- """test that contains_eager() 'chains' by default."""
sess = create_session()
q = sess.query(User).\
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_chaining_aliased_endpoint(self):
+ """test that contains_eager() 'chains' by default and supports
+ an alias at the end."""
+
Dingaling, User, Address = (self.classes.Dingaling,
self.classes.User,
self.classes.Address)
- """test that contains_eager() 'chains' by default and supports
- an alias at the end."""
sess = create_session()
da = aliased(Dingaling, name="foob")
"zxjdbc parses the SQL itself before passing on "
"to PG, doesn't parse this")
def test_values_with_boolean_selects(self):
- User = self.classes.User
-
"""Tests a values clause that works with select boolean
evaluations"""
+
+ User = self.classes.User
+
sess = create_session()
q = sess.query(User)
def test_correlated_subquery(self):
- Address, users, User = (self.classes.Address,
- self.tables.users,
- self.classes.User)
-
"""test that a subquery constructed from ORM attributes doesn't leak out
those entities to the outermost query.
"""
+
+ Address, users, User = (self.classes.Address,
+ self.tables.users,
+ self.classes.User)
+
sess = create_session()
subq = select([func.count()]).\
assert_raises(sa_exc.InvalidRequestError, sess.query(User).add_column, object())
def test_add_multi_columns(self):
+ """test that add_column accepts a FROM clause."""
+
users, User = self.tables.users, self.classes.User
- """test that add_column accepts a FROM clause."""
sess = create_session()
)
def test_multi_columns_2(self):
+ """test aliased/nonalised joins with the usage of add_column()"""
+
User, Address, addresses, users = (self.classes.User,
self.classes.Address,
self.tables.addresses,
self.tables.users)
- """test aliased/nonalised joins with the usage of add_column()"""
sess = create_session()
(user7, user8, user9, user10) = sess.query(User).all()
)
def test_join_mapper_order_by(self):
+ """test that mapper-level order_by is adapted to a selectable."""
+
User, users = self.classes.User, self.tables.users
- """test that mapper-level order_by is adapted to a selectable."""
mapper(User, users, order_by=users.c.id)
)
def test_differentiate_self_external(self):
+ """test some different combinations of joining a table to a subquery of itself."""
+
users, User = self.tables.users, self.classes.User
- """test some different combinations of joining a table to a subquery of itself."""
mapper(User, users)
run_setup_mappers = None
def test_double_same_mappers(self):
+ """test aliasing of joins with a custom join condition"""
+
addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.users)
- """test aliasing of joins with a custom join condition"""
mapper(Address, addresses)
mapper(Order, orders, properties={
'items':relationship(Item, secondary=order_items, lazy='select', order_by=items.c.id),
clear_mappers()
def test_external_columns(self):
+ """test querying mappings that reference external columns or selectables."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test querying mappings that reference external columns or selectables."""
mapper(User, users, properties={
'concat': column_property((users.c.id * 2)),
def test_join(self):
+ """Query.join"""
+
User, Address = self.classes.User, self.classes.Address
- """Query.join"""
session = create_session()
q = (session.query(User).join('orders', 'addresses').
eq_([User(id=7)], q.all())
def test_outer_join(self):
+ """Query.outerjoin"""
+
Order, User, Address = (self.classes.Order,
self.classes.User,
self.classes.Address)
- """Query.outerjoin"""
session = create_session()
q = (session.query(User).outerjoin('orders', 'addresses').
set(q.all()))
def test_outer_join_count(self):
+ """test the join and outerjoin functions on Query"""
+
Order, User, Address = (self.classes.Order,
self.classes.User,
self.classes.Address)
- """test the join and outerjoin functions on Query"""
session = create_session()
)
def test_multiple_adaption(self):
+ """test that multiple filter() adapters get chained together "
+ and work correctly within a multiple-entry join()."""
+
people, Company, Machine, engineers, machines, Engineer = (self.tables.people,
self.classes.Company,
self.classes.Machine,
self.tables.machines,
self.classes.Engineer)
- """test that multiple filter() adapters get chained together "
- and work correctly within a multiple-entry join()."""
sess = create_session()
)
def test_multi_tuple_form(self):
- Item, Order, User = (self.classes.Item,
- self.classes.Order,
- self.classes.User)
-
"""test the 'tuple' form of join, now superseded by the two-element join() form.
Not deprecating this style as of yet.
"""
+ Item, Order, User = (self.classes.Item,
+ self.classes.Order,
+ self.classes.User)
+
+
sess = create_session()
#assert_raises(
)
def test_join_nonmapped_column(self):
+ """test that the search for a 'left' doesn't trip on non-mapped cols"""
+
Order, User = self.classes.Order, self.classes.User
- """test that the search for a 'left' doesn't trip on non-mapped cols"""
sess = create_session()
# intentionally join() with a non-existent "left" side
)
def test_replace_multiple_from_clause(self):
+ """test adding joins onto multiple FROM clauses"""
+
User, Order, Address = (self.classes.User,
self.classes.Order,
self.classes.Address)
- """test adding joins onto multiple FROM clauses"""
sess = create_session()
assert [User(id=7, name='jack')] == result
def test_aliased(self):
+ """test automatic generation of aliased joins."""
+
Item, Order, User, Address = (self.classes.Item,
self.classes.Order,
self.classes.User,
self.classes.Address)
- """test automatic generation of aliased joins."""
sess = create_session()
)
def test_select_from(self):
+ """Test that the left edge of the join can be set reliably with select_from()."""
+
Item, Order, User = (self.classes.Item,
self.classes.Order,
self.classes.User)
- """Test that the left edge of the join can be set reliably with select_from()."""
sess = create_session()
self.assert_compile(
def test_from_self_resets_joinpaths(self):
- Item, Keyword = self.classes.Item, self.classes.Keyword
-
"""test a join from from_self() doesn't confuse joins inside the subquery
with the outside.
"""
+
+ Item, Keyword = self.classes.Item, self.classes.Keyword
+
sess = create_session()
self.assert_compile(
assert node.data == 'n122'
def test_string_or_prop_aliased(self):
- Node = self.classes.Node
-
"""test that join('foo') behaves the same as join(Cls.foo) in a self
referential scenario.
"""
+ Node = self.classes.Node
+
+
sess = create_session()
nalias = aliased(Node, sess.query(Node).filter_by(data='n1').subquery())
)
def test_from_self_inside_excludes_outside(self):
- Node = self.classes.Node
-
"""test the propagation of aliased() from inside to outside
on a from_self()..
"""
+
+ Node = self.classes.Node
+
sess = create_session()
n1 = aliased(Node)
assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(users.c.id == 7).all()
def test_needs_parent(self):
+ """test the error raised when parent object is not bound."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test the error raised when parent object is not bound."""
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses), lazy='select')
] == q.all()
def test_orderby_secondary(self):
+ """tests that a regular mapper select on a single table can order by a relationship to a second table"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """tests that a regular mapper select on a single table can order by a relationship to a second table"""
mapper(Address, addresses)
] == sess.query(User).all()
def test_no_orphan(self):
+ """test that a lazily loaded child object is not marked as an orphan"""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test that a lazily loaded child object is not marked as an orphan"""
mapper(User, users, properties={
'addresses':relationship(Address, cascade="all,delete-orphan", lazy='select')
assert not sa.orm.class_mapper(Address)._is_orphan(attributes.instance_state(user.addresses[0]))
def test_limit(self):
+ """test limit operations combined with lazy-load relationships."""
+
users, items, order_items, orders, Item, User, Address, Order, addresses = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.addresses)
- """test limit operations combined with lazy-load relationships."""
mapper(Item, items)
mapper(Order, orders, properties={
eq_(self.static.user_all_result, l)
def test_uselist_false_warning(self):
+ """test that multiple rows received by a uselist=False raises a warning."""
+
User, users, orders, Order = (self.classes.User,
self.tables.users,
self.tables.orders,
self.classes.Order)
- """test that multiple rows received by a uselist=False raises a warning."""
mapper(User, users, properties={
'order':relationship(Order, uselist=False)
def test_double(self):
+ """tests lazy loading with two relationships simulatneously, from the same table, using aliases. """
+
users, orders, User, Address, Order, addresses = (self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Order,
self.tables.addresses)
- """tests lazy loading with two relationships simulatneously, from the same table, using aliases. """
openorders = sa.alias(orders, 'openorders')
closedorders = sa.alias(orders, 'closedorders')
assert self.static.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all()
def test_uses_get(self):
+ """test that a simple many-to-one lazyload optimizes to use query.get()."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """test that a simple many-to-one lazyload optimizes to use query.get()."""
for pj in (
None,
sa.orm.clear_mappers()
def test_uses_get_compatible_types(self):
+ """test the use_get optimization with compatible but non-identical types"""
+
User, Address = self.classes.User, self.classes.Address
- """test the use_get optimization with compatible but non-identical types"""
class IntDecorator(TypeDecorator):
impl = Integer
run_deletes = None
def test_m2o_noload(self):
+ """test that a NULL foreign key doesn't trigger a lazy load"""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test that a NULL foreign key doesn't trigger a lazy load"""
mapper(User, users)
mapper(Address, addresses, properties={
assert session.query(Foo).one().data == pickleable.Bar(4, 19)
def test_mutated_plus_scalar_state_change_resurrected(self):
- Foo = self.classes.Foo
-
"""test that a non-mutable attribute event subsequent to
a mutable event prevents the object from falling into
resurrected state.
"""
+
+ Foo = self.classes.Foo
+
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
session = Session()
session.add(f1)
assert not attributes.instance_state(f1).modified
def test_scalar_no_net_change_no_update(self):
- Foo = self.classes.Foo
-
"""Test that a no-net-change on a scalar attribute event
doesn't cause an UPDATE for a mutable state.
"""
+ Foo = self.classes.Foo
+
+
f1 = Foo(val=u'hi')
session = Session()
self.sql_count_(0, session.commit)
def test_expire_attribute_set(self):
- Foo = self.classes.Foo
-
"""test no SELECT emitted when assigning to an expired
mutable attribute.
"""
+ Foo = self.classes.Foo
+
+
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
session = Session()
session.add(f1)
eq_(f1.data.x, 10)
def test_expire_mutate(self):
- Foo = self.classes.Foo
-
"""test mutations are detected on an expired mutable
attribute."""
+ Foo = self.classes.Foo
+
+
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
session = Session()
session.add(f1)
eq_(f1.data.x, 10)
def test_deferred_attribute_set(self):
- mutable_t, Foo = self.tables.mutable_t, self.classes.Foo
-
"""test no SELECT emitted when assigning to a deferred
mutable attribute.
"""
+
+ mutable_t, Foo = self.tables.mutable_t, self.classes.Foo
+
sa.orm.clear_mappers()
mapper(Foo, mutable_t, properties={
'data':sa.orm.deferred(mutable_t.c.data)
eq_(f1.data.x, 10)
def test_deferred_mutate(self):
- mutable_t, Foo = self.tables.mutable_t, self.classes.Foo
-
"""test mutations are detected on a deferred mutable
attribute."""
+ mutable_t, Foo = self.tables.mutable_t, self.classes.Foo
+
+
sa.orm.clear_mappers()
mapper(Foo, mutable_t, properties={
'data':sa.orm.deferred(mutable_t.c.data)
mapper(Foo, mutable_t)
def test_dicts(self):
+ """Dictionaries may not pickle the same way twice."""
+
Foo = self.classes.Foo
- """Dictionaries may not pickle the same way twice."""
f1 = Foo()
f1.data = [ {
using the "current" state in all cases except during flush.
"""
+
for loadfk in (True, False):
for loadrel in (True, False):
for autoflush in (True, False):
self.assert_(len(s.courses) == 2)
def test_dupliates_raise(self):
+ """test constraint error is raised for dupe entries in a list"""
+
course, enroll, Student, student, Course = (self.tables.course,
self.tables.enroll,
self.classes.Student,
self.tables.student,
self.classes.Course)
- """test constraint error is raised for dupe entries in a list"""
mapper(Student, student)
mapper(Course, course, properties={
assert_raises(sa.exc.DBAPIError, sess.flush)
def test_delete(self):
+ """A many-to-many table gets cleared out with deletion from the backref side"""
+
course, enroll, Student, student, Course = (self.tables.course,
self.tables.enroll,
self.classes.Student,
self.tables.student,
self.classes.Course)
- """A many-to-many table gets cleared out with deletion from the backref side"""
mapper(Student, student)
mapper(Course, course, properties = {
class MapperTest(_fixtures.FixtureTest):
def test_prop_shadow(self):
+ """A backref name may not shadow an existing property name."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A backref name may not shadow an existing property name."""
mapper(Address, addresses)
mapper(User, users,
assert_raises(sa.exc.ArgumentError, sa.orm.configure_mappers)
def test_update_attr_keys(self):
+ """test that update()/insert() use the correct key when given InstrumentedAttributes."""
+
User, users = self.classes.User, self.tables.users
- """test that update()/insert() use the correct key when given InstrumentedAttributes."""
mapper(User, users, properties={
'foobar':users.c.name
eq_(str(User.addresses), "User.addresses")
def test_exceptions_sticky(self):
- Address, addresses, User = (self.classes.Address,
- self.tables.addresses,
- self.classes.User)
-
"""test preservation of mapper compile errors raised during hasattr(),
as well as for redundant mapper compile calls. Test that
repeated calls don't stack up error messages.
"""
+ Address, addresses, User = (self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
+
mapper(Address, addresses, properties={
'user':relationship(User)
})
assert_raises(sa.exc.ArgumentError, mapper, User, s)
def test_reconfigure_on_other_mapper(self):
+ """A configure trigger on an already-configured mapper
+ still triggers a check against all mappers."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """A configure trigger on an already-configured mapper
- still triggers a check against all mappers."""
mapper(User, users)
sa.orm.configure_mappers()
assert sa.orm.mapperlib._new_mappers is False
session.connection(m)
def test_incomplete_columns(self):
+ """Loading from a select which does not contain all columns"""
+
addresses, Address = self.tables.addresses, self.classes.Address
- """Loading from a select which does not contain all columns"""
mapper(Address, addresses)
s = create_session()
a = s.query(Address).from_statement(
: addresses.c.user_id})
def test_constructor_exc(self):
- users, addresses = self.tables.users, self.tables.addresses
-
"""TypeError is raised for illegal constructor args,
whether or not explicit __init__ is present [ticket:908]."""
+ users, addresses = self.tables.users, self.tables.addresses
+
+
class Foo(object):
def __init__(self):
pass
def test_mapping_to_join_raises(self):
+ """Test implicit merging of two cols raises."""
+
addresses, users, User = (self.tables.addresses,
self.tables.users,
self.classes.User)
- """Test implicit merging of two cols raises."""
usersaddresses = sa.join(users, addresses,
users.c.id == addresses.c.user_id)
)
def test_mapping_to_join_explicit_prop(self):
+ """Mapping to a join"""
+
User, addresses, users = (self.classes.User,
self.tables.addresses,
self.tables.users)
- """Mapping to a join"""
usersaddresses = sa.join(users, addresses, users.c.id
== addresses.c.user_id)
eq_(l, self.static.user_result[:3])
def test_mapping_to_join_exclude_prop(self):
+ """Mapping to a join"""
+
User, addresses, users = (self.classes.User,
self.tables.addresses,
self.tables.users)
- """Mapping to a join"""
usersaddresses = sa.join(users, addresses, users.c.id
== addresses.c.user_id)
eq_(email_bounces.count().scalar(), 5)
def test_mapping_to_outerjoin(self):
+ """Mapping to an outer join with a nullable composite primary key."""
+
users, addresses, User = (self.tables.users,
self.tables.addresses,
self.classes.User)
- """Mapping to an outer join with a nullable composite primary key."""
mapper(User, users.outerjoin(addresses),
User(id=10, address_id=None)])
def test_mapping_to_outerjoin_no_partial_pks(self):
+ """test the allow_partial_pks=False flag."""
+
users, addresses, User = (self.tables.users,
self.tables.addresses,
self.classes.User)
- """test the allow_partial_pks=False flag."""
mapper(User, users.outerjoin(addresses),
def test_custom_join(self):
+ """select_from totally replace the FROM parameters."""
+
users, items, order_items, orders, Item, User, Order = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.classes.User,
self.classes.Order)
- """select_from totally replace the FROM parameters."""
mapper(Item, items)
# 'Raises a "expression evaluation not supported" error at prepare time
@testing.fails_on('firebird', 'FIXME: unknown')
def test_function(self):
+ """Mapping to a SELECT statement that has functions in it."""
+
addresses, users, User = (self.tables.addresses,
self.tables.users,
self.classes.User)
- """Mapping to a SELECT statement that has functions in it."""
s = sa.select([users,
(users.c.id * 2).label('concat'),
eq_(l[idx].concat, total)
def test_count(self):
+ """The count function on Query."""
+
User, users = self.classes.User, self.tables.users
- """The count function on Query."""
mapper(User, users)
eq_(q.count(), 2)
def test_override_1(self):
+ """Overriding a column raises an error."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """Overriding a column raises an error."""
def go():
mapper(User, users,
properties=dict(
assert_raises(sa.exc.ArgumentError, go)
def test_override_2(self):
+ """exclude_properties cancels the error."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """exclude_properties cancels the error."""
mapper(User, users,
exclude_properties=['name'],
assert bool(User.name)
def test_override_3(self):
+ """The column being named elsewhere also cancels the error,"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """The column being named elsewhere also cancels the error,"""
mapper(User, users,
properties=dict(
name=relationship(mapper(Address, addresses)),
go)
def test_column_synonyms(self):
+ """Synonyms which automatically instrument properties, set up aliased column, etc."""
+
addresses, users, Address = (self.tables.addresses,
self.tables.users,
self.classes.Address)
- """Synonyms which automatically instrument properties, set up aliased column, etc."""
assert_col = []
self.assert_sql_count(testing.db, go, 1)
def test_eager_options(self):
+ """A lazy relationship can be upgraded to an eager relationship."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A lazy relationship can be upgraded to an eager relationship."""
mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses),
order_by=addresses.c.id)))
self.sql_count_(1, go)
def test_eager_degrade(self):
+ """An eager relationship automatically degrades to a lazy relationship
+ if eager columns are not available"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """An eager relationship automatically degrades to a lazy relationship
- if eager columns are not available"""
mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses),
lazy='joined', order_by=addresses.c.id)))
self.assert_sql_count(testing.db, go, 6)
def test_lazy_options(self):
+ """An eager relationship can be upgraded to a lazy relationship."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """An eager relationship can be upgraded to a lazy relationship."""
mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses), lazy='joined')
))
self.assert_sql_count(testing.db, go, 3)
def test_deep_options_2(self):
+ """test (joined|subquery)load_all() options"""
+
User = self.classes.User
- """test (joined|subquery)load_all() options"""
sess = create_session()
class DeferredTest(_fixtures.FixtureTest):
def test_basic(self):
+ """A basic deferred load."""
+
Order, orders = self.classes.Order, self.tables.orders
- """A basic deferred load."""
mapper(Order, orders, order_by=orders.c.id, properties={
'description': deferred(orders.c.description)})
{'param_1':3})])
def test_unsaved(self):
+ """Deferred loading does not kick in when just PK cols are set."""
+
Order, orders = self.classes.Order, self.tables.orders
- """Deferred loading does not kick in when just PK cols are set."""
mapper(Order, orders, properties={
'description': deferred(orders.c.description)})
self.sql_count_(0, go)
def test_unsaved_group(self):
+ """Deferred loading doesnt kick in when just PK cols are set"""
+
orders, Order = self.tables.orders, self.classes.Order
- """Deferred loading doesnt kick in when just PK cols are set"""
mapper(Order, orders, order_by=orders.c.id, properties=dict(
description=deferred(orders.c.description, group='primary'),
sess.flush()
def test_group(self):
+ """Deferred load with a group"""
+
orders, Order = self.tables.orders, self.classes.Order
- """Deferred load with a group"""
mapper(Order, orders, properties=util.OrderedDict([
('userident', deferred(orders.c.user_id, group='primary')),
('addrident', deferred(orders.c.address_id, group='primary')),
self.sql_count_(0, go)
def test_preserve_changes(self):
+ """A deferred load operation doesn't revert modifications on attributes"""
+
orders, Order = self.tables.orders, self.classes.Order
- """A deferred load operation doesn't revert modifications on attributes"""
mapper(Order, orders, properties = {
'userident': deferred(orders.c.user_id, group='primary'),
'description': deferred(orders.c.description, group='primary'),
assert o in sess.dirty
def test_commits_state(self):
- orders, Order = self.tables.orders, self.classes.Order
-
"""
When deferred elements are loaded via a group, they get the proper
CommittedState and don't result in changes being committed
"""
+
+ orders, Order = self.tables.orders, self.classes.Order
+
mapper(Order, orders, properties = {
'userident':deferred(orders.c.user_id, group='primary'),
'description':deferred(orders.c.description, group='primary'),
self.assert_sql_count(testing.db, sess.flush, 0)
def test_options(self):
+ """Options on a mapper to create deferred and undeferred columns"""
+
orders, Order = self.tables.orders, self.classes.Order
- """Options on a mapper to create deferred and undeferred columns"""
mapper(Order, orders)
{})])
def test_locates_col(self):
+ """Manually adding a column to the result undefers the column."""
+
orders, Order = self.tables.orders, self.classes.Order
- """Manually adding a column to the result undefers the column."""
mapper(Order, orders, properties={
'description':deferred(orders.c.description)})
self.sql_count_(0, go)
def test_map_selectable_wo_deferred(self):
- Order, orders = self.classes.Order, self.tables.orders
-
"""test mapping to a selectable with deferred cols,
the selectable doesn't include the deferred col.
"""
+ Order, orders = self.classes.Order, self.tables.orders
+
+
order_select = sa.select([
orders.c.id,
orders.c.user_id,
run_deletes = None
def test_basic(self):
+ """A basic one-to-many lazy load"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A basic one-to-many lazy load"""
m = mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses), lazy='noload')
))
# end Py2K
def test_comparison_overrides(self):
- ht6, ht5, ht4, ht3, ht2, ht1 = (self.tables.ht6,
- self.tables.ht5,
- self.tables.ht4,
- self.tables.ht3,
- self.tables.ht2,
- self.tables.ht1)
-
"""Simple tests to ensure users can supply comparison __methods__.
The suite-level test --options are better suited to detect
test run.
"""
+ ht6, ht5, ht4, ht3, ht2, ht1 = (self.tables.ht6,
+ self.tables.ht5,
+ self.tables.ht4,
+ self.tables.ht3,
+ self.tables.ht2,
+ self.tables.ht1)
+
+
# adding these methods directly to each class to avoid decoration
# by the testlib decorators.
class _Base(object):
eq_(sess.query(User).first(), User(id=7, name='fred'))
def test_transient_to_pending_no_pk(self):
+ """test that a transient object with no PK attribute doesn't trigger a needless load."""
+
User, users = self.classes.User, self.tables.users
- """test that a transient object with no PK attribute doesn't trigger a needless load."""
mapper(User, users)
sess = create_session()
u = User(name='fred')
Address(id=3, email_address='fred3')])))
def test_unsaved_cascade(self):
+ """Merge of a transient entity with two child transient entities, with a bidirectional relationship."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """Merge of a transient entity with two child transient entities, with a bidirectional relationship."""
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses),
assert u1.addresses.keys() == ['foo@bar.com']
def test_attribute_cascade(self):
+ """Merge of a persistent entity with two child persistent entities."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """Merge of a persistent entity with two child persistent entities."""
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses), backref='user')
eq_(load.called, 21)
def test_no_relationship_cascade(self):
+ """test that merge doesn't interfere with a relationship()
+ target that specifically doesn't include 'merge' cascade.
+ """
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """test that merge doesn't interfere with a relationship()
- target that specifically doesn't include 'merge' cascade.
- """
mapper(Address, addresses, properties={
'user':relationship(User, cascade="save-update")
})
go()
def test_no_load_with_backrefs(self):
+ """load=False populates relationships in both directions without requiring a load"""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """load=False populates relationships in both directions without requiring a load"""
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses), backref='user')
})
def test_dontload_with_eager(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
-
"""
This test illustrates that with load=False, we can't just copy the
'dirty'.
"""
+
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses))
})
self.assert_sql_count(testing.db, go, 0)
def test_no_load_disallows_dirty(self):
- users, User = self.tables.users, self.classes.User
-
"""load=False doesnt support 'dirty' objects right now
(see test_no_load_with_eager()). Therefore lets assert it.
"""
+
+ users, User = self.tables.users, self.classes.User
+
mapper(User, users)
sess = create_session()
u = User()
self.assert_sql_count(testing.db, go, 0)
def test_no_load_preserves_parents(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
-
"""Merge with load=False does not trigger a 'delete-orphan' operation.
merge with load=False sets attributes without using events. this means
mapper._is_orphan().)
"""
+
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(mapper(Address, addresses),
backref='user', cascade="all, delete-orphan")})
sess.merge(u)
def test_cascade_doesnt_blowaway_manytoone(self):
+ """a merge test that was fixed by [ticket:1202]"""
+
User, Address, addresses, users = (self.classes.User,
self.classes.Address,
self.tables.addresses,
self.tables.users)
- """a merge test that was fixed by [ticket:1202]"""
s = create_session(autoflush=True, autocommit=False)
mapper(User, users, properties={
sess.commit()
def test_dont_expire_pending(self):
+ """test that pending instances aren't expired during a merge."""
+
users, User = self.tables.users, self.classes.User
- """test that pending instances aren't expired during a merge."""
mapper(User, users)
u = User(id=7)
self.assert_sql_count(testing.db, go, 0)
def test_option_state(self):
- users, User = self.tables.users, self.classes.User
-
"""test that the merged takes on the MapperOption characteristics
of that which is merged.
"""
+
+ users, User = self.tables.users, self.classes.User
+
class Option(MapperOption):
propagate_to_loaders = True
run_inserts = None
def test_transient_exception(self):
+ """An object that goes from a pk value to transient/pending
+ doesn't count as a "pk" switch.
+
+ """
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """An object that goes from a pk value to transient/pending
- doesn't count as a "pk" switch.
-
- """
mapper(User, users)
mapper(Address, addresses, properties={'user':relationship(User)})
self._test_o2m_change(False)
def _test_o2m_change(self, passive_updates):
- User, Address, users, addresses = (self.classes.User,
- self.classes.Address,
- self.tables.users,
- self.tables.addresses)
-
"""Change the PK of a related entity to another.
"on update cascade" is not involved here, so the mapper has
to do the UPDATE itself.
"""
+
+ User, Address, users, addresses = (self.classes.User,
+ self.classes.Address,
+ self.tables.users,
+ self.tables.addresses)
+
mapper(User, users, properties={
'addresses':relationship(Address,
passive_updates=passive_updates)})
self._test_o2m_move(False)
def _test_o2m_move(self, passive_updates):
+ """Move the related entity to a different collection,
+ changing its PK.
+
+ """
+
User, Address, users, addresses = (self.classes.User,
self.classes.Address,
self.tables.users,
self.tables.addresses)
- """Move the related entity to a different collection,
- changing its PK.
-
- """
mapper(User, users, properties={
'addresses':relationship(Address,
passive_updates=passive_updates)})
def _test_onetomany(self, passive_updates):
- User, Address, users, addresses = (self.classes.User,
- self.classes.Address,
- self.tables.users,
- self.tables.addresses)
-
"""Change the PK of a related entity via foreign key cascade.
For databases that require "on update cascade", the mapper
it does the update.
"""
+
+ User, Address, users, addresses = (self.classes.User,
+ self.classes.Address,
+ self.tables.users,
+ self.tables.addresses)
+
mapper(User, users, properties={
'addresses':relationship(Address,
passive_updates=passive_updates)})
test_needs_fk=True
)
def test_rebuild_state(self):
- users = self.tables.users
-
"""not much of a 'test', but illustrate how to
remove instance-level state before pickling.
"""
+
+ users = self.tables.users
+
mapper(User, users)
u1 = User()
assert_raises(sa_exc.InvalidRequestError, q.get, (7, 10, 100))
def test_get_null_pk(self):
- users, addresses = self.tables.users, self.tables.addresses
-
"""test that a mapping which can have None in a
PK (i.e. map to an outerjoin) works with get()."""
+ users, addresses = self.tables.users, self.tables.addresses
+
+
s = users.outerjoin(addresses)
class UserThing(fixtures.ComparableEntity):
)
def test_no_criterion(self):
+ """test that get()/load() does not use preexisting filter/etc. criterion"""
+
User, Address = self.classes.User, self.classes.Address
- """test that get()/load() does not use preexisting filter/etc. criterion"""
s = create_session()
assert_raises(sa_exc.InvalidRequestError, q.add_column, object())
def test_distinct(self):
+ """test that a distinct() call is not valid before 'clauseelement' conditions."""
+
User = self.classes.User
- """test that a distinct() call is not valid before 'clauseelement' conditions."""
s = create_session()
q = s.query(User).distinct()
assert_raises(sa_exc.InvalidRequestError, q.with_polymorphic, User)
def test_order_by(self):
+ """test that an order_by() call is not valid before 'clauseelement' conditions."""
+
User = self.classes.User
- """test that an order_by() call is not valid before 'clauseelement' conditions."""
s = create_session()
q = s.query(User).order_by(User.id)
@testing.only_on('sqlite', 'testing execution but db-specific syntax')
def test_limit_offset_applies(self):
- User = self.classes.User
-
"""Test that the expected LIMIT/OFFSET is applied for slices.
The LIMIT/OFFSET syntax differs slightly on all databases, and
SQL strings using sqlite's syntax.
"""
+
+ User = self.classes.User
+
sess = create_session()
q = sess.query(User)
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."""
+
User, Address = self.classes.User, self.classes.Address
- """test comparing a collection to an object instance."""
sess = create_session()
address = sess.query(Address).get(3)
def test_comparison(self):
+ """test scalar comparison to an object instance"""
+
Item, Order, Dingaling, User, Address = (self.classes.Item,
self.classes.Order,
self.classes.Dingaling,
self.classes.User,
self.classes.Address)
- """test scalar comparison to an object instance"""
sess = create_session()
user = sess.query(User).get(8)
)
def test_statement_labels(self):
+ """test that label conflicts don't occur with joins etc."""
+
User, Address = self.classes.User, self.classes.Address
- """test that label conflicts don't occur with joins etc."""
s = create_session()
q1 = s.query(User, Address).join(User.addresses).\
)
def test_union_literal_expressions_compile(self):
- User = self.classes.User
-
"""test that column expressions translate during
the _from_statement() portion of union(), others"""
+ User = self.classes.User
+
+
s = Session()
q1 = s.query(User, literal("x"))
q2 = s.query(User, literal_column("'y'"))
eq_(q.count(), 5)
def test_cols(self):
+ """test that column-based queries always nest."""
+
User, Address = self.classes.User, self.classes.Address
- """test that column-based queries always nest."""
s = create_session()
)
def test_joined(self):
+ """test that orderbys from a joined table get placed into the columns clause when DISTINCT is used"""
+
User, Address = self.classes.User, self.classes.Address
- """test that orderbys from a joined table get placed into the columns clause when DISTINCT is used"""
sess = create_session()
q = sess.query(User).join('addresses').distinct().order_by(desc(Address.email_address))
order_by=pagecomments.c.comment_id))})
def test_basic(self):
+ """A combination of complicated join conditions with post_update."""
+
Job = self.classes.Job
- """A combination of complicated join conditions with post_update."""
j1 = Job(jobno=u'somejob')
j1.create_page(u'page1')
pass
def test_onetoone_switch(self):
+ """test that active history is enabled on a
+ one-to-many/one that has use_get==True"""
+
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
- """test that active history is enabled on a
- one-to-many/one that has use_get==True"""
mapper(A, tableA, properties={
'b':relationship(B, cascade="all,delete-orphan", uselist=False)})
sess.flush()
def test_no_delete_PK_AtoB(self):
+ """A cant be deleted without B because B would have no PK value."""
+
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
- """A cant be deleted without B because B would have no PK value."""
mapper(A, tableA, properties={
'bs':relationship(B, cascade="save-update")})
mapper(B, tableB)
sess.flush()
def test_delete_cascade_BtoA(self):
+ """No 'blank the PK' error when the child is to
+ be deleted as part of a cascade"""
+
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
- """No 'blank the PK' error when the child is to
- be deleted as part of a cascade"""
for cascade in ("save-update, delete",
#"save-update, delete-orphan",
sa.orm.clear_mappers()
def test_delete_cascade_AtoB(self):
+ """No 'blank the PK' error when the child is to
+ be deleted as part of a cascade"""
+
tableB, A, B, tableA = (self.tables.tableB,
self.classes.A,
self.classes.B,
self.tables.tableA)
- """No 'blank the PK' error when the child is to
- be deleted as part of a cascade"""
for cascade in ("save-update, delete",
#"save-update, delete-orphan",
Column('t2c1', MySpecialType(30), ForeignKey('t2.col1')))
def testm2m(self):
+ """Many-to-many tables with special types for candidate keys."""
+
t2, t3, t1 = (self.tables.t2,
self.tables.t3,
self.tables.t1)
- """Many-to-many tables with special types for candidate keys."""
class T1(fixtures.BasicEntity): pass
class T2(fixtures.BasicEntity): pass
Column('t2id', Integer, ForeignKey('t2.id')))
def test_three_table_view(self):
- t2, t3, t1 = (self.tables.t2,
- self.tables.t3,
- self.tables.t1)
-
"""A three table join with overlapping PK names.
A third table is pulled into the primary join condition using
error.
"""
+
+ t2, t3, t1 = (self.tables.t2,
+ self.tables.t3,
+ self.tables.t1)
+
class C1(fixtures.BasicEntity): pass
class C2(fixtures.BasicEntity): pass
class C3(fixtures.BasicEntity): pass
Column('t2id_ref', Integer, ForeignKey('t2.t2id')))
def test_three_table_view(self):
- t2, t3, t1 = (self.tables.t2,
- self.tables.t3,
- self.tables.t1)
-
"""A three table join with overlapping PK names.
A third table is pulled into the primary join condition using unique
PK column names and should not produce 'mapper has no columnX' error.
"""
+
+ t2, t3, t1 = (self.tables.t2,
+ self.tables.t3,
+ self.tables.t1)
+
class C1(fixtures.BasicEntity): pass
class C2(fixtures.BasicEntity): pass
class C3(fixtures.BasicEntity): pass
run_inserts = None
def test_no_close_on_flush(self):
+ """Flush() doesn't close a connection the session didn't open"""
+
User, users = self.classes.User, self.tables.users
- """Flush() doesn't close a connection the session didn't open"""
c = testing.db.connect()
c.execute("select * from users")
c.execute("select * from users")
def test_close(self):
+ """close() doesn't close a connection the session didn't open"""
+
User, users = self.classes.User, self.tables.users
- """close() doesn't close a connection the session didn't open"""
c = testing.db.connect()
c.execute("select * from users")
eq_(sess.query(User).count(), 1)
def test_autoflush_expressions(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
-
"""test that an expression which is dependent on object state is
evaluated after the session autoflushes. This is the lambda
inside of strategies.py lazy_clause.
"""
+
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")})
mapper(Address, addresses)
assert not sess.is_active
def test_textual_execute(self):
+ """test that Session.execute() converts to text()"""
+
users = self.tables.users
- """test that Session.execute() converts to text()"""
sess = create_session(bind=self.metadata.bind)
users.insert().execute(id=7, name='jack')
assert not s.is_modified(u)
def test_weak_ref(self):
- users, User = self.tables.users, self.classes.User
-
"""test the weak-referencing identity map, which strongly-
references modified items."""
+ users, User = self.tables.users, self.classes.User
+
+
s = create_session()
mapper(User, users)
self.assert_sql_count(testing.db, go, 2)
def test_disable_dynamic(self):
+ """test no subquery option on a dynamic."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test no subquery option on a dynamic."""
mapper(User, users, properties={
'addresses':relationship(Address, lazy="dynamic")
], q.order_by(User.id).all())
def test_orderby_related(self):
+ """A regular mapper select on a single table can
+ order by a relationship to a second table"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A regular mapper select on a single table can
- order by a relationship to a second table"""
mapper(Address, addresses)
mapper(User, users, properties = dict(
def test_cyclical(self):
+ """A circular eager relationship breaks the cycle with a lazy loader"""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """A circular eager relationship breaks the cycle with a lazy loader"""
mapper(Address, addresses)
mapper(User, users, properties = dict(
eq_(self.static.user_address_result, sess.query(User).order_by(User.id).all())
def test_double(self):
+ """Eager loading with two relationships simultaneously,
+ from the same table, using aliases."""
+
users, orders, User, Address, Order, addresses = (self.tables.users,
self.tables.orders,
self.classes.User,
self.classes.Order,
self.tables.addresses)
- """Eager loading with two relationships simultaneously,
- from the same table, using aliases."""
openorders = sa.alias(orders, 'openorders')
closedorders = sa.alias(orders, 'closedorders')
self.assert_sql_count(testing.db, go, 4)
def test_double_same_mappers(self):
+ """Eager loading with two relationships simulatneously,
+ from the same table, using aliases."""
+
addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.users)
- """Eager loading with two relationships simulatneously,
- from the same table, using aliases."""
mapper(Address, addresses)
mapper(Order, orders, properties={
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_limit(self):
+ """Limit operations combined with lazy-load relationships."""
+
users, items, order_items, orders, Item, User, Address, Order, addresses = (self.tables.users,
self.tables.items,
self.tables.order_items,
self.classes.Order,
self.tables.addresses)
- """Limit operations combined with lazy-load relationships."""
mapper(Item, items)
mapper(Order, orders, properties={
self.assert_sql_count(testing.db, go, 3)
def test_uselist_false_warning(self):
+ """test that multiple rows received by a
+ uselist=False raises a warning."""
+
User, users, orders, Order = (self.classes.User,
self.tables.users,
self.tables.orders,
self.classes.Order)
- """test that multiple rows received by a
- uselist=False raises a warning."""
mapper(User, users, properties={
'order':relationship(Order, uselist=False)
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_no_depth(self):
+ """no join depth is set, so no eager loading occurs."""
+
nodes = self.tables.nodes
- """no join depth is set, so no eager loading occurs."""
class Node(fixtures.ComparableEntity):
def append(self, node):
self.children.append(node)
def test_accounting_no_select_needed(self):
"""test that flush accounting works on non-expired instances
when autocommit=True/expire_on_commit=True."""
+
User = self.classes.User
sess = create_session(autocommit=True, expire_on_commit=True)
@testing.emits_warning(r".*'passive_deletes' is normally configured on one-to-many")
def test_backwards_pd(self):
- myothertable, MyClass, MyOtherClass, mytable = (self.tables.myothertable,
- self.classes.MyClass,
- self.classes.MyOtherClass,
- self.tables.mytable)
-
"""Test that passive_deletes=True disables a delete from an m2o.
This is not the usual usage and it now raises a warning, but test
that it works nonetheless.
"""
+
+ myothertable, MyClass, MyOtherClass, mytable = (self.tables.myothertable,
+ self.classes.MyClass,
+ self.classes.MyOtherClass,
+ self.tables.mytable)
+
mapper(MyOtherClass, myothertable, properties={
'myclass':relationship(MyClass, cascade="all, delete", passive_deletes=True)
})
@testing.fails_on('firebird', 'Data type unknown on the parameter')
def test_used_in_relationship(self):
+ """A server-side default can be used as the target of a foreign key"""
+
Hoho, hohoval, default_t, secondary_table, Secondary = (self.classes.Hoho,
self.other.hohoval,
self.tables.default_t,
self.tables.secondary_table,
self.classes.Secondary)
- """A server-side default can be used as the target of a foreign key"""
mapper(Hoho, default_t, properties={
'secondaries':relationship(Secondary, order_by=secondary_table.c.id)})
run_inserts = None
def test_one_to_many_1(self):
+ """Basic save of one to many."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """Basic save of one to many."""
m = mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses), lazy='select')
self.assert_(u.id == userid and a2.id == addressid)
def test_one_to_many_2(self):
+ """Modifying the child items of an object."""
+
Address, addresses, users, User = (self.classes.Address,
self.tables.addresses,
self.tables.users,
self.classes.User)
- """Modifying the child items of an object."""
m = mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses), lazy='select')))
{'user_id': u1.id, 'addresses_id': a3.id})])
def test_child_move(self):
- Address, addresses, users, User = (self.classes.Address,
- self.tables.addresses,
- self.tables.users,
- self.classes.User)
-
"""Moving a child from one parent to another, with a delete.
Tests that deleting the first parent properly updates the child with
module.
"""
+
+ Address, addresses, users, User = (self.classes.Address,
+ self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
m = mapper(User, users, properties=dict(
addresses = relationship(mapper(Address, addresses), lazy='select')))
eq_(u.syn_name, 'User:some name:User')
def test_lazyattr_commit(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
-
"""Lazily loaded relationships.
When a lazy-loaded list is unloaded, and a commit occurs, that the
'passive' call on that list does not blow away its value
"""
+
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties = {
'addresses': relationship(mapper(Address, addresses))})
eq_(len(u.addresses), 4)
def test_inherits(self):
+ """a user object that also has the users mailing address."""
+
users, addresses, User = (self.tables.users,
self.tables.addresses,
self.classes.User)
m1 = mapper(User, users)
class AddressUser(User):
- """a user object that also has the users mailing address."""
pass
# define a mapper for AddressUser that inherits the User.mapper, and
eq_(rt.id, rt.id)
def test_deferred(self):
+ """Deferred column operations"""
+
orders, Order = self.tables.orders, self.classes.Order
- """Deferred column operations"""
mapper(Order, orders, properties={
'description': sa.orm.deferred(orders.c.description)})
self.sql_count_(0, session.flush)
def test_multi_table_selectable(self):
- addresses, users, User = (self.tables.addresses,
- self.tables.users,
- self.classes.User)
-
"""Mapped selectables that span tables.
Also tests redefinition of the keynames for the column properties.
"""
+
+ addresses, users, User = (self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
usersaddresses = sa.join(users, addresses,
users.c.id == addresses.c.user_id)
assert u.name == 'imnew'
def test_history_get(self):
+ """The history lazy-fetches data when it wasn't otherwise loaded."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """The history lazy-fetches data when it wasn't otherwise loaded."""
mapper(User, users, properties={
'addresses':relationship(Address, cascade="all, delete-orphan")})
mapper(Address, addresses)
assert addresses.count().scalar() == 0
def test_batch_mode(self):
+ """The 'batch=False' flag on mapper()"""
+
users, User = self.tables.users, self.classes.User
- """The 'batch=False' flag on mapper()"""
names = []
class Events(object):
session.flush()
def test_many_to_many_remove(self):
- keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
- self.tables.items,
- self.tables.item_keywords,
- self.classes.Keyword,
- self.classes.Item)
-
"""Setting a collection to empty deletes many-to-many rows.
Tests that setting a list-based attribute to '[]' properly affects the
history and allows the many-to-many rows to be deleted
"""
+
+ keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
+ self.tables.items,
+ self.tables.item_keywords,
+ self.classes.Keyword,
+ self.classes.Item)
+
mapper(Keyword, keywords)
mapper(Item, items, properties=dict(
keywords = relationship(Keyword, item_keywords, lazy='joined'),
assert item_keywords.count().scalar() == 0
def test_scalar(self):
+ """sa.dependency won't delete an m2m relationship referencing None."""
+
keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
self.tables.items,
self.tables.item_keywords,
self.classes.Keyword,
self.classes.Item)
- """sa.dependency won't delete an m2m relationship referencing None."""
mapper(Keyword, keywords)
session.flush()
def test_many_to_many_update(self):
+ """Assorted history operations on a many to many"""
+
keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
self.tables.items,
self.tables.item_keywords,
self.classes.Keyword,
self.classes.Item)
- """Assorted history operations on a many to many"""
mapper(Keyword, keywords)
mapper(Item, items, properties=dict(
keywords=relationship(Keyword,
assert item.keywords == [k1, k2]
def test_association(self):
+ """Basic test of an association object"""
+
keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
self.tables.items,
self.tables.item_keywords,
self.classes.Keyword,
self.classes.Item)
- """Basic test of an association object"""
class IKAssociation(fixtures.ComparableEntity):
pass
pass
def test_manytomany_xtracol_delete(self):
+ """A many-to-many on a table that has an extra column can properly delete rows from the table without referencing the extra column"""
+
keywords, items, assoc, Keyword, Item = (self.tables.keywords,
self.tables.items,
self.tables.assoc,
self.classes.Keyword,
self.classes.Item)
- """A many-to-many on a table that has an extra column can properly delete rows from the table without referencing the extra column"""
mapper(Keyword, keywords)
mapper(Item, items, properties=dict(
)
def test_natural_ordering(self):
+ """test that unconnected items take relationship() into account regardless."""
+
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.tables.addresses,
self.classes.User)
- """test that unconnected items take relationship() into account regardless."""
mapper(User, users)
mapper(Address, addresses, properties={
)
def test_natural_selfref(self):
+ """test that unconnected items take relationship() into account regardless."""
+
Node, nodes = self.classes.Node, self.tables.nodes
- """test that unconnected items take relationship() into account regardless."""
mapper(Node, nodes, properties={
'children':relationship(Node)
)
def test_batch_interaction(self):
- t = self.tables.t
-
"""test batching groups same-structured, primary
key present statements together.
"""
+
+ t = self.tables.t
+
class T(fixtures.ComparableEntity):
pass
mapper(T, t)
assert Point.max_x is alias.max_x
def test_descriptors(self):
+ """Tortured..."""
+
class descriptor(object):
- """Tortured..."""
def __init__(self, fn):
self.fn = fn
def __get__(self, obj, owner):
@testing.emits_warning(r'.*does not support updated rowcount')
def test_bump_version(self):
- Foo = self.classes.Foo
-
"""test that version number can be bumped.
Ensures that the UPDATE or DELETE is against the
state.
"""
+
+ Foo = self.classes.Foo
+
s1 = self._fixture()
f1 = Foo(value='f1')
s1.add(f1)
@testing.emits_warning(r'.*does not support updated rowcount')
@engines.close_open_connections
def test_versioncheck(self):
+ """query.with_lockmode performs a 'version check' on an already loaded instance"""
+
Foo = self.classes.Foo
- """query.with_lockmode performs a 'version check' on an already loaded instance"""
s1 = self._fixture()
f1s1 = Foo(value='f1 value')
@engines.close_open_connections
@testing.requires.update_nowait
def test_versioncheck_for_update(self):
+ """query.with_lockmode performs a 'version check' on an already loaded instance"""
+
Foo = self.classes.Foo
- """query.with_lockmode performs a 'version check' on an already loaded instance"""
s1 = self._fixture()
f1s1 = Foo(value='f1 value')
@testing.emits_warning(r'.*does not support updated rowcount')
@engines.close_open_connections
def test_noversioncheck(self):
+ """test query.with_lockmode works when the mapper has no version id col"""
+
Foo, version_table = self.classes.Foo, self.tables.version_table
- """test query.with_lockmode works when the mapper has no version id col"""
s1 = create_session(autocommit=False)
mapper(Foo, version_table)
f1s1 = Foo(value="foo", version_id=0)
essentially tests the ANONYMOUS_LABEL regex.
"""
+
s1 = table1.select()
s2 = s1.alias()
s3 = select([s2], use_labels=True)
def test_index_create_camelcase(self):
"""test that mixed-case index identifiers are legal"""
+
employees = Table('companyEmployees', metadata,
Column('id', Integer, primary_key=True),
Column('firstName', String(30)),
"""Using a DefaultGenerator, Sequence, DefaultClause
in the columns, where clause of a select, or in the values
clause of insert, update, raises an informative error"""
+
for const in (
sa.Sequence('y'),
sa.ColumnDefault('y'),
def _assert_seq_result(self, ret):
"""asserts return of next_value is an int"""
+
assert isinstance(ret, (int, long))
assert ret > 0
like this. Testing that all backends do the same thing here.
"""
+
metadata = self.metadata
t = Table('x', metadata,
Column('y', String(10), server_default='key_one', primary_key=True),
@testing.provide_metadata
def test_string_default_on_insert_with_returning(self):
"""With implicit_returning, we get a string PK default back no problem."""
+
metadata = self.metadata
t = Table('x', metadata,
Column('y', String(10), server_default='key_one', primary_key=True),
ORDER BY.
"""
+
users.insert().execute(
{'user_id':7, 'user_name':'jack'},
{'user_id':8, 'user_name':'ed'},
This should be run for DB-APIs with both positional and named
paramstyles.
"""
+
users.insert().execute(user_id = 7, user_name = 'jack')
users.insert().execute(user_id = 8, user_name = 'fred')
@testing.exclude('mysql', '<', (5, 0, 37), 'database bug')
def test_scalar_select(self):
"""test that scalar subqueries with labels get their type propagated to the result set."""
+
# mysql and/or mysqldb has a bug here, type isn't propagated for scalar
# subquery.
datetable = Table('datetable', metadata,
generate ? = ?.
"""
+
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)
more palatable to a wider variety of engines.
"""
+
u = union(
select([t1.c.col3]),
select([t1.c.col3]),
where the "UPPERCASE" column of "LaLa" doesnt exist.
"""
+
x = table1.select(distinct=True).alias("LaLa").select().scalar()
def testlabels2(self):
for default compilation.
"""
+
for type_, expected in (
(String(), "VARCHAR"),
(Integer(), "INTEGER"),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
