PublicArchive/sqlalchemy
2
0
Fork 0
mirror of https://github.com/sqlalchemy/sqlalchemy.git synced 2026-05-20 07:32:05 -04:00
Code Issues Packages Projects Releases Wiki Activity

Clean up (engine|db).execute for mysql

Browse Source 
Change-Id: Ibc3509630a7c418b500ec7c6587f311326b75c93
This commit is contained in:
Gord Thompson
2020-03-31 06:33:55 -06:00
parent 379e878c58
commit 5bcdbacae4
1 changed files with 19 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/dialect/mysql/test_query.py
+19 -18
View File
@@ -130,9 +130,9 @@ class MatchTest(fixtures.TestBase):
)
eq_([3], [r.id for r in results])
def test_return_value(self):
def test_return_value(self, connection):
# test [ticket:3263]
result = testing.db.execute(
result = connection.execute(
select(
[
matchtable.c.title.match("Agile Ruby Programming").label(
@@ -233,35 +233,36 @@ class AnyAllTest(fixtures.TablesTest):
@classmethod
def insert_data(cls):
stuff = cls.tables.stuff
testing.db.execute(
stuff.insert(),
[
{"id": 1, "value": 1},
{"id": 2, "value": 2},
{"id": 3, "value": 3},
{"id": 4, "value": 4},
{"id": 5, "value": 5},
],
)
with testing.db.begin() as conn:
conn.execute(
stuff.insert(),
[
{"id": 1, "value": 1},
{"id": 2, "value": 2},
{"id": 3, "value": 3},
{"id": 4, "value": 4},
{"id": 5, "value": 5},
],
)
def test_any_w_comparator(self):
def test_any_w_comparator(self, connection):
stuff = self.tables.stuff
stmt = select([stuff.c.id]).where(
stuff.c.value > any_(select([stuff.c.value]).scalar_subquery())
)
eq_(testing.db.execute(stmt).fetchall(), [(2,), (3,), (4,), (5,)])
eq_(connection.execute(stmt).fetchall(), [(2,), (3,), (4,), (5,)])
def test_all_w_comparator(self):
def test_all_w_comparator(self, connection):
stuff = self.tables.stuff
stmt = select([stuff.c.id]).where(
stuff.c.value >= all_(select([stuff.c.value]).scalar_subquery())
)
eq_(testing.db.execute(stmt).fetchall(), [(5,)])
eq_(connection.execute(stmt).fetchall(), [(5,)])
def test_any_literal(self):
def test_any_literal(self, connection):
stuff = self.tables.stuff
stmt = select([4 == any_(select([stuff.c.value]).scalar_subquery())])
is_(testing.db.execute(stmt).scalar(), True)
is_(connection.execute(stmt).scalar(), True)