dependencies on those.
from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey, ForeignKeyConstraint, asc, Index
from testlib.sa.orm import relation, create_session, class_mapper, eagerload, compile_mappers, backref, clear_mappers, polymorphic_union, deferred
from testlib.testing import eq_
+
+
from orm._base import ComparableEntity, MappedTest
class DeclarativeTestBase(testing.TestBase, testing.AssertsExecutionResults):
def produce_test(inline, stringbased):
- class ExplicitJoinTest(testing.ORMTest):
+ class ExplicitJoinTest(MappedTest):
def define_tables(self, metadata):
global User, Address
from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey, select, desc, func, util
from testlib.sa.orm import relation, sessionmaker, scoped_session, class_mapper, mapper, eagerload, compile_mappers, aliased
from testlib.testing import eq_
+
from orm._base import ComparableEntity, MappedTest
class Address(ComparableEntity):
pass
-class SerializeTest(testing.ORMTest):
- keep_mappers = True
- keep_data = True
+class SerializeTest(MappedTest):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
def define_tables(self, metadata):
global users, addresses
self_key = sa.orm.attributes.instance_state(self).key
except sa.orm.exc.NO_STATE:
self_key = None
- try:
- other_key = sa.orm.attributes.instance_state(other).key
- except sa.orm.exc.NO_STATE:
- other_key = None
-
- if other_key is None and self_key is not None:
- a, b = other, self
+
+ if other is None:
+ a = self
+ b = other
+ elif self_key is not None:
+ a = other
+ b = self
else:
- a, b = self, other
+ a = self
+ b = other
for attr in a.__dict__.keys():
if attr.startswith('_'):
# TODO: ensure instrumentation registry is empty
class MappedTest(ORMTest):
- # 'once', 'foreach', None
+ # 'once', 'each', None
run_define_tables = 'once'
- # 'once', 'foreach', None
+ # 'once', 'each', None
run_setup_classes = 'once'
- # 'once', 'foreach', None
+ # 'once', 'each', None
run_setup_mappers = 'each'
- # 'once', 'foreach', None
+ # 'once', 'each', None
run_inserts = 'each'
- # 'foreach', None
+ # 'each', None
run_deletes = 'each'
metadata = None
def tearDown(self):
sa.orm.session.Session.close_all()
- if self.run_setup_mappers == 'each':
+ # some tests create mappers in the test bodies
+ # and will define setup_mappers as None -
+ # clear mappers in any case
+ if self.run_setup_mappers != 'once':
sa.orm.clear_mappers()
# no need to run deletes if tables are recreated on setup
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.orm.interfaces import ONETOMANY, MANYTOONE
-from testlib import *
+
+from testlib import testing
+from orm import _base
def produce_test(parent, child, direction):
"""produce a testcase for A->B->C inheritance with a self-referential
relationship between two of the classes, using either one-to-many or
many-to-one."""
- class ABCTest(ORMTest):
+ class ABCTest(_base.MappedTest):
def define_tables(self, meta):
global ta, tb, tc
ta = ["a", meta]
from sqlalchemy import *
from sqlalchemy import util
from sqlalchemy.orm import *
-from testlib import *
-from testlib import fixtures
-class ABCTest(ORMTest):
+from testlib import _function_named
+from orm import _base, _fixtures
+
+class ABCTest(_base.MappedTest):
def define_tables(self, metadata):
global a, b, c
a = Table('a', metadata,
def make_test(fetchtype):
def test_roundtrip(self):
- class A(fixtures.Base):pass
+ class A(_fixtures.Base):pass
class B(A):pass
class C(B):pass
from sqlalchemy import exc as sa_exc, util
from sqlalchemy.orm import *
from sqlalchemy.orm import exc as orm_exc
-from testlib import *
-from testlib import fixtures
+
+#from testlib import *
+#from testlib import fixtures
+from testlib import _function_named, testing, engines
from orm import _base, _fixtures
-class O2MTest(ORMTest):
+class O2MTest(_base.MappedTest):
"""deals with inheritance and one-to-many relationships"""
def define_tables(self, metadata):
global foo, bar, blub
self.assert_(compare == result)
self.assert_(l[0].parent_foo.data == 'foo #1' and l[1].parent_foo.data == 'foo #1')
-class FalseDiscriminatorTest(ORMTest):
+class FalseDiscriminatorTest(_base.MappedTest):
def define_tables(self, metadata):
global t1
t1 = Table('t1', metadata, Column('id', Integer, primary_key=True), Column('type', Integer, nullable=False))
sess.expunge_all()
assert isinstance(sess.query(Foo).one(), Bar)
-class PolymorphicSynonymTest(ORMTest):
+class PolymorphicSynonymTest(_base.MappedTest):
def define_tables(self, metadata):
global t1, t2
t1 = Table('t1', metadata,
Column('data', String(10), nullable=False))
def test_polymorphic_synonym(self):
- class T1(fixtures.Base):
+ class T1(_fixtures.Base):
def info(self):
return "THE INFO IS:" + self._info
def _set_info(self, x):
self.assertEquals(at2.info, "THE INFO IS:at2")
-class CascadeTest(ORMTest):
+class CascadeTest(_base.MappedTest):
"""that cascades on polymorphic relations continue
cascading along the path of the instance's mapper, not
the base mapper."""
Column('data', String(30)))
def test_cascade(self):
- class T1(fixtures.Base):
+ class T1(_fixtures.Base):
pass
- class T2(fixtures.Base):
+ class T2(_fixtures.Base):
pass
class T3(T2):
pass
- class T4(fixtures.Base):
+ class T4(_fixtures.Base):
pass
mapper(T1, t1, properties={
assert t4_1 in sess.deleted
sess.flush()
-class GetTest(ORMTest):
+class GetTest(_base.MappedTest):
def define_tables(self, metadata):
global foo, bar, blub
foo = Table('foo', metadata,
test_get_polymorphic = create_test(True, 'test_get_polymorphic')
test_get_nonpolymorphic = create_test(False, 'test_get_nonpolymorphic')
-class EagerLazyTest(ORMTest):
+class EagerLazyTest(_base.MappedTest):
"""tests eager load/lazy load of child items off inheritance mappers, tests that
LazyLoader constructs the right query condition."""
def define_tables(self, metadata):
self.assert_(len(q.first().eager) == 1)
-class FlushTest(ORMTest):
+class FlushTest(_base.MappedTest):
"""test dependency sorting among inheriting mappers"""
def define_tables(self, metadata):
global users, roles, user_roles, admins
sess.flush()
assert user_roles.count().scalar() == 1
-class VersioningTest(ORMTest):
+class VersioningTest(_base.MappedTest):
def define_tables(self, metadata):
global base, subtable, stuff
base = Table('base', metadata,
@testing.fails_on('mssql', 'FIXME: the flush still happens with the concurrency issue.')
@engines.close_open_connections
def test_save_update(self):
- class Base(fixtures.Base):
+ class Base(_fixtures.Base):
pass
class Sub(Base):
pass
@testing.fails_on('mssql', 'FIXME: the flush still happens with the concurrency issue.')
def test_delete(self):
- class Base(fixtures.Base):
+ class Base(_fixtures.Base):
pass
class Sub(Base):
pass
except orm_exc.ConcurrentModificationError, e:
assert True
-class DistinctPKTest(ORMTest):
+class DistinctPKTest(_base.MappedTest):
"""test the construction of mapper.primary_key when an inheriting relationship
joins on a column other than primary key column."""
- keep_data = True
+
+ run_inserts = 'once'
+ run_deletes = None
def define_tables(self, metadata):
global person_table, employee_table, Person, Employee
assert alice1.name == alice2.name == 'alice'
assert bob.name == 'bob'
-class SyncCompileTest(ORMTest):
+class SyncCompileTest(_base.MappedTest):
"""test that syncrules compile properly on custom inherit conds"""
def define_tables(self, metadata):
global _a_table, _b_table, _c_table
assert len(session.query(B).all()) == 2
assert len(session.query(C).all()) == 1
-class OverrideColKeyTest(ORMTest):
+class OverrideColKeyTest(_base.MappedTest):
"""test overriding of column attributes."""
def define_tables(self, metadata):
assert sess.query(Base).get(b1.base_id).data == "this is base"
assert sess.query(Sub).get(s1.base_id).data == "this is base"
-class OptimizedLoadTest(ORMTest):
+class OptimizedLoadTest(_base.MappedTest):
"""test that the 'optimized load' routine doesn't crash when
a column in the join condition is not available.
assert a.type == 2
-class DeleteOrphanTest(ORMTest):
+class DeleteOrphanTest(_base.MappedTest):
def define_tables(self, metadata):
global single, parent
single = Table('single', metadata,
)
def test_orphan_message(self):
- class Base(fixtures.Base):
+ class Base(_fixtures.Base):
pass
class SubClass(Base):
pass
- class Parent(fixtures.Base):
+ class Parent(_fixtures.Base):
pass
mapper(Base, single, polymorphic_on=single.c.type, polymorphic_identity='base')
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
+from testlib import testing, _function_named
+from orm import _base
class BaseObject(object):
def __init__(self, *args, **kwargs):
pass
-class MagazineTest(ORMTest):
+class MagazineTest(_base.MappedTest):
def define_tables(self, metadata):
global publication_table, issue_table, location_table, location_name_table, magazine_table, \
page_table, magazine_page_table, classified_page_table, page_size_table
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
+from testlib import testing
+from orm import _base
-class InheritTest(ORMTest):
+
+class InheritTest(_base.MappedTest):
"""deals with inheritance and many-to-many relationships"""
def define_tables(self, metadata):
global principals
sess.flush()
# TODO: put an assertion
-class InheritTest2(ORMTest):
+class InheritTest2(_base.MappedTest):
"""deals with inheritance and many-to-many relationships"""
def define_tables(self, metadata):
global foo, bar, foo_bar
{'id':b.id, 'data':'barfoo', 'foos':(Foo, [{'id':f1.id,'data':'subfoo1'}, {'id':f2.id,'data':'subfoo2'}])},
)
-class InheritTest3(ORMTest):
+class InheritTest3(_base.MappedTest):
"""deals with inheritance and many-to-many relationships"""
def define_tables(self, metadata):
global foo, bar, blub, bar_foo, blub_bar, blub_foo
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
+from orm import _base
+from testlib import testing
+
+
+class PolymorphicCircularTest(_base.MappedTest):
+ run_setup_mappers = 'once'
-class PolymorphicCircularTest(ORMTest):
- keep_mappers = True
def define_tables(self, metadata):
global Table1, Table1B, Table2, Table3, Data
table1 = Table('table1', metadata,
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.orm import exc as orm_exc
-from testlib import *
-from testlib import fixtures
+from testlib import _function_named, Column, testing
+from orm import _fixtures, _base
-class Person(fixtures.Base):
+class Person(_fixtures.Base):
pass
class Engineer(Person):
pass
pass
class Boss(Manager):
pass
-class Company(fixtures.Base):
+class Company(_fixtures.Base):
pass
-class PolymorphTest(ORMTest):
+class PolymorphTest(_base.MappedTest):
def define_tables(self, metadata):
global companies, people, engineers, managers, boss
from sqlalchemy import *
from sqlalchemy import util
from sqlalchemy.orm import *
-from testlib import *
-from testlib import fixtures
-from orm import _base
+
+from testlib import _function_named, TestBase, AssertsExecutionResults, testing
+from orm import _base, _fixtures
from testlib.testing import eq_
class AttrSettable(object):
return self.__class__.__name__ + "(%s)" % (hex(id(self)))
-class RelationTest1(ORMTest):
+class RelationTest1(_base.MappedTest):
"""test self-referential relationships on polymorphic mappers"""
def define_tables(self, metadata):
global people, managers
print p, m, m.employee
assert m.employee is p
-class RelationTest2(ORMTest):
+class RelationTest2(_base.MappedTest):
"""test self-referential relationships on polymorphic mappers"""
def define_tables(self, metadata):
global people, managers, data
if usedata:
assert m.data.data == 'ms data'
-class RelationTest3(ORMTest):
+class RelationTest3(_base.MappedTest):
"""test self-referential relationships on polymorphic mappers"""
def define_tables(self, metadata):
global people, managers, data
setattr(RelationTest3, func.__name__, func)
-class RelationTest4(ORMTest):
+class RelationTest4(_base.MappedTest):
def define_tables(self, metadata):
global people, engineers, managers, cars
people = Table('people', metadata,
c = s.join("employee").filter(Person.name=="E4")[0]
assert c.car_id==car1.car_id
-class RelationTest5(ORMTest):
+class RelationTest5(_base.MappedTest):
def define_tables(self, metadata):
global people, engineers, managers, cars
people = Table('people', metadata,
assert carlist[0].manager is None
assert carlist[1].manager.person_id == car2.manager.person_id
-class RelationTest6(ORMTest):
+class RelationTest6(_base.MappedTest):
"""test self-referential relationships on a single joined-table inheritance mapper"""
def define_tables(self, metadata):
global people, managers, data
m2 = sess.query(Manager).get(m2.person_id)
assert m.colleague is m2
-class RelationTest7(ORMTest):
+class RelationTest7(_base.MappedTest):
def define_tables(self, metadata):
global people, engineers, managers, cars, offroad_cars
cars = Table('cars', metadata,
for p in r:
assert p.car_id == p.car.car_id
-class RelationTest8(ORMTest):
+class RelationTest8(_base.MappedTest):
def define_tables(self, metadata):
global taggable, users
taggable = Table('taggable', metadata,
r = session.query(Person).filter(exists([1], Car.owner==Person.person_id))
assert str(list(r)) == "[Engineer E4, field X, status Status dead]"
-class MultiLevelTest(ORMTest):
+class MultiLevelTest(_base.MappedTest):
def define_tables(self, metadata):
global table_Employee, table_Engineer, table_Manager
table_Employee = Table( 'Employee', metadata,
assert set(session.query( Engineer).all()) == set([b,c])
assert session.query( Manager).all() == [c]
-class ManyToManyPolyTest(ORMTest):
+class ManyToManyPolyTest(_base.MappedTest):
def define_tables(self, metadata):
global base_item_table, item_table, base_item_collection_table, collection_table
base_item_table = Table(
class_mapper(BaseItem)
-class CustomPKTest(ORMTest):
+class CustomPKTest(_base.MappedTest):
def define_tables(self, metadata):
global t1, t2
t1 = Table('t1', metadata,
ot1.data = 'hi'
sess.flush()
-class InheritingEagerTest(ORMTest):
+class InheritingEagerTest(_base.MappedTest):
def define_tables(self, metadata):
global people, employees, tags, peopleTags
def test_basic(self):
"""test that Query uses the full set of mapper._eager_loaders when generating SQL"""
- class Person(fixtures.Base):
+ class Person(_fixtures.Base):
pass
class Employee(Person):
def __init__(self, name='bob'):
self.name = name
- class Tag(fixtures.Base):
+ class Tag(_fixtures.Base):
def __init__(self, label):
self.label = label
instance = session.query(Employee).filter_by(id=1).limit(1).first()
assert len(instance.tags) == 2
-class MissingPolymorphicOnTest(ORMTest):
+class MissingPolymorphicOnTest(_base.MappedTest):
def define_tables(self, metadata):
global tablea, tableb, tablec, tabled
tablea = Table('tablea', metadata,
)
def test_polyon_col_setsup(self):
- class A(fixtures.Base):
+ class A(_fixtures.Base):
pass
- class B(fixtures.Base):
+ class B(_fixtures.Base):
pass
class C(A):
pass
from datetime import datetime
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
-class InheritTest(ORMTest):
+from testlib import testing
+from orm import _base
+
+
+class InheritTest(_base.MappedTest):
"""tests some various inheritance round trips involving a particular set of polymorphic inheritance relationships"""
def define_tables(self, metadata):
global products_table, specification_table, documents_table
-"""tests the Query object's ability to work with polymorphic selectables
-and inheriting mappers."""
-
-# TODO: under construction !
-
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy import exc as sa_exc
-from testlib import *
-from testlib import fixtures
-from orm import _base
-from testlib.testing import eq_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine import default
-class Company(fixtures.Base):
+from testlib import AssertsCompiledSQL, testing
+from orm import _base, _fixtures
+from testlib.testing import eq_
+
+class Company(_fixtures.Base):
pass
-class Person(fixtures.Base):
+class Person(_fixtures.Base):
pass
class Engineer(Person):
pass
class Boss(Manager):
pass
-class Machine(fixtures.Base):
+class Machine(_fixtures.Base):
pass
-class Paperwork(fixtures.Base):
+class Paperwork(_fixtures.Base):
pass
def make_test(select_type):
- class PolymorphicQueryTest(ORMTest, AssertsCompiledSQL):
- keep_data = True
- keep_mappers = True
-
+ class PolymorphicQueryTest(_base.MappedTest, AssertsCompiledSQL):
+ run_inserts = 'once'
+ run_setup_mappers = 'once'
+ run_deletes = None
+
def define_tables(self, metadata):
global companies, people, engineers, managers, boss, paperwork, machines
del testclass
-class SelfReferentialTestJoinedToBase(ORMTest):
- keep_mappers = True
+class SelfReferentialTestJoinedToBase(_base.MappedTest):
+ run_setup_mappers = 'once'
def define_tables(self, metadata):
global people, engineers
Column('reports_to_id', Integer, ForeignKey('people.person_id'))
)
+ def setup_mappers(self):
mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
mapper(Engineer, engineers, inherits=Person,
inherit_condition=engineers.c.person_id==people.c.person_id,
sess.query(Engineer).join('reports_to', aliased=True).filter(Person.name=='dogbert').first(),
Engineer(name='dilbert'))
-class SelfReferentialJ2JTest(ORMTest):
- keep_mappers = True
+class SelfReferentialJ2JTest(_base.MappedTest):
+ run_setup_mappers = 'once'
def define_tables(self, metadata):
- global people, engineers
+ global people, engineers, managers
people = Table('people', metadata,
Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
Column('name', String(50)),
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
)
+ def setup_mappers(self):
mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
mapper(Manager, managers, inherits=Person, polymorphic_identity='manager')
-class M2MFilterTest(ORMTest):
- keep_mappers = True
- keep_data = True
+class M2MFilterTest(_base.MappedTest):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
def define_tables(self, metadata):
- global people, engineers, Organization
+ global people, engineers, organizations, engineers_to_org
organizations = Table('organizations', metadata,
Column('id', Integer, Sequence('org_id_seq', optional=True), primary_key=True),
Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
Column('primary_language', String(50)),
)
-
- class Organization(fixtures.Base):
+
+ def setup_mappers(self):
+ global Organization
+ class Organization(_fixtures.Base):
pass
mapper(Organization, organizations, properties={
self.assertEquals(sess.query(Organization).filter(Organization.engineers.of_type(Engineer).any(Engineer.name=='e1')).all(), [Organization(name='org1')])
self.assertEquals(sess.query(Organization).filter(Organization.engineers.any(Engineer.name=='e1')).all(), [Organization(name='org1')])
-class SelfReferentialM2MTest(ORMTest, AssertsCompiledSQL):
- keep_mappers = True
+class SelfReferentialM2MTest(_base.MappedTest, AssertsCompiledSQL):
+ run_setup_mappers = 'once'
def define_tables(self, metadata):
+ global Parent, Child1, Child2
+
Base = declarative_base(metadata=metadata)
secondary_table = Table('secondary', Base.metadata,
Column('left_id', Integer, ForeignKey('parent.id'), nullable=False),
Column('right_id', Integer, ForeignKey('parent.id'), nullable=False))
-
- global Parent, Child1, Child2
+
class Parent(Base):
__tablename__ = 'parent'
id = Column(Integer, primary_key=True)
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
-from testlib.fixtures import Base
+from testlib import testing
+from orm._fixtures import Base
+from orm._base import MappedTest
-class InheritingSelectablesTest(ORMTest):
+
+class InheritingSelectablesTest(MappedTest):
def define_tables(self, metadata):
global foo, bar, baz
foo = Table('foo', metadata,
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
-from testlib.fixtures import Base
+
+from testlib import testing
+from orm import _fixtures
from orm._base import MappedTest, ComparableEntity
)
go()
-class SingleOnJoinedTest(ORMTest):
+class SingleOnJoinedTest(MappedTest):
def define_tables(self, metadata):
global persons_table, employees_table
)
def test_single_on_joined(self):
- class Person(Base):
+ class Person(_fixtures.Base):
pass
class Employee(Person):
pass
from sqlalchemy.orm import *
from sqlalchemy.orm import attributes
-from testlib import *
-from orm import _base
-from testlib import engines
-from testlib.fixtures import *
-from testlib import sa, testing
from testlib.testing import eq_
+
+from testlib import sa, testing, AssertsCompiledSQL, Column, engines
+
from orm import _fixtures
+from orm._fixtures import keywords, addresses, Base, Keyword, FixtureTest, \
+ Dingaling, item_keywords, dingalings, User, items,\
+ orders, Address, users, nodes, \
+ order_items, Item, Order, Node
-from sqlalchemy.orm.util import join, outerjoin, with_parent
+from orm import _base
+from sqlalchemy.orm.util import join, outerjoin, with_parent
-class QueryTest(FixtureTest):
- keep_mappers = True
- keep_data = True
+class QueryTest(_fixtures.FixtureTest):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
def setup_mappers(self):
})
mapper(Keyword, keywords)
- compile_mappers()
-
-class UnicodeSchemaTest(QueryTest):
- keep_mappers = False
-
- def setup_mappers(self):
- pass
-
- def define_tables(self, metadata):
- super(UnicodeSchemaTest, self).define_tables(metadata)
- global uni_meta, uni_users
- uni_meta = MetaData()
- uni_users = Table(u'users', uni_meta,
- Column(u'id', Integer, primary_key=True),
- Column(u'name', String(30), nullable=False))
+ mapper(Node, nodes, properties={
+ 'children':relation(Node,
+ backref=backref('parent', remote_side=[nodes.c.id])
+ )
+ })
- def test_get(self):
- mapper(User, uni_users)
- assert User(id=7) == create_session(bind=testing.db).query(User).get(7)
+ compile_mappers()
class RowTupleTest(QueryTest):
- keep_mappers = False
-
- def setup_mappers(self):
- pass
+ run_setup_mappers = None
def test_custom_names(self):
mapper(User, users, properties={
def _test(self, clause, expected):
self.assert_compile(clause, expected, dialect=default.DefaultDialect())
- def define_tables(self, metadata):
- global nodes
- nodes = Table('nodes', metadata,
- Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
- Column('parent_id', Integer, ForeignKey('nodes.id')),
- Column('data', String(30)))
-
- def insert_data(self):
- global Node
-
- class Node(Base):
- pass
-
- mapper(Node, nodes, properties={
- 'children':relation(Node,
- backref=backref('parent', remote_side=[nodes.c.id])
- )
- })
-
def test_arithmetic(self):
create_session().query(User)
for (py_op, sql_op) in ((operator.add, '+'), (operator.mul, '*'),
def test_multiple_entities(self):
sess = create_session()
- if False:
- self.assertEquals(
- sess.query(User, Address).filter(User.id==Address.user_id).filter(Address.id.in_([2, 5]))._from_self().all(),
- [
- (User(id=8), Address(id=2)),
- (User(id=9), Address(id=5))
- ]
- )
+ self.assertEquals(
+ sess.query(User, Address).filter(User.id==Address.user_id).filter(Address.id.in_([2, 5]))._from_self().all(),
+ [
+ (User(id=8), Address(id=2)),
+ (User(id=9), Address(id=5))
+ ]
+ )
self.assertEquals(
sess.query(User, Address).filter(User.id==Address.user_id).filter(Address.id.in_([2, 5]))._from_self().options(eagerload('addresses')).first(),
]
)
-
-class MultiplePathTest(ORMTest):
+class MultiplePathTest(_base.MappedTest):
def define_tables(self, metadata):
global t1, t2, t1t2_1, t1t2_2
t1 = Table('t1', metadata,
create_session().query(T1).join('t2s_1', aliased=True).filter(t2.c.id==5).reset_joinpoint().join('t2s_2').all()
create_session().query(T1).join('t2s_1').filter(t2.c.id==5).reset_joinpoint().join('t2s_2', aliased=True).all()
-
class SynonymTest(QueryTest):
- keep_mappers = True
- keep_data = True
def setup_mappers(self):
mapper(User, users, properties={
def go():
l = list(q.options(contains_alias('ulist'), contains_eager('addresses')).instances(query.execute()))
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def go():
l = q.options(contains_alias('ulist'), contains_eager('addresses')).from_statement(query).all()
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
# better way. use select_from()
def go():
l = sess.query(User).select_from(query).options(contains_eager('addresses')).all()
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
# same thing, but alias addresses, so that the adapter generated by select_from() is wrapped within
query = users.select(users.c.id==7).union(users.select(users.c.id>7)).alias('ulist').outerjoin(adalias).select(use_labels=True,order_by=['ulist.id', adalias.c.id])
def go():
l = sess.query(User).select_from(query).options(contains_eager('addresses', alias=adalias)).all()
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager(self):
"users.name AS users_name FROM users LEFT OUTER JOIN addresses "\
"ON users.id = addresses.user_id ORDER BY users.id"
, dialect=default.DefaultDialect())
-
+
def go():
- assert fixtures.user_address_result == q.all()
+ assert self.static.user_address_result == q.all()
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
adalias = addresses.alias()
q = sess.query(User).select_from(users.outerjoin(adalias)).options(contains_eager(User.addresses, alias=adalias))
def go():
- self.assertEquals(fixtures.user_address_result, q.order_by(User.id).all())
+ self.assertEquals(self.static.user_address_result, q.order_by(User.id).all())
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def go():
l = list(q.options(contains_eager('addresses')).instances(selectquery.execute()))
- assert fixtures.user_address_result[0:3] == l
+ assert self.static.user_address_result[0:3] == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def go():
l = list(q.options(contains_eager(User.addresses)).instances(selectquery.execute()))
- assert fixtures.user_address_result[0:3] == l
+ assert self.static.user_address_result[0:3] == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def go():
l = q.options(contains_eager('addresses')).from_statement(selectquery).all()
- assert fixtures.user_address_result[0:3] == l
+ assert self.static.user_address_result[0:3] == l
self.assert_sql_count(testing.db, go, 1)
def test_contains_eager_alias(self):
selectquery = users.outerjoin(adalias).select(use_labels=True, order_by=[users.c.id, adalias.c.id])
sess = create_session()
q = sess.query(User)
-
+
# string alias name
def go():
l = list(q.options(contains_eager('addresses', alias="adalias")).instances(selectquery.execute()))
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
# expression.Alias object
def go():
l = list(q.options(contains_eager('addresses', alias=adalias)).instances(selectquery.execute()))
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
adalias = aliased(Address)
def go():
l = q.options(contains_eager('addresses', alias=adalias)).outerjoin((adalias, User.addresses)).order_by(User.id, adalias.id)
- assert fixtures.user_address_result == l.all()
+ assert self.static.user_address_result == l.all()
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
# test using string alias with more than one level deep
def go():
l = list(q.options(contains_eager('orders', alias='o1'), contains_eager('orders.items', alias='i1')).instances(query.execute()))
- assert fixtures.user_order_result == l
+ assert self.static.user_order_result == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
# test using Alias with more than one level deep
def go():
l = list(q.options(contains_eager('orders', alias=oalias), contains_eager('orders.items', alias=ialias)).instances(query.execute()))
- assert fixtures.user_order_result == l
+ assert self.static.user_order_result == l
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def go():
l = q.options(contains_eager(User.orders, alias=oalias), contains_eager(User.orders, Order.items, alias=ialias)).\
outerjoin((oalias, User.orders), (ialias, oalias.items)).order_by(User.id, oalias.id, ialias.id)
- assert fixtures.user_order_result == l.all()
+ assert self.static.user_order_result == l.all()
self.assert_sql_count(testing.db, go, 1)
sess.expunge_all()
def test_mixed_eager_contains_with_limit(self):
sess = create_session()
-
+
q = sess.query(User)
def go():
# outerjoin to User.orders, offset 1/limit 2 so we get user 7 + second two orders.
Order(address_id=None,user_id=7,description=u'order 5',isopen=0,id=5)
])])
self.assert_sql_count(testing.db, go, 1)
-
-
+
+
class MixedEntitiesTest(QueryTest):
def test_values(self):
q = sess.query(User)
q2 = q.select_from(sel).values(User.name)
self.assertEquals(list(q2), [(u'jack',), (u'ed',)])
-
+
q = sess.query(User)
q2 = q.order_by(User.id).values(User.name, User.name + " " + cast(User.id, String))
self.assertEquals(list(q2), [(u'jack', u'jack 7'), (u'ed', u'ed 8'), (u'fred', u'fred 9'), (u'chuck', u'chuck 10')])
-
+
q2 = q.join('addresses').filter(User.name.like('%e%')).order_by(User.id, Address.id).values(User.name, Address.email_address)
self.assertEquals(list(q2), [(u'ed', u'ed@wood.com'), (u'ed', u'ed@bettyboop.com'), (u'ed', u'ed@lala.com'), (u'fred', u'fred@fred.com')])
-
+
q2 = q.join('addresses').filter(User.name.like('%e%')).order_by(desc(Address.email_address)).slice(1, 3).values(User.name, Address.email_address)
self.assertEquals(list(q2), [(u'ed', u'ed@wood.com'), (u'ed', u'ed@lala.com')])
-
+
adalias = aliased(Address)
q2 = q.join(('addresses', adalias)).filter(User.name.like('%e%')).values(User.name, adalias.email_address)
self.assertEquals(list(q2), [(u'ed', u'ed@wood.com'), (u'ed', u'ed@bettyboop.com'), (u'ed', u'ed@lala.com'), (u'fred', u'fred@fred.com')])
-
+
q2 = q.values(func.count(User.name))
assert q2.next() == (4,)
def test_correlated_subquery(self):
"""test that a subquery constructed from ORM attributes doesn't leak out
those entities to the outermost query.
-
+
"""
sess = create_session()
-
+
subq = select([func.count()]).\
where(User.id==Address.user_id).\
correlate(users).\
list(sess.query(User, subq)[0:3]),
[(User(id=7,name=u'jack'), 1), (User(id=8,name=u'ed'), 3), (User(id=9,name=u'fred'), 1)]
)
-
+
def test_tuple_labeling(self):
sess = create_session()
for row in sess.query(User, Address).join(User.addresses).all():
self.assertEquals(set(row.keys()), set(['User', 'Address']))
self.assertEquals(row.User, row[0])
self.assertEquals(row.Address, row[1])
-
+
for row in sess.query(User.name, User.id.label('foobar')):
self.assertEquals(set(row.keys()), set(['name', 'foobar']))
self.assertEquals(row.name, row[0])
sess = create_session()
self.assertEquals(sess.query(User.name).all(), [(u'jack',), (u'ed',), (u'fred',), (u'chuck',)])
-
+
sel = users.select(User.id.in_([7, 8])).alias()
q = sess.query(User.name)
q2 = q.select_from(sel).all()
(u'ed', u'ed@bettyboop.com'), (u'ed', u'ed@lala.com'),
(u'fred', u'fred@fred.com')
])
-
+
self.assertEquals(sess.query(User.name, func.count(Address.email_address)).outerjoin(User.addresses).group_by(User.id, User.name).order_by(User.id).all(),
[(u'jack', 1), (u'ed', 3), (u'fred', 1), (u'chuck', 0)]
)
self.assertEquals(sess.query(func.count(Address.email_address), User).outerjoin(User.addresses).group_by(User).order_by(User.id).all(),
[(1, User(name='jack',id=7)), (3, User(name='ed',id=8)), (1, User(name='fred',id=9)), (0, User(name='chuck',id=10))]
)
-
+
adalias = aliased(Address)
self.assertEquals(sess.query(User, func.count(adalias.email_address)).outerjoin(('addresses', adalias)).group_by(User).order_by(User.id).all(),
[(User(name='jack',id=7), 1), (User(name='ed',id=8), 3), (User(name='fred',id=9), 1), (User(name='chuck',id=10), 0)]
(User(name=u'chuck',id=10), None)
]
)
-
+
# anon + select from aliasing
self.assertEquals(
sess.query(User).join(User.addresses, aliased=True).filter(Address.email_address.like('%ed%')).from_self().all(),
def test_column_from_limited_eagerload(self):
sess = create_session()
-
+
def go():
results = sess.query(User).limit(1).options(eagerload('addresses')).add_column(User.name).all()
self.assertEquals(results, [(User(name='jack'), 'jack')])
self.assert_sql_count(testing.db, go, 1)
-
+
def test_self_referential(self):
-
+
sess = create_session()
oalias = aliased(Order)
for q in [
sess.query(Order, oalias).filter(Order.user_id==oalias.user_id).filter(Order.user_id==7).filter(Order.id>oalias.id).order_by(Order.id, oalias.id),
sess.query(Order, oalias)._from_self().filter(Order.user_id==oalias.user_id).filter(Order.user_id==7).filter(Order.id>oalias.id).order_by(Order.id, oalias.id),
-
+
# same thing, but reversed.
sess.query(oalias, Order)._from_self().filter(oalias.user_id==Order.user_id).filter(oalias.user_id==7).filter(Order.id<oalias.id).order_by(oalias.id, Order.id),
-
+
# here we go....two layers of aliasing
sess.query(Order, oalias).filter(Order.user_id==oalias.user_id).filter(Order.user_id==7).filter(Order.id>oalias.id)._from_self().order_by(Order.id, oalias.id).limit(10).options(eagerload(Order.items)),
sess.query(Order, oalias).filter(Order.user_id==oalias.user_id).filter(Order.user_id==7).filter(Order.id>oalias.id)._from_self()._from_self()._from_self().order_by(Order.id, oalias.id).limit(10).options(eagerload(Order.items)),
]:
-
+
self.assertEquals(
q.all(),
[
(Order(address_id=None,description=u'order 5',isopen=0,user_id=7,id=5), Order(address_id=1,description=u'order 3',isopen=1,user_id=7,id=3))
]
)
-
+
def test_multi_mappers(self):
test_session = create_session()
sess.expunge_all()
self.assertRaises(sa_exc.InvalidRequestError, sess.query(User).add_column, object())
-
+
def test_add_multi_columns(self):
"""test that add_column accepts a FROM clause."""
-
+
sess = create_session()
-
+
eq_(
sess.query(User.id).add_column(users).all(),
[(7, 7, u'jack'), (8, 8, u'ed'), (9, 9, u'fred'), (10, 10, u'chuck')]
)
-
+
def test_multi_columns_2(self):
"""test aliased/nonalised joins with the usage of add_column()"""
sess = create_session()
q = q.group_by([c for c in users.c]).order_by(User.id).outerjoin('addresses').add_column(func.count(Address.id).label('count'))
self.assertEquals(q.all(), expected)
sess.expunge_all()
-
+
adalias = aliased(Address)
q = sess.query(User)
q = q.group_by([c for c in users.c]).order_by(User.id).outerjoin(('addresses', adalias)).add_column(func.count(adalias.id).label('count'))
assert q.all() == expected
sess.expunge_all()
-
class ImmediateTest(_fixtures.FixtureTest):
run_inserts = 'once'
run_deletes = None
class SelectFromTest(QueryTest):
- keep_mappers = False
-
- def setup_mappers(self):
- pass
+ run_setup_mappers = None
def test_replace_with_select(self):
mapper(User, users, properties = {
def test_join_mapper_order_by(self):
"""test that mapper-level order_by is adapted to a selectable."""
-
+
mapper(User, users, order_by=users.c.id)
sel = users.select(users.c.id.in_([7, 8]))
(User(name='ed',id=8), Address(user_id=8,email_address='ed@lala.com',id=4))
]
)
-
+
def test_more_joins(self):
mapper(User, users, properties={
'orders':relation(Order, backref='user'), # o2m, m2o
sel = users.select(users.c.id.in_([7, 8]))
sess = create_session()
-
+
# TODO: remove
sess.query(User).select_from(sel).options(eagerload_all('orders.items.keywords')).join('orders', 'items', 'keywords', aliased=True).filter(Keyword.name.in_(['red', 'big', 'round'])).all()
def go():
self.assertEquals(sess.query(User).options(eagerload('addresses')).select_from(sel).order_by(User.id)[1], User(id=8, addresses=[Address(id=2), Address(id=3), Address(id=4)]))
self.assert_sql_count(testing.db, go, 1)
-
-class CustomJoinTest(QueryTest):
- keep_mappers = False
- def setup_mappers(self):
- pass
+class CustomJoinTest(QueryTest):
+ run_setup_mappers = None
def test_double_same_mappers(self):
"""test aliasing of joins with a custom join condition"""
assert [User(id=7)] == q.join(['open_orders', 'items'], aliased=True).filter(Item.id==4).join(['closed_orders', 'items'], aliased=True).filter(Item.id==3).all()
-class SelfReferentialTest(ORMTest):
- keep_mappers = True
- keep_data = True
-
+class SelfReferentialTest(_base.MappedTest):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
+
def define_tables(self, metadata):
global nodes
nodes = Table('nodes', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('parent_id', Integer, ForeignKey('nodes.id')),
Column('data', String(30)))
-
+
def insert_data(self):
global Node
-
+
class Node(Base):
def append(self, node):
self.children.append(node)
sess.add(n1)
sess.flush()
sess.close()
-
+
def test_join(self):
sess = create_session()
ret = sess.query(Node.data).join(Node.children, aliased=True).filter_by(data='n122').all()
assert ret == [('n12',)]
-
+
node = sess.query(Node).join(['children', 'children'], aliased=True).filter_by(data='n122').first()
assert node.data=='n1'
node = sess.query(Node).filter_by(data='n122').join('parent', aliased=True).filter_by(data='n12').\
join('parent', aliased=True, from_joinpoint=True).filter_by(data='n1').first()
assert node.data == 'n122'
-
+
def test_explicit_join(self):
sess = create_session()
-
+
n1 = aliased(Node)
n2 = aliased(Node)
-
+
node = sess.query(Node).select_from(join(Node, n1, 'children')).filter(n1.data=='n122').first()
assert node.data=='n12'
-
+
node = sess.query(Node).select_from(join(Node, n1, 'children').join(n2, 'children')).\
filter(n2.data=='n122').first()
assert node.data=='n1'
-
+
# mix explicit and named onclauses
node = sess.query(Node).select_from(join(Node, n1, Node.id==n1.parent_id).join(n2, 'children')).\
filter(n2.data=='n122').first()
list(sess.query(Node).select_from(join(Node, n1, 'parent').join(n2, 'parent')).\
filter(and_(Node.data=='n122', n1.data=='n12', n2.data=='n1')).values(Node.data, n1.data, n2.data)),
[('n122', 'n12', 'n1')])
-
+
def test_join_to_nonaliased(self):
sess = create_session()
-
+
n1 = aliased(Node)
# using 'n1.parent' implicitly joins to unaliased Node
sess.query(n1).join(n1.parent).filter(Node.data=='n1').all(),
[Node(parent_id=1,data=u'n11',id=2), Node(parent_id=1,data=u'n12',id=3), Node(parent_id=1,data=u'n13',id=4)]
)
-
+
# explicit (new syntax)
self.assertEquals(
sess.query(n1).join((Node, n1.parent)).filter(Node.data=='n1').all(),
[Node(parent_id=1,data=u'n11',id=2), Node(parent_id=1,data=u'n12',id=3), Node(parent_id=1,data=u'n13',id=4)]
)
-
+
def test_multiple_explicit_entities(self):
sess = create_session()
-
+
parent = aliased(Node)
grandparent = aliased(Node)
self.assertEquals(
options(eagerload(Node.children)).first(),
(Node(data='n122'), Node(data='n12'), Node(data='n1'))
)
-
-
+
+
def test_any(self):
sess = create_session()
self.assertEquals(sess.query(Node).filter(Node.children.any(Node.data=='n1')).all(), [])
def test_has(self):
sess = create_session()
-
+
self.assertEquals(sess.query(Node).filter(Node.parent.has(Node.data=='n12')).all(), [Node(data='n121'),Node(data='n122'),Node(data='n123')])
self.assertEquals(sess.query(Node).filter(Node.parent.has(Node.data=='n122')).all(), [])
self.assertEquals(sess.query(Node).filter(~Node.parent.has()).all(), [Node(data='n1')])
-
+
def test_contains(self):
sess = create_session()
-
+
n122 = sess.query(Node).filter(Node.data=='n122').one()
self.assertEquals(sess.query(Node).filter(Node.children.contains(n122)).all(), [Node(data='n12')])
n13 = sess.query(Node).filter(Node.data=='n13').one()
self.assertEquals(sess.query(Node).filter(Node.children.contains(n13)).all(), [Node(data='n1')])
-
+
def test_eq_ne(self):
sess = create_session()
-
+
n12 = sess.query(Node).filter(Node.data=='n12').one()
self.assertEquals(sess.query(Node).filter(Node.parent==n12).all(), [Node(data='n121'),Node(data='n122'),Node(data='n123')])
-
+
self.assertEquals(sess.query(Node).filter(Node.parent != n12).all(), [Node(data='n1'), Node(data='n11'), Node(data='n12'), Node(data='n13')])
-class SelfReferentialM2MTest(ORMTest):
- keep_mappers = True
- keep_data = True
-
+class SelfReferentialM2MTest(_base.MappedTest):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
+
def define_tables(self, metadata):
global nodes, node_to_nodes
nodes = Table('nodes', metadata,
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
Column('data', String(30)))
-
+
node_to_nodes =Table('node_to_nodes', metadata,
Column('left_node_id', Integer, ForeignKey('nodes.id'),primary_key=True),
Column('right_node_id', Integer, ForeignKey('nodes.id'),primary_key=True),
)
-
+
def insert_data(self):
global Node
-
+
class Node(Base):
pass
n5 = Node(data='n5')
n6 = Node(data='n6')
n7 = Node(data='n7')
-
+
n1.children = [n2, n3, n4]
n2.children = [n3, n6, n7]
n3.children = [n5, n4]
self.assertEquals(sess.query(Node).filter(Node.children.contains(n4)).order_by(Node.data).all(), [Node(data='n1'), Node(data='n3')])
self.assertEquals(sess.query(Node).filter(not_(Node.children.contains(n4))).order_by(Node.data).all(), [Node(data='n2'), Node(data='n4'), Node(data='n5'), Node(data='n6'), Node(data='n7')])
-
+
def test_explicit_join(self):
sess = create_session()
-
+
n1 = aliased(Node)
self.assertEquals(
sess.query(Node).select_from(join(Node, n1, 'children')).filter(n1.data.in_(['n3', 'n7'])).order_by(Node.id).all(),
[Node(data='n1'), Node(data='n2')]
)
-
+
class ExternalColumnsTest(QueryTest):
"""test mappers with SQL-expressions added as column properties."""
-
- keep_mappers = False
- def setup_mappers(self):
- pass
+ run_setup_mappers = None
def test_external_columns_bad(self):
def test_external_columns(self):
"""test querying mappings that reference external columns or selectables."""
-
+
mapper(User, users, properties={
'concat': column_property((users.c.id * 2)),
'count': column_property(select([func.count(addresses.c.id)], users.c.id==addresses.c.user_id).correlate(users).as_scalar())
})
sess = create_session()
-
+
sess.query(Address).options(eagerload('user')).all()
self.assertEquals(sess.query(User).all(),
def go():
self.assertEquals(sess.query(Address).options(eagerload('user')).all(), address_result)
self.assert_sql_count(testing.db, go, 1)
-
+
ualias = aliased(User)
self.assertEquals(
sess.query(Address, ualias).join(('user', ualias)).all(),
Column('id', Integer, ForeignKey('base.id'), ForeignKey('sub1.id'), primary_key=True),
Column('data', String(50))
)
-
+
def test_equivs(self):
class Base(_base.ComparableEntity):
pass
pass
class Sub2(_base.ComparableEntity):
pass
-
+
mapper(Base, base, properties={
'sub1':relation(Sub1),
'sub2':relation(Sub2)
})
-
+
mapper(Sub1, sub1)
mapper(Sub2, sub2)
sess = create_session()
-
+
s11 = Sub1(data='s11')
s12 = Sub1(data='s12')
s2 = Sub2(data='s2')
sess.add(b1)
sess.add(b2)
sess.flush()
-
+
# theres an overlapping ForeignKey here, so not much option except
# to artifically control the flush order
b2.sub2 = [s2]
sess.flush()
-
+
q = sess.query(Base).outerjoin('sub2', aliased=True)
assert sub1.c.id not in q._filter_aliases.equivalents
filter(Sub1.id==1).one(),
b1
)
-
+
class UpdateDeleteTest(_base.MappedTest):
def define_tables(self, metadata):
Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('user_id', None, ForeignKey('users.id')),
Column('title', String(32)))
-
+
def setup_classes(self):
class User(_base.ComparableEntity):
pass
class Document(_base.ComparableEntity):
pass
-
+
@testing.resolve_artifact_names
def insert_data(self):
users.insert().execute([
dict(id=2, user_id=1, title='bar'),
dict(id=3, user_id=2, title='baz'),
])
-
+
@testing.resolve_artifact_names
def setup_mappers(self):
mapper(User, users)
mapper(Document, documents, properties={
'user': relation(User, lazy=False, backref=backref('documents', lazy=True))
})
-
+
@testing.resolve_artifact_names
def test_delete(self):
sess = create_session(bind=testing.db, autocommit=False)
-
+
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).delete()
-
+
assert john not in sess and jill not in sess
-
+
eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
@testing.resolve_artifact_names
assert john not in sess and jill not in sess
sess.rollback()
assert john in sess and jill in sess
-
+
@testing.resolve_artifact_names
def test_delete_without_session_sync(self):
sess = create_session(bind=testing.db, autocommit=False)
-
+
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).delete(synchronize_session=False)
-
+
assert john in sess and jill in sess
-
- eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
+ eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
+
@testing.resolve_artifact_names
def test_delete_with_fetch_strategy(self):
sess = create_session(bind=testing.db, autocommit=False)
-
+
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
sess.query(User).filter(or_(User.name == 'john', User.name == 'jill')).delete(synchronize_session='fetch')
-
+
assert john not in sess and jill not in sess
-
- eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
+ eq_(sess.query(User).order_by(User.id).all(), [jack,jane])
+
@testing.fails_on('mysql', 'FIXME: unknown')
@testing.resolve_artifact_names
def test_delete_fallback(self):
sess = create_session(bind=testing.db, autocommit=False)
-
+
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
sess.query(User).filter(User.name == select([func.max(User.name)])).delete(synchronize_session='evaluate')
-
+
assert john not in sess
-
- eq_(sess.query(User).order_by(User.id).all(), [jack,jill,jane])
+ eq_(sess.query(User).order_by(User.id).all(), [jack,jill,jane])
+
@testing.resolve_artifact_names
def test_update(self):
sess = create_session(bind=testing.db, autocommit=False)
-
+
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
sess.query(User).filter(User.age > 29).update({'age': User.age - 10}, synchronize_session='evaluate')
-
+
eq_([john.age, jack.age, jill.age, jane.age], [25,37,29,27])
eq_(sess.query(User.age).order_by(User.id).all(), zip([25,37,29,27]))
sess = create_session(bind=testing.db, autocommit=False, autoflush=False)
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
-
+
john.age = 50
jack.age = 37
-
+
# autoflush is false. therefore our '50' and '37' are getting blown away by this operation.
-
+
sess.query(User).filter(User.age > 29).update({'age': User.age - 10}, synchronize_session='evaluate')
for x in (john, jack, jill, jane):
assert not sess.is_modified(x)
eq_([john.age, jack.age, jill.age, jane.age], [25,37,29,27])
-
+
john.age = 25
assert john in sess.dirty
assert jack in sess.dirty
assert jill not in sess.dirty
assert sess.is_modified(john)
assert not sess.is_modified(jack)
-
-
+
+
@testing.resolve_artifact_names
def test_update_with_expire_strategy(self):
sess = create_session(bind=testing.db, autocommit=False)
-
+
john,jack,jill,jane = sess.query(User).order_by(User.id).all()
sess.query(User).filter(User.age > 29).update({'age': User.age - 10}, synchronize_session='expire')
-
+
eq_([john.age, jack.age, jill.age, jane.age], [25,37,29,27])
eq_(sess.query(User.age).order_by(User.id).all(), zip([25,37,29,27]))
assert b in sess
assert len(list(sess)) == 1
-class DisposedStates(testing.ORMTest):
- keep_mappers = True
- keep_tables = True
+class DisposedStates(_base.MappedTest):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
def define_tables(self, metadata):
global t1
import testenv; testenv.configure_for_tests()
-import operator
+
from sqlalchemy import *
from sqlalchemy.orm import attributes
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import *
-from testlib import *
-from testlib.fixtures import *
+
+from testlib import testing
+from orm import _base
+from orm._fixtures import FixtureTest, User, Address, users, addresses
+
import gc
class TransactionTest(FixtureTest):
- keep_mappers = True
+ run_setup_mappers = 'once'
+ run_inserts = None
session = sessionmaker()
def setup_mappers(self):
class FixtureDataTest(TransactionTest):
- refresh_data = True
-
+ run_inserts = 'each'
+
def test_attrs_on_rollback(self):
sess = self.session()
u1 = sess.query(User).get(7)
assert u1.name == 'will'
class AutoExpireTest(TransactionTest):
- tables_only = True
def test_expunge_pending_on_rollback(self):
sess = self.session()
assert u1.name == 'will'
class TwoPhaseTest(TransactionTest):
- only_tables = True
@testing.requires.two_phase_transactions
def test_rollback_on_prepare(self):
assert u not in s
class RollbackRecoverTest(TransactionTest):
- only_tables = True
def test_pk_violation(self):
s = self.session()
class SavepointTest(TransactionTest):
- only_tables = True
-
@testing.requires.savepoints
def test_savepoint_rollback(self):
s = self.session()
class AccountingFlagsTest(TransactionTest):
-
def test_no_expire_on_commit(self):
sess = sessionmaker(expire_on_commit=False)()
u1 = User(name='ed')
assert testing.db.execute(select([users.c.name])).fetchall() == [('ed',)]
-class AutocommitTest(TransactionTest):
+class AutoCommitTest(TransactionTest):
def test_begin_nested_requires_trans(self):
sess = create_session(autocommit=True)
self.assertRaises(sa_exc.InvalidRequestError, sess.begin_nested)
import testenv; testenv.configure_for_tests()
from sqlalchemy.orm import interfaces, util
-from testlib import *
-from testlib import fixtures
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import Table
from sqlalchemy.orm import aliased
from sqlalchemy.orm import mapper, create_session
+
+
+from testlib import TestBase, testing
+
from orm import _fixtures
from testlib.testing import eq_
import testenv; testenv.configure_for_tests()
from sqlalchemy import *
from sqlalchemy.orm import *
-from testlib import *
-from testlib import fixtures
import gc
+from testlib import TestBase, AssertsExecutionResults, profiling, testing
+from orm import _fixtures
+
# in this test we are specifically looking for time spent in the attributes.InstanceState.__cleanup() method.
ITERATIONS = 100
l.append({'c2':'this is t2 #%d' % y, 't1id':x})
t2.insert().execute(*l)
- class T1(fixtures.Base):
+ class T1(_fixtures.Base):
pass
- class T2(fixtures.Base):
+ class T2(_fixtures.Base):
pass
mapper(T1, t1, properties={
AssertsCompiledSQL, \
AssertsExecutionResults, \
ComparesTables, \
- ORMTest, \
TestBase, \
rowset
from testlib.orm import mapper
'mapper',
'Table', 'Column',
'rowset',
- 'TestBase', 'AssertsExecutionResults', 'ORMTest',
+ 'TestBase', 'AssertsExecutionResults',
'AssertsCompiledSQL', 'ComparesTables',
'profiling', 'engines',
'_function_named')
+++ /dev/null
-from testlib.sa import MetaData, Table, Column, Integer, String, ForeignKey
-from testlib.sa.orm import attributes
-from testlib.testing import ORMTest
-
-
-__all__ = ['keywords', 'addresses', 'Base', 'Keyword', 'FixtureTest',
- 'Dingaling', 'item_keywords', 'dingalings', 'User', 'items',
- 'Fixtures', 'orders', 'install_fixture_data', 'Address', 'users',
- 'order_items', 'Item', 'Order', 'fixtures']
-
-
-_recursion_stack = set()
-class Base(object):
- def __init__(self, **kwargs):
- for k in kwargs:
- setattr(self, k, kwargs[k])
-
- # TODO: add recursion checks to this
- def __repr__(self):
- return "%s(%s)" % (
- (self.__class__.__name__),
- ','.join(["%s=%s" % (key, repr(getattr(self, key))) for key in self.__dict__ if not key.startswith('_')])
- )
-
- def __ne__(self, other):
- return not self.__eq__(other)
-
- __hash__ = object.__hash__
-
- def __eq__(self, other):
- """'passively' compare this object to another.
-
- only look at attributes that are present on the source object.
-
- """
-
- if self in _recursion_stack:
- return True
- _recursion_stack.add(self)
- try:
- # pick the entity thats not SA persisted as the source
- try:
- state = attributes.instance_state(self)
- key = state.key
- except (KeyError, AttributeError):
- key = None
- if other is None:
- a = self
- b = other
- elif key is not None:
- a = other
- b = self
- else:
- a = self
- b = other
-
- for attr in a.__dict__.keys():
- if attr[0] == '_':
- continue
- value = getattr(a, attr)
- #print "looking at attr:", attr, "start value:", value
- if hasattr(value, '__iter__') and not isinstance(value, basestring):
- try:
- # catch AttributeError so that lazy loaders trigger
- battr = getattr(b, attr)
- except AttributeError:
- #print "b class does not have attribute named '%s'" % attr
- #raise
- return False
-
- if list(value) == list(battr):
- continue
- else:
- return False
- else:
- if value is not None:
- if value != getattr(b, attr, None):
- #print "2. Attribute named '%s' does not match that of b" % attr
- return False
- else:
- return True
- finally:
- _recursion_stack.remove(self)
-
-class User(Base):pass
-class Order(Base):pass
-class Item(Base):pass
-class Keyword(Base):pass
-class Address(Base):pass
-class Dingaling(Base):pass
-
-metadata = MetaData()
-
-users = Table('users', metadata,
- Column('id', Integer, primary_key=True),
- Column('name', String(30), nullable=False),
- test_needs_acid=True,
- test_needs_fk=True
- )
-
-orders = Table('orders', metadata,
- Column('id', Integer, primary_key=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
- )
-
-addresses = Table('addresses', metadata,
- Column('id', Integer, primary_key=True),
- Column('user_id', None, ForeignKey('users.id')),
- Column('email_address', String(50), nullable=False),
- test_needs_acid=True,
- test_needs_fk=True)
-
-dingalings = Table("dingalings", metadata,
- Column('id', Integer, primary_key=True),
- Column('address_id', None, ForeignKey('addresses.id')),
- Column('data', String(30)),
- test_needs_acid=True,
- test_needs_fk=True
- )
-
-items = Table('items', metadata,
- Column('id', Integer, primary_key=True),
- Column('description', String(30), nullable=False),
- test_needs_acid=True,
- test_needs_fk=True
- )
-
-order_items = Table('order_items', metadata,
- Column('item_id', None, ForeignKey('items.id')),
- Column('order_id', None, ForeignKey('orders.id')),
- test_needs_acid=True,
- test_needs_fk=True)
-
-item_keywords = Table('item_keywords', metadata,
- Column('item_id', None, ForeignKey('items.id')),
- Column('keyword_id', None, ForeignKey('keywords.id')),
- test_needs_acid=True,
- test_needs_fk=True)
-
-keywords = Table('keywords', metadata,
- Column('id', Integer, primary_key=True),
- Column('name', String(30), nullable=False),
- test_needs_acid=True,
- test_needs_fk=True
- )
-
-def install_fixture_data():
- users.insert().execute(
- dict(id = 7, name = 'jack'),
- dict(id = 8, name = 'ed'),
- dict(id = 9, name = 'fred'),
- dict(id = 10, name = 'chuck'),
-
- )
- addresses.insert().execute(
- dict(id = 1, user_id = 7, email_address = "jack@bean.com"),
- dict(id = 2, user_id = 8, email_address = "ed@wood.com"),
- dict(id = 3, user_id = 8, email_address = "ed@bettyboop.com"),
- dict(id = 4, user_id = 8, email_address = "ed@lala.com"),
- dict(id = 5, user_id = 9, email_address = "fred@fred.com"),
- )
- dingalings.insert().execute(
- dict(id=1, address_id=2, data='ding 1/2'),
- dict(id=2, address_id=5, data='ding 2/5'),
- )
- orders.insert().execute(
- dict(id = 1, user_id = 7, description = 'order 1', isopen=0, address_id=1),
- dict(id = 2, user_id = 9, description = 'order 2', isopen=0, address_id=4),
- dict(id = 3, user_id = 7, description = 'order 3', isopen=1, address_id=1),
- dict(id = 4, user_id = 9, description = 'order 4', isopen=1, address_id=4),
- dict(id = 5, user_id = 7, description = 'order 5', isopen=0, address_id=None)
- )
- items.insert().execute(
- dict(id=1, description='item 1'),
- dict(id=2, description='item 2'),
- dict(id=3, description='item 3'),
- dict(id=4, description='item 4'),
- dict(id=5, description='item 5'),
- )
- order_items.insert().execute(
- dict(item_id=1, order_id=1),
- dict(item_id=2, order_id=1),
- dict(item_id=3, order_id=1),
-
- dict(item_id=1, order_id=2),
- dict(item_id=2, order_id=2),
- dict(item_id=3, order_id=2),
-
- dict(item_id=3, order_id=3),
- dict(item_id=4, order_id=3),
- dict(item_id=5, order_id=3),
-
- dict(item_id=1, order_id=4),
- dict(item_id=5, order_id=4),
-
- dict(item_id=5, order_id=5),
- )
- keywords.insert().execute(
- dict(id=1, name='blue'),
- dict(id=2, name='red'),
- dict(id=3, name='green'),
- dict(id=4, name='big'),
- dict(id=5, name='small'),
- dict(id=6, name='round'),
- dict(id=7, name='square')
- )
-
- # this many-to-many table has the keywords inserted
- # in primary key order, to appease the unit tests.
- # this is because postgres, oracle, and sqlite all support
- # true insert-order row id, but of course our pal MySQL does not,
- # so the best it can do is order by, well something, so there you go.
- item_keywords.insert().execute(
- dict(keyword_id=2, item_id=1),
- dict(keyword_id=2, item_id=2),
- dict(keyword_id=4, item_id=1),
- dict(keyword_id=6, item_id=1),
- dict(keyword_id=5, item_id=2),
- dict(keyword_id=3, item_id=3),
- dict(keyword_id=4, item_id=3),
- dict(keyword_id=7, item_id=2),
- dict(keyword_id=6, item_id=3)
- )
-
-class FixtureTest(ORMTest):
- refresh_data = False
- only_tables = False
-
- def setUpAll(self):
- super(FixtureTest, self).setUpAll()
- if not self.only_tables and self.keep_data:
- install_fixture_data()
-
- def setUp(self):
- if not self.only_tables and self.refresh_data:
- install_fixture_data()
-
- def define_tables(self, meta):
- pass
-FixtureTest.metadata = metadata
-
-class Fixtures(object):
- @property
- def user_address_result(self):
- return [
- User(id=7, addresses=[
- Address(id=1)
- ]),
- User(id=8, addresses=[
- Address(id=2, email_address='ed@wood.com'),
- Address(id=3, email_address='ed@bettyboop.com'),
- Address(id=4, email_address='ed@lala.com'),
- ]),
- User(id=9, addresses=[
- Address(id=5)
- ]),
- User(id=10, addresses=[])
- ]
-
- @property
- def user_all_result(self):
- return [
- User(id=7, addresses=[
- Address(id=1)
- ], orders=[
- Order(description='order 1', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]),
- Order(description='order 3'),
- Order(description='order 5'),
- ]),
- User(id=8, addresses=[
- Address(id=2),
- Address(id=3),
- Address(id=4)
- ]),
- User(id=9, addresses=[
- Address(id=5)
- ], orders=[
- Order(description='order 2', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]),
- Order(description='order 4', items=[Item(description='item 1'), Item(description='item 5')]),
- ]),
- User(id=10, addresses=[])
- ]
-
- @property
- def user_order_result(self):
- return [
- User(id=7, orders=[
- Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]),
- Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]),
- Order(id=5, items=[Item(id=5)]),
- ]),
- User(id=8, orders=[]),
- User(id=9, orders=[
- Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]),
- Order(id=4, items=[Item(id=1), Item(id=5)]),
- ]),
- User(id=10)
- ]
-
- @property
- def item_keyword_result(self):
- return [
- Item(id=1, keywords=[Keyword(name='red'), Keyword(name='big'), Keyword(name='round')]),
- Item(id=2, keywords=[Keyword(name='red'), Keyword(name='small'), Keyword(name='square')]),
- Item(id=3, keywords=[Keyword(name='green'), Keyword(name='big'), Keyword(name='round')]),
- Item(id=4, keywords=[]),
- Item(id=5, keywords=[]),
- ]
-fixtures = Fixtures()
sa = None
orm = None
-
__all__ = 'mapper',
self.assert_sql_execution(db, callable_, assertsql.CountStatements(count))
-_otest_metadata = None
-class ORMTest(TestBase, AssertsExecutionResults):
- keep_mappers = False
- keep_data = False
- metadata = None
-
- def setUpAll(self):
- global MetaData, _otest_metadata
-
- if MetaData is None:
- from sqlalchemy import MetaData
-
- if self.metadata is None:
- _otest_metadata = MetaData(config.db)
- else:
- _otest_metadata = self.metadata
- if self.metadata.bind is None:
- _otest_metadata.bind = config.db
- self.define_tables(_otest_metadata)
- _otest_metadata.create_all()
- self.setup_mappers()
- self.insert_data()
-
- def define_tables(self, _otest_metadata):
- raise NotImplementedError()
-
- def setup_mappers(self):
- pass
-
- def insert_data(self):
- pass
-
- def get_metadata(self):
- return _otest_metadata
-
- def tearDownAll(self):
- global clear_mappers
- if clear_mappers is None:
- from sqlalchemy.orm import clear_mappers
-
- clear_mappers()
- _otest_metadata.drop_all()
-
- def tearDown(self):
- global Session
- if Session is None:
- from sqlalchemy.orm.session import Session
- Session.close_all()
- global clear_mappers
- if clear_mappers is None:
- from sqlalchemy.orm import clear_mappers
-
- if not self.keep_mappers:
- clear_mappers()
- if not self.keep_data:
- for t in reversed(_otest_metadata.sorted_tables):
- try:
- t.delete().execute().close()
- except Exception, e:
- print "EXCEPTION DELETING...", e
-
class TTestSuite(unittest.TestSuite):
"""A TestSuite with once per TestCase setUpAll() and tearDownAll()"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
