summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/orm/inheritance/test_polymorphic_rel.py12
-rw-r--r--test/orm/test_cache_key.py2
-rw-r--r--test/orm/test_cascade.py742
-rw-r--r--test/orm/test_composites.py2
-rw-r--r--test/orm/test_froms.py4
-rw-r--r--test/orm/test_lambdas.py14
-rw-r--r--test/orm/test_query.py16
-rw-r--r--test/orm/test_transaction.py4
-rw-r--r--test/orm/test_update_delete.py18
9 files changed, 504 insertions, 310 deletions
diff --git a/test/orm/inheritance/test_polymorphic_rel.py b/test/orm/inheritance/test_polymorphic_rel.py
index d13b2f945..d8214465a 100644
--- a/test/orm/inheritance/test_polymorphic_rel.py
+++ b/test/orm/inheritance/test_polymorphic_rel.py
@@ -230,7 +230,7 @@ class _PolymorphicTestBase(object):
)
def test_multi_join_future(self):
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
e = aliased(Person)
c = aliased(Company)
@@ -283,7 +283,7 @@ class _PolymorphicTestBase(object):
eq_(sess.query(Engineer).all()[0], Engineer(name="dilbert"))
def test_filter_on_subclass_one_future(self):
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
eq_(
sess.execute(select(Engineer)).scalar(), Engineer(name="dilbert"),
)
@@ -337,7 +337,7 @@ class _PolymorphicTestBase(object):
)
def test_join_from_polymorphic_nonaliased_one_future(self):
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
eq_(
sess.execute(
select(Person)
@@ -396,7 +396,7 @@ class _PolymorphicTestBase(object):
)
def test_join_from_polymorphic_flag_aliased_one_future(self):
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
pa = aliased(Paperwork)
eq_(
@@ -496,7 +496,7 @@ class _PolymorphicTestBase(object):
)
def test_join_from_with_polymorphic_nonaliased_one_future(self):
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
pm = with_polymorphic(Person, [Manager])
eq_(
@@ -1544,7 +1544,7 @@ class _PolymorphicTestBase(object):
# TODO: this is the first test *EVER* of an aliased class of
# an aliased class. we should add many more tests for this.
# new case added in Id810f485c5f7ed971529489b84694e02a3356d6d
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
expected = [(m1, e1), (m1, e2), (m1, b1)]
p1 = aliased(Person)
diff --git a/test/orm/test_cache_key.py b/test/orm/test_cache_key.py
index 4156d606c..02b1b9fbf 100644
--- a/test/orm/test_cache_key.py
+++ b/test/orm/test_cache_key.py
@@ -472,7 +472,7 @@ class RoundTripTest(QueryTest, AssertsCompiledSQL):
# query.
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
def query(names):
stmt = (
diff --git a/test/orm/test_cascade.py b/test/orm/test_cascade.py
index 7e6db3b89..8a21297f1 100644
--- a/test/orm/test_cascade.py
+++ b/test/orm/test_cascade.py
@@ -12,7 +12,6 @@ from sqlalchemy.orm import attributes
from sqlalchemy.orm import backref
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import configure_mappers
-from sqlalchemy.orm import create_session
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import foreign
from sqlalchemy.orm import mapper
@@ -271,68 +270,71 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
def test_list_assignment_new(self):
User, Order = self.classes.User, self.classes.Order
- sess = Session()
- u = User(
- name="jack",
- orders=[
- Order(description="order 1"),
- Order(description="order 2"),
- ],
- )
- sess.add(u)
- sess.commit()
-
- eq_(
- u,
- User(
+ with Session() as sess:
+ u = User(
name="jack",
orders=[
Order(description="order 1"),
Order(description="order 2"),
],
- ),
- )
+ )
+ sess.add(u)
+ sess.commit()
+
+ eq_(
+ u,
+ User(
+ name="jack",
+ orders=[
+ Order(description="order 1"),
+ Order(description="order 2"),
+ ],
+ ),
+ )
def test_list_assignment_replace(self):
User, Order = self.classes.User, self.classes.Order
- sess = Session()
- u = User(
- name="jack",
- orders=[
- Order(description="someorder"),
- Order(description="someotherorder"),
- ],
- )
- sess.add(u)
-
- u.orders = [Order(description="order 3"), Order(description="order 4")]
- sess.commit()
-
- eq_(
- u,
- User(
+ with Session() as sess:
+ u = User(
name="jack",
orders=[
- Order(description="order 3"),
- Order(description="order 4"),
+ Order(description="someorder"),
+ Order(description="someotherorder"),
],
- ),
- )
+ )
+ sess.add(u)
- # order 1, order 2 have been deleted
- eq_(
- sess.query(Order).order_by(Order.id).all(),
- [Order(description="order 3"), Order(description="order 4")],
- )
+ u.orders = [
+ Order(description="order 3"),
+ Order(description="order 4"),
+ ]
+ sess.commit()
+
+ eq_(
+ u,
+ User(
+ name="jack",
+ orders=[
+ Order(description="order 3"),
+ Order(description="order 4"),
+ ],
+ ),
+ )
+
+ # order 1, order 2 have been deleted
+ eq_(
+ sess.query(Order).order_by(Order.id).all(),
+ [Order(description="order 3"), Order(description="order 4")],
+ )
def test_standalone_orphan(self):
Order = self.classes.Order
- sess = Session()
- o5 = Order(description="order 5")
- sess.add(o5)
- assert_raises(sa_exc.DBAPIError, sess.flush)
+ with Session() as sess:
+ o5 = Order(description="order 5")
+ sess.add(o5)
+ assert_raises(sa_exc.DBAPIError, sess.flush)
def test_save_update_sends_pending(self):
"""test that newly added and deleted collection items are
@@ -361,41 +363,41 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
def test_remove_pending_from_collection(self):
User, Order = self.classes.User, self.classes.Order
- sess = Session()
+ with Session() as sess:
- u = User(name="jack")
- sess.add(u)
- sess.commit()
+ u = User(name="jack")
+ sess.add(u)
+ sess.commit()
- o1 = Order()
- u.orders.append(o1)
- assert o1 in sess
- u.orders.remove(o1)
- assert o1 not in sess
+ o1 = Order()
+ u.orders.append(o1)
+ assert o1 in sess
+ u.orders.remove(o1)
+ assert o1 not in sess
def test_remove_pending_from_pending_parent(self):
# test issue #4040
User, Order = self.classes.User, self.classes.Order
- sess = Session()
+ with Session() as sess:
- u = User(name="jack")
+ u = User(name="jack")
- o1 = Order()
- sess.add(o1)
+ o1 = Order()
+ sess.add(o1)
- # object becomes an orphan, but parent is not in session
- u.orders.append(o1)
- u.orders.remove(o1)
+ # object becomes an orphan, but parent is not in session
+ u.orders.append(o1)
+ u.orders.remove(o1)
- sess.add(u)
+ sess.add(u)
- assert o1 in sess
+ assert o1 in sess
- sess.flush()
+ sess.flush()
- assert o1 not in sess
+ assert o1 not in sess
def test_delete(self):
User, users, orders, Order = (
@@ -405,21 +407,31 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
self.classes.Order,
)
- sess = create_session()
- u = User(
- name="jack",
- orders=[
- Order(description="someorder"),
- Order(description="someotherorder"),
- ],
- )
- sess.add(u)
- sess.flush()
-
- sess.delete(u)
- sess.flush()
- eq_(select(func.count("*")).select_from(users).scalar(), 0)
- eq_(select(func.count("*")).select_from(orders).scalar(), 0)
+ with Session() as sess:
+ u = User(
+ name="jack",
+ orders=[
+ Order(description="someorder"),
+ Order(description="someotherorder"),
+ ],
+ )
+ sess.add(u)
+ sess.flush()
+
+ sess.delete(u)
+ sess.flush()
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 0,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(orders)
+ ).scalar(),
+ 0,
+ )
def test_delete_unloaded_collections(self):
"""Unloaded collections are still included in a delete-cascade
@@ -432,27 +444,47 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
self.classes.Address,
)
- sess = create_session()
- u = User(
- name="jack",
- addresses=[
- Address(email_address="address1"),
- Address(email_address="address2"),
- ],
- )
- sess.add(u)
- sess.flush()
- sess.expunge_all()
- eq_(select(func.count("*")).select_from(addresses).scalar(), 2)
- eq_(select(func.count("*")).select_from(users).scalar(), 1)
+ with Session() as sess:
+ u = User(
+ name="jack",
+ addresses=[
+ Address(email_address="address1"),
+ Address(email_address="address2"),
+ ],
+ )
+ sess.add(u)
+ sess.flush()
+ sess.expunge_all()
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(addresses)
+ ).scalar(),
+ 2,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 1,
+ )
- u = sess.query(User).get(u.id)
+ u = sess.get(User, u.id)
- assert "addresses" not in u.__dict__
- sess.delete(u)
- sess.flush()
- eq_(select(func.count("*")).select_from(addresses).scalar(), 0)
- eq_(select(func.count("*")).select_from(users).scalar(), 0)
+ assert "addresses" not in u.__dict__
+ sess.delete(u)
+ sess.flush()
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(addresses)
+ ).scalar(),
+ 0,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 0,
+ )
def test_cascades_onlycollection(self):
"""Cascade only reaches instances that are still part of the
@@ -465,34 +497,48 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
self.tables.orders,
)
- sess = create_session()
- u = User(
- name="jack",
- orders=[
- Order(description="someorder"),
- Order(description="someotherorder"),
- ],
- )
- sess.add(u)
- sess.flush()
-
- o = u.orders[0]
- del u.orders[0]
- sess.delete(u)
- assert u in sess.deleted
- assert o not in sess.deleted
- assert o in sess
-
- u2 = User(name="newuser", orders=[o])
- sess.add(u2)
- sess.flush()
- sess.expunge_all()
- eq_(select(func.count("*")).select_from(users).scalar(), 1)
- eq_(select(func.count("*")).select_from(orders).scalar(), 1)
- eq_(
- sess.query(User).all(),
- [User(name="newuser", orders=[Order(description="someorder")])],
- )
+ with Session(autoflush=False) as sess:
+ u = User(
+ name="jack",
+ orders=[
+ Order(description="someorder"),
+ Order(description="someotherorder"),
+ ],
+ )
+ sess.add(u)
+ sess.flush()
+
+ o = u.orders[0]
+ del u.orders[0]
+ sess.delete(u)
+ assert u in sess.deleted
+ assert o not in sess.deleted
+ assert o in sess
+
+ u2 = User(name="newuser", orders=[o])
+ sess.add(u2)
+ sess.flush()
+ sess.expunge_all()
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 1,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(orders)
+ ).scalar(),
+ 1,
+ )
+ eq_(
+ sess.query(User).all(),
+ [
+ User(
+ name="newuser", orders=[Order(description="someorder")]
+ )
+ ],
+ )
def test_cascade_nosideeffects(self):
"""test that cascade leaves the state of unloaded
@@ -504,7 +550,7 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
self.classes.Address,
)
- sess = create_session()
+ sess = Session()
u = User(name="jack")
sess.add(u)
assert "orders" not in u.__dict__
@@ -534,7 +580,7 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
self.classes.Order,
)
- sess = create_session()
+ sess = Session()
u = User(
name="jack",
orders=[
@@ -544,14 +590,26 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
)
sess.add(u)
sess.flush()
- eq_(select(func.count("*")).select_from(users).scalar(), 1)
- eq_(select(func.count("*")).select_from(orders).scalar(), 2)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(users)).scalar(),
+ 1,
+ )
+ eq_(
+ sess.execute(select(func.count("*")).select_from(orders)).scalar(),
+ 2,
+ )
del u.orders[0]
sess.delete(u)
sess.flush()
- eq_(select(func.count("*")).select_from(users).scalar(), 0)
- eq_(select(func.count("*")).select_from(orders).scalar(), 0)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(users)).scalar(),
+ 0,
+ )
+ eq_(
+ sess.execute(select(func.count("*")).select_from(orders)).scalar(),
+ 0,
+ )
def test_collection_orphans(self):
User, users, orders, Order = (
@@ -561,26 +619,46 @@ class O2MCascadeDeleteOrphanTest(fixtures.MappedTest):
self.classes.Order,
)
- sess = create_session()
- u = User(
- name="jack",
- orders=[
- Order(description="someorder"),
- Order(description="someotherorder"),
- ],
- )
- sess.add(u)
- sess.flush()
-
- eq_(select(func.count("*")).select_from(users).scalar(), 1)
- eq_(select(func.count("*")).select_from(orders).scalar(), 2)
+ with Session() as sess:
+ u = User(
+ name="jack",
+ orders=[
+ Order(description="someorder"),
+ Order(description="someotherorder"),
+ ],
+ )
+ sess.add(u)
+ sess.flush()
+
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 1,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(orders)
+ ).scalar(),
+ 2,
+ )
- u.orders[:] = []
+ u.orders[:] = []
- sess.flush()
+ sess.flush()
- eq_(select(func.count("*")).select_from(users).scalar(), 1)
- eq_(select(func.count("*")).select_from(orders).scalar(), 0)
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 1,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(orders)
+ ).scalar(),
+ 0,
+ )
class O2MCascadeTest(fixtures.MappedTest):
@@ -715,24 +793,44 @@ class O2MCascadeDeleteNoOrphanTest(fixtures.MappedTest):
self.tables.users,
)
- sess = create_session()
- u = User(
- name="jack",
- orders=[
- Order(description="someorder"),
- Order(description="someotherorder"),
- ],
- )
- sess.add(u)
- sess.flush()
- eq_(select(func.count("*")).select_from(users).scalar(), 1)
- eq_(select(func.count("*")).select_from(orders).scalar(), 2)
+ with Session() as sess:
+ u = User(
+ name="jack",
+ orders=[
+ Order(description="someorder"),
+ Order(description="someotherorder"),
+ ],
+ )
+ sess.add(u)
+ sess.flush()
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 1,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(orders)
+ ).scalar(),
+ 2,
+ )
- del u.orders[0]
- sess.delete(u)
- sess.flush()
- eq_(select(func.count("*")).select_from(users).scalar(), 0)
- eq_(select(func.count("*")).select_from(orders).scalar(), 1)
+ del u.orders[0]
+ sess.delete(u)
+ sess.flush()
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(users)
+ ).scalar(),
+ 0,
+ )
+ eq_(
+ sess.execute(
+ select(func.count("*")).select_from(orders)
+ ).scalar(),
+ 1,
+ )
class O2OSingleParentTest(_fixtures.FixtureTest):
@@ -1183,18 +1281,44 @@ class NoSaveCascadeFlushTest(_fixtures.FixtureTest):
User, Address = self.classes.User, self.classes.Address
self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
- sess = Session()
- u1 = User(name="u1")
- sess.add(u1)
- sess.flush()
+ with Session() as sess:
+ u1 = User(name="u1")
+ sess.add(u1)
+ sess.flush()
+
+ a1 = Address(email_address="a1")
+ with testing.expect_deprecated(
+ '"Address" object is being merged into a Session along '
+ 'the backref cascade path for relationship "User.addresses"'
+ ):
+ a1.user = u1
+ sess.add(a1)
+ sess.expunge(u1)
+ assert u1 not in sess
+ assert a1 in sess
+ assert_raises_message(
+ sa_exc.SAWarning, "not in session", sess.flush
+ )
- a1 = Address(email_address="a1")
- a1.user = u1
- sess.add(a1)
- sess.expunge(u1)
- assert u1 not in sess
- assert a1 in sess
- assert_raises_message(sa_exc.SAWarning, "not in session", sess.flush)
+ def test_m2o_backref_future_child_expunged(self):
+ User, Address = self.classes.User, self.classes.Address
+
+ self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
+ with Session(testing.db, future=True) as sess:
+ u1 = User(name="u1")
+ sess.add(u1)
+ sess.flush()
+
+ a1 = Address(email_address="a1")
+ a1.user = u1
+ assert a1 not in sess
+ sess.add(a1)
+ sess.expunge(u1)
+ assert u1 not in sess
+ assert a1 in sess
+ assert_raises_message(
+ sa_exc.SAWarning, "not in session", sess.flush
+ )
def test_m2o_backref_child_pending_nochange(self):
User, Address = self.classes.User, self.classes.Address
@@ -1221,25 +1345,56 @@ class NoSaveCascadeFlushTest(_fixtures.FixtureTest):
User, Address = self.classes.User, self.classes.Address
self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
- sess = Session()
- u1 = User(name="u1")
- sess.add(u1)
- sess.flush()
- a1 = Address(email_address="a1")
- a1.user = u1
- sess.add(a1)
- sess.expunge(u1)
- assert u1 not in sess
- assert a1 in sess
+ with Session() as sess:
+ u1 = User(name="u1")
+ sess.add(u1)
+ sess.flush()
+
+ a1 = Address(email_address="a1")
+ with testing.expect_deprecated(
+ '"Address" object is being merged into a Session along the '
+ 'backref cascade path for relationship "User.addresses"'
+ ):
+ a1.user = u1
+ sess.add(a1)
+ sess.expunge(u1)
+ assert u1 not in sess
+ assert a1 in sess
+
+ @testing.emits_warning(r".*not in session")
+ def go():
+ sess.commit()
+
+ go()
+ # didn't get flushed
+ assert a1.user is None
+
+ def test_m2o_backref_future_child_expunged_nochange(self):
+ User, Address = self.classes.User, self.classes.Address
- @testing.emits_warning(r".*not in session")
- def go():
- sess.commit()
+ self._one_to_many_fixture(o2m=True, m2o=True, m2o_cascade=False)
- go()
- # didn't get flushed
- assert a1.user is None
+ with Session(testing.db, future=True) as sess:
+ u1 = User(name="u1")
+ sess.add(u1)
+ sess.flush()
+
+ a1 = Address(email_address="a1")
+ a1.user = u1
+ assert a1 not in sess
+ sess.add(a1)
+ sess.expunge(u1)
+ assert u1 not in sess
+ assert a1 in sess
+
+ @testing.emits_warning(r".*not in session")
+ def go():
+ sess.commit()
+
+ go()
+ # didn't get flushed
+ assert a1.user is None
def test_m2m_only_child_pending(self):
Item, Keyword = self.classes.Item, self.classes.Keyword
@@ -1394,7 +1549,7 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
),
)
- sess = create_session()
+ sess = Session()
o1 = Order()
sess.add(o1)
@@ -1429,7 +1584,7 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
)
mapper(User, users)
- sess = create_session()
+ sess = Session()
u1 = User()
sess.add(u1)
@@ -1470,7 +1625,7 @@ class NoSaveCascadeBackrefTest(_fixtures.FixtureTest):
)
mapper(Keyword, keywords)
- sess = create_session()
+ sess = Session()
i1 = Item()
k1 = Keyword()
@@ -1586,7 +1741,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
u1 = User(name="ed", pref=Pref(data="pref 1", extra=[Extra()]))
u2 = User(name="jack", pref=Pref(data="pref 2", extra=[Extra()]))
u3 = User(name="foo", pref=Pref(data="pref 3", extra=[Extra()]))
- sess = create_session(connection)
+ sess = Session(connection)
sess.add_all((u1, u2, u3))
sess.flush()
sess.close()
@@ -1598,14 +1753,26 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
self.tables.extra,
)
- sess = create_session()
- eq_(select(func.count("*")).select_from(prefs).scalar(), 3)
- eq_(select(func.count("*")).select_from(extra).scalar(), 3)
+ sess = Session()
+ eq_(
+ sess.execute(select(func.count("*")).select_from(prefs)).scalar(),
+ 3,
+ )
+ eq_(
+ sess.execute(select(func.count("*")).select_from(extra)).scalar(),
+ 3,
+ )
jack = sess.query(User).filter_by(name="jack").one()
jack.pref = None
sess.flush()
- eq_(select(func.count("*")).select_from(prefs).scalar(), 2)
- eq_(select(func.count("*")).select_from(extra).scalar(), 2)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(prefs)).scalar(),
+ 2,
+ )
+ eq_(
+ sess.execute(select(func.count("*")).select_from(extra)).scalar(),
+ 2,
+ )
def test_cascade_on_deleted(self):
"""test a bug introduced by r6711"""
@@ -1657,7 +1824,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
self.tables.extra,
)
- sess = create_session()
+ sess = Session()
jack = sess.query(User).filter_by(name="jack").one()
p = jack.pref
e = jack.pref.extra[0]
@@ -1670,13 +1837,19 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
assert p in sess
assert e in sess
sess.flush()
- eq_(select(func.count("*")).select_from(prefs).scalar(), 2)
- eq_(select(func.count("*")).select_from(extra).scalar(), 2)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(prefs)).scalar(),
+ 2,
+ )
+ eq_(
+ sess.execute(select(func.count("*")).select_from(extra)).scalar(),
+ 2,
+ )
def test_pending_expunge(self):
Pref, User = self.classes.Pref, self.classes.User
- sess = create_session()
+ sess = Session()
someuser = User(name="someuser")
sess.add(someuser)
sess.flush()
@@ -1695,7 +1868,7 @@ class M2OCascadeDeleteOrphanTestOne(fixtures.MappedTest):
Pref, User = self.classes.Pref, self.classes.User
- sess = create_session()
+ sess = Session()
jack = sess.query(User).filter_by(name="jack").one()
newpref = Pref(data="newpref")
@@ -1788,7 +1961,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
def test_cascade_delete(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
sess.add(x)
sess.flush()
@@ -1802,7 +1975,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
def test_deletes_orphans_onelevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x2 = T1(data="t1b", t2=T2(data="t2b", t3=T3(data="t3b")))
sess.add(x2)
sess.flush()
@@ -1817,7 +1990,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
def test_deletes_orphans_twolevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
sess.add(x)
sess.flush()
@@ -1832,7 +2005,7 @@ class M2OCascadeDeleteOrphanTestTwo(fixtures.MappedTest):
def test_finds_orphans_twolevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
sess.add(x)
sess.flush()
@@ -1929,7 +2102,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
def test_cascade_delete(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
sess.add(x)
sess.flush()
@@ -1943,7 +2116,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
def test_cascade_delete_postappend_onelevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x1 = T1(data="t1")
x2 = T2(data="t2")
x3 = T3(data="t3")
@@ -1961,7 +2134,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
def test_cascade_delete_postappend_twolevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x1 = T1(data="t1", t2=T2(data="t2"))
x3 = T3(data="t3")
sess.add_all((x1, x3))
@@ -1977,7 +2150,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
def test_preserves_orphans_onelevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x2 = T1(data="t1b", t2=T2(data="t2b", t3=T3(data="t3b")))
sess.add(x2)
sess.flush()
@@ -1993,7 +2166,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
def test_preserves_orphans_onelevel_postremove(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x2 = T1(data="t1b", t2=T2(data="t2b", t3=T3(data="t3b")))
sess.add(x2)
sess.flush()
@@ -2008,7 +2181,7 @@ class M2OCascadeDeleteNoOrphanTest(fixtures.MappedTest):
def test_preserves_orphans_twolevel(self):
T2, T3, T1 = (self.classes.T2, self.classes.T3, self.classes.T1)
- sess = create_session()
+ sess = Session()
x = T1(data="t1a", t2=T2(data="t2a", t3=T3(data="t3a")))
sess.add(x)
sess.flush()
@@ -2097,7 +2270,7 @@ class M2MCascadeTest(fixtures.MappedTest):
)
mapper(B, b)
- sess = create_session()
+ sess = Session()
b1 = B(data="b1")
a1 = A(data="a1", bs=[b1])
sess.add(a1)
@@ -2105,9 +2278,11 @@ class M2MCascadeTest(fixtures.MappedTest):
a1.bs.remove(b1)
sess.flush()
- eq_(select(func.count("*")).select_from(atob).scalar(), 0)
- eq_(select(func.count("*")).select_from(b).scalar(), 0)
- eq_(select(func.count("*")).select_from(a).scalar(), 1)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+ )
+ eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+ eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 1)
def test_delete_orphan_dynamic(self):
a, A, B, b, atob = (
@@ -2135,7 +2310,7 @@ class M2MCascadeTest(fixtures.MappedTest):
# failed until [ticket:427] was fixed
mapper(B, b)
- sess = create_session()
+ sess = Session()
b1 = B(data="b1")
a1 = A(data="a1", bs=[b1])
sess.add(a1)
@@ -2143,9 +2318,11 @@ class M2MCascadeTest(fixtures.MappedTest):
a1.bs.remove(b1)
sess.flush()
- eq_(select(func.count("*")).select_from(atob).scalar(), 0)
- eq_(select(func.count("*")).select_from(b).scalar(), 0)
- eq_(select(func.count("*")).select_from(a).scalar(), 1)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+ )
+ eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+ eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 1)
def test_delete_orphan_cascades(self):
a, A, c, b, C, B, atob = (
@@ -2179,7 +2356,7 @@ class M2MCascadeTest(fixtures.MappedTest):
)
mapper(C, c)
- sess = create_session()
+ sess = Session()
b1 = B(data="b1", cs=[C(data="c1")])
a1 = A(data="a1", bs=[b1])
sess.add(a1)
@@ -2187,10 +2364,12 @@ class M2MCascadeTest(fixtures.MappedTest):
a1.bs.remove(b1)
sess.flush()
- eq_(select(func.count("*")).select_from(atob).scalar(), 0)
- eq_(select(func.count("*")).select_from(b).scalar(), 0)
- eq_(select(func.count("*")).select_from(a).scalar(), 1)
- eq_(select(func.count("*")).select_from(c).scalar(), 0)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+ )
+ eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+ eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 1)
+ eq_(sess.execute(select(func.count("*")).select_from(c)).scalar(), 0)
def test_cascade_delete(self):
a, A, B, b, atob = (
@@ -2215,16 +2394,18 @@ class M2MCascadeTest(fixtures.MappedTest):
)
mapper(B, b)
- sess = create_session()
+ sess = Session()
a1 = A(data="a1", bs=[B(data="b1")])
sess.add(a1)
sess.flush()
sess.delete(a1)
sess.flush()
- eq_(select(func.count("*")).select_from(atob).scalar(), 0)
- eq_(select(func.count("*")).select_from(b).scalar(), 0)
- eq_(select(func.count("*")).select_from(a).scalar(), 0)
+ eq_(
+ sess.execute(select(func.count("*")).select_from(atob)).scalar(), 0
+ )
+ eq_(sess.execute(select(func.count("*")).select_from(b)).scalar(), 0)
+ eq_(sess.execute(select(func.count("*")).select_from(a)).scalar(), 0)
def test_single_parent_error(self):
a, A, B, b, atob = (
@@ -2492,7 +2673,11 @@ class NoBackrefCascadeTest(_fixtures.FixtureTest):
sess.add(a1)
d1 = Dingaling()
- d1.address = a1
+ with testing.expect_deprecated(
+ '"Dingaling" object is being merged into a Session along the '
+ 'backref cascade path for relationship "Address.dingalings"'
+ ):
+ d1.address = a1
assert d1 in a1.dingalings
assert d1 in sess
@@ -2519,7 +2704,11 @@ class NoBackrefCascadeTest(_fixtures.FixtureTest):
sess.add(a1)
u1 = User(name="u1")
- u1.addresses.append(a1)
+ with testing.expect_deprecated(
+ '"User" object is being merged into a Session along the backref '
+ 'cascade path for relationship "Address.user"'
+ ):
+ u1.addresses.append(a1)
assert u1 in sess
def test_m2o_commit_warns(self):
@@ -2666,7 +2855,7 @@ class PendingOrphanTestSingleLevel(fixtures.MappedTest):
)
),
)
- s = create_session()
+ s = Session()
u = User()
s.add(u)
@@ -2702,7 +2891,7 @@ class PendingOrphanTestSingleLevel(fixtures.MappedTest):
)
),
)
- s = create_session()
+ s = Session()
u = User(name="u1", addresses=[Address(email_address="ad1")])
s.add(u)
a1 = u.addresses[0]
@@ -2928,12 +3117,12 @@ class DoubleParentO2MOrphanTest(fixtures.MappedTest):
)
),
)
- s = create_session()
+ s = Session(expire_on_commit=False, autoflush=False)
a = Account(balance=0)
sr = SalesRep(name="John")
s.add_all((a, sr))
- s.flush()
+ s.commit()
c = Customer(name="Jane")
@@ -3093,7 +3282,7 @@ class DoubleParentM2OOrphanTest(fixtures.MappedTest):
},
)
- session = create_session()
+ session = Session()
h1 = Home(description="home1", address=Address(street="address1"))
b1 = Business(
description="business1", address=Address(street="address2")
@@ -3103,11 +3292,11 @@ class DoubleParentM2OOrphanTest(fixtures.MappedTest):
session.expunge_all()
eq_(
- session.query(Home).get(h1.id),
+ session.get(Home, h1.id),
Home(description="home1", address=Address(street="address1")),
)
eq_(
- session.query(Business).get(b1.id),
+ session.get(Business, b1.id),
Business(
description="business1", address=Address(street="address2")
),
@@ -3152,7 +3341,7 @@ class DoubleParentM2OOrphanTest(fixtures.MappedTest):
)
},
)
- session = create_session()
+ session = Session()
a1 = Address()
session.add(a1)
session.flush()
@@ -3197,18 +3386,18 @@ class CollectionAssignmentOrphanTest(fixtures.MappedTest):
a1 = A(name="a1", bs=[B(name="b1"), B(name="b2"), B(name="b3")])
- sess = create_session()
+ sess = Session()
sess.add(a1)
sess.flush()
sess.expunge_all()
eq_(
- sess.query(A).get(a1.id),
+ sess.get(A, a1.id),
A(name="a1", bs=[B(name="b1"), B(name="b2"), B(name="b3")]),
)
- a1 = sess.query(A).get(a1.id)
+ a1 = sess.get(A, a1.id)
assert not class_mapper(B)._is_orphan(
attributes.instance_state(a1.bs[0])
)
@@ -3218,7 +3407,7 @@ class CollectionAssignmentOrphanTest(fixtures.MappedTest):
sess.expunge_all()
eq_(
- sess.query(A).get(a1.id),
+ sess.get(A, a1.id),
A(name="a1", bs=[B(name="b1"), B(name="b2"), B(name="b3")]),
)
@@ -3440,27 +3629,26 @@ class O2MConflictTest(fixtures.MappedTest):
def _do_move_test(self, delete_old):
Parent, Child = self.classes.Parent, self.classes.Child
- sess = create_session()
-
- p1, p2, c1 = Parent(), Parent(), Child()
- if Parent.child.property.uselist:
- p1.child.append(c1)
- else:
- p1.child = c1
- sess.add_all([p1, c1])
- sess.flush()
+ with Session(autoflush=False) as sess:
+ p1, p2, c1 = Parent(), Parent(), Child()
+ if Parent.child.property.uselist:
+ p1.child.append(c1)
+ else:
+ p1.child = c1
+ sess.add_all([p1, c1])
+ sess.flush()
- if delete_old:
- sess.delete(p1)
+ if delete_old:
+ sess.delete(p1)
- if Parent.child.property.uselist:
- p2.child.append(c1)
- else:
- p2.child = c1
- sess.add(p2)
+ if Parent.child.property.uselist:
+ p2.child.append(c1)
+ else:
+ p2.child = c1
+ sess.add(p2)
- sess.flush()
- eq_(sess.query(Child).filter(Child.parent_id == p2.id).all(), [c1])
+ sess.flush()
+ eq_(sess.query(Child).filter(Child.parent_id == p2.id).all(), [c1])
def test_o2o_delete_old(self):
Child, Parent, parent, child = (
@@ -3508,7 +3696,11 @@ class O2MConflictTest(fixtures.MappedTest):
Parent,
parent,
properties={
- "child": relationship(Child, uselist=False, backref="parent")
+ "child": relationship(
+ Child,
+ uselist=False,
+ backref=backref("parent", cascade_backrefs=False),
+ )
},
)
mapper(Child, child)
@@ -3573,7 +3765,7 @@ class O2MConflictTest(fixtures.MappedTest):
Child,
uselist=False,
cascade="all, delete, delete-orphan",
- backref="parent",
+ backref=backref("parent", cascade_backrefs=False),
)
},
)
@@ -3600,6 +3792,7 @@ class O2MConflictTest(fixtures.MappedTest):
single_parent=True,
backref=backref("child", uselist=False),
cascade="all,delete,delete-orphan",
+ cascade_backrefs=False,
)
},
)
@@ -3625,6 +3818,7 @@ class O2MConflictTest(fixtures.MappedTest):
single_parent=True,
backref=backref("child", uselist=True),
cascade="all,delete,delete-orphan",
+ cascade_backrefs=False,
)
},
)
@@ -3684,7 +3878,7 @@ class PartialFlushTest(fixtures.MappedTest):
)
mapper(Child, noninh_child)
- sess = create_session()
+ sess = Session()
c1, c2 = Child(), Child()
b1 = Base(descr="b1", children=[c1, c2])
@@ -3701,7 +3895,7 @@ class PartialFlushTest(fixtures.MappedTest):
assert c2 in sess and c2 not in sess.new
assert b1 in sess and b1 not in sess.new
- sess = create_session()
+ sess = Session()
c1, c2 = Child(), Child()
b1 = Base(descr="b1", children=[c1, c2])
sess.add(b1)
@@ -3711,7 +3905,7 @@ class PartialFlushTest(fixtures.MappedTest):
assert c2 in sess and c2 in sess.new
assert b1 in sess and b1 in sess.new
- sess = create_session()
+ sess = Session()
c1, c2 = Child(), Child()
b1 = Base(descr="b1", children=[c1, c2])
sess.add(b1)
@@ -3756,7 +3950,7 @@ class PartialFlushTest(fixtures.MappedTest):
mapper(Parent, parent, inherits=Base)
- sess = create_session()
+ sess = Session()
p1 = Parent()
c1, c2, c3 = Child(), Child(), Child()
diff --git a/test/orm/test_composites.py b/test/orm/test_composites.py
index b4c0c2dc1..046a6acb9 100644
--- a/test/orm/test_composites.py
+++ b/test/orm/test_composites.py
@@ -95,7 +95,7 @@ class PointTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
self.classes.Point,
)
- sess = Session(future=future)
+ sess = Session(testing.db, future=future)
g = Graph(
id=1,
edges=[
diff --git a/test/orm/test_froms.py b/test/orm/test_froms.py
index 4b4c2bf73..e5a53df4b 100644
--- a/test/orm/test_froms.py
+++ b/test/orm/test_froms.py
@@ -1219,7 +1219,7 @@ class InstancesTest(QueryTest, AssertsCompiledSQL):
self.classes.User,
)
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
selectquery = users.outerjoin(addresses).select(
users.c.id < 10,
@@ -2154,7 +2154,7 @@ class MixedEntitiesTest(QueryTest, AssertsCompiledSQL):
(user10, None),
]
- sess = create_session(future=True)
+ sess = create_session(testing.db, future=True)
selectquery = users.outerjoin(addresses).select(
use_labels=True, order_by=[users.c.id, addresses.c.id]
diff --git a/test/orm/test_lambdas.py b/test/orm/test_lambdas.py
index 2aac956ca..d4fae7f6f 100644
--- a/test/orm/test_lambdas.py
+++ b/test/orm/test_lambdas.py
@@ -92,7 +92,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
def test_cols_round_trip(self, plain_fixture):
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
# note this does a traversal + _clone of the InstrumentedAttribute
# for the first time ever
@@ -135,7 +135,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
def test_entity_round_trip(self, plain_fixture):
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
def query(names):
stmt = lambda_stmt(
@@ -182,7 +182,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
def test_subqueryload_internal_lambda(self, plain_fixture):
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
def query(names):
stmt = (
@@ -220,7 +220,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
def test_subqueryload_external_lambda_caveats(self, plain_fixture):
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
def query(names):
stmt = lambda_stmt(
@@ -263,7 +263,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
def test_does_filter_aliasing_work(self, plain_fixture):
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
# aliased=True is to be deprecated, other filter lambdas
# that go into effect include polymorphic filtering.
@@ -314,7 +314,7 @@ class LambdaTest(QueryTest, AssertsCompiledSQL):
def test_join_entity_arg(self, plain_fixture, test_case):
User, Address = plain_fixture
- s = Session(future=True)
+ s = Session(testing.db, future=True)
stmt = testing.resolve_lambda(test_case, **locals())
self.assert_compile(
@@ -411,7 +411,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_update(self):
User, Address = self.classes("User", "Address")
- s = Session(future=True)
+ s = Session(testing.db, future=True)
def go(ids, values):
stmt = lambda_stmt(lambda: update(User).where(User.id.in_(ids)))
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index b6d014936..4663789c4 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -187,7 +187,7 @@ class RowTupleTest(QueryTest):
mapper(User, users)
- s = Session(future=True)
+ s = Session(testing.db, future=True)
q = testing.resolve_lambda(test_case, **locals())
@@ -876,7 +876,7 @@ class GetTest(QueryTest):
def test_populate_existing_future(self):
User, Address = self.classes.User, self.classes.Address
- s = Session(future=True, autoflush=False)
+ s = Session(testing.db, future=True, autoflush=False)
userlist = s.query(User).all()
@@ -4533,7 +4533,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
def test_select_star_future(self):
User = self.classes.User
- sess = Session(future=True)
+ sess = Session(testing.db, future=True)
eq_(
sess.execute(
select(User).from_statement(
@@ -4581,7 +4581,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
# ordering doesn't matter
User = self.classes.User
- s = create_session(future=True)
+ s = create_session(testing.db, future=True)
q = select(User).from_statement(
text(
"select name, 27 as foo, id as users_id from users order by id"
@@ -4628,7 +4628,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
User = self.classes.User
Address = self.classes.Address
- s = create_session(future=True)
+ s = create_session(testing.db, future=True)
q = select(User, Address).from_statement(
text(
"select users.name AS users_name, users.id AS users_id, "
@@ -4679,7 +4679,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
User = self.classes.User
Address = self.classes.Address
- s = create_session(future=True)
+ s = create_session(testing.db, future=True)
q = (
select(User)
.from_statement(
@@ -4731,7 +4731,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
User = self.classes.User
Address = self.classes.Address
- s = create_session(future=True)
+ s = create_session(testing.db, future=True)
q = (
select(User)
.from_statement(
@@ -4838,7 +4838,7 @@ class TextTest(QueryTest, AssertsCompiledSQL):
def test_whereclause_future(self):
User = self.classes.User
- s = create_session(future=True)
+ s = create_session(testing.db, future=True)
eq_(
s.execute(select(User).filter(text("id in (8, 9)")))
.scalars()
diff --git a/test/orm/test_transaction.py b/test/orm/test_transaction.py
index 660bc7a5d..6095d8642 100644
--- a/test/orm/test_transaction.py
+++ b/test/orm/test_transaction.py
@@ -483,7 +483,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
User, users = self.classes.User, self.tables.users
mapper(User, users)
- sess = create_session(autocommit=False, future=True)
+ sess = create_session(testing.db, autocommit=False, future=True)
u = User(name="u1")
sess.add(u)
sess.flush()
@@ -1032,7 +1032,7 @@ class SessionTransactionTest(fixtures.RemovesEvents, FixtureTest):
User, users = self.classes.User, self.tables.users
mapper(User, users)
- session = create_session(autocommit=False, future=True)
+ session = create_session(testing.db, autocommit=False, future=True)
session.add(User(name="ed"))
session.transaction.commit()
diff --git a/test/orm/test_update_delete.py b/test/orm/test_update_delete.py
index e4396d919..75dca1c99 100644
--- a/test/orm/test_update_delete.py
+++ b/test/orm/test_update_delete.py
@@ -154,7 +154,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
User = self.classes.User
- s = Session(future=True)
+ s = Session(testing.db, future=True)
jill = s.query(User).filter(User.name == "jill").one()
@@ -179,7 +179,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
User = self.classes.User
- s = Session(future=True)
+ s = Session(testing.db, future=True)
jill = s.query(User).filter(User.name == "jill").one()
@@ -435,7 +435,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_update_future(self):
User, users = self.classes.User, self.tables.users
- sess = Session(future=True)
+ sess = Session(testing.db, future=True)
john, jack, jill, jane = (
sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -487,7 +487,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_update_future_lambda(self):
User, users = self.classes.User, self.tables.users
- sess = Session(future=True)
+ sess = Session(testing.db, future=True)
john, jack, jill, jane = (
sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -623,7 +623,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_update_fetch_returning_lambda(self):
User = self.classes.User
- sess = Session(future=True)
+ sess = Session(testing.db, future=True)
john, jack, jill, jane = (
sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -711,7 +711,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_delete_fetch_returning_lambda(self):
User = self.classes.User
- sess = Session(future=True)
+ sess = Session(testing.db, future=True)
john, jack, jill, jane = (
sess.execute(select(User).order_by(User.id)).scalars().all()
@@ -1068,7 +1068,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_update_multi_values_error_future(self):
User = self.classes.User
- session = Session(future=True)
+ session = Session(testing.db, future=True)
# Do update using a tuple and check that order is preserved
@@ -1087,7 +1087,7 @@ class UpdateDeleteTest(fixtures.MappedTest):
def test_update_preserve_parameter_order_future(self):
User = self.classes.User
- session = Session(future=True)
+ session = Session(testing.db, future=True)
# Do update using a tuple and check that order is preserved
@@ -1585,7 +1585,7 @@ class InheritTest(fixtures.DeclarativeMappedTest):
person = self.classes.Person.__table__
engineer = self.classes.Engineer.__table__
- sess = Session(future=True)
+ sess = Session(testing.db, future=True)
sess.query(person.join(engineer)).filter(person.c.name == "e2").update(
{person.c.name: "updated", engineer.c.engineer_name: "e2a"},
)
View Lorry Controller Status