summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/dialect/postgresql/test_query.py6
-rw-r--r--test/dialect/test_sqlite.py2
-rw-r--r--test/engine/test_logging.py4
-rw-r--r--test/orm/dml/test_bulk_statements.py6
-rw-r--r--test/orm/test_dynamic.py66
-rw-r--r--test/orm/test_merge.py2
-rw-r--r--test/orm/test_unitofworkv2.py2
-rw-r--r--test/requirements.py11
-rw-r--r--test/sql/test_compiler.py366
-rw-r--r--test/sql/test_cte.py20
-rw-r--r--test/sql/test_insert.py95
-rw-r--r--test/sql/test_resultset.py2
-rw-r--r--test/sql/test_types.py2
-rw-r--r--test/sql/test_update.py66
14 files changed, 498 insertions, 152 deletions
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index 42ec20743..2b32d6db7 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -990,19 +990,19 @@ class MatchTest(fixtures.TablesTest, AssertsCompiledSQL):
"matchtable.title @@ plainto_tsquery(%(title_1)s)",
)
- @testing.requires.format_paramstyle
+ @testing.only_if("+asyncpg")
def test_expression_positional(self, connection):
matchtable = self.tables.matchtable
if self._strs_render_bind_casts(connection):
self.assert_compile(
matchtable.c.title.match("somstr"),
- "matchtable.title @@ plainto_tsquery(%s::VARCHAR(200))",
+ "matchtable.title @@ plainto_tsquery($1::VARCHAR(200))",
)
else:
self.assert_compile(
matchtable.c.title.match("somstr"),
- "matchtable.title @@ plainto_tsquery(%s)",
+ "matchtable.title @@ plainto_tsquery($1)",
)
def test_simple_match(self, connection):
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index c5147e37f..07117b862 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -2916,6 +2916,8 @@ class OnConflictTest(AssertsCompiledSQL, fixtures.TablesTest):
)
@testing.combinations("control", "excluded", "dict")
+ @testing.skip_if("+pysqlite_numeric")
+ @testing.skip_if("+pysqlite_dollar")
def test_set_excluded(self, scenario):
"""test #8014, sending all of .excluded to set"""
diff --git a/test/engine/test_logging.py b/test/engine/test_logging.py
index 277248617..19c26f43c 100644
--- a/test/engine/test_logging.py
+++ b/test/engine/test_logging.py
@@ -28,7 +28,7 @@ def exec_sql(engine, sql, *args, **kwargs):
class LogParamsTest(fixtures.TestBase):
- __only_on__ = "sqlite"
+ __only_on__ = "sqlite+pysqlite"
__requires__ = ("ad_hoc_engines",)
def setup_test(self):
@@ -704,7 +704,7 @@ class LoggingNameTest(fixtures.TestBase):
class TransactionContextLoggingTest(fixtures.TestBase):
- __only_on__ = "sqlite"
+ __only_on__ = "sqlite+pysqlite"
@testing.fixture()
def plain_assert_buf(self, plain_logging_engine):
diff --git a/test/orm/dml/test_bulk_statements.py b/test/orm/dml/test_bulk_statements.py
index 557b5e9da..78607e03d 100644
--- a/test/orm/dml/test_bulk_statements.py
+++ b/test/orm/dml/test_bulk_statements.py
@@ -958,7 +958,7 @@ class BulkDMLReturningJoinedInhTest(
BulkDMLReturningInhTest, fixtures.DeclarativeMappedTest
):
- __requires__ = ("insert_returning",)
+ __requires__ = ("insert_returning", "insert_executemany_returning")
__backend__ = True
@classmethod
@@ -1044,7 +1044,7 @@ class BulkDMLReturningJoinedInhTest(
class BulkDMLReturningSingleInhTest(
BulkDMLReturningInhTest, fixtures.DeclarativeMappedTest
):
- __requires__ = ("insert_returning",)
+ __requires__ = ("insert_returning", "insert_executemany_returning")
__backend__ = True
@classmethod
@@ -1075,7 +1075,7 @@ class BulkDMLReturningSingleInhTest(
class BulkDMLReturningConcreteInhTest(
BulkDMLReturningInhTest, fixtures.DeclarativeMappedTest
):
- __requires__ = ("insert_returning",)
+ __requires__ = ("insert_returning", "insert_executemany_returning")
__backend__ = True
@classmethod
diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py
index df335f0f6..714878f4e 100644
--- a/test/orm/test_dynamic.py
+++ b/test/orm/test_dynamic.py
@@ -1791,14 +1791,33 @@ class WriteOnlyBulkTest(
"INSERT INTO users (name) VALUES (:name)",
[{"name": "x"}],
),
- CompiledSQL(
- "INSERT INTO addresses (user_id, email_address) "
- "VALUES (:user_id, :email_address) "
- "RETURNING addresses.id",
+ Conditional(
+ testing.requires.insert_executemany_returning.enabled,
+ [
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address) "
+ "RETURNING addresses.id",
+ [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ],
+ )
+ ],
[
- {"user_id": uid, "email_address": "e1"},
- {"user_id": uid, "email_address": "e2"},
- {"user_id": uid, "email_address": "e3"},
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address)",
+ param,
+ )
+ for param in [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ]
],
),
],
@@ -1863,14 +1882,33 @@ class WriteOnlyBulkTest(
"INSERT INTO users (name) VALUES (:name)",
[{"name": "x"}],
),
- CompiledSQL(
- "INSERT INTO addresses (user_id, email_address) "
- "VALUES (:user_id, :email_address) "
- "RETURNING addresses.id",
+ Conditional(
+ testing.requires.insert_executemany_returning.enabled,
+ [
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address) "
+ "RETURNING addresses.id",
+ [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ],
+ )
+ ],
[
- {"user_id": uid, "email_address": "e1"},
- {"user_id": uid, "email_address": "e2"},
- {"user_id": uid, "email_address": "e3"},
+ CompiledSQL(
+ "INSERT INTO addresses "
+ "(user_id, email_address) "
+ "VALUES (:user_id, :email_address)",
+ param,
+ )
+ for param in [
+ {"user_id": uid, "email_address": "e1"},
+ {"user_id": uid, "email_address": "e2"},
+ {"user_id": uid, "email_address": "e3"},
+ ]
],
),
],
diff --git a/test/orm/test_merge.py b/test/orm/test_merge.py
index 36c47e27b..eb5a795e2 100644
--- a/test/orm/test_merge.py
+++ b/test/orm/test_merge.py
@@ -1458,7 +1458,7 @@ class MergeTest(_fixtures.FixtureTest):
)
attrname = "user"
else:
- assert False
+ direction.fail()
assert attrname in obj_to_merge.__dict__
diff --git a/test/orm/test_unitofworkv2.py b/test/orm/test_unitofworkv2.py
index f204e954c..468d43063 100644
--- a/test/orm/test_unitofworkv2.py
+++ b/test/orm/test_unitofworkv2.py
@@ -3077,7 +3077,7 @@ class EagerDefaultsTest(fixtures.MappedTest):
asserter.assert_(
Conditional(
- testing.db.dialect.insert_executemany_returning,
+ testing.db.dialect.insert_returning,
[
CompiledSQL(
"INSERT INTO test (id) VALUES (:id) "
diff --git a/test/requirements.py b/test/requirements.py
index 5276593c9..83cd65cd8 100644
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -232,7 +232,6 @@ class DefaultRequirements(SuiteRequirements):
"mariadb+pymysql",
"mariadb+cymysql",
"mariadb+mysqlconnector",
- "postgresql+asyncpg",
"postgresql+pg8000",
]
)
@@ -388,6 +387,14 @@ class DefaultRequirements(SuiteRequirements):
)
@property
+ def predictable_gc(self):
+ """target platform must remove all cycles unconditionally when
+ gc.collect() is called, as well as clean out unreferenced subclasses.
+
+ """
+ return self.cpython + skip_if("+aiosqlite")
+
+ @property
def memory_process_intensive(self):
"""Driver is able to handle the memory tests which run in a subprocess
and iterate through hundreds of connections
@@ -969,6 +976,8 @@ class DefaultRequirements(SuiteRequirements):
"mariadb",
"sqlite+aiosqlite",
"sqlite+pysqlite",
+ "sqlite+pysqlite_numeric",
+ "sqlite+pysqlite_dollar",
"sqlite+pysqlcipher",
"mssql",
)
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 205ce5157..d342b9248 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -79,6 +79,7 @@ from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql.elements import BooleanClauseList
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.elements import CompilerColumnElement
+from sqlalchemy.sql.elements import Grouping
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import ClauseList
from sqlalchemy.sql.selectable import LABEL_STYLE_NONE
@@ -4915,88 +4916,259 @@ class BindParameterTest(AssertsCompiledSQL, fixtures.TestBase):
dialect="default",
)
- @standalone_escape
- @testing.variation("use_assert_compile", [True, False])
@testing.variation("use_positional", [True, False])
- def test_standalone_bindparam_escape_expanding(
- self, paramname, expected, use_assert_compile, use_positional
+ def test_standalone_bindparam_escape_collision(self, use_positional):
+ """this case is currently not supported
+
+ it's kinda bad since positional takes the unescaped param
+ while non positional takes the escaped one.
+ """
+ stmt = select(table1.c.myid).where(
+ table1.c.name == bindparam("[brackets]", value="x"),
+ table1.c.description == bindparam("_brackets_", value="y"),
+ )
+
+ if use_positional:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name = ? "
+ "AND mytable.description = ?",
+ params={"[brackets]": "a", "_brackets_": "b"},
+ checkpositional=("a", "a"),
+ dialect="sqlite",
+ )
+ else:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name = "
+ ":_brackets_ AND mytable.description = :_brackets_",
+ params={"[brackets]": "a", "_brackets_": "b"},
+ checkparams={"_brackets_": "b"},
+ dialect="default",
+ )
+
+ paramstyle = testing.variation("paramstyle", ["named", "qmark", "numeric"])
+
+ @standalone_escape
+ @paramstyle
+ def test_standalone_bindparam_escape_expanding_compile(
+ self, paramname, expected, paramstyle
):
stmt = select(table1.c.myid).where(
table1.c.name.in_(bindparam(paramname, value=["a", "b"]))
)
- if use_assert_compile:
- if use_positional:
- self.assert_compile(
- stmt,
- "SELECT mytable.myid FROM mytable "
- "WHERE mytable.name IN (?, ?)",
- params={paramname: ["y", "z"]},
- # NOTE: this is what render_postcompile will do right now
- # if you run construct_params(). render_postcompile mode
- # is not actually used by the execution internals, it's for
- # user-facing compilation code. So this is likely a
- # current limitation of construct_params() which is not
- # doing the full blown postcompile; just assert that's
- # what it does for now. it likely should be corrected
- # to make more sense.
- checkpositional=(["y", "z"], ["y", "z"]),
- dialect="sqlite",
- render_postcompile=True,
- )
- else:
- self.assert_compile(
- stmt,
- "SELECT mytable.myid FROM mytable WHERE mytable.name IN "
- "(:%s_1, :%s_2)" % (expected, expected),
- params={paramname: ["y", "z"]},
- # NOTE: this is what render_postcompile will do right now
- # if you run construct_params(). render_postcompile mode
- # is not actually used by the execution internals, it's for
- # user-facing compilation code. So this is likely a
- # current limitation of construct_params() which is not
- # doing the full blown postcompile; just assert that's
- # what it does for now. it likely should be corrected
- # to make more sense.
- checkparams={
- "%s_1" % expected: ["y", "z"],
- "%s_2" % expected: ["y", "z"],
- },
- dialect="default",
- render_postcompile=True,
- )
+ # NOTE: below the rendered params are just what
+ # render_postcompile will do right now
+ # if you run construct_params(). render_postcompile mode
+ # is not actually used by the execution internals, it's for
+ # user-facing compilation code. So this is likely a
+ # current limitation of construct_params() which is not
+ # doing the full blown postcompile; just assert that's
+ # what it does for now. it likely should be corrected
+ # to make more sense.
+ if paramstyle.qmark:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable "
+ "WHERE mytable.name IN (?, ?)",
+ params={paramname: ["y", "z"]},
+ checkpositional=(["y", "z"], ["y", "z"]),
+ dialect="sqlite",
+ render_postcompile=True,
+ )
+ elif paramstyle.numeric:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable "
+ "WHERE mytable.name IN (:1, :2)",
+ params={paramname: ["y", "z"]},
+ checkpositional=(["y", "z"], ["y", "z"]),
+ dialect=sqlite.dialect(paramstyle="numeric"),
+ render_postcompile=True,
+ )
+ elif paramstyle.named:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name IN "
+ "(:%s_1, :%s_2)" % (expected, expected),
+ params={paramname: ["y", "z"]},
+ checkparams={
+ "%s_1" % expected: ["y", "z"],
+ "%s_2" % expected: ["y", "z"],
+ },
+ dialect="default",
+ render_postcompile=True,
+ )
else:
- # this is what DefaultDialect actually does.
- # this should be matched to DefaultDialect._init_compiled()
- if use_positional:
- compiled = stmt.compile(
- dialect=default.DefaultDialect(paramstyle="qmark")
- )
- else:
- compiled = stmt.compile(dialect=default.DefaultDialect())
+ paramstyle.fail()
- checkparams = compiled.construct_params(
- {paramname: ["y", "z"]}, escape_names=False
- )
+ @standalone_escape
+ @paramstyle
+ def test_standalone_bindparam_escape_expanding(
+ self, paramname, expected, paramstyle
+ ):
+ stmt = select(table1.c.myid).where(
+ table1.c.name.in_(bindparam(paramname, value=["a", "b"]))
+ )
+ # this is what DefaultDialect actually does.
+ # this should be matched to DefaultDialect._init_compiled()
+ if paramstyle.qmark:
+ dialect = default.DefaultDialect(paramstyle="qmark")
+ elif paramstyle.numeric:
+ dialect = default.DefaultDialect(paramstyle="numeric")
+ else:
+ dialect = default.DefaultDialect()
- # nothing actually happened. if the compiler had
- # render_postcompile set, the
- # above weird param thing happens
- eq_(checkparams, {paramname: ["y", "z"]})
+ compiled = stmt.compile(dialect=dialect)
+ checkparams = compiled.construct_params(
+ {paramname: ["y", "z"]}, escape_names=False
+ )
- expanded_state = compiled._process_parameters_for_postcompile(
- checkparams
- )
+ # nothing actually happened. if the compiler had
+ # render_postcompile set, the
+ # above weird param thing happens
+ eq_(checkparams, {paramname: ["y", "z"]})
+
+ expanded_state = compiled._process_parameters_for_postcompile(
+ checkparams
+ )
+ eq_(
+ expanded_state.additional_parameters,
+ {f"{expected}_1": "y", f"{expected}_2": "z"},
+ )
+
+ if paramstyle.qmark or paramstyle.numeric:
eq_(
- expanded_state.additional_parameters,
- {f"{expected}_1": "y", f"{expected}_2": "z"},
+ expanded_state.positiontup,
+ [f"{expected}_1", f"{expected}_2"],
)
- if use_positional:
- eq_(
- expanded_state.positiontup,
- [f"{expected}_1", f"{expected}_2"],
+ @paramstyle
+ def test_expanding_in_repeated(self, paramstyle):
+ stmt = (
+ select(table1)
+ .where(
+ table1.c.name.in_(
+ bindparam("uname", value=["h", "e"], expanding=True)
+ )
+ | table1.c.name.in_(
+ bindparam("uname2", value=["y"], expanding=True)
+ )
+ )
+ .where(table1.c.myid == 8)
+ )
+ stmt = stmt.union(
+ select(table1)
+ .where(
+ table1.c.name.in_(
+ bindparam("uname", value=["h", "e"], expanding=True)
+ )
+ | table1.c.name.in_(
+ bindparam("uname2", value=["y"], expanding=True)
)
+ )
+ .where(table1.c.myid == 9)
+ ).order_by("myid")
+
+ # NOTE: below the rendered params are just what
+ # render_postcompile will do right now
+ # if you run construct_params(). render_postcompile mode
+ # is not actually used by the execution internals, it's for
+ # user-facing compilation code. So this is likely a
+ # current limitation of construct_params() which is not
+ # doing the full blown postcompile; just assert that's
+ # what it does for now. it likely should be corrected
+ # to make more sense.
+
+ if paramstyle.qmark:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (?, ?) OR "
+ "mytable.name IN (?)) "
+ "AND mytable.myid = ? "
+ "UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (?, ?) OR "
+ "mytable.name IN (?)) "
+ "AND mytable.myid = ? ORDER BY myid",
+ params={"uname": ["y", "z"], "uname2": ["a"]},
+ checkpositional=(
+ ["y", "z"],
+ ["y", "z"],
+ ["a"],
+ 8,
+ ["y", "z"],
+ ["y", "z"],
+ ["a"],
+ 9,
+ ),
+ dialect="sqlite",
+ render_postcompile=True,
+ )
+ elif paramstyle.numeric:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:3, :4) OR "
+ "mytable.name IN (:5)) "
+ "AND mytable.myid = :1 "
+ "UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:3, :4) OR "
+ "mytable.name IN (:5)) "
+ "AND mytable.myid = :2 ORDER BY myid",
+ params={"uname": ["y", "z"], "uname2": ["a"]},
+ checkpositional=(8, 9, ["y", "z"], ["y", "z"], ["a"]),
+ dialect=sqlite.dialect(paramstyle="numeric"),
+ render_postcompile=True,
+ )
+ elif paramstyle.named:
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:uname_1, :uname_2) OR "
+ "mytable.name IN (:uname2_1)) "
+ "AND mytable.myid = :myid_1 "
+ "UNION SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE (mytable.name IN (:uname_1, :uname_2) OR "
+ "mytable.name IN (:uname2_1)) "
+ "AND mytable.myid = :myid_2 ORDER BY myid",
+ params={"uname": ["y", "z"], "uname2": ["a"]},
+ checkparams={
+ "uname": ["y", "z"],
+ "uname2": ["a"],
+ "uname_1": ["y", "z"],
+ "uname_2": ["y", "z"],
+ "uname2_1": ["a"],
+ "myid_1": 8,
+ "myid_2": 9,
+ },
+ dialect="default",
+ render_postcompile=True,
+ )
+ else:
+ paramstyle.fail()
+
+ def test_numeric_dollar_bindparam(self):
+ stmt = table1.select().where(
+ table1.c.name == "a", table1.c.myid.in_([1, 2])
+ )
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable "
+ "WHERE mytable.name = $1 "
+ "AND mytable.myid IN ($2, $3)",
+ checkpositional=("a", 1, 2),
+ dialect=default.DefaultDialect(paramstyle="numeric_dollar"),
+ render_postcompile=True,
+ )
class UnsupportedTest(fixtures.TestBase):
@@ -5096,6 +5268,28 @@ class StringifySpecialTest(fixtures.TestBase):
"INSERT INTO mytable (myid) VALUES (:myid_m0), (:myid_m1)",
)
+ def test_multirow_insert_positional(self):
+ stmt = table1.insert().values([{"myid": 1}, {"myid": 2}])
+ eq_ignore_whitespace(
+ stmt.compile(dialect=sqlite.dialect()).string,
+ "INSERT INTO mytable (myid) VALUES (?), (?)",
+ )
+
+ def test_multirow_insert_numeric(self):
+ stmt = table1.insert().values([{"myid": 1}, {"myid": 2}])
+ eq_ignore_whitespace(
+ stmt.compile(dialect=sqlite.dialect(paramstyle="numeric")).string,
+ "INSERT INTO mytable (myid) VALUES (:1), (:2)",
+ )
+
+ def test_insert_noparams_numeric(self):
+ ii = table1.insert().returning(table1.c.myid)
+ eq_ignore_whitespace(
+ ii.compile(dialect=sqlite.dialect(paramstyle="numeric")).string,
+ "INSERT INTO mytable (myid, name, description) VALUES "
+ "(:1, :2, :3) RETURNING myid",
+ )
+
def test_cte(self):
# stringify of these was supported anyway by defaultdialect.
stmt = select(table1.c.myid).cte()
@@ -5153,6 +5347,42 @@ class StringifySpecialTest(fixtures.TestBase):
"SELECT CAST(mytable.myid AS MyType()) AS myid FROM mytable",
)
+ def test_dialect_sub_compile(self):
+ class Widget(ClauseElement):
+ __visit_name__ = "widget"
+ stringify_dialect = "sqlite"
+
+ def visit_widget(self, element, **kw):
+ return "widget"
+
+ with mock.patch(
+ "sqlalchemy.dialects.sqlite.base.SQLiteCompiler.visit_widget",
+ visit_widget,
+ create=True,
+ ):
+ eq_(str(Grouping(Widget())), "(widget)")
+
+ def test_dialect_sub_compile_w_binds(self):
+ """test sub-compile into a new compiler where
+ state != CompilerState.COMPILING, but we have to render a bindparam
+ string. has to render the correct template.
+
+ """
+
+ class Widget(ClauseElement):
+ __visit_name__ = "widget"
+ stringify_dialect = "sqlite"
+
+ def visit_widget(self, element, **kw):
+ return f"widget {self.process(bindparam('q'), **kw)}"
+
+ with mock.patch(
+ "sqlalchemy.dialects.sqlite.base.SQLiteCompiler.visit_widget",
+ visit_widget,
+ create=True,
+ ):
+ eq_(str(Grouping(Widget())), "(widget ?)")
+
def test_within_group(self):
# stringify of these was supported anyway by defaultdialect.
from sqlalchemy import within_group
diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index b89d18de6..502104dae 100644
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -993,20 +993,20 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
s,
'WITH regional_sales AS (SELECT orders."order" '
- 'AS "order", :1 AS anon_2 FROM orders) SELECT '
- 'regional_sales."order", :2 AS anon_1 FROM regional_sales',
- checkpositional=("x", "y"),
+ 'AS "order", :2 AS anon_2 FROM orders) SELECT '
+ 'regional_sales."order", :1 AS anon_1 FROM regional_sales',
+ checkpositional=("y", "x"),
dialect=dialect,
)
self.assert_compile(
s.union(s),
'WITH regional_sales AS (SELECT orders."order" '
- 'AS "order", :1 AS anon_2 FROM orders) SELECT '
- 'regional_sales."order", :2 AS anon_1 FROM regional_sales '
- 'UNION SELECT regional_sales."order", :3 AS anon_1 '
+ 'AS "order", :2 AS anon_2 FROM orders) SELECT '
+ 'regional_sales."order", :1 AS anon_1 FROM regional_sales '
+ 'UNION SELECT regional_sales."order", :1 AS anon_1 '
"FROM regional_sales",
- checkpositional=("x", "y", "y"),
+ checkpositional=("y", "x"),
dialect=dialect,
)
@@ -1057,8 +1057,8 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
s3,
'WITH regional_sales_1 AS (SELECT orders."order" AS "order" '
- 'FROM orders WHERE orders."order" = :1), regional_sales_2 AS '
- '(SELECT orders."order" = :2 AS anon_1, '
+ 'FROM orders WHERE orders."order" = :2), regional_sales_2 AS '
+ '(SELECT orders."order" = :1 AS anon_1, '
'anon_2."order" AS "order", '
'orders."order" AS order_1, '
'regional_sales_1."order" AS order_2 FROM orders, '
@@ -1067,7 +1067,7 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
'WHERE orders."order" = :3) SELECT regional_sales_2.anon_1, '
'regional_sales_2."order", regional_sales_2.order_1, '
"regional_sales_2.order_2 FROM regional_sales_2",
- checkpositional=("x", "y", "z"),
+ checkpositional=("y", "x", "z"),
dialect=dialect,
)
diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index ac9ac4022..1c24d4c79 100644
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -488,7 +488,8 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
dialect=postgresql.dialect(),
)
- def test_heterogeneous_multi_values(self):
+ @testing.variation("paramstyle", ["pg", "qmark", "numeric", "dollar"])
+ def test_heterogeneous_multi_values(self, paramstyle):
"""for #6047, originally I thought we'd take any insert().values()
and be able to convert it to a "many" style execution that we can
cache.
@@ -519,33 +520,81 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
]
)
+ pos_par = (
+ 1,
+ 1,
+ 2,
+ 2,
+ 1,
+ 2,
+ 2,
+ 3,
+ 1,
+ 2,
+ 2,
+ 10,
+ )
+
# SQL expressions in the params at arbitrary locations means
# we have to scan them at compile time, and the shape of the bound
# parameters is not predictable. so for #6047 where I originally
# thought all of values() could be rewritten, this makes it not
# really worth it.
- self.assert_compile(
- stmt,
- "INSERT INTO t (x, y, z) VALUES "
- "(%(x_m0)s, sum(%(sum_1)s, %(sum_2)s), %(z_m0)s), "
- "(sum(%(sum_3)s, %(sum_4)s), %(y_m1)s, %(z_m1)s), "
- "(sum(%(sum_5)s, %(sum_6)s), %(y_m2)s, foo(%(foo_1)s))",
- checkparams={
- "x_m0": 1,
- "sum_1": 1,
- "sum_2": 2,
- "z_m0": 2,
- "sum_3": 1,
- "sum_4": 2,
- "y_m1": 2,
- "z_m1": 3,
- "sum_5": 1,
- "sum_6": 2,
- "y_m2": 2,
- "foo_1": 10,
- },
- dialect=postgresql.dialect(),
- )
+ if paramstyle.pg:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "(%(x_m0)s, sum(%(sum_1)s, %(sum_2)s), %(z_m0)s), "
+ "(sum(%(sum_3)s, %(sum_4)s), %(y_m1)s, %(z_m1)s), "
+ "(sum(%(sum_5)s, %(sum_6)s), %(y_m2)s, foo(%(foo_1)s))",
+ checkparams={
+ "x_m0": 1,
+ "sum_1": 1,
+ "sum_2": 2,
+ "z_m0": 2,
+ "sum_3": 1,
+ "sum_4": 2,
+ "y_m1": 2,
+ "z_m1": 3,
+ "sum_5": 1,
+ "sum_6": 2,
+ "y_m2": 2,
+ "foo_1": 10,
+ },
+ dialect=postgresql.dialect(),
+ )
+ elif paramstyle.qmark:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "(?, sum(?, ?), ?), "
+ "(sum(?, ?), ?, ?), "
+ "(sum(?, ?), ?, foo(?))",
+ checkpositional=pos_par,
+ dialect=sqlite.dialect(),
+ )
+ elif paramstyle.numeric:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "(:1, sum(:2, :3), :4), "
+ "(sum(:5, :6), :7, :8), "
+ "(sum(:9, :10), :11, foo(:12))",
+ checkpositional=pos_par,
+ dialect=sqlite.dialect(paramstyle="numeric"),
+ )
+ elif paramstyle.dollar:
+ self.assert_compile(
+ stmt,
+ "INSERT INTO t (x, y, z) VALUES "
+ "($1, sum($2, $3), $4), "
+ "(sum($5, $6), $7, $8), "
+ "(sum($9, $10), $11, foo($12))",
+ checkpositional=pos_par,
+ dialect=sqlite.dialect(paramstyle="numeric_dollar"),
+ )
+ else:
+ paramstyle.fail()
def test_insert_seq_pk_multi_values_seq_not_supported(self):
m = MetaData()
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py
index b856acfd3..7f1124c84 100644
--- a/test/sql/test_resultset.py
+++ b/test/sql/test_resultset.py
@@ -107,7 +107,7 @@ class CursorResultTest(fixtures.TablesTest):
Column("y", String(50)),
)
- @testing.requires.insert_returning
+ @testing.requires.insert_executemany_returning
def test_splice_horizontally(self, connection):
users = self.tables.users
addresses = self.tables.addresses
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 91413ff35..59519a5ec 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -3322,7 +3322,7 @@ class ExpressionTest(
elif expression_type.right_side:
expr = (column("x", Integer) == Widget(52)).right
else:
- assert False
+ expression_type.fail()
if secondary_adapt:
is_(expr.type._type_affinity, String)
diff --git a/test/sql/test_update.py b/test/sql/test_update.py
index cd7f992e2..66971f64e 100644
--- a/test/sql/test_update.py
+++ b/test/sql/test_update.py
@@ -907,7 +907,8 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
dialect=dialect,
)
- def test_update_bound_ordering(self):
+ @testing.variation("paramstyle", ["qmark", "format", "numeric"])
+ def test_update_bound_ordering(self, paramstyle):
"""test that bound parameters between the UPDATE and FROM clauses
order correctly in different SQL compilation scenarios.
@@ -921,30 +922,47 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
.values(name="foo")
)
- dialect = default.StrCompileDialect()
- dialect.positional = True
- self.assert_compile(
- upd,
- "UPDATE mytable SET name=:name FROM (SELECT "
- "myothertable.otherid AS otherid, "
- "myothertable.othername AS othername "
- "FROM myothertable "
- "WHERE myothertable.otherid = :otherid_1) AS anon_1 "
- "WHERE mytable.name = anon_1.othername",
- checkpositional=("foo", 5),
- dialect=dialect,
- )
+ if paramstyle.qmark:
- self.assert_compile(
- upd,
- "UPDATE mytable, (SELECT myothertable.otherid AS otherid, "
- "myothertable.othername AS othername "
- "FROM myothertable "
- "WHERE myothertable.otherid = %s) AS anon_1 SET mytable.name=%s "
- "WHERE mytable.name = anon_1.othername",
- checkpositional=(5, "foo"),
- dialect=mysql.dialect(),
- )
+ dialect = default.StrCompileDialect(paramstyle="qmark")
+ self.assert_compile(
+ upd,
+ "UPDATE mytable SET name=? FROM (SELECT "
+ "myothertable.otherid AS otherid, "
+ "myothertable.othername AS othername "
+ "FROM myothertable "
+ "WHERE myothertable.otherid = ?) AS anon_1 "
+ "WHERE mytable.name = anon_1.othername",
+ checkpositional=("foo", 5),
+ dialect=dialect,
+ )
+ elif paramstyle.format:
+ self.assert_compile(
+ upd,
+ "UPDATE mytable, (SELECT myothertable.otherid AS otherid, "
+ "myothertable.othername AS othername "
+ "FROM myothertable "
+ "WHERE myothertable.otherid = %s) AS anon_1 "
+ "SET mytable.name=%s "
+ "WHERE mytable.name = anon_1.othername",
+ checkpositional=(5, "foo"),
+ dialect=mysql.dialect(),
+ )
+ elif paramstyle.numeric:
+ dialect = default.StrCompileDialect(paramstyle="numeric")
+ self.assert_compile(
+ upd,
+ "UPDATE mytable SET name=:1 FROM (SELECT "
+ "myothertable.otherid AS otherid, "
+ "myothertable.othername AS othername "
+ "FROM myothertable "
+ "WHERE myothertable.otherid = :2) AS anon_1 "
+ "WHERE mytable.name = anon_1.othername",
+ checkpositional=("foo", 5),
+ dialect=dialect,
+ )
+ else:
+ paramstyle.fail()
class UpdateFromCompileTest(
View Lorry Controller Status