summaryrefslogtreecommitdiff
path: root/test/sql/test_update.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/sql/test_update.py')
-rw-r--r--test/sql/test_update.py113
1 files changed, 104 insertions, 9 deletions
diff --git a/test/sql/test_update.py b/test/sql/test_update.py
index 1eed37832..964e3ee6b 100644
--- a/test/sql/test_update.py
+++ b/test/sql/test_update.py
@@ -1,3 +1,6 @@
+import itertools
+import random
+
from sqlalchemy import bindparam
from sqlalchemy import column
from sqlalchemy import exc
@@ -23,6 +26,7 @@ from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
@@ -856,34 +860,125 @@ class UpdateTest(_UpdateFromTestBase, fixtures.TablesTest, AssertsCompiledSQL):
"UPDATE mytable SET foo(myid)=:param_1",
)
- def test_update_to_expression_two(self):
- """test update from an expression.
+ @testing.fixture
+ def randomized_param_order_update(self):
+ from sqlalchemy.sql.dml import UpdateDMLState
- this logic is triggered currently by a left side that doesn't
- have a key. The current supported use case is updating the index
- of a PostgreSQL ARRAY type.
+ super_process_ordered_values = UpdateDMLState._process_ordered_values
- """
+ # this fixture is needed for Python 3.6 and above to work around
+ # dictionaries being insert-ordered. in python 2.7 the previous
+ # logic fails pretty easily without this fixture.
+ def _process_ordered_values(self, statement):
+ super_process_ordered_values(self, statement)
+
+ tuples = list(self._dict_parameters.items())
+ random.shuffle(tuples)
+ self._dict_parameters = dict(tuples)
+
+ dialect = default.StrCompileDialect()
+ dialect.paramstyle = "qmark"
+ dialect.positional = True
+
+ with mock.patch.object(
+ UpdateDMLState, "_process_ordered_values", _process_ordered_values
+ ):
+ yield
+ def random_update_order_parameters():
from sqlalchemy import ARRAY
t = table(
"foo",
column("data1", ARRAY(Integer)),
column("data2", ARRAY(Integer)),
+ column("data3", ARRAY(Integer)),
+ column("data4", ARRAY(Integer)),
)
+ idx_to_value = [
+ (t.c.data1, 5, 7),
+ (t.c.data2, 10, 18),
+ (t.c.data3, 8, 4),
+ (t.c.data4, 12, 14),
+ ]
+
+ def combinations():
+ while True:
+ random.shuffle(idx_to_value)
+ yield list(idx_to_value)
+
+ return testing.combinations(
+ *[
+ (t, combination)
+ for i, combination in zip(range(10), combinations())
+ ],
+ argnames="t, idx_to_value"
+ )
+
+ @random_update_order_parameters()
+ def test_update_to_expression_two(
+ self, randomized_param_order_update, t, idx_to_value
+ ):
+ """test update from an expression.
+
+ this logic is triggered currently by a left side that doesn't
+ have a key. The current supported use case is updating the index
+ of a PostgreSQL ARRAY type.
+
+ """
+
+ dialect = default.StrCompileDialect()
+ dialect.paramstyle = "qmark"
+ dialect.positional = True
+
stmt = t.update().ordered_values(
- (t.c.data1[5], 7), (t.c.data2[10], 18)
+ *[(col[idx], val) for col, idx, val in idx_to_value]
)
+
+ self.assert_compile(
+ stmt,
+ "UPDATE foo SET %s"
+ % (
+ ", ".join(
+ "%s[?]=?" % col.key for col, idx, val in idx_to_value
+ )
+ ),
+ dialect=dialect,
+ checkpositional=tuple(
+ itertools.chain.from_iterable(
+ (idx, val) for col, idx, val in idx_to_value
+ )
+ ),
+ )
+
+ @random_update_order_parameters()
+ def test_update_to_expression_ppo(
+ self, randomized_param_order_update, t, idx_to_value
+ ):
dialect = default.StrCompileDialect()
dialect.paramstyle = "qmark"
dialect.positional = True
+
+ # deprecated pattern here
+ stmt = t.update(preserve_parameter_order=True).values(
+ [(col[idx], val) for col, idx, val in idx_to_value]
+ )
+
self.assert_compile(
stmt,
- "UPDATE foo SET data1[?]=?, data2[?]=?",
+ "UPDATE foo SET %s"
+ % (
+ ", ".join(
+ "%s[?]=?" % col.key for col, idx, val in idx_to_value
+ )
+ ),
dialect=dialect,
- checkpositional=(5, 7, 10, 18),
+ checkpositional=tuple(
+ itertools.chain.from_iterable(
+ (idx, val) for col, idx, val in idx_to_value
+ )
+ ),
)
def test_update_to_expression_three(self):