-from sqlalchemy.testing import eq_, is_
-from sqlalchemy.orm import backref, configure_mappers
-from sqlalchemy import testing
-from sqlalchemy import desc, select, func, exc, cast, Integer
-from sqlalchemy.orm import mapper, relationship, create_session, Query, \
- attributes, exc as orm_exc, Session
+from sqlalchemy import testing, desc, select, func, exc, cast, Integer
+from sqlalchemy.orm import (
+ mapper, relationship, create_session, Query, attributes, exc as orm_exc,
+ Session, backref, configure_mappers)
from sqlalchemy.orm.dynamic import AppenderMixin
-from sqlalchemy.testing import AssertsCompiledSQL, \
- assert_raises_message, assert_raises
+from sqlalchemy.testing import (
+ AssertsCompiledSQL, assert_raises_message, assert_raises, eq_, is_)
from test.orm import _fixtures
-
from sqlalchemy.testing.assertsql import CompiledSQL
-
class _DynamicFixture(object):
def _user_address_fixture(self, addresses_args={}):
users, Address, addresses, User = (self.tables.users,
self.tables.addresses,
self.classes.User)
- mapper(User, users, properties={
- 'addresses': relationship(Address, lazy="dynamic",
- **addresses_args)
- })
+ mapper(
+ User, users, properties={
+ 'addresses': relationship(
+ Address, lazy="dynamic", **addresses_args)})
mapper(Address, addresses)
return User, Address
self.tables.order_items,
self.classes.Item)
- mapper(Order, orders, properties={
- 'items': relationship(Item,
- secondary=order_items,
- lazy="dynamic",
- **items_args
- )
- })
+ mapper(
+ Order, orders, properties={
+ 'items': relationship(
+ Item, secondary=order_items, lazy="dynamic",
+ **items_args)})
mapper(Item, items)
return Order, Item
+
class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
def test_basic(self):
def test_no_uselist_false(self):
User, Address = self._user_address_fixture(
- addresses_args={"uselist": False})
+ addresses_args={"uselist": False})
assert_raises_message(
exc.InvalidRequestError,
"On relationship User.addresses, 'dynamic' loaders cannot be "
self.classes.Address,
self.tables.addresses,
self.classes.User)
- mapper(Address, addresses, properties={
- 'user': relationship(User, lazy='dynamic')
- })
+ mapper(
+ Address, addresses, properties={
+ 'user': relationship(User, lazy='dynamic')})
mapper(User, users)
assert_raises_message(
exc.InvalidRequestError,
u = sess.query(User).get(8)
eq_(
list(u.addresses.order_by(desc(Address.email_address))),
- [
+ [
Address(email_address='ed@wood.com'),
Address(email_address='ed@lala.com'),
Address(email_address='ed@bettyboop.com')
def test_configured_order_by(self):
addresses = self.tables.addresses
User, Address = self._user_address_fixture(
- addresses_args={
- "order_by":
- addresses.c.email_address.desc()})
+ addresses_args={"order_by": addresses.c.email_address.desc()})
sess = create_session()
u = sess.query(User).get(8)
sess = create_session()
ad = sess.query(Address).get(1)
+
def go():
ad.user = None
self.assert_sql_count(testing.db, go, 0)
eq_(
q.filter(User.id == 7).all(),
[
- User(id=7,
- addresses=[
- Address(id=1, email_address='jack@bean.com')
- ])
- ]
- )
+ User(
+ id=7, addresses=[
+ Address(id=1, email_address='jack@bean.com')])])
self.assert_sql_count(testing.db, go, 2)
def test_no_populate(self):
)
def test_m2m(self):
- Order, Item = self._order_item_fixture(items_args={
- "backref": backref("orders", lazy="dynamic")
- })
+ Order, Item = self._order_item_fixture(
+ items_args={"backref": backref("orders", lazy="dynamic")})
sess = create_session()
o1 = Order(id=15, description="order 10")
assert o1 in i1.orders.all()
assert i1 in o1.items.all()
- @testing.exclude('mysql', 'between',
- ((5, 1, 49), (5, 1, 52)),
- 'https://bugs.launchpad.net/ubuntu/+source/mysql-5.1/+bug/706988')
+ @testing.exclude(
+ 'mysql', 'between', ((5, 1, 49), (5, 1, 52)),
+ 'https://bugs.launchpad.net/ubuntu/+source/mysql-5.1/+bug/706988')
def test_association_nonaliased(self):
items, Order, orders, order_items, Item = (self.tables.items,
self.classes.Order,
class MyQuery(Query):
pass
User, Address = self._user_address_fixture(
- addresses_args={"query_class": MyQuery})
+ addresses_args={"query_class": MyQuery})
sess = create_session()
u = User()
query_class = MyQuery
User, Address = self._user_address_fixture(
- addresses_args={
- "query_class": MyAppenderQuery})
-
+ addresses_args={"query_class": MyAppenderQuery})
sess = create_session()
u = User()
eq_(type(q).__name__, 'MyQuery')
-class UOWTest(_DynamicFixture, _fixtures.FixtureTest,
- testing.AssertsExecutionResults):
+class UOWTest(
+ _DynamicFixture, _fixtures.FixtureTest,
+ testing.AssertsExecutionResults):
+
run_inserts = None
def test_persistence(self):
testing.db.scalar(
select(
[func.count(cast(1, Integer))]).
- where(addresses.c.user_id != None)),
- 0
- )
+ where(addresses.c.user_id != None)),
+ 0)
u1 = sess.query(User).get(u1.id)
u1.addresses.append(a1)
sess.flush()
eq_(
testing.db.execute(
- select([addresses]).where(addresses.c.user_id != None)
+ select([addresses]).where(addresses.c.user_id != None)
).fetchall(),
[(a1.id, u1.id, 'foo')]
)
testing.db.scalar(
select(
[func.count(cast(1, Integer))]).
- where(addresses.c.user_id != None)),
+ where(addresses.c.user_id != None)),
0
)
[(a2.id, u1.id, 'bar')]
)
-
def test_merge(self):
addresses = self.tables.addresses
User, Address = self._user_address_fixture(
- addresses_args={
- "order_by": addresses.c.email_address})
+ addresses_args={"order_by": addresses.c.email_address})
sess = create_session()
u1 = User(name='jack')
a1 = Address(email_address='a1')
def test_collection_set(self):
addresses = self.tables.addresses
User, Address = self._user_address_fixture(
- addresses_args={
- "order_by": addresses.c.email_address})
+ addresses_args={"order_by": addresses.c.email_address})
sess = create_session(autoflush=True, autocommit=False)
u1 = User(name='jack')
a1 = Address(email_address='a1')
def test_rollback(self):
User, Address = self._user_address_fixture()
sess = create_session(
- expire_on_commit=False, autocommit=False, autoflush=True)
+ expire_on_commit=False, autocommit=False, autoflush=True)
u1 = User(name='jack')
u1.addresses.append(Address(email_address='lala@hoho.com'))
sess.add(u1)
def _test_delete_cascade(self, expected):
addresses = self.tables.addresses
- User, Address = self._user_address_fixture(addresses_args={
- "order_by": addresses.c.id,
- "backref": "user",
- "cascade": "save-update" if expected \
- else "all, delete"
- })
+ User, Address = self._user_address_fixture(
+ addresses_args={
+ "order_by": addresses.c.id,
+ "backref": "user",
+ "cascade": "save-update" if expected else "all, delete"})
sess = create_session(autoflush=True, autocommit=False)
u = User(name='ed')
)
sess.add(u)
sess.commit()
- eq_(testing.db.scalar(
- addresses.count(addresses.c.user_id == None)), 0)
- eq_(testing.db.scalar(
- addresses.count(addresses.c.user_id != None)), 6)
+ eq_(testing.db.scalar(addresses.count(addresses.c.user_id == None)), 0)
+ eq_(testing.db.scalar(addresses.count(addresses.c.user_id != None)), 6)
sess.delete(u)
sess.commit()
if expected:
- eq_(testing.db.scalar(
+ eq_(
+ testing.db.scalar(
addresses.count(addresses.c.user_id == None)), 6)
- eq_(testing.db.scalar(
+ eq_(
+ testing.db.scalar(
addresses.count(addresses.c.user_id != None)), 0)
else:
eq_(testing.db.scalar(addresses.count()), 0)
def test_self_referential(self):
Node, nodes = self.classes.Node, self.tables.nodes
-
- mapper(Node, nodes, properties={
- 'children': relationship(Node, lazy="dynamic", order_by=nodes.c.id)
- })
+ mapper(
+ Node, nodes, properties={
+ 'children': relationship(
+ Node, lazy="dynamic", order_by=nodes.c.id)})
sess = Session()
n2, n3 = Node(), Node()
eq_(n1.children.all(), [n2, n3])
-
def test_remove_orphans(self):
addresses = self.tables.addresses
- User, Address = self._user_address_fixture(addresses_args={
- "order_by": addresses.c.id,
- "backref": "user",
- "cascade": "all, delete-orphan"
- })
+ User, Address = self._user_address_fixture(
+ addresses_args={
+ "order_by": addresses.c.id,
+ "backref": "user",
+ "cascade": "all, delete-orphan"})
sess = create_session(autoflush=True, autocommit=False)
u = User(name='ed')
)
def _backref_test(self, autoflush, saveuser):
- User, Address = self._user_address_fixture(addresses_args={
- "backref": "user",
- })
+ User, Address = self._user_address_fixture(
+ addresses_args={"backref": "user"})
sess = create_session(autoflush=autoflush, autocommit=False)
u = User(name='buffy')
self._backref_test(False, False)
def test_backref_events(self):
- User, Address = self._user_address_fixture(addresses_args={
- "backref": "user",
- })
+ User, Address = self._user_address_fixture(
+ addresses_args={"backref": "user"})
u1 = User()
a1 = Address()
is_(a1.user, u1)
def test_no_deref(self):
- User, Address = self._user_address_fixture(addresses_args={
- "backref": "user",
- })
+ User, Address = self._user_address_fixture(
+ addresses_args={"backref": "user", })
session = create_session()
user = User()
def query3():
session = create_session(testing.db)
- user = session.query(User).first()
return session.query(User).first().addresses.all()
eq_(query1(), [Address(email_address='joe@joesdomain.example')])
eq_(query2(), [Address(email_address='joe@joesdomain.example')])
eq_(query3(), [Address(email_address='joe@joesdomain.example')])
+
class HistoryTest(_DynamicFixture, _fixtures.FixtureTest):
run_inserts = None
def _transient_fixture(self, addresses_args={}):
User, Address = self._user_address_fixture(
- addresses_args=addresses_args)
+ addresses_args=addresses_args)
u1 = User()
a1 = Address()
def _persistent_fixture(self, autoflush=True, addresses_args={}):
User, Address = self._user_address_fixture(
- addresses_args=addresses_args)
+ addresses_args=addresses_args)
u1 = User(name='u1')
a1 = Address(email_address='a1')
def test_backref_pop_persistent_autoflush_o2m_active_hist(self):
u1, a1, s = self._persistent_fixture(
- addresses_args={"backref":
- backref("user", active_history=True)})
+ addresses_args={"backref": backref("user", active_history=True)})
u1.addresses.append(a1)
s.flush()
s.expire_all()
def test_backref_pop_persistent_autoflush_m2m(self):
o1, i1, s = self._persistent_m2m_fixture(
- items_args={"backref": "orders"})
+ items_args={"backref": "orders"})
o1.items.append(i1)
s.flush()
s.expire_all()
def test_backref_pop_persistent_noflush_m2m(self):
o1, i1, s = self._persistent_m2m_fixture(
- items_args={"backref": "orders"}, autoflush=False)
+ items_args={"backref": "orders"}, autoflush=False)
o1.items.append(i1)
s.flush()
s.expire_all()
u1, a1 = self._transient_fixture()
a2, a3, a4, a5 = Address(email_address='a2'), \
- Address(email_address='a3'), \
- Address(email_address='a4'), \
- Address(email_address='a5')
+ Address(email_address='a3'), Address(email_address='a4'), \
+ Address(email_address='a5')
u1.addresses = [a1, a2]
u1.addresses = [a2, a3, a4, a5]
u1, a1, s = self._persistent_fixture(autoflush=False)
a2, a3, a4, a5 = Address(email_address='a2'), \
- Address(email_address='a3'), \
- Address(email_address='a4'), \
- Address(email_address='a5')
+ Address(email_address='a3'), Address(email_address='a4'), \
+ Address(email_address='a5')
u1.addresses = [a1, a2]
u1.addresses = [a2, a3, a4, a5]
u1, a1, s = self._persistent_fixture(autoflush=True)
a2, a3, a4, a5 = Address(email_address='a2'), \
- Address(email_address='a3'), \
- Address(email_address='a4'), \
- Address(email_address='a5')
+ Address(email_address='a3'), Address(email_address='a4'), \
+ Address(email_address='a5')
u1.addresses = [a1, a2]
u1.addresses = [a2, a3, a4, a5]
compare_passive=([a3, a4, a5], [], [a1])
)
-
def test_persistent_but_readded_noflush(self):
u1, a1, s = self._persistent_fixture(autoflush=False)
u1.addresses.append(a1)
u1.addresses.remove(a1)
- self._assert_history(u1,
- ([], [], []),
- compare_passive=([], [], [a1])
- )
+ self._assert_history(u1, ([], [], []), compare_passive=([], [], [a1]))
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
