summaryrefslogtreecommitdiff
path: root/test/sql
diff options
context:
space:
mode:
Diffstat (limited to 'test/sql')
-rw-r--r--test/sql/test_compiler.py4
-rw-r--r--test/sql/test_deprecations.py79
-rw-r--r--test/sql/test_external_traversal.py39
-rw-r--r--test/sql/test_operators.py29
-rw-r--r--test/sql/test_selectable.py12
-rw-r--r--test/sql/test_utils.py20
6 files changed, 74 insertions, 109 deletions
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 6b1f443e2..033da10a3 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -4863,7 +4863,7 @@ class ResultMapTest(fixtures.TestBase):
with self._nested_result() as nested:
contexts[stmt2.element] = nested
text = super(MyCompiler, self).visit_select(
- stmt2.element
+ stmt2.element,
)
self._add_to_result_map("k1", "k1", (1, 2, 3), int_)
else:
@@ -4970,7 +4970,7 @@ class ResultMapTest(fixtures.TestBase):
eq_(len(stmt.subquery().c), 7)
# will render 7 as well
- eq_(len(stmt._columns_plus_names), 7)
+ eq_(len(stmt._compile_state_factory(stmt, None).columns_plus_names), 7)
wrapped = stmt._generate()
wrapped = wrapped.add_columns(
diff --git a/test/sql/test_deprecations.py b/test/sql/test_deprecations.py
index 5b7b3bd1f..93b29847f 100644
--- a/test/sql/test_deprecations.py
+++ b/test/sql/test_deprecations.py
@@ -112,20 +112,6 @@ class DeprecationWarningsTest(fixtures.TestBase, AssertsCompiledSQL):
assert t1t2.onclause.compare(join_cond)
- def test_select_autocommit(self):
- with testing.expect_deprecated(
- "The select.autocommit parameter is deprecated and "
- "will be removed in a future release."
- ):
- select([column("x")], autocommit=True)
-
- def test_select_for_update(self):
- with testing.expect_deprecated(
- "The select.for_update parameter is deprecated and "
- "will be removed in a future release."
- ):
- select([column("x")], for_update=True)
-
def test_empty_and_or(self):
with testing.expect_deprecated(
r"Invoking and_\(\) without arguments is deprecated, and "
@@ -197,63 +183,6 @@ class ConvertUnicodeDeprecationTest(fixtures.TestBase):
eq_(proc(utfdata), unicodedata.encode("ascii", "ignore").decode())
-class ForUpdateTest(fixtures.TestBase, AssertsCompiledSQL):
- __dialect__ = "default"
-
- def _assert_legacy(self, leg, read=False, nowait=False):
- t = table("t", column("c"))
-
- with testing.expect_deprecated(
- "The select.for_update parameter is deprecated and "
- "will be removed in a future release."
- ):
- s1 = select([t], for_update=leg)
-
- if leg is False:
- assert s1._for_update_arg is None
- assert s1.for_update is None
- else:
- eq_(s1._for_update_arg.read, read)
- eq_(s1._for_update_arg.nowait, nowait)
- eq_(s1.for_update, leg)
-
- def test_false_legacy(self):
- self._assert_legacy(False)
-
- def test_plain_true_legacy(self):
- self._assert_legacy(True)
-
- def test_read_legacy(self):
- self._assert_legacy("read", read=True)
-
- def test_nowait_legacy(self):
- self._assert_legacy("nowait", nowait=True)
-
- def test_read_nowait_legacy(self):
- self._assert_legacy("read_nowait", read=True, nowait=True)
-
- def test_unknown_mode(self):
- t = table("t", column("c"))
-
- with testing.expect_deprecated(
- "The select.for_update parameter is deprecated and "
- "will be removed in a future release."
- ):
- assert_raises_message(
- exc.ArgumentError,
- "Unknown for_update argument: 'unknown_mode'",
- t.select,
- t.c.c == 7,
- for_update="unknown_mode",
- )
-
- def test_legacy_setter(self):
- t = table("t", column("c"))
- s = select([t])
- s.for_update = "nowait"
- eq_(s._for_update_arg.nowait, True)
-
-
class SubqueryCoercionsTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
@@ -606,12 +535,6 @@ class TextTest(fixtures.TestBase, AssertsCompiledSQL):
},
)
- def test_autocommit(self):
- with testing.expect_deprecated(
- "The text.autocommit parameter is deprecated"
- ):
- text("select id, name from user", autocommit=True)
-
class SelectableTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = "default"
@@ -1723,7 +1646,7 @@ class DefaultTest(fixtures.TestBase):
metadata.create_all(testing.db)
with testing.db.connect() as conn:
- with testing.expect_deprecated(
+ with testing.expect_deprecated_20(
r"The .close\(\) method on a so-called 'branched' "
r"connection is deprecated as of 1.4, as are "
r"'branched' connections overall"
diff --git a/test/sql/test_external_traversal.py b/test/sql/test_external_traversal.py
index 2a82c2cc1..ac423931f 100644
--- a/test/sql/test_external_traversal.py
+++ b/test/sql/test_external_traversal.py
@@ -1005,8 +1005,10 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
s = select(
[literal_column("*")], from_obj=[t1alias, t2alias]
).scalar_subquery()
- assert t2alias in s._froms
- assert t1alias in s._froms
+
+ froms = list(s._iterate_from_elements())
+ assert t2alias in froms
+ assert t1alias in froms
self.assert_compile(
select([literal_column("*")], t2alias.c.col1 == s),
@@ -1016,8 +1018,9 @@ class ClauseAdapterTest(fixtures.TestBase, AssertsCompiledSQL):
)
s = vis.traverse(s)
- assert t2alias in s._froms # present because it was not cloned
- assert t1alias in s._froms # present because the adapter placed
+ froms = list(s._iterate_from_elements())
+ assert t2alias in froms # present because it was not cloned
+ assert t1alias in froms # present because the adapter placed
# it there and was also not cloned
# correlate list on "s" needs to take into account the full
@@ -1853,7 +1856,7 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
"SELECT table1.col1, table1.col2, "
"table1.col3 FROM table1, table2",
)
- assert s._froms is not select_copy._froms
+
self.assert_compile(
s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
)
@@ -1961,13 +1964,13 @@ class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
def test_add_kwarg(self):
i = t1.insert()
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
eq_(compile_state._dict_parameters, None)
i = i.values(col1=5)
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(compile_state._dict_parameters, {"col1": 5})
i = i.values(col2=7)
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(
compile_state._dict_parameters, {"col1": 5, "col2": 7}
)
@@ -1975,11 +1978,11 @@ class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
def test_via_tuple_single(self):
i = t1.insert()
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
eq_(compile_state._dict_parameters, None)
i = i.values((5, 6, 7))
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(
compile_state._dict_parameters, {"col1": 5, "col2": 6, "col3": 7},
@@ -1997,11 +2000,11 @@ class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
def test_via_tuple_multi(self):
i = t1.insert()
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
eq_(compile_state._dict_parameters, None)
i = i.values([(5, 6, 7), (8, 9, 10)])
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
eq_(
compile_state._dict_parameters, {"col1": 5, "col2": 6, "col3": 7},
)
@@ -2017,7 +2020,7 @@ class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
def test_inline_values_single(self):
i = t1.insert(values={"col1": 5})
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(compile_state._dict_parameters, {"col1": 5})
is_(compile_state._has_multi_parameters, False)
@@ -2025,7 +2028,7 @@ class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
def test_inline_values_multi(self):
i = t1.insert(values=[{"col1": 5}, {"col1": 6}])
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
# multiparams are not converted to bound parameters
eq_(compile_state._dict_parameters, {"col1": 5})
@@ -2050,25 +2053,25 @@ class ValuesBaseTest(fixtures.TestBase, AssertsCompiledSQL):
def test_add_dictionary(self):
i = t1.insert()
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
eq_(compile_state._dict_parameters, None)
i = i.values({"col1": 5})
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(compile_state._dict_parameters, {"col1": 5})
is_(compile_state._has_multi_parameters, False)
i = i.values({"col1": 6})
# note replaces
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(compile_state._dict_parameters, {"col1": 6})
is_(compile_state._has_multi_parameters, False)
i = i.values({"col2": 7})
- compile_state = i._compile_state_cls(i, None, isinsert=True)
+ compile_state = i._compile_state_factory(i, None)
self._compare_param_dict(
compile_state._dict_parameters, {"col1": 6, "col2": 7}
)
diff --git a/test/sql/test_operators.py b/test/sql/test_operators.py
index d3afc2dee..ea3a1f773 100644
--- a/test/sql/test_operators.py
+++ b/test/sql/test_operators.py
@@ -1106,6 +1106,7 @@ class ConjunctionTest(fixtures.TestBase, testing.AssertsCompiledSQL):
def test_empty_clauses(self, op, str_op, str_continue):
# these warning classes will change to ArgumentError when the
# deprecated behavior is disabled
+
assert_raises_message(
exc.SADeprecationWarning,
r"Invoking %(str_op)s\(\) without arguments is deprecated, and "
@@ -1199,21 +1200,35 @@ class ConjunctionTest(fixtures.TestBase, testing.AssertsCompiledSQL):
select([x]).where(~null()), "SELECT x WHERE NOT NULL"
)
- def test_constant_non_singleton(self):
- is_not_(null(), null())
- is_not_(false(), false())
- is_not_(true(), true())
+ def test_constants_are_singleton(self):
+ is_(null(), null())
+ is_(false(), false())
+ is_(true(), true())
def test_constant_render_distinct(self):
self.assert_compile(
- select([null(), null()]), "SELECT NULL AS anon_1, NULL AS anon_2"
+ select([null(), null()]), "SELECT NULL AS anon_1, NULL AS anon__1"
)
self.assert_compile(
- select([true(), true()]), "SELECT true AS anon_1, true AS anon_2"
+ select([true(), true()]), "SELECT true AS anon_1, true AS anon__1"
)
self.assert_compile(
select([false(), false()]),
- "SELECT false AS anon_1, false AS anon_2",
+ "SELECT false AS anon_1, false AS anon__1",
+ )
+
+ def test_constant_render_distinct_use_labels(self):
+ self.assert_compile(
+ select([null(), null()]).apply_labels(),
+ "SELECT NULL AS anon_1, NULL AS anon__1",
+ )
+ self.assert_compile(
+ select([true(), true()]).apply_labels(),
+ "SELECT true AS anon_1, true AS anon__1",
+ )
+ self.assert_compile(
+ select([false(), false()]).apply_labels(),
+ "SELECT false AS anon_1, false AS anon__1",
)
def test_is_true_literal(self):
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index cc2df16a9..30dfc0630 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -1133,12 +1133,14 @@ class SelectableTest(
s3 = sql_util.ClauseAdapter(ta).traverse(s2)
- assert s1 not in s3._froms
+ froms = list(s3._iterate_from_elements())
+
+ assert s1 not in froms
# these are new assumptions with the newer approach that
# actively swaps out whereclause and others
assert s3._whereclause.left.table is not s1
- assert s3._whereclause.left.table in s3._froms
+ assert s3._whereclause.left.table in froms
class RefreshForNewColTest(fixtures.TestBase):
@@ -2540,7 +2542,8 @@ class AnnotationsTest(fixtures.TestBase):
):
# the columns clause isn't changed at all
assert sel._raw_columns[0].table is a1
- assert sel._froms[0].element is sel._froms[1].left.element
+ froms = list(sel._iterate_from_elements())
+ assert froms[0].element is froms[1].left.element
eq_(str(s), str(sel))
@@ -2551,7 +2554,8 @@ class AnnotationsTest(fixtures.TestBase):
sql_util._deep_deannotate(s, {"foo": "bar"}),
sql_util._deep_annotate(s, {"foo": "bar"}),
):
- assert sel._froms[0] is not sel._froms[1].left
+ froms = list(sel._iterate_from_elements())
+ assert froms[0] is not froms[1].left
# but things still work out due to
# re49563072578
diff --git a/test/sql/test_utils.py b/test/sql/test_utils.py
index 48d6de6db..1ccd1e123 100644
--- a/test/sql/test_utils.py
+++ b/test/sql/test_utils.py
@@ -1,3 +1,9 @@
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import select
+from sqlalchemy import String
+from sqlalchemy import Table
from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.testing import eq_
@@ -10,3 +16,17 @@ class MiscTest(fixtures.TestBase):
_traverse_internals = []
eq_(sql_util.find_tables(MyElement(), check_columns=True), [])
+
+ def test_find_tables_selectable(self):
+ metadata = MetaData()
+ common = Table(
+ "common",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", Integer),
+ Column("extra", String(45)),
+ )
+
+ subset_select = select([common.c.id, common.c.data]).alias()
+
+ eq_(sql_util.find_tables(subset_select), [common])