summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--test/engine/_base.py24
-rw-r--r--test/lib/pickleable.py31
-rw-r--r--test/orm/_fixtures.py102
-rw-r--r--test/orm/inheritance/test_abc_polymorphic.py2
-rw-r--r--test/orm/inheritance/test_basic.py16
-rw-r--r--test/orm/inheritance/test_polymorph.py4
-rw-r--r--test/orm/inheritance/test_polymorph2.py8
-rw-r--r--test/orm/inheritance/test_query.py10
-rw-r--r--test/orm/inheritance/test_selects.py4
-rw-r--r--test/orm/inheritance/test_single.py2
-rw-r--r--test/orm/test_cascade.py60
-rw-r--r--test/orm/test_froms.py240
-rw-r--r--test/orm/test_joins.py129
-rw-r--r--test/orm/test_mapper.py2
-rw-r--r--test/orm/test_pickled.py243
-rw-r--r--test/orm/test_query.py320
-rw-r--r--test/orm/test_session.py119
-rw-r--r--test/orm/test_transaction.py36
-rw-r--r--test/orm/test_unitofworkv2.py107
-rw-r--r--test/perf/sessions.py4
20 files changed, 1075 insertions, 388 deletions
diff --git a/test/engine/_base.py b/test/engine/_base.py
index ead668782..387c937b6 100644
--- a/test/engine/_base.py
+++ b/test/engine/_base.py
@@ -2,7 +2,7 @@ import sqlalchemy as sa
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."""
@@ -14,9 +14,12 @@ class TablesTest(testing.TestBase):
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
@@ -37,6 +40,11 @@ class TablesTest(testing.TestBase):
@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()
@@ -62,16 +70,19 @@ class TablesTest(testing.TestBase):
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):
@@ -108,7 +119,8 @@ class TablesTest(testing.TestBase):
@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)
diff --git a/test/lib/pickleable.py b/test/lib/pickleable.py
index acc07ceba..58422a914 100644
--- a/test/lib/pickleable.py
+++ b/test/lib/pickleable.py
@@ -1,10 +1,33 @@
-"""
+"""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):
diff --git a/test/orm/_fixtures.py b/test/orm/_fixtures.py
index 366c4ad5a..41c354e4a 100644
--- a/test/orm/_fixtures.py
+++ b/test/orm/_fixtures.py
@@ -1,4 +1,5 @@
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
@@ -8,63 +9,45 @@ from test.orm import _base
__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):
@@ -149,11 +132,6 @@ class FixtureTest(_base.MappedTest):
)
@classmethod
- def setup_classes(cls):
- for cl in cls.fixture_classes.values():
- cls.register_class(cl)
-
- @classmethod
def setup_mappers(cls):
pass
@@ -267,11 +245,20 @@ class FixtureTest(_base.MappedTest):
)
)
+ @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),
@@ -280,6 +267,8 @@ class CannedResults(object):
@property
def user_address_result(self):
+ User, Address = self.test.classes.User, self.test.classes.Address
+
return [
User(id=7, addresses=[
Address(id=1)
@@ -296,6 +285,10 @@ class CannedResults(object):
@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=[
@@ -330,6 +323,8 @@ class CannedResults(object):
@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=[
@@ -363,6 +358,7 @@ class CannedResults(object):
@property
def item_keyword_result(self):
+ Item, Keyword = self.test.classes.Item, self.test.classes.Keyword
return [
Item(id=1,
keywords=[
@@ -386,6 +382,9 @@ class CannedResults(object):
@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=[
@@ -426,5 +425,4 @@ class CannedResults(object):
User(id=10, orders=[])]
return user_result
-FixtureTest.static = CannedResults()
diff --git a/test/orm/inheritance/test_abc_polymorphic.py b/test/orm/inheritance/test_abc_polymorphic.py
index 58569ab9c..f18adac7a 100644
--- a/test/orm/inheritance/test_abc_polymorphic.py
+++ b/test/orm/inheritance/test_abc_polymorphic.py
@@ -24,7 +24,7 @@ class ABCTest(_base.MappedTest):
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
diff --git a/test/orm/inheritance/test_basic.py b/test/orm/inheritance/test_basic.py
index ea7cf9a69..372adce56 100644
--- a/test/orm/inheritance/test_basic.py
+++ b/test/orm/inheritance/test_basic.py
@@ -178,7 +178,7 @@ class PolymorphicSynonymTest(_base.MappedTest):
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):
@@ -318,13 +318,13 @@ class CascadeTest(_base.MappedTest):
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={
@@ -743,7 +743,7 @@ class VersioningTest(_base.MappedTest):
self.tables.base,
self.tables.stuff)
- class Base(_fixtures.Base):
+ class Base(_base.ComparableEntity):
pass
class Sub(Base):
pass
@@ -794,7 +794,7 @@ class VersioningTest(_base.MappedTest):
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
@@ -1661,13 +1661,13 @@ class DeleteOrphanTest(_base.MappedTest):
)
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')
diff --git a/test/orm/inheritance/test_polymorph.py b/test/orm/inheritance/test_polymorph.py
index 9cc58af92..1282fbf15 100644
--- a/test/orm/inheritance/test_polymorph.py
+++ b/test/orm/inheritance/test_polymorph.py
@@ -9,7 +9,7 @@ from test.lib import Column, testing
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
@@ -17,7 +17,7 @@ class Manager(Person):
pass
class Boss(Manager):
pass
-class Company(_fixtures.Base):
+class Company(_base.ComparableEntity):
pass
class PolymorphTest(_base.MappedTest):
diff --git a/test/orm/inheritance/test_polymorph2.py b/test/orm/inheritance/test_polymorph2.py
index 92af79ae1..8f707c1e8 100644
--- a/test/orm/inheritance/test_polymorph2.py
+++ b/test/orm/inheritance/test_polymorph2.py
@@ -1037,14 +1037,14 @@ class InheritingEagerTest(_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
@@ -1094,9 +1094,9 @@ class MissingPolymorphicOnTest(_base.MappedTest):
)
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
diff --git a/test/orm/inheritance/test_query.py b/test/orm/inheritance/test_query.py
index 08131d702..ce7aea68a 100644
--- a/test/orm/inheritance/test_query.py
+++ b/test/orm/inheritance/test_query.py
@@ -11,10 +11,10 @@ from test.orm import _base, _fixtures
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
@@ -23,10 +23,10 @@ class Manager(Person):
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):
@@ -1117,7 +1117,7 @@ class M2MFilterTest(_base.MappedTest):
cls.tables.engineers,
cls.tables.engineers_to_org)
- class Organization(_fixtures.Base):
+ class Organization(_base.ComparableEntity):
pass
mapper(Organization, organizations, properties={
diff --git a/test/orm/inheritance/test_selects.py b/test/orm/inheritance/test_selects.py
index 5a7389386..f06482b5a 100644
--- a/test/orm/inheritance/test_selects.py
+++ b/test/orm/inheritance/test_selects.py
@@ -2,8 +2,8 @@ from sqlalchemy import *
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):
diff --git a/test/orm/inheritance/test_single.py b/test/orm/inheritance/test_single.py
index 2815806c6..2d65ecbed 100644
--- a/test/orm/inheritance/test_single.py
+++ b/test/orm/inheritance/test_single.py
@@ -479,7 +479,7 @@ class SingleOnJoinedTest(MappedTest):
)
def test_single_on_joined(self):
- class Person(_fixtures.Base):
+ class Person(ComparableEntity):
pass
class Employee(Person):
pass
diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index 4ba18a875..63b4c881a 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -1007,13 +1007,13 @@ class M2OCascadeDeleteOrphanTestOne(_base.MappedTest):
@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
@@ -1187,11 +1187,11 @@ class M2OCascadeDeleteOrphanTestTwo(_base.MappedTest):
@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
@@ -1328,11 +1328,11 @@ class M2OCascadeDeleteNoOrphanTest(_base.MappedTest):
@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
@@ -1489,11 +1489,11 @@ class M2MCascadeTest(_base.MappedTest):
@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):
@@ -1790,11 +1790,11 @@ class PendingOrphanTestSingleLevel(_base.MappedTest):
)
@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):
@@ -2025,11 +2025,11 @@ class DoubleParentO2MOrphanTest(_base.MappedTest):
"""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)
@@ -2069,11 +2069,11 @@ class DoubleParentO2MOrphanTest(_base.MappedTest):
"""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)
@@ -2144,11 +2144,11 @@ class DoubleParentM2OOrphanTest(_base.MappedTest):
"""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)
@@ -2182,13 +2182,13 @@ class DoubleParentM2OOrphanTest(_base.MappedTest):
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)
@@ -2219,9 +2219,9 @@ class CollectionAssignmentOrphanTest(_base.MappedTest):
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={
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index fd0b2fd95..28133b456 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -13,12 +13,6 @@ import sqlalchemy as sa
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
@@ -29,9 +23,21 @@ class QueryTest(_fixtures.FixtureTest):
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
@@ -67,6 +73,8 @@ class RawSelectTest(QueryTest, AssertsCompiledSQL):
"""
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,
@@ -104,6 +112,8 @@ class RawSelectTest(QueryTest, AssertsCompiledSQL):
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()
@@ -111,6 +121,8 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
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)),
@@ -120,6 +132,8 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
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(),
@@ -134,6 +148,8 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
)
def test_no_joinedload(self):
+ User = self.classes.User
+
"""test that joinedloads are pushed outwards and not rendered in subqueries."""
s = create_session()
@@ -150,6 +166,8 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -187,6 +205,8 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
def test_multiple_entities(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
eq_(
@@ -212,6 +232,8 @@ class FromSelfTest(QueryTest, AssertsCompiledSQL):
)
def test_multiple_with_column_entities(self):
+ User = self.classes.User
+
sess = create_session()
eq_(
@@ -261,10 +283,10 @@ class AddEntityEquivalenceTest(_base.MappedTest, AssertsCompiledSQL):
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):
@@ -339,6 +361,10 @@ class AddEntityEquivalenceTest(_base.MappedTest, AssertsCompiledSQL):
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)).\
@@ -389,6 +415,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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
@@ -453,6 +483,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -470,6 +504,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -487,6 +525,8 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -502,6 +542,12 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -523,6 +569,12 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -544,6 +596,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -562,6 +618,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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()
@@ -591,6 +651,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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."""
@@ -622,6 +686,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -673,6 +741,10 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
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()
@@ -731,6 +803,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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")
@@ -744,6 +818,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
@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()
@@ -771,6 +847,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
"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()
@@ -786,6 +864,10 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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.
@@ -817,59 +899,12 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
(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',)])
@@ -987,6 +1022,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
)
def test_column_from_limited_joinedload(self):
+ User = self.classes.User
+
sess = create_session()
def go():
@@ -998,6 +1035,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
@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)
@@ -1062,6 +1101,11 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1107,6 +1151,11 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1132,6 +1181,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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)
@@ -1150,6 +1201,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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()]
@@ -1161,6 +1214,8 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1171,6 +1226,11 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -1204,6 +1264,10 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
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 = [
@@ -1256,6 +1320,11 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
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)
})
@@ -1281,6 +1350,8 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
)
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)
@@ -1295,6 +1366,8 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
)
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)
@@ -1353,6 +1426,8 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
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]))
@@ -1365,6 +1440,11 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
)
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)
})
@@ -1396,6 +1476,17 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
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
})
@@ -1492,6 +1583,11 @@ class SelectFromTest(QueryTest, AssertsCompiledSQL):
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)
})
@@ -1525,6 +1621,16 @@ class CustomJoinTest(QueryTest):
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={
@@ -1550,6 +1656,8 @@ class ExternalColumnsTest(QueryTest):
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),
@@ -1557,6 +1665,11 @@ class ExternalColumnsTest(QueryTest):
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={
@@ -1657,6 +1770,13 @@ class ExternalColumnsTest(QueryTest):
)
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.
diff --git a/test/orm/test_joins.py b/test/orm/test_joins.py
index acbf36a57..cae48947f 100644
--- a/test/orm/test_joins.py
+++ b/test/orm/test_joins.py
@@ -13,12 +13,6 @@ import sqlalchemy as sa
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
@@ -32,6 +26,19 @@ class QueryTest(_fixtures.FixtureTest):
@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
@@ -112,9 +119,9 @@ class InheritedJoinTest(_base.MappedTest, AssertsCompiledSQL):
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
@@ -122,9 +129,9 @@ class InheritedJoinTest(_base.MappedTest, AssertsCompiledSQL):
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={
@@ -346,6 +353,8 @@ class InheritedJoinTest(_base.MappedTest, AssertsCompiledSQL):
class JoinTest(QueryTest, AssertsCompiledSQL):
def test_single_name(self):
+ User = self.classes.User
+
sess = create_session()
self.assert_compile(
@@ -379,6 +388,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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.
@@ -420,6 +433,11 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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),
@@ -550,6 +568,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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).\
@@ -557,11 +577,17 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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)]:
@@ -601,6 +627,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -615,6 +643,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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.
@@ -648,6 +678,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
def test_multiple_with_aliases(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = create_session()
ualias = aliased(User)
@@ -664,6 +696,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
def test_select_from_orm_joins(self):
+ User, Order = self.classes.User, self.classes.Order
+
sess = create_session()
ualias = aliased(User)
@@ -728,6 +762,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
def test_overlapping_backwards_joins(self):
+ User, Order = self.classes.User, self.classes.Order
+
sess = create_session()
oalias1 = aliased(Order)
@@ -743,6 +779,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -758,6 +798,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
def test_multiple_adaption(self):
+ Item, Order, User = (self.classes.Item,
+ self.classes.Order,
+ self.classes.User)
+
sess = create_session()
self.assert_compile(
@@ -769,6 +813,12 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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
@@ -804,17 +854,27 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
# 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_(
@@ -834,6 +894,12 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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_(
@@ -883,6 +949,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
def test_aliased_classes(self):
+ User, Address = self.classes.User, self.classes.Address
+
sess = create_session()
(user7, user8, user9, user10) = sess.query(User).all()
@@ -922,6 +990,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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()
@@ -955,6 +1025,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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)
@@ -982,6 +1056,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -1018,6 +1094,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
def test_joins_from_adapted_entities(self):
+ User = self.classes.User
+
# test for #1853
@@ -1056,6 +1134,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1065,6 +1145,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1074,6 +1158,11 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1110,6 +1199,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
assert [User(id=7)] == q.all()
def test_aliased_order_by(self):
+ User = self.classes.User
+
sess = create_session()
ualias = aliased(User)
@@ -1126,6 +1217,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
def test_plain_table(self):
+ addresses, User = self.tables.addresses, self.classes.User
+
sess = create_session()
@@ -1135,6 +1228,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
def test_no_joinpoint_expr(self):
+ User, users = self.classes.User, self.tables.users
+
sess = create_session()
# these are consistent regardless of
@@ -1153,6 +1248,10 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -1177,6 +1276,8 @@ class JoinTest(QueryTest, AssertsCompiledSQL):
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.
"""
@@ -1270,10 +1371,10 @@ class SelfRefMixedTest(_base.MappedTest, AssertsCompiledSQL):
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={
@@ -1345,7 +1446,7 @@ class SelfReferentialTest(_base.MappedTest, AssertsCompiledSQL):
@classmethod
def setup_classes(cls):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@@ -1700,7 +1801,7 @@ class SelfReferentialM2MTest(_base.MappedTest):
@classmethod
def setup_classes(cls):
- class Node(Base):
+ class Node(_base.ComparableEntity):
pass
@classmethod
diff --git a/test/orm/test_mapper.py b/test/orm/test_mapper.py
index 3edd72a4c..b4c338ddd 100644
--- a/test/orm/test_mapper.py
+++ b/test/orm/test_mapper.py
@@ -1601,8 +1601,6 @@ class OptionsTest(_fixtures.FixtureTest):
assert opt2a not in ustate.load_options
assert opt2b not in ustate.load_options
- import pickle
- pickle.dumps(u1)
class DeepOptionsTest(_fixtures.FixtureTest):
@classmethod
diff --git a/test/orm/test_pickled.py b/test/orm/test_pickled.py
index c979ed334..3291d2ff9 100644
--- a/test/orm/test_pickled.py
+++ b/test/orm/test_pickled.py
@@ -1,5 +1,5 @@
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
@@ -10,21 +10,42 @@ from sqlalchemy.orm import mapper, relationship, create_session, \
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")
@@ -44,7 +65,7 @@ class PickleTest(_fixtures.FixtureTest):
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)
@@ -55,11 +76,11 @@ class PickleTest(_fixtures.FixtureTest):
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)
@@ -76,10 +97,8 @@ class PickleTest(_fixtures.FixtureTest):
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")
@@ -109,10 +128,8 @@ class PickleTest(_fixtures.FixtureTest):
)
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),
@@ -144,10 +161,8 @@ class PickleTest(_fixtures.FixtureTest):
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')
@@ -175,10 +190,8 @@ class PickleTest(_fixtures.FixtureTest):
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")
@@ -224,10 +237,8 @@ class PickleTest(_fixtures.FixtureTest):
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")
@@ -247,11 +258,10 @@ class PickleTest(_fixtures.FixtureTest):
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")
@@ -282,8 +292,6 @@ class PickleTest(_fixtures.FixtureTest):
"""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,
@@ -296,14 +304,6 @@ class PickleTest(_fixtures.FixtureTest):
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),
@@ -311,10 +311,6 @@ class PickleTest(_fixtures.FixtureTest):
})
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)
@@ -333,26 +329,11 @@ class PolymorphicDeferredTest(_base.MappedTest):
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')
@@ -370,9 +351,95 @@ class PolymorphicDeferredTest(_base.MappedTest):
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.
@@ -387,41 +454,3 @@ class CustomSetupTeardownTest(_fixtures.FixtureTest):
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)
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 1861fd7dd..9a799bba2 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -13,12 +13,6 @@ import sqlalchemy as sa
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
@@ -32,6 +26,19 @@ class QueryTest(_fixtures.FixtureTest):
@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
@@ -63,6 +70,8 @@ class RowTupleTest(QueryTest):
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
})
@@ -74,6 +83,11 @@ class RowTupleTest(QueryTest):
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()
@@ -128,6 +142,8 @@ class RowTupleTest(QueryTest):
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)
@@ -138,10 +154,14 @@ class GetTest(QueryTest):
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
@@ -149,21 +169,29 @@ class GetTest(QueryTest):
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()."""
@@ -183,6 +211,8 @@ class GetTest(QueryTest):
)
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()
@@ -195,6 +225,8 @@ class GetTest(QueryTest):
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')
@@ -205,6 +237,8 @@ class GetTest(QueryTest):
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
@@ -246,7 +280,7 @@ class GetTest(QueryTest):
# 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),
@@ -255,6 +289,8 @@ class GetTest(QueryTest):
metadata.drop_all()
def test_populate_existing(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
userlist = s.query(User).all()
@@ -288,6 +324,8 @@ class GetTest(QueryTest):
@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_(
@@ -297,6 +335,8 @@ class GetTest(QueryTest):
class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
def test_no_limit_offset(self):
+ User = self.classes.User
+
s = create_session()
for q in (
@@ -322,6 +362,8 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
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)
@@ -341,23 +383,33 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
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()
@@ -367,6 +419,8 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
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()
@@ -376,6 +430,8 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
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)
@@ -401,12 +457,16 @@ class InvalidGenerationsTest(QueryTest, AssertsCompiledSQL):
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)
@@ -431,6 +491,8 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
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, '-'),
@@ -453,6 +515,8 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
self._test(py_op(lhs, rhs), res % sql_op)
def test_comparison(self):
+ User = self.classes.User
+
create_session().query(User)
ualias = aliased(User)
@@ -488,6 +552,8 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
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")
@@ -498,6 +564,8 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
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 "
@@ -516,6 +584,8 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
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
@@ -588,29 +658,43 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
)
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"),
@@ -625,6 +709,10 @@ class OperatorTest(QueryTest, AssertsCompiledSQL):
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
@@ -633,6 +721,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
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(
@@ -648,6 +738,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
)
def test_scalar_subquery(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==7).subquery()
@@ -657,6 +749,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
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')
@@ -668,6 +762,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
def test_label(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==7).label('foo')
@@ -678,6 +774,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
)
def test_as_scalar(self):
+ User = self.classes.User
+
session = create_session()
q = session.query(User.id).filter(User.id==7).as_scalar()
@@ -691,6 +789,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
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()
@@ -700,6 +800,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
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_(
@@ -708,6 +810,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
)
def test_union(self):
+ User = self.classes.User
+
s = create_session()
q1 = s.query(User).filter(User.name=='ed').with_labels()
@@ -718,6 +822,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
)
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
@@ -732,6 +838,8 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -746,12 +854,16 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
# 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
@@ -794,10 +906,14 @@ class SliceTest(QueryTest):
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])
@@ -809,15 +925,21 @@ class FilterTest(QueryTest):
@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()
@@ -841,6 +963,8 @@ class FilterTest(QueryTest):
#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()
@@ -862,6 +986,10 @@ class FilterTest(QueryTest):
@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()
@@ -883,6 +1011,8 @@ class FilterTest(QueryTest):
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()
@@ -894,6 +1024,12 @@ class FilterTest(QueryTest):
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()
@@ -917,6 +1053,8 @@ class FilterTest(QueryTest):
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()
@@ -930,6 +1068,10 @@ class FilterTest(QueryTest):
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
@@ -959,6 +1101,8 @@ class FilterTest(QueryTest):
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()
@@ -973,6 +1117,8 @@ class FilterTest(QueryTest):
class SetOpsTest(QueryTest, AssertsCompiledSQL):
def test_union(self):
+ User = self.classes.User
+
s = create_session()
fred = s.query(User).filter(User.name=='fred')
@@ -988,6 +1134,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
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()
@@ -1006,6 +1154,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
def test_union_literal_expressions_compile(self):
+ User = self.classes.User
+
"""test that column expressions translate during
the _from_statement() portion of union(), others"""
@@ -1024,6 +1174,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
def test_union_literal_expressions_results(self):
+ User = self.classes.User
+
s = Session()
q1 = s.query(User, literal("x"))
@@ -1054,6 +1206,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
def test_union_labeled_anonymous_columns(self):
+ User = self.classes.User
+
s = Session()
c1, c2 = column('c1'), column('c2')
@@ -1079,6 +1233,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
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)
@@ -1102,6 +1258,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
@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')
@@ -1116,6 +1274,8 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
)
def test_eager_load(self):
+ User, Address = self.classes.User, self.classes.Address
+
s = create_session()
fred = s.query(User).filter(User.name=='fred')
@@ -1136,16 +1296,22 @@ class SetOpsTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1153,6 +1319,8 @@ class AggregateTest(QueryTest):
class CountTest(QueryTest):
def test_basic(self):
+ users, User = self.tables.users, self.classes.User
+
s = create_session()
eq_(s.query(User).count(), 4)
@@ -1160,6 +1328,8 @@ class CountTest(QueryTest):
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
@@ -1168,6 +1338,8 @@ class CountTest(QueryTest):
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)
@@ -1179,6 +1351,8 @@ class CountTest(QueryTest):
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()
@@ -1202,6 +1376,8 @@ class CountTest(QueryTest):
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()
@@ -1212,6 +1388,8 @@ class DistinctTest(QueryTest):
)
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()
@@ -1238,6 +1416,8 @@ class DistinctTest(QueryTest):
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"))
@@ -1257,6 +1437,8 @@ class YieldTest(QueryTest):
class HintsTest(QueryTest, AssertsCompiledSQL):
def test_hints(self):
+ User = self.classes.User
+
from sqlalchemy.dialects import mysql
dialect = mysql.dialect()
@@ -1290,12 +1472,16 @@ class HintsTest(QueryTest, AssertsCompiledSQL):
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()
@@ -1303,9 +1489,13 @@ class TextTest(QueryTest):
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"))
@@ -1313,6 +1503,10 @@ class TextTest(QueryTest):
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)
@@ -1341,6 +1535,8 @@ class ParentTest(QueryTest):
#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)
@@ -1355,12 +1551,16 @@ class ParentTest(QueryTest):
"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)
@@ -1379,6 +1579,8 @@ class ParentTest(QueryTest):
)
def test_with_pending_autoflush(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = Session()
o1 = sess.query(Order).first()
@@ -1394,6 +1596,8 @@ class ParentTest(QueryTest):
)
def test_with_pending_no_autoflush(self):
+ Order, User = self.classes.Order, self.classes.User
+
sess = Session(autoflush=False)
o1 = sess.query(Order).first()
@@ -1411,6 +1615,14 @@ class SynonymTest(QueryTest):
@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),
@@ -1431,6 +1643,8 @@ class SynonymTest(QueryTest):
@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'],
@@ -1443,6 +1657,8 @@ class SynonymTest(QueryTest):
@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'),
@@ -1927,6 +2143,8 @@ class UpdateDeleteTest(_base.MappedTest):
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)
@@ -1975,6 +2193,8 @@ class OptionsTest(QueryTest):
)
def test_get_path_one_level_string(self):
+ User = self.classes.User
+
sess = Session()
q = sess.query(User)
@@ -1982,6 +2202,8 @@ class OptionsTest(QueryTest):
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)
@@ -1989,6 +2211,8 @@ class OptionsTest(QueryTest):
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]
@@ -1999,6 +2223,8 @@ class OptionsTest(QueryTest):
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)
@@ -2006,6 +2232,10 @@ class OptionsTest(QueryTest):
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)
@@ -2018,6 +2248,10 @@ class OptionsTest(QueryTest):
[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)
@@ -2030,6 +2264,10 @@ class OptionsTest(QueryTest):
[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'])
@@ -2041,6 +2279,10 @@ class OptionsTest(QueryTest):
], [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'])
@@ -2052,6 +2294,10 @@ class OptionsTest(QueryTest):
], [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'])
@@ -2064,6 +2310,10 @@ class OptionsTest(QueryTest):
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'])
@@ -2076,6 +2326,8 @@ class OptionsTest(QueryTest):
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
@@ -2089,6 +2341,8 @@ class OptionsTest(QueryTest):
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
@@ -2102,6 +2356,8 @@ class OptionsTest(QueryTest):
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
@@ -2115,6 +2371,8 @@ class OptionsTest(QueryTest):
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
@@ -2129,6 +2387,10 @@ class OptionsTest(QueryTest):
], [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
@@ -2145,6 +2407,8 @@ class OptionsTest(QueryTest):
], [User, SubAddr])
def test_aliased_single(self):
+ User = self.classes.User
+
sess = Session()
ualias = aliased(User)
q = sess.query(ualias)
@@ -2152,6 +2416,8 @@ class OptionsTest(QueryTest):
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(
@@ -2161,6 +2427,8 @@ class OptionsTest(QueryTest):
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(
@@ -2171,6 +2439,8 @@ class OptionsTest(QueryTest):
@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(
@@ -2179,31 +2449,49 @@ class OptionsTest(QueryTest):
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 "\
@@ -2212,6 +2500,8 @@ class OptionsNoPropTest(_base.MappedTest):
'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\."
@@ -2219,17 +2509,21 @@ class OptionsNoPropTest(_base.MappedTest):
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'))
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 15a931331..042a51d7c 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -1,8 +1,9 @@
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
@@ -95,35 +96,6 @@ class SessionTest(_fixtures.FixtureTest):
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):
@@ -423,34 +395,6 @@ class SessionTest(_fixtures.FixtureTest):
).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
@@ -1035,7 +979,7 @@ class SessionTest(_fixtures.FixtureTest):
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)
@@ -1232,7 +1176,7 @@ class SessionTest(_fixtures.FixtureTest):
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()
@@ -1320,6 +1264,61 @@ class SessionTest(_fixtures.FixtureTest):
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'
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index af52f1f2e..396ed4e54 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -8,7 +8,7 @@ from sqlalchemy.orm import *
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'
@@ -17,6 +17,8 @@ class TransactionTest(FixtureTest):
@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),
@@ -28,6 +30,7 @@ class FixtureDataTest(TransactionTest):
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'
@@ -35,6 +38,7 @@ class FixtureDataTest(TransactionTest):
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'
@@ -43,6 +47,7 @@ class FixtureDataTest(TransactionTest):
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'
@@ -59,6 +64,7 @@ class FixtureDataTest(TransactionTest):
class AutoExpireTest(TransactionTest):
def test_expunge_pending_on_rollback(self):
+ User = self.classes.User
sess = self.session()
u2= User(name='newuser')
sess.add(u2)
@@ -67,6 +73,7 @@ class AutoExpireTest(TransactionTest):
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)
@@ -80,6 +87,7 @@ class AutoExpireTest(TransactionTest):
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)
@@ -95,6 +103,8 @@ class AutoExpireTest(TransactionTest):
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)
@@ -122,6 +132,7 @@ class AutoExpireTest(TransactionTest):
s.commit()
def test_trans_deleted_cleared_on_rollback(self):
+ User = self.classes.User
s = self.session()
u1 = User(name='ed')
s.add(u1)
@@ -134,6 +145,8 @@ class AutoExpireTest(TransactionTest):
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)
@@ -148,6 +161,8 @@ class AutoExpireTest(TransactionTest):
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)
@@ -163,6 +178,7 @@ class AutoExpireTest(TransactionTest):
assert u1.addresses == [a1]
def test_commit_pending(self):
+ User = self.classes.User
sess = self.session()
u1 = User(name='newuser')
sess.add(u1)
@@ -172,6 +188,7 @@ class AutoExpireTest(TransactionTest):
def test_concurrent_commit_pending(self):
+ User = self.classes.User
s1 = self.session()
u1 = User(name='edward')
s1.add(u1)
@@ -188,6 +205,7 @@ class TwoPhaseTest(TransactionTest):
@testing.requires.two_phase_transactions
def test_rollback_on_prepare(self):
+ User = self.classes.User
s = self.session(twophase=True)
u = User(name='ed')
@@ -200,6 +218,7 @@ class TwoPhaseTest(TransactionTest):
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])
@@ -231,6 +250,7 @@ class RollbackRecoverTest(TransactionTest):
@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])
@@ -320,6 +340,7 @@ class SavepointTest(TransactionTest):
@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)
@@ -351,6 +372,7 @@ class SavepointTest(TransactionTest):
@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)
@@ -410,6 +432,8 @@ class SavepointTest(TransactionTest):
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)
@@ -422,6 +446,8 @@ class AccountingFlagsTest(TransactionTest):
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)
@@ -437,6 +463,8 @@ class AccountingFlagsTest(TransactionTest):
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)
@@ -457,6 +485,8 @@ class AccountingFlagsTest(TransactionTest):
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')
@@ -480,6 +510,7 @@ class AutoCommitTest(TransactionTest):
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')
@@ -494,6 +525,7 @@ class AutoCommitTest(TransactionTest):
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
@@ -524,6 +556,7 @@ class AutoCommitTest(TransactionTest):
)
def test_accounting_commit_fails_delete(self):
+ User = self.classes.User
sess = create_session(autocommit=True)
fail = False
@@ -557,6 +590,7 @@ class AutoCommitTest(TransactionTest):
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')
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index aaecd2af4..dc818a8a6 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -8,12 +8,6 @@ from sqlalchemy.orm import mapper, relationship, backref, \
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)
@@ -26,10 +20,7 @@ class AssertsUOW(object):
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
@@ -42,6 +33,11 @@ class UOWTest(_fixtures.FixtureTest,
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),
})
@@ -72,6 +68,11 @@ class RudimentaryFlushTest(UOWTest):
)
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),
})
@@ -99,6 +100,11 @@ class RudimentaryFlushTest(UOWTest):
)
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),
})
@@ -130,6 +136,11 @@ class RudimentaryFlushTest(UOWTest):
)
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={
@@ -162,6 +173,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -191,6 +207,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -225,6 +246,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -295,6 +321,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -348,6 +379,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -401,6 +437,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -443,6 +484,8 @@ class RudimentaryFlushTest(UOWTest):
)
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={
@@ -472,6 +515,12 @@ class RudimentaryFlushTest(UOWTest):
)
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)
})
@@ -513,6 +562,11 @@ class RudimentaryFlushTest(UOWTest):
)
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)
@@ -523,6 +577,11 @@ class RudimentaryFlushTest(UOWTest):
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),
})
@@ -561,6 +620,8 @@ class RudimentaryFlushTest(UOWTest):
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)
})
@@ -595,6 +656,8 @@ class SingleCycleTest(UOWTest):
)
def test_one_to_many_delete_all(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
@@ -619,6 +682,8 @@ class SingleCycleTest(UOWTest):
)
def test_one_to_many_delete_parent(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
@@ -647,6 +712,8 @@ class SingleCycleTest(UOWTest):
)
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)
})
@@ -681,6 +748,8 @@ class SingleCycleTest(UOWTest):
)
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)
})
@@ -705,6 +774,8 @@ class SingleCycleTest(UOWTest):
)
def test_cycle_rowswitch(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
@@ -721,6 +792,8 @@ class SingleCycleTest(UOWTest):
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',
@@ -741,6 +814,8 @@ class SingleCycleTest(UOWTest):
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)
@@ -797,6 +872,8 @@ class SingleCycleTest(UOWTest):
)
def test_singlecycle_flush_size(self):
+ Node, nodes = self.classes.Node, self.tables.nodes
+
mapper(Node, nodes, properties={
'children':relationship(Node)
})
@@ -831,6 +908,8 @@ class SingleCycleTest(UOWTest):
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)
})
@@ -920,9 +999,9 @@ class SingleCyclePlusAttributeTest(_base.MappedTest,
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={
@@ -970,7 +1049,7 @@ class SingleCycleM2MTest(_base.MappedTest,
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={
@@ -1149,7 +1228,7 @@ class BatchInsertsTest(_base.MappedTest, testing.AssertsExecutionResults):
key present statements together.
"""
- class T(Base):
+ class T(_base.ComparableEntity):
pass
mapper(T, t)
sess = Session()
diff --git a/test/perf/sessions.py b/test/perf/sessions.py
index c1258e732..3384736a1 100644
--- a/test/perf/sessions.py
+++ b/test/perf/sessions.py
@@ -36,9 +36,9 @@ class SessionTest(TestBase, AssertsExecutionResults):
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={
View Lorry Controller Status