from test.lib import testing
from test.lib.testing import adict
from test.lib.engines import drop_all_tables
-
+import sys
class TablesTest(testing.TestBase):
"""An integration test that creates and uses tables."""
# 'once', 'each', None
run_define_tables = 'once'
+ # 'once', 'each', None
+ run_create_tables = 'once'
+
# 'once', 'each', None
run_inserts = 'each'
- # 'foreach', None
+ # 'each', None
run_deletes = 'each'
# 'once', 'each', None
@classmethod
def _init_class(cls):
+ if cls.run_define_tables == 'each':
+ if cls.run_create_tables == 'once':
+ cls.run_create_tables = 'each'
+ assert cls.run_inserts in ('each', None)
+
if cls.other is None:
cls.other = adict()
def _setup_once_tables(cls):
if cls.run_define_tables == 'once':
cls.define_tables(cls.metadata)
- cls.metadata.create_all()
+ if cls.run_create_tables == 'once':
+ cls.metadata.create_all()
cls.tables.update(cls.metadata.tables)
def _setup_each_tables(self):
if self.run_define_tables == 'each':
self.tables.clear()
- drop_all_tables(self.metadata)
+ if self.run_create_tables == 'each':
+ drop_all_tables(self.metadata)
self.metadata.clear()
self.define_tables(self.metadata)
- self.metadata.create_all()
+ if self.run_create_tables == 'each':
+ self.metadata.create_all()
self.tables.update(self.metadata.tables)
def _setup_each_inserts(self):
@classmethod
def _teardown_once_metadata_bind(cls):
- cls.metadata.drop_all()
+ if cls.run_create_tables:
+ cls.metadata.drop_all()
if cls.run_dispose_bind == 'once':
cls.dispose_bind(cls.bind)
-"""
+"""Classes used in pickling tests, need to be at the module level for unpickling."""
-some objects used for pickle tests, declared in their own module so that they
-are easily pickleable.
+from test.orm import _base
-"""
+class User(_base.ComparableEntity):
+ pass
+class Order(_base.ComparableEntity):
+ pass
+
+class EmailUser(User):
+ pass
+
+class Address(_base.ComparableEntity):
+ pass
+
+# TODO: these are kind of arbitrary....
+class Child1(_base.ComparableEntity):
+ pass
+
+class Child2(_base.ComparableEntity):
+ pass
+
+class Parent(_base.ComparableEntity):
+ pass
+
+class Screen(object):
+ def __init__(self, obj, parent=None):
+ self.obj = obj
+ self.parent = parent
class Foo(object):
def __init__(self, moredata):
from sqlalchemy import MetaData, Integer, String, ForeignKey
+from sqlalchemy import util
from test.lib.schema import Table
from test.lib.schema import Column
from sqlalchemy.orm import attributes
__all__ = ()
-class Base(_base.ComparableEntity):
- pass
-
-
-class User(Base):
- pass
-
-
-class Order(Base):
- pass
-
-
-class Item(Base):
- pass
-
-
-class Keyword(Base):
- pass
-
-
-class Address(Base):
- pass
+class FixtureTest(_base.MappedTest):
+ """A MappedTest pre-configured with a common set of fixtures.
+ """
-class Dingaling(Base):
- pass
+ run_define_tables = 'once'
+ run_setup_classes = 'once'
+ run_setup_mappers = 'each'
+ run_inserts = 'each'
+ run_deletes = 'each'
+ @classmethod
+ def setup_classes(cls):
+ class Base(_base.ComparableEntity):
+ pass
-class Node(Base):
- pass
+ class User(Base):
+ pass
-class CompositePk(Base):
- pass
+ class Order(Base):
+ pass
-class FixtureTest(_base.MappedTest):
- """A MappedTest pre-configured for fixtures.
+ class Item(Base):
+ pass
- All fixture tables are pre-loaded into cls.tables, as are all fixture
- lasses in cls.classes and as cls.ClassName.
+ class Keyword(Base):
+ pass
- Fixture.mapper() still functions and willregister non-fixture classes into
- cls.classes.
+ class Address(Base):
+ pass
- """
+ class Dingaling(Base):
+ pass
- run_define_tables = 'once'
- run_setup_classes = 'once'
- run_setup_mappers = 'each'
- run_inserts = 'each'
- run_deletes = 'each'
+ class Node(Base):
+ pass
- fixture_classes = dict(User=User,
- Order=Order,
- Item=Item,
- Keyword=Keyword,
- Address=Address,
- Dingaling=Dingaling)
+ class CompositePk(Base):
+ pass
@classmethod
def define_tables(cls, metadata):
Column('k', Integer, nullable=False),
)
- @classmethod
- def setup_classes(cls):
- for cl in cls.fixture_classes.values():
- cls.register_class(cl)
-
@classmethod
def setup_mappers(cls):
pass
)
)
+ @util.memoized_property
+ def static(self):
+ return CannedResults(self)
+
class CannedResults(object):
"""Built on demand, instances use mappers in effect at time of call."""
+ def __init__(self, test):
+ self.test = test
+
@property
def user_result(self):
+ User = self.test.classes.User
+
return [
User(id=7),
User(id=8),
@property
def user_address_result(self):
+ User, Address = self.test.classes.User, self.test.classes.Address
+
return [
User(id=7, addresses=[
Address(id=1)
@property
def user_all_result(self):
+ User, Address, Order, Item = self.test.classes.User, \
+ self.test.classes.Address, self.test.classes.Order, \
+ self.test.classes.Item
+
return [
User(id=7,
addresses=[
@property
def user_order_result(self):
+ User, Order, Item = self.test.classes.User, \
+ self.test.classes.Order, self.test.classes.Item
return [
User(id=7,
orders=[
@property
def item_keyword_result(self):
+ Item, Keyword = self.test.classes.Item, self.test.classes.Keyword
return [
Item(id=1,
keywords=[
@property
def user_item_keyword_result(self):
+ Item, Keyword = self.test.classes.Item, self.test.classes.Keyword
+ User, Order = self.test.classes.User, self.test.classes.Order
+
item1, item2, item3, item4, item5 = \
Item(id=1,
keywords=[
User(id=10, orders=[])]
return user_result
-FixtureTest.static = CannedResults()
def make_test(fetchtype):
def test_roundtrip(self):
- class A(_fixtures.Base):pass
+ class A(_base.ComparableEntity):pass
class B(A):pass
class C(B):pass
Column('data', String(10), nullable=False))
def test_polymorphic_synonym(self):
- class T1(_fixtures.Base):
+ class T1(_base.ComparableEntity):
def info(self):
return "THE INFO IS:" + self._info
def _set_info(self, x):
Column('data', String(30)))
def test_cascade(self):
- class T1(_fixtures.Base):
+ class T1(_base.ComparableEntity):
pass
- class T2(_fixtures.Base):
+ class T2(_base.ComparableEntity):
pass
class T3(T2):
pass
- class T4(_fixtures.Base):
+ class T4(_base.ComparableEntity):
pass
mapper(T1, t1, properties={
self.tables.base,
self.tables.stuff)
- class Base(_fixtures.Base):
+ class Base(_base.ComparableEntity):
pass
class Sub(Base):
pass
def test_delete(self):
subtable, base = self.tables.subtable, self.tables.base
- class Base(_fixtures.Base):
+ class Base(_base.ComparableEntity):
pass
class Sub(Base):
pass
)
def test_orphan_message(self):
- class Base(_fixtures.Base):
+ class Base(_base.ComparableEntity):
pass
class SubClass(Base):
pass
- class Parent(_fixtures.Base):
+ class Parent(_base.ComparableEntity):
pass
mapper(Base, single, polymorphic_on=single.c.type, polymorphic_identity='base')
from test.lib.util import function_named
from test.orm import _fixtures, _base
-class Person(_fixtures.Base):
+class Person(_base.ComparableEntity):
pass
class Engineer(Person):
pass
pass
class Boss(Manager):
pass
-class Company(_fixtures.Base):
+class Company(_base.ComparableEntity):
pass
class PolymorphTest(_base.MappedTest):
def test_basic(self):
"""test that Query uses the full set of mapper._eager_loaders when generating SQL"""
- class Person(_fixtures.Base):
+ class Person(_base.ComparableEntity):
pass
class Employee(Person):
def __init__(self, name='bob'):
self.name = name
- class Tag(_fixtures.Base):
+ class Tag(_base.ComparableEntity):
def __init__(self, label):
self.label = label
)
def test_polyon_col_setsup(self):
- class A(_fixtures.Base):
+ class A(_base.ComparableEntity):
pass
- class B(_fixtures.Base):
+ class B(_base.ComparableEntity):
pass
class C(A):
pass
from test.lib.testing import eq_
from test.lib.schema import Table, Column
-class Company(_fixtures.Base):
+class Company(_base.ComparableEntity):
pass
-class Person(_fixtures.Base):
+class Person(_base.ComparableEntity):
pass
class Engineer(Person):
pass
class Boss(Manager):
pass
-class Machine(_fixtures.Base):
+class Machine(_base.ComparableEntity):
pass
-class Paperwork(_fixtures.Base):
+class Paperwork(_base.ComparableEntity):
pass
def _produce_test(select_type):
cls.tables.engineers,
cls.tables.engineers_to_org)
- class Organization(_fixtures.Base):
+ class Organization(_base.ComparableEntity):
pass
mapper(Organization, organizations, properties={
from sqlalchemy.orm import *
from test.lib import testing
-from test.orm._fixtures import Base
-from test.orm._base import MappedTest
+
+from test.orm._base import MappedTest, ComparableEntity as Base
class InheritingSelectablesTest(MappedTest):
)
def test_single_on_joined(self):
- class Person(_fixtures.Base):
+ class Person(ComparableEntity):
pass
class Employee(Person):
pass
@classmethod
def setup_classes(cls):
- class User(_fixtures.Base):
+ class User(_base.ComparableEntity):
pass
- class Pref(_fixtures.Base):
+ class Pref(_base.ComparableEntity):
pass
- class Extra(_fixtures.Base):
+ class Extra(_base.ComparableEntity):
pass
- class Foo(_fixtures.Base):
+ class Foo(_base.ComparableEntity):
pass
@classmethod
@classmethod
def setup_classes(cls):
- class T1(_fixtures.Base):
+ class T1(_base.ComparableEntity):
pass
- class T2(_fixtures.Base):
+ class T2(_base.ComparableEntity):
pass
- class T3(_fixtures.Base):
+ class T3(_base.ComparableEntity):
pass
@classmethod
@classmethod
def setup_classes(cls):
- class T1(_fixtures.Base):
+ class T1(_base.ComparableEntity):
pass
- class T2(_fixtures.Base):
+ class T2(_base.ComparableEntity):
pass
- class T3(_fixtures.Base):
+ class T3(_base.ComparableEntity):
pass
@classmethod
@classmethod
def setup_classes(cls):
- class A(_fixtures.Base):
+ class A(_base.ComparableEntity):
pass
- class B(_fixtures.Base):
+ class B(_base.ComparableEntity):
pass
- class C(_fixtures.Base):
+ class C(_base.ComparableEntity):
pass
def test_delete_orphan(self):
)
@classmethod
def setup_classes(cls):
- class User(_fixtures.Base):
+ class User(_base.ComparableEntity):
pass
- class Address(_fixtures.Base):
+ class Address(_base.ComparableEntity):
pass
- class Order(_fixtures.Base):
+ class Order(_base.ComparableEntity):
pass
def test_pending_standalone_orphan(self):
"""test the delete-orphan uow event for multiple delete-orphan
parent relationships."""
- class Customer(_fixtures.Base):
+ class Customer(_base.ComparableEntity):
pass
- class Account(_fixtures.Base):
+ class Account(_base.ComparableEntity):
pass
- class SalesRep(_fixtures.Base):
+ class SalesRep(_base.ComparableEntity):
pass
mapper(Customer, customers)
"""test the delete-orphan uow event for multiple delete-orphan
parent relationships."""
- class Customer(_fixtures.Base):
+ class Customer(_base.ComparableEntity):
pass
- class Account(_fixtures.Base):
+ class Account(_base.ComparableEntity):
pass
- class SalesRep(_fixtures.Base):
+ class SalesRep(_base.ComparableEntity):
pass
mapper(Customer, customers)
"""test that an entity can have two parent delete-orphan
cascades, and persists normally."""
- class Address(_fixtures.Base):
+ class Address(_base.ComparableEntity):
pass
- class Home(_fixtures.Base):
+ class Home(_base.ComparableEntity):
pass
- class Business(_fixtures.Base):
+ class Business(_base.ComparableEntity):
pass
mapper(Address, addresses)
cascades, and is detected as an orphan when saved without a
parent."""
- class Address(_fixtures.Base):
+ class Address(_base.ComparableEntity):
pass
- class Home(_fixtures.Base):
+ class Home(_base.ComparableEntity):
pass
- class Business(_fixtures.Base):
+ class Business(_base.ComparableEntity):
pass
mapper(Address, addresses)
def test_basic(self):
table_b, table_a = self.tables.table_b, self.tables.table_a
- class A(_fixtures.Base):
+ class A(_base.ComparableEntity):
pass
- class B(_fixtures.Base):
+ class B(_base.ComparableEntity):
pass
mapper(A, table_a, properties={
from test.lib import testing, AssertsCompiledSQL, Column, engines
from test.orm import _fixtures
-from test.orm._fixtures import keywords, addresses, Base, \
- Keyword, FixtureTest, \
- Dingaling, item_keywords, dingalings, User, items,\
- orders, Address, users, nodes, \
- order_items, Item, Order, Node, \
- composite_pk_table, CompositePk
from test.orm import _base
run_inserts = 'once'
run_deletes = None
-
@classmethod
def setup_mappers(cls):
+ Node, composite_pk_table, users, Keyword, items, Dingaling, \
+ order_items, item_keywords, Item, User, dingalings, \
+ Address, keywords, CompositePk, nodes, Order, orders, \
+ addresses = cls.classes.Node, \
+ cls.tables.composite_pk_table, cls.tables.users, \
+ cls.classes.Keyword, cls.tables.items, \
+ cls.classes.Dingaling, cls.tables.order_items, \
+ cls.tables.item_keywords, cls.classes.Item, \
+ cls.classes.User, cls.tables.dingalings, \
+ cls.classes.Address, cls.tables.keywords, \
+ cls.classes.CompositePk, cls.tables.nodes, \
+ cls.classes.Order, cls.tables.orders, cls.tables.addresses
+
mapper(User, users, properties={
'addresses':relationship(Address, backref='user', order_by=addresses.c.id),
'orders':relationship(Order, backref='user', order_by=orders.c.id), # o2m, m2o
"""
def test_select(self):
+ addresses, users = self.tables.addresses, self.tables.users
+
sess = create_session()
self.assert_compile(sess.query(users).select_from(users.select()).with_labels().statement,
class FromSelfTest(QueryTest, AssertsCompiledSQL):
def test_filter(self):
+ User = self.classes.User
+
assert [User(id=8), User(id=9)] == create_session().query(User).filter(User.id.in_([8,9])).from_self().all()
assert [User(id=8)] == list(create_session().query(User).filter(User.id.in_([8,9])).from_self().order_by(User.id)[0:1])
def test_join(self):
+ User, Address = self.classes.User, self.classes.Address
+
assert [
(User(id=8), Address(id=2)),
(User(id=8), Address(id=3)),
join('addresses').add_entity(Address).order_by(User.id, Address.id).all()
def test_group_by(self):
+ Address = self.classes.Address
+
eq_(
create_session().query(Address.user_id, func.count(Address.id).label('count')).\
group_by(Address.user_id).order_by(Address.user_id).all(),
)
def test_no_joinedload(self):
+ User = self.classes.User
+
"""test that joinedloads are pushed outwards and not rendered in subqueries."""
s = create_session()
)
def test_aliases(self):
+ User, Address = self.classes.User, self.classes.Address
+
"""test that aliased objects are accessible externally to a from_self() call."""
s = create_session()
def test_multiple_entities(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
eq_(
)
def test_multiple_with_column_entities(self):
+ User = self.classes.User
+
sess = create_session()
eq_(
cls.tables.b,
cls.tables.d)
- class A(_fixtures.Base):
+ class A(_base.ComparableEntity):
pass
- class B(_fixtures.Base):
+ class B(_base.ComparableEntity):
pass
class C(B):
class InstancesTest(QueryTest, AssertsCompiledSQL):
def test_from_alias(self):
+ User, addresses, users = (self.classes.User,
+ self.tables.addresses,
+ self.tables.users)
+
query = users.select(users.c.id==7).\
union(users.select(users.c.id>7)).\
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager(self):
+ users, addresses, User = (self.tables.users,
+ self.tables.addresses,
+ self.classes.User)
+
sess = create_session()
# test that contains_eager suppresses the normal outer join rendering
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_string_alias(self):
+ addresses, users, User = (self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
sess = create_session()
q = sess.query(User)
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_aliased_instances(self):
+ addresses, users, User = (self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
sess = create_session()
q = sess.query(User)
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_aliased(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
q = sess.query(User)
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_multi_string_alias(self):
+ orders, items, users, order_items, User = (self.tables.orders,
+ self.tables.items,
+ self.tables.users,
+ self.tables.order_items,
+ self.classes.User)
+
sess = create_session()
q = sess.query(User)
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_multi_alias(self):
+ orders, items, users, order_items, User = (self.tables.orders,
+ self.tables.items,
+ self.tables.users,
+ self.tables.order_items,
+ self.classes.User)
+
sess = create_session()
q = sess.query(User)
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_multi_aliased(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = create_session()
q = sess.query(User)
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_chaining(self):
+ Dingaling, User, Address = (self.classes.Dingaling,
+ self.classes.User,
+ self.classes.Address)
+
"""test that contains_eager() 'chains' by default."""
sess = create_session()
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_chaining_aliased_endpoint(self):
+ 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."""
self.assert_sql_count(testing.db, go, 1)
def test_mixed_eager_contains_with_limit(self):
+ Order, User, Address = (self.classes.Order,
+ self.classes.User,
+ self.classes.Address)
+
sess = create_session()
q = sess.query(User)
class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
def test_values(self):
+ Address, users, User = (self.classes.Address,
+ self.tables.users,
+ self.classes.User)
+
sess = create_session()
assert list(sess.query(User).values()) == list()
eq_(list(q2), [(u'ed', u'jack', u'jack')])
def test_alias_naming(self):
+ User = self.classes.User
+
sess = create_session()
ua = aliased(User, name="foobar")
@testing.fails_on('mssql', 'FIXME: unknown')
def test_values_specific_order_by(self):
+ users, User = self.tables.users, self.classes.User
+
sess = create_session()
assert list(sess.query(User).values()) == list()
"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"""
sess = create_session()
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.
(User(id=9,name=u'fred'), 1)]
)
- def test_tuple_labeling(self):
- sess = create_session()
-
- # test pickle + all the protocols !
- for pickled in False, -1, 0, 1, 2:
- for row in sess.query(User, Address).join(User.addresses).all():
- if pickled is not False:
- row = util.pickle.loads(util.pickle.dumps(row, pickled))
-
- eq_(row.keys(), ['User', 'Address'])
- eq_(row.User, row[0])
- eq_(row.Address, row[1])
-
- for row in sess.query(User.name, User.id.label('foobar')):
- if pickled is not False:
- row = util.pickle.loads(util.pickle.dumps(row, pickled))
- eq_(row.keys(), ['name', 'foobar'])
- eq_(row.name, row[0])
- eq_(row.foobar, row[1])
-
- for row in sess.query(User).values(User.name, User.id.label('foobar')):
- if pickled is not False:
- row = util.pickle.loads(util.pickle.dumps(row, pickled))
- eq_(row.keys(), ['name', 'foobar'])
- eq_(row.name, row[0])
- eq_(row.foobar, row[1])
-
- oalias = aliased(Order)
- for row in sess.query(User, oalias).join(User.orders).all():
- if pickled is not False:
- row = util.pickle.loads(util.pickle.dumps(row, pickled))
- eq_(row.keys(), ['User'])
- eq_(row.User, row[0])
-
- oalias = aliased(Order, name='orders')
- for row in sess.query(User, oalias).join(oalias, User.orders).all():
- if pickled is not False:
- row = util.pickle.loads(util.pickle.dumps(row, pickled))
- eq_(row.keys(), ['User', 'orders'])
- eq_(row.User, row[0])
- eq_(row.orders, row[1])
-
- # test here that first col is not labeled, only
- # one name in keys, matches correctly
- for row in sess.query(User.name + 'hoho', User.name):
- eq_(row.keys(), ['name'])
- eq_(row[0], row.name + 'hoho')
-
- if pickled is not False:
- ret = sess.query(User, Address).join(User.addresses).all()
- util.pickle.loads(util.pickle.dumps(ret, pickled))
def test_column_queries(self):
+ Address, users, User = (self.classes.Address,
+ self.tables.users,
+ self.classes.User)
+
sess = create_session()
eq_(sess.query(User.name).all(), [(u'jack',), (u'ed',), (u'fred',), (u'chuck',)])
)
def test_column_from_limited_joinedload(self):
+ User = self.classes.User
+
sess = create_session()
def go():
@testing.fails_on('postgresql+pg8000', "'type oid 705 not mapped to py type' (due to literal)")
def test_self_referential(self):
+ Order = self.classes.Order
+
sess = create_session()
oalias = aliased(Order)
def test_multi_mappers(self):
+ Address, addresses, users, User = (self.classes.Address,
+ self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
test_session = create_session()
sess.expunge_all()
def test_aliased_multi_mappers(self):
+ User, addresses, users, Address = (self.classes.User,
+ self.tables.addresses,
+ self.tables.users,
+ self.classes.Address)
+
sess = create_session()
(user7, user8, user9, user10) = sess.query(User).all()
assert l == [(user8, address3)]
def test_with_entities(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
q = sess.query(User).filter(User.id==7).order_by(User.name)
def test_multi_columns(self):
+ users, User = self.tables.users, self.classes.User
+
sess = create_session()
expected = [(u, u.name) for u in sess.query(User).all()]
assert_raises(sa_exc.InvalidRequestError, sess.query(User).add_column, object())
def test_add_multi_columns(self):
+ 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):
+ 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()
def test_raw_columns(self):
+ addresses, users, User = (self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
sess = create_session()
(user7, user8, user9, user10) = sess.query(User).all()
expected = [
run_setup_mappers = None
def test_replace_with_select(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties = {
'addresses':relationship(Address)
})
)
def test_join_mapper_order_by(self):
+ 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):
+ users, User = self.tables.users, self.classes.User
+
"""test some different combinations of joining a table to a subquery of itself."""
mapper(User, users)
def test_join_no_order_by(self):
+ User, users = self.classes.User, self.tables.users
+
mapper(User, users)
sel = users.select(users.c.id.in_([7, 8]))
)
def test_join(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties = {
'addresses':relationship(Address)
})
def test_more_joins(self):
+ users, Keyword, orders, items, order_items, Order, Item, User, keywords, item_keywords = (self.tables.users,
+ self.classes.Keyword,
+ self.tables.orders,
+ self.tables.items,
+ self.tables.order_items,
+ self.classes.Order,
+ self.classes.Item,
+ self.classes.User,
+ self.tables.keywords,
+ self.tables.item_keywords)
+
mapper(User, users, properties={
'orders':relationship(Order, backref='user'), # o2m, m2o
})
def test_replace_with_eager(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties = {
'addresses':relationship(Address, order_by=addresses.c.id)
})
run_setup_mappers = None
def test_double_same_mappers(self):
+ addresses, items, order_items, orders, Item, User, Address, Order, users = (self.tables.addresses,
+ self.tables.items,
+ self.tables.order_items,
+ self.tables.orders,
+ self.classes.Item,
+ self.classes.User,
+ self.classes.Address,
+ self.classes.Order,
+ self.tables.users)
+
"""test aliasing of joins with a custom join condition"""
mapper(Address, addresses)
mapper(Order, orders, properties={
run_setup_mappers = None
def test_external_columns_bad(self):
+ users, User = self.tables.users, self.classes.User
+
assert_raises_message(sa_exc.ArgumentError, "not represented in the mapper's table", mapper, User, users, properties={
'concat': (users.c.id * 2),
clear_mappers()
def test_external_columns(self):
+ 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={
)
def test_external_columns_joinedload(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)
+
# in this test, we have a subquery on User that accesses "addresses", underneath
# an joinedload for "addresses". So the "addresses" alias adapter needs to *not* hit
# the "addresses" table within the "user" subquery, but "user" still needs to be adapted.
from test.lib import testing, AssertsCompiledSQL, Column, engines
from test.orm import _fixtures
-from test.orm._fixtures import keywords, addresses, Base, \
- Keyword, FixtureTest, \
- Dingaling, item_keywords, dingalings, User, items,\
- orders, Address, users, nodes, \
- order_items, Item, Order, Node, \
- composite_pk_table, CompositePk
from test.orm import _base
@classmethod
def setup_mappers(cls):
+ Node, composite_pk_table, users, Keyword, items, Dingaling, \
+ order_items, item_keywords, Item, User, dingalings, \
+ Address, keywords, CompositePk, nodes, Order, orders, \
+ addresses = cls.classes.Node, \
+ cls.tables.composite_pk_table, cls.tables.users, \
+ cls.classes.Keyword, cls.tables.items, \
+ cls.classes.Dingaling, cls.tables.order_items, \
+ cls.tables.item_keywords, cls.classes.Item, \
+ cls.classes.User, cls.tables.dingalings, \
+ cls.classes.Address, cls.tables.keywords, \
+ cls.classes.CompositePk, cls.tables.nodes, \
+ cls.classes.Order, cls.tables.orders, cls.tables.addresses
+
mapper(User, users, properties={
'addresses':relationship(Address, backref='user', order_by=addresses.c.id),
'orders':relationship(Order, backref='user', order_by=orders.c.id), # o2m, m2o
cls.tables.machines,
cls.tables.engineers)
- class Company(_fixtures.Base):
+ class Company(_base.ComparableEntity):
pass
- class Person(_fixtures.Base):
+ class Person(_base.ComparableEntity):
pass
class Engineer(Person):
pass
pass
class Boss(Manager):
pass
- class Machine(_fixtures.Base):
+ class Machine(_base.ComparableEntity):
pass
- class Paperwork(_fixtures.Base):
+ class Paperwork(_base.ComparableEntity):
pass
mapper(Company, companies, properties={
class JoinTest(QueryTest, AssertsCompiledSQL):
def test_single_name(self):
+ User = self.classes.User
+
sess = create_session()
self.assert_compile(
)
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.
)
def test_single_prop(self):
+ Item, Order, User, Address = (self.classes.Item,
+ self.classes.Order,
+ self.classes.User,
+ self.classes.Address)
+
sess = create_session()
self.assert_compile(
sess.query(User).join(User.orders),
)
def test_overlapping_paths(self):
+ User = self.classes.User
+
for aliased in (True,False):
# load a user who has an order that contains item id 3 and address id 1 (order 3, owned by jack)
result = create_session().query(User).join('orders', 'items', aliased=aliased).\
assert [User(id=7, name='jack')] == result
def test_overlapping_paths_outerjoin(self):
+ User = self.classes.User
+
result = create_session().query(User).outerjoin('orders', 'items').\
filter_by(id=3).outerjoin('orders','address').filter_by(id=1).all()
assert [User(id=7, name='jack')] == result
def test_from_joinpoint(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = create_session()
for oalias,ialias in [(True, True), (False, False), (True, False), (False, True)]:
)
def test_join_nonmapped_column(self):
+ 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()
def test_backwards_join(self):
+ User, Address = self.classes.User, self.classes.Address
+
# a more controversial feature. join from
# User->Address, but the onclause is Address.user.
)
def test_multiple_with_aliases(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = create_session()
ualias = aliased(User)
)
def test_select_from_orm_joins(self):
+ User, Order = self.classes.User, self.classes.Order
+
sess = create_session()
ualias = aliased(User)
def test_overlapping_backwards_joins(self):
+ User, Order = self.classes.User, self.classes.Order
+
sess = create_session()
oalias1 = aliased(Order)
)
def test_replace_multiple_from_clause(self):
+ User, Order, Address = (self.classes.User,
+ self.classes.Order,
+ self.classes.Address)
+
"""test adding joins onto multiple FROM clauses"""
sess = create_session()
)
def test_multiple_adaption(self):
+ Item, Order, User = (self.classes.Item,
+ self.classes.Order,
+ self.classes.User)
+
sess = create_session()
self.assert_compile(
)
def test_onclause_conditional_adaption(self):
+ Item, Order, orders, order_items, User = (self.classes.Item,
+ self.classes.Order,
+ self.tables.orders,
+ self.tables.order_items,
+ self.classes.User)
+
sess = create_session()
# this is now a very weird test, nobody should really
# or: query.join(path_to_some_joined_table_mapper).join(target, sql_expression)
def test_pure_expression_error(self):
+ addresses, users = self.tables.addresses, self.tables.users
+
sess = create_session()
assert_raises_message(sa.exc.InvalidRequestError, "Could not find a FROM clause to join from", sess.query(users).join, addresses)
def test_orderby_arg_bug(self):
+ User, users, Order = (self.classes.User,
+ self.tables.users,
+ self.classes.Order)
+
sess = create_session()
# no arg error
result = sess.query(User).join('orders', aliased=True).order_by(Order.id).reset_joinpoint().order_by(users.c.id).all()
def test_no_onclause(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = create_session()
eq_(
)
def test_clause_onclause(self):
+ Item, Order, users, order_items, User = (self.classes.Item,
+ self.classes.Order,
+ self.tables.users,
+ self.tables.order_items,
+ self.classes.User)
+
sess = create_session()
eq_(
def test_aliased_classes(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
(user7, user8, user9, user10) = sess.query(User).all()
eq_(l.all(), [(user8, address2),(user8, address3),(user8, address4),])
def test_expression_onclauses(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = create_session()
subq = sess.query(User).subquery()
def test_implicit_joins_from_aliases(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = create_session()
OrderAlias = aliased(Order)
)
def test_aliased_classes_m2m(self):
+ Item, Order = self.classes.Item, self.classes.Order
+
sess = create_session()
(order1, order2, order3, order4, order5) = sess.query(Order).all()
)
def test_joins_from_adapted_entities(self):
+ User = self.classes.User
+
# test for #1853
use_default_dialect=True)
def test_reset_joinpoint(self):
+ User = self.classes.User
+
for aliased in (True, False):
# load a user who has an order that contains item id 3 and address id 1 (order 3, owned by jack)
result = create_session().query(User).join('orders', 'items', aliased=aliased).filter_by(id=3).reset_joinpoint().join('orders','address', aliased=aliased).filter_by(id=1).all()
assert [User(id=7, name='jack')] == result
def test_overlap_with_aliases(self):
+ orders, User, users = (self.tables.orders,
+ self.classes.User,
+ self.tables.users)
+
oalias = orders.alias('oalias')
result = create_session().query(User).select_from(users.join(oalias)).filter(oalias.c.description.in_(["order 1", "order 2", "order 3"])).join('orders', 'items').order_by(User.id).all()
assert [User(id=7, name='jack')] == result
def test_aliased(self):
+ 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()
assert [User(id=7)] == q.all()
def test_aliased_order_by(self):
+ User = self.classes.User
+
sess = create_session()
ualias = aliased(User)
)
def test_plain_table(self):
+ addresses, User = self.tables.addresses, self.classes.User
+
sess = create_session()
)
def test_no_joinpoint_expr(self):
+ User, users = self.classes.User, self.tables.users
+
sess = create_session()
# these are consistent regardless of
)
def test_select_from(self):
+ 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()
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.
"""
cls.tables.assoc_table,
cls.tables.sub_table)
- class Node(Base):
+ class Node(_base.ComparableEntity):
pass
- class Sub(Base):
+ class Sub(_base.ComparableEntity):
pass
mapper(Node, nodes, properties={
@classmethod
def setup_classes(cls):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@classmethod
def setup_classes(cls):
- class Node(Base):
+ class Node(_base.ComparableEntity):
pass
@classmethod
assert opt2a not in ustate.load_options
assert opt2b not in ustate.load_options
- import pickle
- pickle.dumps(u1)
class DeepOptionsTest(_fixtures.FixtureTest):
@classmethod
from test.lib.testing import eq_
-import pickle
+from sqlalchemy.util import pickle
import sqlalchemy as sa
from test.lib import testing
from test.lib.util import picklers
sessionmaker, attributes, interfaces,\
clear_mappers, exc as orm_exc,\
configure_mappers, Session, lazyload_all,\
- lazyload
+ lazyload, aliased
from test.orm import _base, _fixtures
+from test.lib.pickleable import User, Address, Order, Child1, Child2, Parent, Screen, EmailUser
-User, EmailUser = None, None
+class PickleTest(_base.MappedTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('users', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(30), nullable=False),
+ test_needs_acid=True,
+ test_needs_fk=True
+ )
+ Table('addresses', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('user_id', None, ForeignKey('users.id')),
+ Column('email_address', String(50), nullable=False),
+ test_needs_acid=True,
+ test_needs_fk=True
+ )
+ Table('orders', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('user_id', None, ForeignKey('users.id')),
+ Column('address_id', None, ForeignKey('addresses.id')),
+ Column('description', String(30)),
+ Column('isopen', Integer),
+ test_needs_acid=True,
+ test_needs_fk=True
+ )
-class PickleTest(_fixtures.FixtureTest):
- run_inserts = None
def test_transient(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
+ users, addresses = (self.tables.users,
+ self.tables.addresses)
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
eq_(u1, sess.query(User).get(u2.id))
def test_no_mappers(self):
- users, User = self.tables.users, self.classes.User
+ users = self.tables.users
umapper = mapper(User, users)
assert_raises_message(
orm_exc.UnmappedInstanceError,
- "Cannot deserialize object of type <class 'test.orm._fixtures.User'> - no mapper()",
+ "Cannot deserialize object of type <class 'test.lib.pickleable.User'> - no mapper()",
pickle.loads, u1_pickled)
def test_no_instrumentation(self):
- users, User = self.tables.users, self.classes.User
+ users = self.tables.users
umapper = mapper(User, users)
eq_(str(u1), "User(name='ed')")
def test_serialize_path(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
+ users, addresses = (self.tables.users,
+ self.tables.addresses)
umapper = mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
)
def test_class_deferred_cols(self):
- Address, addresses, users, User = (self.classes.Address,
- self.tables.addresses,
- self.tables.users,
- self.classes.User)
+ addresses, users = (self.tables.addresses,
+ self.tables.users)
mapper(User, users, properties={
'name':sa.orm.deferred(users.c.name),
eq_(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
def test_instance_lazy_relation_loaders(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
+ users, addresses = (self.tables.users,
+ self.tables.addresses)
mapper(User, users, properties={
'addresses':relationship(Address, lazy='noload')
assert u2.addresses
def test_instance_deferred_cols(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
+ users, addresses = (self.tables.users,
+ self.tables.addresses)
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
eq_(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')]))
def test_pickle_protocols(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
+ users, addresses = (self.tables.users,
+ self.tables.addresses)
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
u2 = loads(dumps(u1))
eq_(u1, u2)
+
def test_options_with_descriptors(self):
- users, Address, addresses, User = (self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User)
+ users, addresses = (self.tables.users,
+ self.tables.addresses)
mapper(User, users, properties={
'addresses':relationship(Address, backref="user")
"""test a particular cycle that requires CollectionAdapter
to not rely upon InstanceState to deserialize."""
- global Child1, Child2, Parent, Screen
-
m = MetaData()
c1 = Table('c1', m,
Column('parent_id', String,
p = Table('p', m,
Column('id', String, primary_key=True)
)
- class Child1(_base.ComparableEntity):
- pass
-
- class Child2(_base.ComparableEntity):
- pass
-
- class Parent(_base.ComparableEntity):
- pass
mapper(Parent, p, properties={
'children1':relationship(Child1),
})
mapper(Child1, c1)
mapper(Child2, c2)
- class Screen(object):
- def __init__(self, obj, parent=None):
- self.obj = obj
- self.parent = parent
obj = Parent()
screen1 = Screen(obj)
Column('id', Integer, ForeignKey('users.id'), primary_key=True),
Column('email_address', String(30)))
- @classmethod
- def setup_classes(cls):
- global User, EmailUser
- class User(_base.BasicEntity):
- pass
-
- class EmailUser(User):
- pass
-
- @classmethod
- def teardown_class(cls):
- global User, EmailUser
- User, EmailUser = None, None
- super(PolymorphicDeferredTest, cls).teardown_class()
def test_polymorphic_deferred(self):
- EmailUser, email_users, users, User = (self.classes.EmailUser,
- self.tables.email_users,
+ email_users, users = (self.tables.email_users,
self.tables.users,
- self.classes.User)
+ )
mapper(User, users, polymorphic_identity='user', polymorphic_on=users.c.type)
mapper(EmailUser, email_users, inherits=User, polymorphic_identity='emailuser')
assert 'email_address' not in eu2.__dict__
eq_(eu2.email_address, 'foo@bar.com')
-class CustomSetupTeardownTest(_fixtures.FixtureTest):
+class TupleLabelTest(_fixtures.FixtureTest):
+ @classmethod
+ def setup_classes(cls):
+ pass
+
+ @classmethod
+ def setup_mappers(cls):
+ users, addresses, orders = cls.tables.users, cls.tables.addresses, cls.tables.orders
+ mapper(User, users, properties={
+ 'addresses':relationship(Address, backref='user', order_by=addresses.c.id),
+ 'orders':relationship(Order, backref='user', order_by=orders.c.id), # o2m, m2o
+ })
+ mapper(Address, addresses)
+ mapper(Order, orders, properties={
+ 'address':relationship(Address), # m2o
+ })
+
+ def test_tuple_labeling(self):
+ users = self.tables.users
+ sess = create_session()
+
+ # test pickle + all the protocols !
+ for pickled in False, -1, 0, 1, 2:
+ for row in sess.query(User, Address).join(User.addresses).all():
+ if pickled is not False:
+ row = pickle.loads(pickle.dumps(row, pickled))
+
+ eq_(row.keys(), ['User', 'Address'])
+ eq_(row.User, row[0])
+ eq_(row.Address, row[1])
+
+ for row in sess.query(User.name, User.id.label('foobar')):
+ if pickled is not False:
+ row = pickle.loads(pickle.dumps(row, pickled))
+ eq_(row.keys(), ['name', 'foobar'])
+ eq_(row.name, row[0])
+ eq_(row.foobar, row[1])
+
+ for row in sess.query(User).values(User.name, User.id.label('foobar')):
+ if pickled is not False:
+ row = pickle.loads(pickle.dumps(row, pickled))
+ eq_(row.keys(), ['name', 'foobar'])
+ eq_(row.name, row[0])
+ eq_(row.foobar, row[1])
+
+ oalias = aliased(Order)
+ for row in sess.query(User, oalias).join(User.orders).all():
+ if pickled is not False:
+ row = pickle.loads(pickle.dumps(row, pickled))
+ eq_(row.keys(), ['User'])
+ eq_(row.User, row[0])
+
+ oalias = aliased(Order, name='orders')
+ for row in sess.query(User, oalias).join(oalias, User.orders).all():
+ if pickled is not False:
+ row = pickle.loads(pickle.dumps(row, pickled))
+ eq_(row.keys(), ['User', 'orders'])
+ eq_(row.User, row[0])
+ eq_(row.orders, row[1])
+
+ # test here that first col is not labeled, only
+ # one name in keys, matches correctly
+ for row in sess.query(User.name + 'hoho', User.name):
+ eq_(row.keys(), ['name'])
+ eq_(row[0], row.name + 'hoho')
+
+ if pickled is not False:
+ ret = sess.query(User, Address).join(User.addresses).all()
+ pickle.loads(pickle.dumps(ret, pickled))
+
+class CustomSetupTeardownTest(_base.MappedTest):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('users', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(30), nullable=False),
+ test_needs_acid=True,
+ test_needs_fk=True
+ )
+
+ Table('addresses', metadata,
+ Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('user_id', None, ForeignKey('users.id')),
+ Column('email_address', String(50), nullable=False),
+ test_needs_acid=True,
+ test_needs_fk=True
+ )
def test_rebuild_state(self):
- User, users = self.classes.User, self.tables.users
+ users = self.tables.users
"""not much of a 'test', but illustrate how to
remove instance-level state before pickling.
attributes.manager_of_class(User).setup_instance(u2)
assert attributes.instance_state(u2)
-class UnpickleSA05Test(_fixtures.FixtureTest):
- """test loading picklestrings from SQLA 0.5."""
-
- __requires__ = ('python2',)
-
- def test_one(self):
- 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)
- data = \
- '\x80\x02]q\x00(ctest.orm._fixtures\nUser\nq\x01)\x81q\x02}q\x03(U\x12_sa_instance_stateq\x04csqlalchemy.orm.state\nInstanceState\nq\x05)\x81q\x06}q\x07(U\x08instanceq\x08h\x02U\x03keyq\th\x01K\x07\x85q\n\x86q\x0bubU\taddressesq\x0ccsqlalchemy.orm.collections\nInstrumentedList\nq\r)\x81q\x0ectest.orm._fixtures\nAddress\nq\x0f)\x81q\x10}q\x11(U\remail_addressq\x12X\r\x00\x00\x00jack@bean.comq\x13h\x04h\x05)\x81q\x14}q\x15(h\x08h\x10h\th\x0fK\x01\x85q\x16\x86q\x17ubU\x07user_idq\x18K\x07U\x02idq\x19K\x01uba}q\x1aU\x0b_sa_adapterq\x1bcsqlalchemy.orm.collections\nCollectionAdapter\nq\x1c)\x81q\x1d}q\x1e(U\x04dataq\x1fh\x0eU\x0bowner_stateq h\x06U\x03keyq!h\x0cubsbh\x19K\x07U\x04nameq"X\x04\x00\x00\x00jackq#ubh\x01)\x81q$}q%(h\x04h\x05)\x81q&}q\'(h\x08h$h\th\x01K\x08\x85q(\x86q)ubh\x0ch\r)\x81q*(h\x0f)\x81q+}q,(h\x12X\x0b\x00\x00\x00ed@wood.comq-h\x04h\x05)\x81q.}q/(h\x08h+h\th\x0fK\x02\x85q0\x86q1ubh\x18K\x08h\x19K\x02ubh\x0f)\x81q2}q3(h\x12X\x10\x00\x00\x00ed@bettyboop.comq4h\x04h\x05)\x81q5}q6(h\x08h2h\th\x0fK\x03\x85q7\x86q8ubh\x18K\x08h\x19K\x03ubh\x0f)\x81q9}q:(h\x12X\x0b\x00\x00\x00ed@lala.comq;h\x04h\x05)\x81q<}q=(h\x08h9h\th\x0fK\x04\x85q>\x86q?ubh\x18K\x08h\x19K\x04ube}q@h\x1bh\x1c)\x81qA}qB(h\x1fh*h h&h!h\x0cubsbh\x19K\x08h"X\x02\x00\x00\x00edqCubh\x01)\x81qD}qE(h\x04h\x05)\x81qF}qG(h\x08hDh\th\x01K\t\x85qH\x86qIubh\x0ch\r)\x81qJh\x0f)\x81qK}qL(h\x12X\r\x00\x00\x00fred@fred.comqMh\x04h\x05)\x81qN}qO(h\x08hKh\th\x0fK\x05\x85qP\x86qQubh\x18K\th\x19K\x05uba}qRh\x1bh\x1c)\x81qS}qT(h\x1fhJh hFh!h\x0cubsbh\x19K\th"X\x04\x00\x00\x00fredqUubh\x01)\x81qV}qW(h\x04h\x05)\x81qX}qY(h\x08hVh\th\x01K\n\x85qZ\x86q[ubh\x0ch\r)\x81q\\}q]h\x1bh\x1c)\x81q^}q_(h\x1fh\\h hXh!h\x0cubsbh\x19K\nh"X\x05\x00\x00\x00chuckq`ube.'
-
- sess = create_session()
- result = list(sess.query(User).merge_result(pickle.loads(data)))
- eq_(result, self.static.user_address_result)
-
- def test_two(self):
- 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)
- data = \
- '\x80\x02]q\x00(ctest.orm._fixtures\nUser\nq\x01)\x81q\x02}q\x03(U\x12_sa_instance_stateq\x04csqlalchemy.orm.state\nInstanceState\nq\x05)\x81q\x06}q\x07(U\x08instanceq\x08h\x02U\tload_pathq\t]q\nU\x03keyq\x0bh\x01K\x07\x85q\x0c\x86q\rU\x0cload_optionsq\x0ec__builtin__\nset\nq\x0f]q\x10csqlalchemy.orm.strategies\nEagerLazyOption\nq\x11)\x81q\x12}q\x13(U\x06mapperq\x14NU\x04lazyq\x15\x89U\x14propagate_to_loadersq\x16\x88U\x03keyq\x17]q\x18h\x01U\taddressesq\x19\x86q\x1aaU\x07chainedq\x1b\x89uba\x85q\x1cRq\x1dubh\x19csqlalchemy.orm.collections\nInstrumentedList\nq\x1e)\x81q\x1fctest.orm._fixtures\nAddress\nq )\x81q!}q"(U\remail_addressq#X\r\x00\x00\x00jack@bean.comq$h\x04h\x05)\x81q%}q&(h\x08h!h\t]q\'h\x01h\x19\x86q(ah\x0bh K\x01\x85q)\x86q*h\x0eh\x1dubU\x07user_idq+K\x07U\x02idq,K\x01uba}q-U\x0b_sa_adapterq.csqlalchemy.orm.collections\nCollectionAdapter\nq/)\x81q0}q1(U\x04dataq2h\x1fU\x0bowner_stateq3h\x06h\x17h\x19ubsbU\x04nameq4X\x04\x00\x00\x00jackq5h,K\x07ubh\x01)\x81q6}q7(h\x04h\x05)\x81q8}q9(h\x08h6h\t]q:h\x0bh\x01K\x08\x85q;\x86q<h\x0eh\x1dubh\x19h\x1e)\x81q=(h )\x81q>}q?(h#X\x0b\x00\x00\x00ed@wood.comq@h\x04h\x05)\x81qA}qB(h\x08h>h\t]qCh\x01h\x19\x86qDah\x0bh K\x02\x85qE\x86qFh\x0eh\x1dubh+K\x08h,K\x02ubh )\x81qG}qH(h#X\x10\x00\x00\x00ed@bettyboop.comqIh\x04h\x05)\x81qJ}qK(h\x08hGh\t]qLh\x01h\x19\x86qMah\x0bh K\x03\x85qN\x86qOh\x0eh\x1dubh+K\x08h,K\x03ubh )\x81qP}qQ(h#X\x0b\x00\x00\x00ed@lala.comqRh\x04h\x05)\x81qS}qT(h\x08hPh\t]qUh\x01h\x19\x86qVah\x0bh K\x04\x85qW\x86qXh\x0eh\x1dubh+K\x08h,K\x04ube}qYh.h/)\x81qZ}q[(h2h=h3h8h\x17h\x19ubsbh4X\x02\x00\x00\x00edq\\h,K\x08ubh\x01)\x81q]}q^(h\x04h\x05)\x81q_}q`(h\x08h]h\t]qah\x0bh\x01K\t\x85qb\x86qch\x0eh\x1dubh\x19h\x1e)\x81qdh )\x81qe}qf(h#X\r\x00\x00\x00fred@fred.comqgh\x04h\x05)\x81qh}qi(h\x08heh\t]qjh\x01h\x19\x86qkah\x0bh K\x05\x85ql\x86qmh\x0eh\x1dubh+K\th,K\x05uba}qnh.h/)\x81qo}qp(h2hdh3h_h\x17h\x19ubsbh4X\x04\x00\x00\x00fredqqh,K\tubh\x01)\x81qr}qs(h\x04h\x05)\x81qt}qu(h\x08hrh\t]qvh\x0bh\x01K\n\x85qw\x86qxh\x0eh\x1dubh\x19h\x1e)\x81qy}qzh.h/)\x81q{}q|(h2hyh3hth\x17h\x19ubsbh4X\x05\x00\x00\x00chuckq}h,K\nube.'
-
- sess = create_session()
- result = list(sess.query(User).merge_result(pickle.loads(data)))
- eq_(result, self.static.user_address_result)
from test.lib import testing, AssertsCompiledSQL, Column, engines
from test.orm import _fixtures
-from test.orm._fixtures import keywords, addresses, Base, \
- Keyword, FixtureTest, \
- Dingaling, item_keywords, dingalings, User, items,\
- orders, Address, users, nodes, \
- order_items, Item, Order, Node, \
- composite_pk_table, CompositePk
from test.orm import _base
@classmethod
def setup_mappers(cls):
+ Node, composite_pk_table, users, Keyword, items, Dingaling, \
+ order_items, item_keywords, Item, User, dingalings, \
+ Address, keywords, CompositePk, nodes, Order, orders, \
+ addresses = cls.classes.Node, \
+ cls.tables.composite_pk_table, cls.tables.users, \
+ cls.classes.Keyword, cls.tables.items, \
+ cls.classes.Dingaling, cls.tables.order_items, \
+ cls.tables.item_keywords, cls.classes.Item, \
+ cls.classes.User, cls.tables.dingalings, \
+ cls.classes.Address, cls.tables.keywords, \
+ cls.classes.CompositePk, cls.tables.nodes, \
+ cls.classes.Order, cls.tables.orders, cls.tables.addresses
+
mapper(User, users, properties={
'addresses':relationship(Address, backref='user', order_by=addresses.c.id),
'orders':relationship(Order, backref='user', order_by=orders.c.id), # o2m, m2o
run_setup_mappers = None
def test_custom_names(self):
+ User, users = self.classes.User, self.tables.users
+
mapper(User, users, properties={
'uname':users.c.name
})
assert row.uname == 'jack'
def test_column_metadata(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses)
sess = create_session()
class GetTest(QueryTest):
def test_get(self):
+ User = self.classes.User
+
s = create_session()
assert s.query(User).get(19) is None
u = s.query(User).get(7)
assert u is not u2
def test_get_composite_pk_no_result(self):
+ CompositePk = self.classes.CompositePk
+
s = Session()
assert s.query(CompositePk).get((100,100)) is None
def test_get_composite_pk_result(self):
+ CompositePk = self.classes.CompositePk
+
s = Session()
one_two = s.query(CompositePk).get((1,2))
assert one_two.i == 1
assert one_two.k == 3
def test_get_too_few_params(self):
+ CompositePk = self.classes.CompositePk
+
s = Session()
q = s.query(CompositePk)
assert_raises(sa_exc.InvalidRequestError, q.get, 7)
def test_get_too_few_params_tuple(self):
+ CompositePk = self.classes.CompositePk
+
s = Session()
q = s.query(CompositePk)
assert_raises(sa_exc.InvalidRequestError, q.get, (7,))
def test_get_too_many_params(self):
+ CompositePk = self.classes.CompositePk
+
s = Session()
q = s.query(CompositePk)
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()."""
)
def test_no_criterion(self):
+ User, Address = self.classes.User, self.classes.Address
+
"""test that get()/load() does not use preexisting filter/etc. criterion"""
s = create_session()
s.query(User).order_by(User.id).get(8)
def test_unique_param_names(self):
+ users = self.tables.users
+
class SomeUser(object):
pass
s = users.select(users.c.id!=12).alias('users')
assert sess.query(SomeUser).get(7).name == 'jack'
def test_load(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
assert s.query(User).populate_existing().get(19) is None
# end Py2K
table.insert().execute(id=ustring, data=ustring)
- class LocalFoo(Base):
+ class LocalFoo(self.classes.Base):
pass
mapper(LocalFoo, table)
eq_(create_session().query(LocalFoo).get(ustring),
metadata.drop_all()
def test_populate_existing(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
userlist = s.query(User).all()
@testing.fails_on_everything_except('sqlite', '+pyodbc', '+zxjdbc', 'mysql+oursql')
def test_query_str(self):
+ User = self.classes.User
+
s = create_session()
q = s.query(User).filter(User.id==1)
eq_(
class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
def test_no_limit_offset(self):
+ User = self.classes.User
+
s = create_session()
for q in (
q.enable_assertions(False).group_by('foo')
def test_no_from(self):
+ users, User = self.tables.users, self.classes.User
+
s = create_session()
q = s.query(User).select_from(users)
q.from_self()
def test_invalid_select_from(self):
+ User = self.classes.User
+
s = create_session()
q = s.query(User)
assert_raises(sa_exc.ArgumentError, q.select_from, User.id==5)
assert_raises(sa_exc.ArgumentError, q.select_from, User.id)
def test_invalid_from_statement(self):
+ User, addresses, users = (self.classes.User,
+ self.tables.addresses,
+ self.tables.users)
+
s = create_session()
q = s.query(User)
assert_raises(sa_exc.ArgumentError, q.from_statement, User.id==5)
assert_raises(sa_exc.ArgumentError, q.from_statement, users.join(addresses))
def test_invalid_column(self):
+ User = self.classes.User
+
s = create_session()
q = s.query(User)
assert_raises(sa_exc.InvalidRequestError, q.add_column, object())
def test_distinct(self):
+ User = self.classes.User
+
"""test that a distinct() call is not valid before 'clauseelement' conditions."""
s = create_session()
assert_raises(sa_exc.InvalidRequestError, q.with_polymorphic, User)
def test_order_by(self):
+ User = self.classes.User
+
"""test that an order_by() call is not valid before 'clauseelement' conditions."""
s = create_session()
assert_raises(sa_exc.InvalidRequestError, q.with_polymorphic, User)
def test_cancel_order_by(self):
+ User = self.classes.User
+
s = create_session()
q = s.query(User).order_by(User.id)
q._no_select_modifiers("foo")
def test_mapper_zero(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
q = s.query(User, Address)
assert_raises(sa_exc.InvalidRequestError, q.get, 5)
def test_from_statement(self):
+ User = self.classes.User
+
s = create_session()
q = s.query(User).filter(User.id==5)
self.assert_compile(clause, expected, dialect=default.DefaultDialect())
def test_arithmetic(self):
+ User = self.classes.User
+
create_session().query(User)
for (py_op, sql_op) in ((operator.add, '+'), (operator.mul, '*'),
(operator.sub, '-'),
self._test(py_op(lhs, rhs), res % sql_op)
def test_comparison(self):
+ User = self.classes.User
+
create_session().query(User)
ualias = aliased(User)
fwd_sql + "'\n or\n'" + rev_sql + "'")
def test_negated_null(self):
+ User, Address = self.classes.User, self.classes.Address
+
self._test(User.id == None, "users.id IS NULL")
self._test(~(User.id==None), "users.id IS NOT NULL")
self._test(None == User.id, "users.id IS NULL")
self._test(~(None == Address.user), "addresses.user_id IS NOT NULL")
def test_relationship(self):
+ User, Address = self.classes.User, self.classes.Address
+
self._test(User.addresses.any(Address.id==17),
"EXISTS (SELECT 1 "
"FROM addresses "
self._test(Address.user != None, "addresses.user_id IS NOT NULL")
def test_selfref_relationship(self):
+ Node = self.classes.Node
+
nalias = aliased(Node)
# auto self-referential aliasing
)
def test_op(self):
+ User = self.classes.User
+
self._test(User.name.op('ilike')('17'), "users.name ilike :name_1")
def test_in(self):
- self._test(User.id.in_(['a', 'b']),
+ User = self.classes.User
+
+ self._test(User.id.in_(['a', 'b']),
"users.id IN (:id_1, :id_2)")
def test_in_on_relationship_not_supported(self):
+ User, Address = self.classes.User, self.classes.Address
+
assert_raises(NotImplementedError, Address.user.in_, [User(id=5)])
def test_neg(self):
+ User = self.classes.User
+
self._test(-User.id, "-users.id")
self._test(User.id + -User.id, "users.id + -users.id")
def test_between(self):
+ User = self.classes.User
+
self._test(User.id.between('a', 'b'),
"users.id BETWEEN :id_1 AND :id_2")
def test_selfref_between(self):
+ User = self.classes.User
+
ualias = aliased(User)
self._test(User.id.between(ualias.id, ualias.id), "users.id BETWEEN users_1.id AND users_1.id")
self._test(ualias.id.between(User.id, User.id), "users_1.id BETWEEN users.id AND users.id")
def test_clauses(self):
+ User, Address = self.classes.User, self.classes.Address
+
for (expr, compare) in (
(func.max(User.id), "max(users.id)"),
(User.id.desc(), "users.id DESC"),
class ExpressionTest(QueryTest, AssertsCompiledSQL):
def test_deferred_instances(self):
+ User, addresses, Address = (self.classes.User,
+ self.tables.addresses,
+ self.classes.Address)
+
session = create_session()
s = session.query(User).filter(and_(addresses.c.email_address == bindparam('emailad'),
Address.user_id==User.id)).statement
eq_([User(id=7)], l)
def test_aliased_sql_construct(self):
+ User, Address = self.classes.User, self.classes.Address
+
j = join(User, Address)
a1 = aliased(j)
self.assert_compile(
)
def test_scalar_subquery(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==7).subquery()
eq_(User(id=7), q.one())
def test_named_subquery(self):
+ User = self.classes.User
+
session = create_session()
a1 = session.query(User.id).filter(User.id==7).subquery('foo1')
a2 = session.query(User.id).filter(User.id==7).subquery(name='foo2')
def test_label(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==7).label('foo')
)
def test_as_scalar(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==7).as_scalar()
def test_param_transfer(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==bindparam('foo')).params(foo=7).subquery()
eq_(User(id=7), q.one())
def test_in(self):
+ User, Address = self.classes.User, self.classes.Address
+
session = create_session()
s = session.query(User.id).join(User.addresses).group_by(User.id).having(func.count(Address.id) > 2)
eq_(
)
def test_union(self):
+ User = self.classes.User
+
s = create_session()
q1 = s.query(User).filter(User.name=='ed').with_labels()
)
def test_select(self):
+ User = self.classes.User
+
s = create_session()
# this is actually not legal on most DBs since the subquery has no alias
)
def test_join(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
# TODO: do we want aliased() to detect a query and convert to subquery()
# more slice tests are available in test/orm/generative.py
class SliceTest(QueryTest):
def test_first(self):
+ User = self.classes.User
+
assert User(id=7) == create_session().query(User).first()
assert create_session().query(User).filter(User.id==27).first() is None
@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
class FilterTest(QueryTest):
def test_basic(self):
+ User = self.classes.User
+
assert [User(id=7), User(id=8), User(id=9),User(id=10)] == create_session().query(User).all()
@testing.fails_on('maxdb', 'FIXME: unknown')
def test_limit(self):
+ User = self.classes.User
+
assert [User(id=8), User(id=9)] == create_session().query(User).order_by(User.id).limit(2).offset(1).all()
assert [User(id=8), User(id=9)] == list(create_session().query(User).order_by(User.id)[1:3])
@testing.requires.boolean_col_expressions
def test_exists(self):
+ User = self.classes.User
+
sess = create_session(testing.db)
assert sess.query(exists().where(User.id==9)).scalar()
assert not sess.query(exists().where(User.id==29)).scalar()
def test_one_filter(self):
+ User = self.classes.User
+
assert [User(id=8), User(id=9)] == create_session().query(User).filter(User.name.endswith('ed')).all()
def test_contains(self):
+ User, Address = self.classes.User, self.classes.Address
+
"""test comparing a collection to an object instance."""
sess = create_session()
#assert [User(id=7), User(id=9), User(id=10)] == sess.query(User).filter(User.addresses!=address).all()
def test_any(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
assert [User(id=8), User(id=9)] == sess.query(User).filter(User.addresses.any(Address.email_address.like('%ed%'))).all()
@testing.crashes('maxdb', 'can dump core')
def test_has(self):
+ Dingaling, User, Address = (self.classes.Dingaling,
+ self.classes.User,
+ self.classes.Address)
+
sess = create_session()
assert [Address(id=5)] == sess.query(Address).filter(Address.user.has(name='fred')).all()
assert [User(id=9)] == sess.query(User).filter(User.addresses.any(Address.dingaling==dingaling)).all()
def test_contains_m2m(self):
+ Item, Order = self.classes.Item, self.classes.Order
+
sess = create_session()
item = sess.query(Item).get(3)
assert [Order(id=1), Order(id=2), Order(id=3)] == sess.query(Order).filter(Order.items.contains(item)).all()
def test_comparison(self):
+ 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()
eq_(sess.query(Item).filter(Item.keywords!=None).order_by(Item.id).all(), [Item(id=1),Item(id=2), Item(id=3)])
def test_filter_by(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
user = sess.query(User).get(8)
assert [Address(id=2), Address(id=3), Address(id=4)] == sess.query(Address).filter_by(user=user).all()
assert [User(name='chuck')] == sess.query(User).filter_by(addresses = null()).all()
def test_none_comparison(self):
+ Order, User, Address = (self.classes.Order,
+ self.classes.User,
+ self.classes.Address)
+
sess = create_session()
# scalar
eq_([User(id=7),User(id=8),User(id=9)], sess.query(User).filter(User.addresses!=None).order_by(User.id).all())
def test_blank_filter_by(self):
+ User = self.classes.User
+
eq_(
[(7,), (8,), (9,), (10,)],
create_session().query(User.id).filter_by().order_by(User.id).all()
class SetOpsTest(QueryTest, AssertsCompiledSQL):
def test_union(self):
+ User = self.classes.User
+
s = create_session()
fred = s.query(User).filter(User.name=='fred')
)
def test_statement_labels(self):
+ User, Address = self.classes.User, self.classes.Address
+
"""test that label conflicts don't occur with joins etc."""
s = create_session()
)
def test_union_literal_expressions_compile(self):
+ User = self.classes.User
+
"""test that column expressions translate during
the _from_statement() portion of union(), others"""
)
def test_union_literal_expressions_results(self):
+ User = self.classes.User
+
s = Session()
q1 = s.query(User, literal("x"))
)
def test_union_labeled_anonymous_columns(self):
+ User = self.classes.User
+
s = Session()
c1, c2 = column('c1'), column('c2')
)
def test_union_mapped_colnames_preserved_across_subquery(self):
+ User = self.classes.User
+
s = Session()
q1 = s.query(User.name)
q2 = s.query(User.name)
@testing.fails_on('mysql', "mysql doesn't support intersect")
def test_intersect(self):
+ User = self.classes.User
+
s = create_session()
fred = s.query(User).filter(User.name=='fred')
)
def test_eager_load(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
fred = s.query(User).filter(User.name=='fred')
class AggregateTest(QueryTest):
def test_sum(self):
+ Order = self.classes.Order
+
sess = create_session()
orders = sess.query(Order).filter(Order.id.in_([2, 3, 4]))
eq_(orders.values(func.sum(Order.user_id * Order.address_id)).next(), (79,))
eq_(orders.value(func.sum(Order.user_id * Order.address_id)), 79)
def test_apply(self):
+ Order = self.classes.Order
+
sess = create_session()
assert sess.query(func.sum(Order.user_id * Order.address_id)).filter(Order.id.in_([2, 3, 4])).one() == (79,)
def test_having(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
assert [User(name=u'ed',id=8)] == sess.query(User).order_by(User.id).group_by(User).join('addresses').having(func.count(Address.id)> 2).all()
class CountTest(QueryTest):
def test_basic(self):
+ users, User = self.tables.users, self.classes.User
+
s = create_session()
eq_(s.query(User).count(), 4)
eq_(s.query(User).filter(users.c.name.endswith('ed')).count(), 2)
def test_multiple_entity(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
q = s.query(User, Address)
eq_(q.count(), 20) # cartesian product
eq_(q.count(), 5)
def test_nested(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
q = s.query(User, Address).limit(2)
eq_(q.count(), 2)
eq_(q.count(), 5)
def test_cols(self):
+ User, Address = self.classes.User, self.classes.Address
+
"""test that column-based queries always nest."""
s = create_session()
class DistinctTest(QueryTest):
def test_basic(self):
+ User = self.classes.User
+
eq_(
[User(id=7), User(id=8), User(id=9),User(id=10)],
create_session().query(User).order_by(User.id).distinct().all()
)
def test_joined(self):
+ 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()
class YieldTest(QueryTest):
def test_basic(self):
+ User = self.classes.User
+
sess = create_session()
q = iter(sess.query(User).yield_per(1).from_statement("select * from users"))
class HintsTest(QueryTest, AssertsCompiledSQL):
def test_hints(self):
+ User = self.classes.User
+
from sqlalchemy.dialects import mysql
dialect = mysql.dialect()
class TextTest(QueryTest):
def test_fulltext(self):
+ User = self.classes.User
+
assert [User(id=7), User(id=8), User(id=9),User(id=10)] == create_session().query(User).from_statement("select * from users order by id").all()
assert User(id=7) == create_session().query(User).from_statement("select * from users order by id").first()
assert None == create_session().query(User).from_statement("select * from users where name='nonexistent'").first()
def test_fragment(self):
+ User = self.classes.User
+
assert [User(id=8), User(id=9)] == create_session().query(User).filter("id in (8, 9)").all()
assert [User(id=9)] == create_session().query(User).filter("name='fred'").filter("id=9").all()
assert [User(id=9)] == create_session().query(User).filter("name='fred'").filter(User.id==9).all()
def test_binds(self):
+ User = self.classes.User
+
assert [User(id=8), User(id=9)] == create_session().query(User).filter("id in (:id1, :id2)").params(id1=8, id2=9).all()
def test_as_column(self):
+ User = self.classes.User
+
s = create_session()
assert_raises(sa_exc.InvalidRequestError, s.query, User.id, text("users.name"))
class ParentTest(QueryTest):
def test_o2m(self):
+ User, orders, Order = (self.classes.User,
+ self.tables.orders,
+ self.classes.Order)
+
sess = create_session()
q = sess.query(User)
#assert sess.query(Order).with_parent(None, property='addresses').all() == [Order(description="order 5")]
def test_noparent(self):
+ Item, User = self.classes.Item, self.classes.User
+
sess = create_session()
q = sess.query(User)
"instances of class 'Item' to instances of class 'User'"
def test_m2m(self):
+ Item, Keyword = self.classes.Item, self.classes.Keyword
+
sess = create_session()
i1 = sess.query(Item).filter_by(id=2).one()
k = sess.query(Keyword).with_parent(i1).all()
assert [Keyword(name='red'), Keyword(name='small'), Keyword(name='square')] == k
def test_with_transient(self):
+ User, Order = self.classes.User, self.classes.Order
+
sess = Session()
q = sess.query(User)
)
def test_with_pending_autoflush(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = Session()
o1 = sess.query(Order).first()
)
def test_with_pending_no_autoflush(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = Session(autoflush=False)
o1 = sess.query(Order).first()
@classmethod
def setup_mappers(cls):
+ users, Keyword, items, order_items, orders, Item, User, \
+ Address, keywords, Order, item_keywords, addresses = \
+ cls.tables.users, cls.classes.Keyword, cls.tables.items, \
+ cls.tables.order_items, cls.tables.orders, \
+ cls.classes.Item, cls.classes.User, cls.classes.Address, \
+ cls.tables.keywords, cls.classes.Order, \
+ cls.tables.item_keywords, cls.tables.addresses
+
mapper(User, users, properties={
'name_syn':synonym('name'),
'addresses':relationship(Address),
@testing.fails_if(lambda: True, "0.7 regression, may not support "
"synonyms for relationship")
def test_joins(self):
+ User = self.classes.User
+
for j in (
['orders', 'items'],
['orders_syn', 'items'],
@testing.fails_if(lambda: True, "0.7 regression, may not support "
"synonyms for relationship")
def test_with_parent(self):
+ Order, User = self.classes.Order, self.classes.User
+
for nameprop, orderprop in (
('name', 'orders'),
('name_syn', 'orders'),
class StatementOptionsTest(QueryTest):
def test_query_with_statement_option(self):
+ User = self.classes.User
+
sess = create_session(bind=testing.db, autocommit=False)
q1 = sess.query(User)
)
def test_get_path_one_level_string(self):
+ User = self.classes.User
+
sess = Session()
q = sess.query(User)
self._assert_path_result(opt, q, [(User, 'addresses')], [User])
def test_get_path_one_level_attribute(self):
+ User = self.classes.User
+
sess = Session()
q = sess.query(User)
self._assert_path_result(opt, q, [(User, 'addresses')], [User])
def test_path_on_entity_but_doesnt_match_currentpath(self):
+ User, Address = self.classes.User, self.classes.Address
+
# ensure "current path" is fully consumed before
# matching against current entities.
# see [ticket:2098]
self._assert_path_result(opt, q, [], [])
def test_get_path_one_level_with_unrelated(self):
+ Order = self.classes.Order
+
sess = Session()
q = sess.query(Order)
self._assert_path_result(opt, q, [], [])
def test_path_multilevel_string(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = Session()
q = sess.query(User)
[User, Order, Item])
def test_path_multilevel_attribute(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = Session()
q = sess.query(User)
[User, Order, Item])
def test_with_current_matching_string(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = Session()
q = sess.query(Item)._with_current_path(
self._make_path([User, 'orders', Order, 'items'])
], [Item])
def test_with_current_matching_attribute(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = Session()
q = sess.query(Item)._with_current_path(
self._make_path([User, 'orders', Order, 'items'])
], [Item])
def test_with_current_nonmatching_string(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = Session()
q = sess.query(Item)._with_current_path(
self._make_path([User, 'orders', Order, 'items'])
self._assert_path_result(opt, q, [], [])
def test_with_current_nonmatching_attribute(self):
+ Item, User, Order = (self.classes.Item,
+ self.classes.User,
+ self.classes.Order)
+
sess = Session()
q = sess.query(Item)._with_current_path(
self._make_path([User, 'orders', Order, 'items'])
self._assert_path_result(opt, q, [], [])
def test_from_base_to_subclass_attr(self):
+ Dingaling, Address = self.classes.Dingaling, self.classes.Address
+
sess = Session()
class SubAddr(Address):
pass
self._assert_path_result(opt, q, [(Address, 'flub')], [SubAddr])
def test_from_subclass_to_subclass_attr(self):
+ Dingaling, Address = self.classes.Dingaling, self.classes.Address
+
sess = Session()
class SubAddr(Address):
pass
self._assert_path_result(opt, q, [(SubAddr, 'flub')], [SubAddr])
def test_from_base_to_base_attr_via_subclass(self):
+ Dingaling, Address = self.classes.Dingaling, self.classes.Address
+
sess = Session()
class SubAddr(Address):
pass
self._assert_path_result(opt, q, [(Address, 'user')], [Address])
def test_of_type(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = Session()
class SubAddr(Address):
pass
], [User, Address])
def test_of_type_plus_level(self):
+ Dingaling, User, Address = (self.classes.Dingaling,
+ self.classes.User,
+ self.classes.Address)
+
sess = Session()
class SubAddr(Address):
pass
], [User, SubAddr])
def test_aliased_single(self):
+ User = self.classes.User
+
sess = Session()
ualias = aliased(User)
q = sess.query(ualias)
self._assert_path_result(opt, q, [(ualias, 'addresses')], [User])
def test_with_current_aliased_single(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = Session()
ualias = aliased(User)
q = sess.query(ualias)._with_current_path(
self._assert_path_result(opt, q, [(ualias, 'addresses')], [User])
def test_with_current_aliased_single_nonmatching_option(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = Session()
ualias = aliased(User)
q = sess.query(User)._with_current_path(
@testing.fails_if(lambda: True, "Broken feature")
def test_with_current_aliased_single_nonmatching_entity(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = Session()
ualias = aliased(User)
q = sess.query(ualias)._with_current_path(
opt = self._option_fixture(Address.user, User.addresses)
self._assert_path_result(opt, q, [], [])
-class OptionsNoPropTest(_base.MappedTest):
+class OptionsNoPropTest(_fixtures.FixtureTest):
"""test the error messages emitted when using property
options in conjunection with column-only entities.
"""
+ run_create_tables = False
+ run_inserts = None
+ run_deletes = None
+
def test_option_with_mapper_using_basestring(self):
+ Item = self.classes.Item
+
self._assert_option([Item], 'keywords')
def test_option_with_mapper_using_PropCompatator(self):
+ Item = self.classes.Item
+
self._assert_option([Item], Item.keywords)
def test_option_with_mapper_then_column_using_basestring(self):
+ Item = self.classes.Item
+
self._assert_option([Item, Item.id], 'keywords')
def test_option_with_mapper_then_column_using_PropComparator(self):
+ Item = self.classes.Item
+
self._assert_option([Item, Item.id], Item.keywords)
def test_option_with_column_then_mapper_using_basestring(self):
+ Item = self.classes.Item
+
self._assert_option([Item.id, Item], 'keywords')
def test_option_with_column_then_mapper_using_PropComparator(self):
+ Item = self.classes.Item
+
self._assert_option([Item.id, Item], Item.keywords)
def test_option_with_column_using_basestring(self):
+ Item = self.classes.Item
+
message = \
"Can't find property named 'keywords' on the first mapped "\
"entity in this Query. Consider using an attribute object "\
'keywords', message)
def test_option_with_column_using_PropComparator(self):
+ Item = self.classes.Item
+
message = \
"Can't find property 'keywords' on any entity specified "\
"in this Query\."
self._assert_eager_with_just_column_exception(Item.id,
Item.keywords, message)
- @classmethod
- def define_tables(cls, metadata):
- pass
-
@classmethod
def setup_mappers(cls):
+ keywords, items, item_keywords, Keyword, Item = (cls.tables.keywords,
+ cls.tables.items,
+ cls.tables.item_keywords,
+ cls.classes.Keyword,
+ cls.classes.Item)
+
mapper(Keyword, keywords)
mapper(Item, items,
properties=dict(keywords=relationship(Keyword,
secondary=item_keywords)))
def _assert_option(self, entity_list, option):
+ Item = self.classes.Item
+
q = create_session().query(*entity_list).\
options(eagerload(option))
key = ('loaderstrategy', (class_mapper(Item), 'keywords'))
from test.lib.testing import eq_, assert_raises, \
assert_raises_message
from test.lib.util import gc_collect
+from test.lib import pickleable
+from sqlalchemy.util import pickle
import inspect
-import pickle
from sqlalchemy.orm import create_session, sessionmaker, attributes, \
make_transient, Session
import sqlalchemy as sa
seq.drop(testing.db)
- def test_expunge_cascade(self):
- Address, addresses, users, User = (self.classes.Address,
- self.tables.addresses,
- self.tables.users,
- self.classes.User)
-
- mapper(Address, addresses)
- mapper(User, users, properties={
- 'addresses':relationship(Address,
- backref=backref("user", cascade="all"),
- cascade="all")})
-
- _fixtures.run_inserts_for(users)
- _fixtures.run_inserts_for(addresses)
-
- session = create_session()
- u = session.query(User).filter_by(id=7).one()
-
- # get everything to load in both directions
- print [a.user for a in u.addresses]
-
- # then see if expunge fails
- session.expunge(u)
-
- assert sa.orm.object_session(u) is None
- assert sa.orm.attributes.instance_state(u).session_id is None
- for a in u.addresses:
- assert sa.orm.object_session(a) is None
- assert sa.orm.attributes.instance_state(a).session_id is None
@engines.close_open_connections
def test_mapped_binds(self):
).scalar() == 1
sess.commit()
- def test_autoflush_rollback(self):
- Address, addresses, users, User = (self.classes.Address,
- self.tables.addresses,
- self.tables.users,
- self.classes.User)
-
- mapper(Address, addresses)
- mapper(User, users, properties={
- 'addresses':relationship(Address)})
-
- _fixtures.run_inserts_for(users)
- _fixtures.run_inserts_for(addresses)
-
- sess = create_session(autocommit=False, autoflush=True)
- u = sess.query(User).get(8)
- newad = Address(email_address='a new address')
- u.addresses.append(newad)
- u.name = 'some new name'
- assert u.name == 'some new name'
- assert len(u.addresses) == 4
- assert newad in u.addresses
- sess.rollback()
- assert u.name == 'ed'
- assert len(u.addresses) == 3
-
- assert newad not in u.addresses
- # pending objects dont get expired
- assert newad.email_address == 'a new address'
def test_autocommit_doesnt_raise_on_pending(self):
User, users = self.classes.User, self.tables.users
assert s.identity_map
def test_weak_ref_pickled(self):
- users, User = self.tables.users, self.classes.User
+ users, User = self.tables.users, pickleable.User
s = create_session()
mapper(User, users)
def test_pickled_update(self):
- users, User = self.tables.users, self.classes.User
+ users, User = self.tables.users, pickleable.User
mapper(User, users)
sess1 = create_session()
del u3
gc_collect()
+class SessionDataTest(_fixtures.FixtureTest):
+ def test_expunge_cascade(self):
+ Address, addresses, users, User = (self.classes.Address,
+ self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
+ mapper(Address, addresses)
+ mapper(User, users, properties={
+ 'addresses':relationship(Address,
+ backref=backref("user", cascade="all"),
+ cascade="all")})
+
+ session = create_session()
+ u = session.query(User).filter_by(id=7).one()
+
+ # get everything to load in both directions
+ print [a.user for a in u.addresses]
+
+ # then see if expunge fails
+ session.expunge(u)
+
+ assert sa.orm.object_session(u) is None
+ assert sa.orm.attributes.instance_state(u).session_id is None
+ for a in u.addresses:
+ assert sa.orm.object_session(a) is None
+ assert sa.orm.attributes.instance_state(a).session_id is None
+
+ def test_autoflush_rollback(self):
+ Address, addresses, users, User = (self.classes.Address,
+ self.tables.addresses,
+ self.tables.users,
+ self.classes.User)
+
+ mapper(Address, addresses)
+ mapper(User, users, properties={
+ 'addresses':relationship(Address)})
+
+ sess = create_session(autocommit=False, autoflush=True)
+ u = sess.query(User).get(8)
+ newad = Address(email_address='a new address')
+ u.addresses.append(newad)
+ u.name = 'some new name'
+ assert u.name == 'some new name'
+ assert len(u.addresses) == 4
+ assert newad in u.addresses
+ sess.rollback()
+ assert u.name == 'ed'
+ assert len(u.addresses) == 3
+
+ assert newad not in u.addresses
+ # pending objects dont get expired
+ assert newad.email_address == 'a new address'
+
+
class DisposedStates(_base.MappedTest):
run_setup_mappers = 'once'
run_inserts = 'once'
from test.lib.util import gc_collect
from test.lib import testing
from test.orm import _base
-from test.orm._fixtures import FixtureTest, User, Address, users, addresses
+from test.orm._fixtures import FixtureTest
class TransactionTest(FixtureTest):
run_setup_mappers = 'once'
@classmethod
def setup_mappers(cls):
+ User, Address = cls.classes.User, cls.classes.Address
+ users, addresses = cls.tables.users, cls.tables.addresses
mapper(User, users, properties={
'addresses':relationship(Address, backref='user',
cascade="all, delete-orphan", order_by=addresses.c.id),
run_inserts = 'each'
def test_attrs_on_rollback(self):
+ User = self.classes.User
sess = self.session()
u1 = sess.query(User).get(7)
u1.name = 'ed'
eq_(u1.name, 'jack')
def test_commit_persistent(self):
+ User = self.classes.User
sess = self.session()
u1 = sess.query(User).get(7)
u1.name = 'ed'
eq_(u1.name, 'ed')
def test_concurrent_commit_persistent(self):
+ User = self.classes.User
s1 = self.session()
u1 = s1.query(User).get(7)
u1.name = 'ed'
class AutoExpireTest(TransactionTest):
def test_expunge_pending_on_rollback(self):
+ User = self.classes.User
sess = self.session()
u2= User(name='newuser')
sess.add(u2)
assert u2 not in sess
def test_trans_pending_cleared_on_commit(self):
+ User = self.classes.User
sess = self.session()
u2= User(name='newuser')
sess.add(u2)
assert u2 in sess
def test_update_deleted_on_rollback(self):
+ User = self.classes.User
s = self.session()
u1 = User(name='ed')
s.add(u1)
assert u1 not in s.deleted
def test_gced_delete_on_rollback(self):
+ User, users = self.classes.User, self.tables.users
+
s = self.session()
u1 = User(name='ed')
s.add(u1)
s.commit()
def test_trans_deleted_cleared_on_rollback(self):
+ User = self.classes.User
s = self.session()
u1 = User(name='ed')
s.add(u1)
assert u1 not in s
def test_update_deleted_on_rollback_cascade(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = self.session()
u1 = User(name='ed', addresses=[Address(email_address='foo')])
s.add(u1)
assert u1.addresses[0] not in s.deleted
def test_update_deleted_on_rollback_orphan(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = self.session()
u1 = User(name='ed', addresses=[Address(email_address='foo')])
s.add(u1)
assert u1.addresses == [a1]
def test_commit_pending(self):
+ User = self.classes.User
sess = self.session()
u1 = User(name='newuser')
sess.add(u1)
def test_concurrent_commit_pending(self):
+ User = self.classes.User
s1 = self.session()
u1 = User(name='edward')
s1.add(u1)
@testing.requires.two_phase_transactions
def test_rollback_on_prepare(self):
+ User = self.classes.User
s = self.session(twophase=True)
u = User(name='ed')
class RollbackRecoverTest(TransactionTest):
def test_pk_violation(self):
+ User, Address = self.classes.User, self.classes.Address
s = self.session()
a1 = Address(email_address='foo')
u1 = User(id=1, name='ed', addresses=[a1])
@testing.requires.savepoints
def test_pk_violation_with_savepoint(self):
+ User, Address = self.classes.User, self.classes.Address
s = self.session()
a1 = Address(email_address='foo')
u1 = User(id=1, name='ed', addresses=[a1])
@testing.requires.savepoints
def test_savepoint_rollback_collections(self):
+ User, Address = self.classes.User, self.classes.Address
s = self.session()
u1 = User(name='ed', addresses=[Address(email_address='foo')])
s.add(u1)
@testing.requires.savepoints
def test_savepoint_commit_collections(self):
+ User, Address = self.classes.User, self.classes.Address
s = self.session()
u1 = User(name='ed', addresses=[Address(email_address='foo')])
s.add(u1)
class AccountingFlagsTest(TransactionTest):
def test_no_expire_on_commit(self):
+ User, users = self.classes.User, self.tables.users
+
sess = sessionmaker(expire_on_commit=False)()
u1 = User(name='ed')
sess.add(u1)
assert u1.name == 'edward'
def test_rollback_no_accounting(self):
+ User, users = self.classes.User, self.tables.users
+
sess = sessionmaker(_enable_transaction_accounting=False)()
u1 = User(name='ed')
sess.add(u1)
assert u1.name == 'edward'
def test_commit_no_accounting(self):
+ User, users = self.classes.User, self.tables.users
+
sess = sessionmaker(_enable_transaction_accounting=False)()
u1 = User(name='ed')
sess.add(u1)
sess.commit()
def test_preflush_no_accounting(self):
+ User, users = self.classes.User, self.tables.users
+
sess = Session(_enable_transaction_accounting=False,
autocommit=True, autoflush=False)
u1 = User(name='ed')
assert_raises(sa_exc.InvalidRequestError, sess.begin_nested)
def test_begin_preflush(self):
+ User = self.classes.User
sess = create_session(autocommit=True)
u1 = User(name='ed')
assert sess.query(User).filter_by(name='ed').one() is u1
def test_accounting_commit_fails_add(self):
+ User = self.classes.User
sess = create_session(autocommit=True)
fail = False
)
def test_accounting_commit_fails_delete(self):
+ User = self.classes.User
sess = create_session(autocommit=True)
fail = False
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)
u1 = User(id=1, name='ed')
Session
from test.lib.assertsql import AllOf, CompiledSQL
-from test.orm._fixtures import keywords, addresses, Base, Keyword, \
- Dingaling, item_keywords, dingalings, User, items,\
- orders, Address, users, nodes, \
- order_items, Item, Order, Node, \
- composite_pk_table, CompositePk
-
class AssertsUOW(object):
def _get_test_uow(self, session):
uow = unitofwork.UOWTransaction(session)
uow.register_object(d, isdelete=True)
return uow
- def _assert_uow_size(self,
- session,
- expected
- ):
+ def _assert_uow_size(self, session, expected ):
uow = self._get_test_uow(session)
postsort_actions = uow._generate_actions()
print postsort_actions
class RudimentaryFlushTest(UOWTest):
def test_one_to_many_save(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(Address),
})
)
def test_one_to_many_delete_all(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(Address),
})
)
def test_one_to_many_delete_parent(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(Address),
})
)
def test_many_to_one_save(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
)
def test_many_to_one_delete_all(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
'user':relationship(User)
)
def test_many_to_one_delete_target(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
'user':relationship(User)
)
def test_many_to_one_delete_unloaded(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
'parent':relationship(User)
)
def test_many_to_one_delete_childonly_unloaded(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
'parent':relationship(User)
)
def test_many_to_one_delete_childonly_unloaded_expired(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
'parent':relationship(User)
)
def test_natural_ordering(self):
+ 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)
)
def test_natural_selfref(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
"""test that unconnected items take relationship() into account regardless."""
mapper(Node, nodes, properties={
)
def test_many_to_many(self):
+ keywords, items, item_keywords, Keyword, Item = (self.tables.keywords,
+ self.tables.items,
+ self.tables.item_keywords,
+ self.classes.Keyword,
+ self.classes.Item)
+
mapper(Item, items, properties={
'keywords':relationship(Keyword, secondary=item_keywords)
})
)
def test_m2o_flush_size(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users)
mapper(Address, addresses, properties={
'user':relationship(User, passive_updates=True)
self._assert_uow_size(sess, 2)
def test_o2m_flush_size(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+
mapper(User, users, properties={
'addresses':relationship(Address),
})
class SingleCycleTest(UOWTest):
def test_one_to_many_save(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
)
def test_one_to_many_delete_all(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
)
def test_one_to_many_delete_parent(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
)
def test_many_to_one_save(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'parent':relationship(Node, remote_side=nodes.c.id)
})
)
def test_many_to_one_delete_all(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'parent':relationship(Node, remote_side=nodes.c.id)
})
)
def test_cycle_rowswitch(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
sess.flush()
def test_bidirectional_mutations_one(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node,
backref=backref('parent',
sess.flush()
def test_bidirectional_multilevel_save(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node,
backref=backref('parent', remote_side=nodes.c.id)
)
def test_singlecycle_flush_size(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
self._assert_uow_size(sess, 2)
def test_delete_unloaded_m2o(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'parent':relationship(Node, remote_side=nodes.c.id)
})
def test_flush_size(self):
foobars, nodes = self.tables.foobars, self.tables.nodes
- class Node(Base):
+ class Node(_base.ComparableEntity):
pass
- class FooBar(Base):
+ class FooBar(_base.ComparableEntity):
pass
mapper(Node, nodes, properties={
def test_many_to_many_one(self):
nodes, node_to_nodes = self.tables.nodes, self.tables.node_to_nodes
- class Node(Base):
+ class Node(_base.ComparableEntity):
pass
mapper(Node, nodes, properties={
key present statements together.
"""
- class T(Base):
+ class T(_base.ComparableEntity):
pass
mapper(T, t)
sess = Session()
l.append({'c2':'this is t2 #%d' % y, 't1id':x})
t2.insert().execute(*l)
- class T1(_fixtures.Base):
+ class T1(_base.ComparableEntity):
pass
- class T2(_fixtures.Base):
+ class T2(_base.ComparableEntity):
pass
mapper(T1, t1, properties={
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
