summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy
diff options
context:
space:
mode:
Diffstat (limited to 'lib/sqlalchemy')
-rw-r--r--lib/sqlalchemy/testing/suite/test_cte.py21
-rw-r--r--lib/sqlalchemy/testing/suite/test_ddl.py20
-rw-r--r--lib/sqlalchemy/testing/suite/test_deprecations.py63
-rw-r--r--lib/sqlalchemy/testing/suite/test_insert.py100
-rw-r--r--lib/sqlalchemy/testing/suite/test_results.py77
-rw-r--r--lib/sqlalchemy/testing/suite/test_select.py188
-rw-r--r--lib/sqlalchemy/testing/suite/test_sequence.py72
-rw-r--r--lib/sqlalchemy/testing/suite/test_types.py12
-rw-r--r--lib/sqlalchemy/testing/suite/test_update_delete.py31
9 files changed, 305 insertions, 279 deletions
diff --git a/lib/sqlalchemy/testing/suite/test_cte.py b/lib/sqlalchemy/testing/suite/test_cte.py
index c7e6a266c..fab457606 100644
--- a/lib/sqlalchemy/testing/suite/test_cte.py
+++ b/lib/sqlalchemy/testing/suite/test_cte.py
@@ -37,16 +37,17 @@ class CTETest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "data": "d1", "parent_id": None},
- {"id": 2, "data": "d2", "parent_id": 1},
- {"id": 3, "data": "d3", "parent_id": 1},
- {"id": 4, "data": "d4", "parent_id": 3},
- {"id": 5, "data": "d5", "parent_id": 3},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "data": "d1", "parent_id": None},
+ {"id": 2, "data": "d2", "parent_id": 1},
+ {"id": 3, "data": "d3", "parent_id": 1},
+ {"id": 4, "data": "d4", "parent_id": 3},
+ {"id": 5, "data": "d5", "parent_id": 3},
+ ],
+ )
def test_select_nonrecursive_round_trip(self):
some_table = self.tables.some_table
diff --git a/lib/sqlalchemy/testing/suite/test_ddl.py b/lib/sqlalchemy/testing/suite/test_ddl.py
index 81a55e18a..1f49106fb 100644
--- a/lib/sqlalchemy/testing/suite/test_ddl.py
+++ b/lib/sqlalchemy/testing/suite/test_ddl.py
@@ -67,25 +67,27 @@ class TableDDLTest(fixtures.TestBase):
@requirements.comment_reflection
@util.provide_metadata
- def test_add_table_comment(self):
+ def test_add_table_comment(self, connection):
table = self._simple_fixture()
- table.create(config.db, checkfirst=False)
+ table.create(connection, checkfirst=False)
table.comment = "a comment"
- config.db.execute(schema.SetTableComment(table))
+ connection.execute(schema.SetTableComment(table))
eq_(
- inspect(config.db).get_table_comment("test_table"),
+ inspect(connection).get_table_comment("test_table"),
{"text": "a comment"},
)
@requirements.comment_reflection
@util.provide_metadata
- def test_drop_table_comment(self):
+ def test_drop_table_comment(self, connection):
table = self._simple_fixture()
- table.create(config.db, checkfirst=False)
+ table.create(connection, checkfirst=False)
table.comment = "a comment"
- config.db.execute(schema.SetTableComment(table))
- config.db.execute(schema.DropTableComment(table))
- eq_(inspect(config.db).get_table_comment("test_table"), {"text": None})
+ connection.execute(schema.SetTableComment(table))
+ connection.execute(schema.DropTableComment(table))
+ eq_(
+ inspect(connection).get_table_comment("test_table"), {"text": None}
+ )
__all__ = ("TableDDLTest",)
diff --git a/lib/sqlalchemy/testing/suite/test_deprecations.py b/lib/sqlalchemy/testing/suite/test_deprecations.py
index d0202a0a9..126d82fe9 100644
--- a/lib/sqlalchemy/testing/suite/test_deprecations.py
+++ b/lib/sqlalchemy/testing/suite/test_deprecations.py
@@ -24,20 +24,21 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "x": 1, "y": 2},
- {"id": 2, "x": 2, "y": 3},
- {"id": 3, "x": 3, "y": 4},
- {"id": 4, "x": 4, "y": 5},
- ],
- )
-
- def _assert_result(self, select, result, params=()):
- eq_(config.db.execute(select, params).fetchall(), result)
-
- def test_plain_union(self):
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "x": 1, "y": 2},
+ {"id": 2, "x": 2, "y": 3},
+ {"id": 3, "x": 3, "y": 4},
+ {"id": 4, "x": 4, "y": 5},
+ ],
+ )
+
+ def _assert_result(self, conn, select, result, params=()):
+ eq_(conn.execute(select, params).fetchall(), result)
+
+ def test_plain_union(self, connection):
table = self.tables.some_table
s1 = select([table]).where(table.c.id == 2)
s2 = select([table]).where(table.c.id == 3)
@@ -47,7 +48,9 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
"The SelectBase.c and SelectBase.columns "
"attributes are deprecated"
):
- self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)])
+ self._assert_result(
+ connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
+ )
# note we've had to remove one use case entirely, which is this
# one. the Select gets its FROMS from the WHERE clause and the
@@ -56,7 +59,7 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
# ORDER BY without adding the SELECT into the FROM and breaking the
# query. Users will have to adjust for this use case if they were doing
# it before.
- def _dont_test_select_from_plain_union(self):
+ def _dont_test_select_from_plain_union(self, connection):
table = self.tables.some_table
s1 = select([table]).where(table.c.id == 2)
s2 = select([table]).where(table.c.id == 3)
@@ -66,11 +69,13 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
"The SelectBase.c and SelectBase.columns "
"attributes are deprecated"
):
- self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)])
+ self._assert_result(
+ connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
+ )
@testing.requires.order_by_col_from_union
@testing.requires.parens_in_union_contained_select_w_limit_offset
- def test_limit_offset_selectable_in_unions(self):
+ def test_limit_offset_selectable_in_unions(self, connection):
table = self.tables.some_table
s1 = (
select([table])
@@ -90,10 +95,12 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
"The SelectBase.c and SelectBase.columns "
"attributes are deprecated"
):
- self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)])
+ self._assert_result(
+ connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
+ )
@testing.requires.parens_in_union_contained_select_wo_limit_offset
- def test_order_by_selectable_in_unions(self):
+ def test_order_by_selectable_in_unions(self, connection):
table = self.tables.some_table
s1 = select([table]).where(table.c.id == 2).order_by(table.c.id)
s2 = select([table]).where(table.c.id == 3).order_by(table.c.id)
@@ -103,9 +110,11 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
"The SelectBase.c and SelectBase.columns "
"attributes are deprecated"
):
- self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)])
+ self._assert_result(
+ connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
+ )
- def test_distinct_selectable_in_unions(self):
+ def test_distinct_selectable_in_unions(self, connection):
table = self.tables.some_table
s1 = select([table]).where(table.c.id == 2).distinct()
s2 = select([table]).where(table.c.id == 3).distinct()
@@ -115,9 +124,11 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
"The SelectBase.c and SelectBase.columns "
"attributes are deprecated"
):
- self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)])
+ self._assert_result(
+ connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
+ )
- def test_limit_offset_aliased_selectable_in_unions(self):
+ def test_limit_offset_aliased_selectable_in_unions(self, connection):
table = self.tables.some_table
s1 = (
select([table])
@@ -141,4 +152,6 @@ class DeprecatedCompoundSelectTest(fixtures.TablesTest):
"The SelectBase.c and SelectBase.columns "
"attributes are deprecated"
):
- self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)])
+ self._assert_result(
+ connection, u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]
+ )
diff --git a/lib/sqlalchemy/testing/suite/test_insert.py b/lib/sqlalchemy/testing/suite/test_insert.py
index 931b0ef65..f449b2fe6 100644
--- a/lib/sqlalchemy/testing/suite/test_insert.py
+++ b/lib/sqlalchemy/testing/suite/test_insert.py
@@ -41,28 +41,28 @@ class LastrowidTest(fixtures.TablesTest):
def _assert_round_trip(self, table, conn):
row = conn.execute(table.select()).first()
- eq_(row, (config.db.dialect.default_sequence_base, "some data"))
+ eq_(row, (conn.engine.dialect.default_sequence_base, "some data"))
- def test_autoincrement_on_insert(self):
+ def test_autoincrement_on_insert(self, connection):
- config.db.execute(self.tables.autoinc_pk.insert(), data="some data")
- self._assert_round_trip(self.tables.autoinc_pk, config.db)
+ connection.execute(self.tables.autoinc_pk.insert(), data="some data")
+ self._assert_round_trip(self.tables.autoinc_pk, connection)
- def test_last_inserted_id(self):
+ def test_last_inserted_id(self, connection):
- r = config.db.execute(
+ r = connection.execute(
self.tables.autoinc_pk.insert(), data="some data"
)
- pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
+ pk = connection.scalar(select([self.tables.autoinc_pk.c.id]))
eq_(r.inserted_primary_key, [pk])
@requirements.dbapi_lastrowid
- def test_native_lastrowid_autoinc(self):
- r = config.db.execute(
+ def test_native_lastrowid_autoinc(self, connection):
+ r = connection.execute(
self.tables.autoinc_pk.insert(), data="some data"
)
lastrowid = r.lastrowid
- pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
+ pk = connection.scalar(select([self.tables.autoinc_pk.c.id]))
eq_(lastrowid, pk)
@@ -117,8 +117,8 @@ class InsertBehaviorTest(fixtures.TablesTest):
assert not r.returns_rows
@requirements.returning
- def test_autoclose_on_insert_implicit_returning(self):
- r = config.db.execute(
+ def test_autoclose_on_insert_implicit_returning(self, connection):
+ r = connection.execute(
self.tables.autoinc_pk.insert(), data="some data"
)
assert r._soft_closed
@@ -127,12 +127,12 @@ class InsertBehaviorTest(fixtures.TablesTest):
assert not r.returns_rows
@requirements.empty_inserts
- def test_empty_insert(self):
- r = config.db.execute(self.tables.autoinc_pk.insert())
+ def test_empty_insert(self, connection):
+ r = connection.execute(self.tables.autoinc_pk.insert())
assert r._soft_closed
assert not r.closed
- r = config.db.execute(
+ r = connection.execute(
self.tables.autoinc_pk.select().where(
self.tables.autoinc_pk.c.id != None
)
@@ -141,10 +141,10 @@ class InsertBehaviorTest(fixtures.TablesTest):
assert len(r.fetchall())
@requirements.insert_from_select
- def test_insert_from_select_autoinc(self):
+ def test_insert_from_select_autoinc(self, connection):
src_table = self.tables.manual_pk
dest_table = self.tables.autoinc_pk
- config.db.execute(
+ connection.execute(
src_table.insert(),
[
dict(id=1, data="data1"),
@@ -153,7 +153,7 @@ class InsertBehaviorTest(fixtures.TablesTest):
],
)
- result = config.db.execute(
+ result = connection.execute(
dest_table.insert().from_select(
("data",),
select([src_table.c.data]).where(
@@ -164,17 +164,17 @@ class InsertBehaviorTest(fixtures.TablesTest):
eq_(result.inserted_primary_key, [None])
- result = config.db.execute(
+ result = connection.execute(
select([dest_table.c.data]).order_by(dest_table.c.data)
)
eq_(result.fetchall(), [("data2",), ("data3",)])
@requirements.insert_from_select
- def test_insert_from_select_autoinc_no_rows(self):
+ def test_insert_from_select_autoinc_no_rows(self, connection):
src_table = self.tables.manual_pk
dest_table = self.tables.autoinc_pk
- result = config.db.execute(
+ result = connection.execute(
dest_table.insert().from_select(
("data",),
select([src_table.c.data]).where(
@@ -184,16 +184,16 @@ class InsertBehaviorTest(fixtures.TablesTest):
)
eq_(result.inserted_primary_key, [None])
- result = config.db.execute(
+ result = connection.execute(
select([dest_table.c.data]).order_by(dest_table.c.data)
)
eq_(result.fetchall(), [])
@requirements.insert_from_select
- def test_insert_from_select(self):
+ def test_insert_from_select(self, connection):
table = self.tables.manual_pk
- config.db.execute(
+ connection.execute(
table.insert(),
[
dict(id=1, data="data1"),
@@ -202,7 +202,7 @@ class InsertBehaviorTest(fixtures.TablesTest):
],
)
- config.db.execute(
+ connection.execute(
table.insert(inline=True).from_select(
("id", "data"),
select([table.c.id + 5, table.c.data]).where(
@@ -212,16 +212,16 @@ class InsertBehaviorTest(fixtures.TablesTest):
)
eq_(
- config.db.execute(
+ connection.execute(
select([table.c.data]).order_by(table.c.data)
).fetchall(),
[("data1",), ("data2",), ("data2",), ("data3",), ("data3",)],
)
@requirements.insert_from_select
- def test_insert_from_select_with_defaults(self):
+ def test_insert_from_select_with_defaults(self, connection):
table = self.tables.includes_defaults
- config.db.execute(
+ connection.execute(
table.insert(),
[
dict(id=1, data="data1"),
@@ -230,7 +230,7 @@ class InsertBehaviorTest(fixtures.TablesTest):
],
)
- config.db.execute(
+ connection.execute(
table.insert(inline=True).from_select(
("id", "data"),
select([table.c.id + 5, table.c.data]).where(
@@ -240,7 +240,7 @@ class InsertBehaviorTest(fixtures.TablesTest):
)
eq_(
- config.db.execute(
+ connection.execute(
select([table]).order_by(table.c.data, table.c.id)
).fetchall(),
[
@@ -262,7 +262,7 @@ class ReturningTest(fixtures.TablesTest):
def _assert_round_trip(self, table, conn):
row = conn.execute(table.select()).first()
- eq_(row, (config.db.dialect.default_sequence_base, "some data"))
+ eq_(row, (conn.engine.dialect.default_sequence_base, "some data"))
@classmethod
def define_tables(cls, metadata):
@@ -276,39 +276,35 @@ class ReturningTest(fixtures.TablesTest):
)
@requirements.fetch_rows_post_commit
- def test_explicit_returning_pk_autocommit(self):
- engine = config.db
+ def test_explicit_returning_pk_autocommit(self, connection):
table = self.tables.autoinc_pk
- with engine.begin() as conn:
- r = conn.execute(
- table.insert().returning(table.c.id), data="some data"
- )
+ r = connection.execute(
+ table.insert().returning(table.c.id), data="some data"
+ )
pk = r.first()[0]
- fetched_pk = config.db.scalar(select([table.c.id]))
+ fetched_pk = connection.scalar(select([table.c.id]))
eq_(fetched_pk, pk)
- def test_explicit_returning_pk_no_autocommit(self):
- engine = config.db
+ def test_explicit_returning_pk_no_autocommit(self, connection):
table = self.tables.autoinc_pk
- with engine.begin() as conn:
- r = conn.execute(
- table.insert().returning(table.c.id), data="some data"
- )
- pk = r.first()[0]
- fetched_pk = config.db.scalar(select([table.c.id]))
+ r = connection.execute(
+ table.insert().returning(table.c.id), data="some data"
+ )
+ pk = r.first()[0]
+ fetched_pk = connection.scalar(select([table.c.id]))
eq_(fetched_pk, pk)
- def test_autoincrement_on_insert_implicit_returning(self):
+ def test_autoincrement_on_insert_implicit_returning(self, connection):
- config.db.execute(self.tables.autoinc_pk.insert(), data="some data")
- self._assert_round_trip(self.tables.autoinc_pk, config.db)
+ connection.execute(self.tables.autoinc_pk.insert(), data="some data")
+ self._assert_round_trip(self.tables.autoinc_pk, connection)
- def test_last_inserted_id_implicit_returning(self):
+ def test_last_inserted_id_implicit_returning(self, connection):
- r = config.db.execute(
+ r = connection.execute(
self.tables.autoinc_pk.insert(), data="some data"
)
- pk = config.db.scalar(select([self.tables.autoinc_pk.c.id]))
+ pk = connection.scalar(select([self.tables.autoinc_pk.c.id]))
eq_(r.inserted_primary_key, [pk])
diff --git a/lib/sqlalchemy/testing/suite/test_results.py b/lib/sqlalchemy/testing/suite/test_results.py
index 7a3e92564..0cff7b61d 100644
--- a/lib/sqlalchemy/testing/suite/test_results.py
+++ b/lib/sqlalchemy/testing/suite/test_results.py
@@ -38,46 +38,47 @@ class RowFetchTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.plain_pk.insert(),
- [
- {"id": 1, "data": "d1"},
- {"id": 2, "data": "d2"},
- {"id": 3, "data": "d3"},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.plain_pk.insert(),
+ [
+ {"id": 1, "data": "d1"},
+ {"id": 2, "data": "d2"},
+ {"id": 3, "data": "d3"},
+ ],
+ )
- config.db.execute(
- cls.tables.has_dates.insert(),
- [{"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)}],
- )
+ conn.execute(
+ cls.tables.has_dates.insert(),
+ [{"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)}],
+ )
- def test_via_attr(self):
- row = config.db.execute(
+ def test_via_attr(self, connection):
+ row = connection.execute(
self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
).first()
eq_(row.id, 1)
eq_(row.data, "d1")
- def test_via_string(self):
- row = config.db.execute(
+ def test_via_string(self, connection):
+ row = connection.execute(
self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
).first()
eq_(row._mapping["id"], 1)
eq_(row._mapping["data"], "d1")
- def test_via_int(self):
- row = config.db.execute(
+ def test_via_int(self, connection):
+ row = connection.execute(
self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
).first()
eq_(row[0], 1)
eq_(row[1], "d1")
- def test_via_col_object(self):
- row = config.db.execute(
+ def test_via_col_object(self, connection):
+ row = connection.execute(
self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id)
).first()
@@ -85,8 +86,8 @@ class RowFetchTest(fixtures.TablesTest):
eq_(row._mapping[self.tables.plain_pk.c.data], "d1")
@requirements.duplicate_names_in_cursor_description
- def test_row_with_dupe_names(self):
- result = config.db.execute(
+ def test_row_with_dupe_names(self, connection):
+ result = connection.execute(
select(
[
self.tables.plain_pk.c.data,
@@ -98,7 +99,7 @@ class RowFetchTest(fixtures.TablesTest):
eq_(result.keys(), ["data", "data"])
eq_(row, ("d1", "d1"))
- def test_row_w_scalar_select(self):
+ def test_row_w_scalar_select(self, connection):
"""test that a scalar select as a column is returned as such
and that type conversion works OK.
@@ -109,7 +110,7 @@ class RowFetchTest(fixtures.TablesTest):
datetable = self.tables.has_dates
s = select([datetable.alias("x").c.today]).scalar_subquery()
s2 = select([datetable.c.id, s.label("somelabel")])
- row = config.db.execute(s2).first()
+ row = connection.execute(s2).first()
eq_(row.somelabel, datetime.datetime(2006, 5, 12, 12, 0, 0))
@@ -141,7 +142,7 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
sql.column("spaces % more spaces"),
)
- def test_single_roundtrip(self):
+ def test_single_roundtrip(self, connection):
percent_table = self.tables.percent_table
for params in [
{"percent%": 5, "spaces % more spaces": 12},
@@ -149,15 +150,15 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
{"percent%": 9, "spaces % more spaces": 10},
{"percent%": 11, "spaces % more spaces": 9},
]:
- config.db.execute(percent_table.insert(), params)
- self._assert_table()
+ connection.execute(percent_table.insert(), params)
+ self._assert_table(connection)
- def test_executemany_roundtrip(self):
+ def test_executemany_roundtrip(self, connection):
percent_table = self.tables.percent_table
- config.db.execute(
+ connection.execute(
percent_table.insert(), {"percent%": 5, "spaces % more spaces": 12}
)
- config.db.execute(
+ connection.execute(
percent_table.insert(),
[
{"percent%": 7, "spaces % more spaces": 11},
@@ -165,9 +166,9 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
{"percent%": 11, "spaces % more spaces": 9},
],
)
- self._assert_table()
+ self._assert_table(connection)
- def _assert_table(self):
+ def _assert_table(self, conn):
percent_table = self.tables.percent_table
lightweight_percent_table = self.tables.lightweight_percent_table
@@ -179,16 +180,14 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
):
eq_(
list(
- config.db.execute(
- table.select().order_by(table.c["percent%"])
- )
+ conn.execute(table.select().order_by(table.c["percent%"]))
),
[(5, 12), (7, 11), (9, 10), (11, 9)],
)
eq_(
list(
- config.db.execute(
+ conn.execute(
table.select()
.where(table.c["spaces % more spaces"].in_([9, 10]))
.order_by(table.c["percent%"])
@@ -197,7 +196,7 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
[(9, 10), (11, 9)],
)
- row = config.db.execute(
+ row = conn.execute(
table.select().order_by(table.c["percent%"])
).first()
eq_(row._mapping["percent%"], 5)
@@ -206,7 +205,7 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
eq_(row._mapping[table.c["percent%"]], 5)
eq_(row._mapping[table.c["spaces % more spaces"]], 12)
- config.db.execute(
+ conn.execute(
percent_table.update().values(
{percent_table.c["spaces % more spaces"]: 15}
)
@@ -214,7 +213,7 @@ class PercentSchemaNamesTest(fixtures.TablesTest):
eq_(
list(
- config.db.execute(
+ conn.execute(
percent_table.select().order_by(
percent_table.c["percent%"]
)
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index 852bfcf0d..23f78961f 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -42,16 +42,18 @@ class CollateTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "data": "collate data1"},
- {"id": 2, "data": "collate data2"},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "data": "collate data1"},
+ {"id": 2, "data": "collate data2"},
+ ],
+ )
def _assert_result(self, select, result):
- eq_(config.db.execute(select).fetchall(), result)
+ with config.db.connect() as conn:
+ eq_(conn.execute(select).fetchall(), result)
@testing.requires.order_by_collation
def test_collate_order_by(self):
@@ -90,17 +92,19 @@ class OrderByLabelTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "x": 1, "y": 2, "q": "q1", "p": "p3"},
- {"id": 2, "x": 2, "y": 3, "q": "q2", "p": "p2"},
- {"id": 3, "x": 3, "y": 4, "q": "q3", "p": "p1"},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "x": 1, "y": 2, "q": "q1", "p": "p3"},
+ {"id": 2, "x": 2, "y": 3, "q": "q2", "p": "p2"},
+ {"id": 3, "x": 3, "y": 4, "q": "q3", "p": "p1"},
+ ],
+ )
def _assert_result(self, select, result):
- eq_(config.db.execute(select).fetchall(), result)
+ with config.db.connect() as conn:
+ eq_(conn.execute(select).fetchall(), result)
def test_plain(self):
table = self.tables.some_table
@@ -162,18 +166,20 @@ class LimitOffsetTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "x": 1, "y": 2},
- {"id": 2, "x": 2, "y": 3},
- {"id": 3, "x": 3, "y": 4},
- {"id": 4, "x": 4, "y": 5},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "x": 1, "y": 2},
+ {"id": 2, "x": 2, "y": 3},
+ {"id": 3, "x": 3, "y": 4},
+ {"id": 4, "x": 4, "y": 5},
+ ],
+ )
def _assert_result(self, select, result, params=()):
- eq_(config.db.execute(select, params).fetchall(), result)
+ with config.db.connect() as conn:
+ eq_(conn.execute(select, params).fetchall(), result)
def _assert_result_str(self, select, result, params=()):
conn = config.db.connect(close_with_result=True)
@@ -303,7 +309,8 @@ class JoinTest(fixtures.TablesTest):
__backend__ = True
def _assert_result(self, select, result, params=()):
- eq_(config.db.execute(select, params).fetchall(), result)
+ with config.db.connect() as conn:
+ eq_(conn.execute(select, params).fetchall(), result)
@classmethod
def define_tables(cls, metadata):
@@ -317,20 +324,21 @@ class JoinTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.a.insert(),
- [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.a.insert(),
+ [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}, {"id": 5}],
+ )
- config.db.execute(
- cls.tables.b.insert(),
- [
- {"id": 1, "a_id": 1},
- {"id": 2, "a_id": 1},
- {"id": 4, "a_id": 2},
- {"id": 5, "a_id": 3},
- ],
- )
+ conn.execute(
+ cls.tables.b.insert(),
+ [
+ {"id": 1, "a_id": 1},
+ {"id": 2, "a_id": 1},
+ {"id": 4, "a_id": 2},
+ {"id": 5, "a_id": 3},
+ ],
+ )
def test_inner_join_fk(self):
a, b = self.tables("a", "b")
@@ -413,18 +421,20 @@ class CompoundSelectTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "x": 1, "y": 2},
- {"id": 2, "x": 2, "y": 3},
- {"id": 3, "x": 3, "y": 4},
- {"id": 4, "x": 4, "y": 5},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "x": 1, "y": 2},
+ {"id": 2, "x": 2, "y": 3},
+ {"id": 3, "x": 3, "y": 4},
+ {"id": 4, "x": 4, "y": 5},
+ ],
+ )
def _assert_result(self, select, result, params=()):
- eq_(config.db.execute(select, params).fetchall(), result)
+ with config.db.connect() as conn:
+ eq_(conn.execute(select, params).fetchall(), result)
def test_plain_union(self):
table = self.tables.some_table
@@ -556,15 +566,16 @@ class PostCompileParamsTest(
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "x": 1, "y": 2, "z": "z1"},
- {"id": 2, "x": 2, "y": 3, "z": "z2"},
- {"id": 3, "x": 3, "y": 4, "z": "z3"},
- {"id": 4, "x": 4, "y": 5, "z": "z4"},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "x": 1, "y": 2, "z": "z1"},
+ {"id": 2, "x": 2, "y": 3, "z": "z2"},
+ {"id": 3, "x": 3, "y": 4, "z": "z3"},
+ {"id": 4, "x": 4, "y": 5, "z": "z4"},
+ ],
+ )
def test_compile(self):
table = self.tables.some_table
@@ -673,18 +684,20 @@ class ExpandingBoundInTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "x": 1, "y": 2, "z": "z1"},
- {"id": 2, "x": 2, "y": 3, "z": "z2"},
- {"id": 3, "x": 3, "y": 4, "z": "z3"},
- {"id": 4, "x": 4, "y": 5, "z": "z4"},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "x": 1, "y": 2, "z": "z1"},
+ {"id": 2, "x": 2, "y": 3, "z": "z2"},
+ {"id": 3, "x": 3, "y": 4, "z": "z3"},
+ {"id": 4, "x": 4, "y": 5, "z": "z4"},
+ ],
+ )
def _assert_result(self, select, result, params=()):
- eq_(config.db.execute(select, params).fetchall(), result)
+ with config.db.connect() as conn:
+ eq_(conn.execute(select, params).fetchall(), result)
def test_multiple_empty_sets(self):
# test that any anonymous aliasing used by the dialect
@@ -837,7 +850,7 @@ class ExpandingBoundInTest(fixtures.TablesTest):
self._assert_result(stmt, [(1,), (2,), (3,), (4,)], params={"q": []})
- def test_null_in_empty_set_is_false(self):
+ def test_null_in_empty_set_is_false(self, connection):
stmt = select(
[
case(
@@ -853,7 +866,7 @@ class ExpandingBoundInTest(fixtures.TablesTest):
)
]
)
- in_(config.db.execute(stmt).fetchone()[0], (False, 0))
+ in_(connection.execute(stmt).fetchone()[0], (False, 0))
class LikeFunctionsTest(fixtures.TablesTest):
@@ -873,21 +886,22 @@ class LikeFunctionsTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.some_table.insert(),
- [
- {"id": 1, "data": "abcdefg"},
- {"id": 2, "data": "ab/cdefg"},
- {"id": 3, "data": "ab%cdefg"},
- {"id": 4, "data": "ab_cdefg"},
- {"id": 5, "data": "abcde/fg"},
- {"id": 6, "data": "abcde%fg"},
- {"id": 7, "data": "ab#cdefg"},
- {"id": 8, "data": "ab9cdefg"},
- {"id": 9, "data": "abcde#fg"},
- {"id": 10, "data": "abcd9fg"},
- ],
- )
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.some_table.insert(),
+ [
+ {"id": 1, "data": "abcdefg"},
+ {"id": 2, "data": "ab/cdefg"},
+ {"id": 3, "data": "ab%cdefg"},
+ {"id": 4, "data": "ab_cdefg"},
+ {"id": 5, "data": "abcde/fg"},
+ {"id": 6, "data": "abcde%fg"},
+ {"id": 7, "data": "ab#cdefg"},
+ {"id": 8, "data": "ab9cdefg"},
+ {"id": 9, "data": "abcde#fg"},
+ {"id": 10, "data": "abcd9fg"},
+ ],
+ )
def _test(self, expr, expected):
some_table = self.tables.some_table
diff --git a/lib/sqlalchemy/testing/suite/test_sequence.py b/lib/sqlalchemy/testing/suite/test_sequence.py
index 22ae7d43c..db5582c21 100644
--- a/lib/sqlalchemy/testing/suite/test_sequence.py
+++ b/lib/sqlalchemy/testing/suite/test_sequence.py
@@ -39,21 +39,21 @@ class SequenceTest(fixtures.TablesTest):
Column("data", String(50)),
)
- def test_insert_roundtrip(self):
- config.db.execute(self.tables.seq_pk.insert(), data="some data")
- self._assert_round_trip(self.tables.seq_pk, config.db)
+ def test_insert_roundtrip(self, connection):
+ connection.execute(self.tables.seq_pk.insert(), data="some data")
+ self._assert_round_trip(self.tables.seq_pk, connection)
- def test_insert_lastrowid(self):
- r = config.db.execute(self.tables.seq_pk.insert(), data="some data")
+ def test_insert_lastrowid(self, connection):
+ r = connection.execute(self.tables.seq_pk.insert(), data="some data")
eq_(r.inserted_primary_key, [1])
- def test_nextval_direct(self):
- r = config.db.execute(self.tables.seq_pk.c.id.default)
+ def test_nextval_direct(self, connection):
+ r = connection.execute(self.tables.seq_pk.c.id.default)
eq_(r, 1)
@requirements.sequences_optional
- def test_optional_seq(self):
- r = config.db.execute(
+ def test_optional_seq(self, connection):
+ r = connection.execute(
self.tables.seq_opt_pk.insert(), data="some data"
)
eq_(r.inserted_primary_key, [1])
@@ -67,7 +67,7 @@ class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
__requires__ = ("sequences",)
__backend__ = True
- def test_literal_binds_inline_compile(self):
+ def test_literal_binds_inline_compile(self, connection):
table = Table(
"x",
MetaData(),
@@ -77,14 +77,14 @@ class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
stmt = table.insert().values(q=5)
- seq_nextval = testing.db.dialect.statement_compiler(
- statement=None, dialect=testing.db.dialect
+ seq_nextval = connection.dialect.statement_compiler(
+ statement=None, dialect=connection.dialect
).visit_sequence(Sequence("y_seq"))
self.assert_compile(
stmt,
"INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval,),
literal_binds=True,
- dialect=testing.db.dialect,
+ dialect=connection.dialect,
)
@@ -92,65 +92,65 @@ class HasSequenceTest(fixtures.TestBase):
__requires__ = ("sequences",)
__backend__ = True
- def test_has_sequence(self):
+ def test_has_sequence(self, connection):
s1 = Sequence("user_id_seq")
- testing.db.execute(schema.CreateSequence(s1))
+ connection.execute(schema.CreateSequence(s1))
try:
eq_(
- testing.db.dialect.has_sequence(testing.db, "user_id_seq"),
+ connection.dialect.has_sequence(connection, "user_id_seq"),
True,
)
finally:
- testing.db.execute(schema.DropSequence(s1))
+ connection.execute(schema.DropSequence(s1))
@testing.requires.schemas
- def test_has_sequence_schema(self):
+ def test_has_sequence_schema(self, connection):
s1 = Sequence("user_id_seq", schema=config.test_schema)
- testing.db.execute(schema.CreateSequence(s1))
+ connection.execute(schema.CreateSequence(s1))
try:
eq_(
- testing.db.dialect.has_sequence(
- testing.db, "user_id_seq", schema=config.test_schema
+ connection.dialect.has_sequence(
+ connection, "user_id_seq", schema=config.test_schema
),
True,
)
finally:
- testing.db.execute(schema.DropSequence(s1))
+ connection.execute(schema.DropSequence(s1))
- def test_has_sequence_neg(self):
- eq_(testing.db.dialect.has_sequence(testing.db, "user_id_seq"), False)
+ def test_has_sequence_neg(self, connection):
+ eq_(connection.dialect.has_sequence(connection, "user_id_seq"), False)
@testing.requires.schemas
- def test_has_sequence_schemas_neg(self):
+ def test_has_sequence_schemas_neg(self, connection):
eq_(
- testing.db.dialect.has_sequence(
- testing.db, "user_id_seq", schema=config.test_schema
+ connection.dialect.has_sequence(
+ connection, "user_id_seq", schema=config.test_schema
),
False,
)
@testing.requires.schemas
- def test_has_sequence_default_not_in_remote(self):
+ def test_has_sequence_default_not_in_remote(self, connection):
s1 = Sequence("user_id_seq")
- testing.db.execute(schema.CreateSequence(s1))
+ connection.execute(schema.CreateSequence(s1))
try:
eq_(
- testing.db.dialect.has_sequence(
- testing.db, "user_id_seq", schema=config.test_schema
+ connection.dialect.has_sequence(
+ connection, "user_id_seq", schema=config.test_schema
),
False,
)
finally:
- testing.db.execute(schema.DropSequence(s1))
+ connection.execute(schema.DropSequence(s1))
@testing.requires.schemas
- def test_has_sequence_remote_not_in_default(self):
+ def test_has_sequence_remote_not_in_default(self, connection):
s1 = Sequence("user_id_seq", schema=config.test_schema)
- testing.db.execute(schema.CreateSequence(s1))
+ connection.execute(schema.CreateSequence(s1))
try:
eq_(
- testing.db.dialect.has_sequence(testing.db, "user_id_seq"),
+ connection.dialect.has_sequence(connection, "user_id_seq"),
False,
)
finally:
- testing.db.execute(schema.DropSequence(s1))
+ connection.execute(schema.DropSequence(s1))
diff --git a/lib/sqlalchemy/testing/suite/test_types.py b/lib/sqlalchemy/testing/suite/test_types.py
index 9dabdbd65..7719a3b3c 100644
--- a/lib/sqlalchemy/testing/suite/test_types.py
+++ b/lib/sqlalchemy/testing/suite/test_types.py
@@ -519,10 +519,10 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase):
filter_=lambda n: n is not None and round(n, 5) or None,
)
- def test_float_coerce_round_trip(self):
+ def test_float_coerce_round_trip(self, connection):
expr = 15.7563
- val = testing.db.scalar(select([literal(expr)]))
+ val = connection.scalar(select([literal(expr)]))
eq_(val, expr)
# this does not work in MySQL, see #4036, however we choose not
@@ -530,17 +530,17 @@ class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase):
@testing.requires.implicit_decimal_binds
@testing.emits_warning(r".*does \*not\* support Decimal objects natively")
- def test_decimal_coerce_round_trip(self):
+ def test_decimal_coerce_round_trip(self, connection):
expr = decimal.Decimal("15.7563")
- val = testing.db.scalar(select([literal(expr)]))
+ val = connection.scalar(select([literal(expr)]))
eq_(val, expr)
@testing.emits_warning(r".*does \*not\* support Decimal objects natively")
- def test_decimal_coerce_round_trip_w_cast(self):
+ def test_decimal_coerce_round_trip_w_cast(self, connection):
expr = decimal.Decimal("15.7563")
- val = testing.db.scalar(select([cast(expr, Numeric(10, 4))]))
+ val = connection.scalar(select([cast(expr, Numeric(10, 4))]))
eq_(val, expr)
@testing.requires.precision_numerics_general
diff --git a/lib/sqlalchemy/testing/suite/test_update_delete.py b/lib/sqlalchemy/testing/suite/test_update_delete.py
index 97bdf0ad7..6003a0994 100644
--- a/lib/sqlalchemy/testing/suite/test_update_delete.py
+++ b/lib/sqlalchemy/testing/suite/test_update_delete.py
@@ -22,33 +22,34 @@ class SimpleUpdateDeleteTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
- config.db.execute(
- cls.tables.plain_pk.insert(),
- [
- {"id": 1, "data": "d1"},
- {"id": 2, "data": "d2"},
- {"id": 3, "data": "d3"},
- ],
- )
-
- def test_update(self):
+ with config.db.connect() as conn:
+ conn.execute(
+ cls.tables.plain_pk.insert(),
+ [
+ {"id": 1, "data": "d1"},
+ {"id": 2, "data": "d2"},
+ {"id": 3, "data": "d3"},
+ ],
+ )
+
+ def test_update(self, connection):
t = self.tables.plain_pk
- r = config.db.execute(t.update().where(t.c.id == 2), data="d2_new")
+ r = connection.execute(t.update().where(t.c.id == 2), data="d2_new")
assert not r.is_insert
assert not r.returns_rows
eq_(
- config.db.execute(t.select().order_by(t.c.id)).fetchall(),
+ connection.execute(t.select().order_by(t.c.id)).fetchall(),
[(1, "d1"), (2, "d2_new"), (3, "d3")],
)
- def test_delete(self):
+ def test_delete(self, connection):
t = self.tables.plain_pk
- r = config.db.execute(t.delete().where(t.c.id == 2))
+ r = connection.execute(t.delete().where(t.c.id == 2))
assert not r.is_insert
assert not r.returns_rows
eq_(
- config.db.execute(t.select().order_by(t.c.id)).fetchall(),
+ connection.execute(t.select().order_by(t.c.id)).fetchall(),
[(1, "d1"), (3, "d3")],
)
View Lorry Controller Status