summaryrefslogtreecommitdiff
path: root/test/sql
diff options
context:
space:
mode:
Diffstat (limited to 'test/sql')
-rw-r--r--test/sql/test_compare.py2
-rw-r--r--test/sql/test_compiler.py13
-rw-r--r--test/sql/test_cte.py31
-rw-r--r--test/sql/test_selectable.py18
-rw-r--r--test/sql/test_text.py43
-rw-r--r--test/sql/test_values.py25
6 files changed, 119 insertions, 13 deletions
diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index dd073d2a5..8d6dc7553 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -208,6 +208,8 @@ class CoreFixtures:
column("q") == column("x"),
column("q") == column("y"),
column("z") == column("x"),
+ (column("z") == column("x")).self_group(),
+ (column("q") == column("x")).self_group(),
column("z") + column("x"),
column("z") - column("x"),
column("x") - column("z"),
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index d20037e92..6ca06dc0e 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -82,7 +82,6 @@ from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.elements import CompilerColumnElement
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import ClauseList
-from sqlalchemy.sql.expression import HasPrefixes
from sqlalchemy.sql.selectable import LABEL_STYLE_NONE
from sqlalchemy.sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.testing import assert_raises
@@ -270,18 +269,6 @@ class SelectTest(fixtures.TestBase, AssertsCompiledSQL):
"columns",
)
- def test_prefix_constructor(self):
- class Pref(HasPrefixes):
- def _generate(self):
- return self
-
- assert_raises(
- exc.ArgumentError,
- Pref().prefix_with,
- "some prefix",
- not_a_dialect=True,
- )
-
def test_table_select(self):
self.assert_compile(
table1.select(),
diff --git a/test/sql/test_cte.py b/test/sql/test_cte.py
index 686d4928d..d1d01a5c7 100644
--- a/test/sql/test_cte.py
+++ b/test/sql/test_cte.py
@@ -1186,6 +1186,37 @@ class CTETest(fixtures.TestBase, AssertsCompiledSQL):
dialect="postgresql",
)
+ def test_recursive_dml_syntax(self):
+ orders = table(
+ "orders",
+ column("region"),
+ column("amount"),
+ column("product"),
+ column("quantity"),
+ )
+
+ upsert = (
+ orders.update()
+ .where(orders.c.region == "Region1")
+ .values(amount=1.0, product="Product1", quantity=1)
+ .returning(*(orders.c._all_columns))
+ .cte("upsert", recursive=True)
+ )
+ stmt = select(upsert)
+
+ # This statement probably makes no sense, just want to see that the
+ # column generation aspect needed by RECURSIVE works (new in 2.0)
+ self.assert_compile(
+ stmt,
+ "WITH RECURSIVE upsert(region, amount, product, quantity) "
+ "AS (UPDATE orders SET amount=:param_1, product=:param_2, "
+ "quantity=:param_3 WHERE orders.region = :region_1 "
+ "RETURNING orders.region, orders.amount, orders.product, "
+ "orders.quantity) "
+ "SELECT upsert.region, upsert.amount, upsert.product, "
+ "upsert.quantity FROM upsert",
+ )
+
def test_upsert_from_select(self):
orders = table(
"orders",
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index ca5f43bb6..9fdc51938 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -432,6 +432,24 @@ class SelectableTest(
):
select(stmt.subquery()).compile()
+ def test_correlate_none_arg_error(self):
+ stmt = select(table1)
+ with expect_raises_message(
+ exc.ArgumentError,
+ "additional FROM objects not accepted when passing "
+ "None/False to correlate",
+ ):
+ stmt.correlate(None, table2)
+
+ def test_correlate_except_none_arg_error(self):
+ stmt = select(table1)
+ with expect_raises_message(
+ exc.ArgumentError,
+ "additional FROM objects not accepted when passing "
+ "None/False to correlate_except",
+ ):
+ stmt.correlate_except(None, table2)
+
def test_select_label_grouped_still_corresponds(self):
label = select(table1.c.col1).label("foo")
label2 = label.self_group()
diff --git a/test/sql/test_text.py b/test/sql/test_text.py
index 81b20f86f..0f645a2d2 100644
--- a/test/sql/test_text.py
+++ b/test/sql/test_text.py
@@ -688,6 +688,19 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
mapping = self._mapping(s)
assert x not in mapping
+ def test_subquery_accessors(self):
+ t = self._xy_table_fixture()
+
+ s = text("SELECT x from t").columns(t.c.x)
+
+ self.assert_compile(
+ select(s.scalar_subquery()), "SELECT (SELECT x from t) AS anon_1"
+ )
+ self.assert_compile(
+ select(s.subquery()),
+ "SELECT anon_1.x FROM (SELECT x from t) AS anon_1",
+ )
+
def test_select_label_alt_name_table_alias_column(self):
t = self._xy_table_fixture()
x = t.c.x
@@ -716,6 +729,36 @@ class AsFromTest(fixtures.TestBase, AssertsCompiledSQL):
"FROM mytable, t WHERE mytable.myid = t.id",
)
+ def test_cte_recursive(self):
+ t = (
+ text("select id, name from user")
+ .columns(id=Integer, name=String)
+ .cte("t", recursive=True)
+ )
+
+ s = select(table1).where(table1.c.myid == t.c.id)
+ self.assert_compile(
+ s,
+ "WITH RECURSIVE t(id, name) AS (select id, name from user) "
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable, t WHERE mytable.myid = t.id",
+ )
+
+ def test_unions(self):
+ s1 = text("select id, name from user where id > 5").columns(
+ id=Integer, name=String
+ )
+ s2 = text("select id, name from user where id < 15").columns(
+ id=Integer, name=String
+ )
+ stmt = union(s1, s2)
+ eq_(stmt.selected_columns.keys(), ["id", "name"])
+ self.assert_compile(
+ stmt,
+ "select id, name from user where id > 5 UNION "
+ "select id, name from user where id < 15",
+ )
+
def test_subquery(self):
t = (
text("select id, name from user")
diff --git a/test/sql/test_values.py b/test/sql/test_values.py
index f5ae9ea53..d14de9aee 100644
--- a/test/sql/test_values.py
+++ b/test/sql/test_values.py
@@ -294,6 +294,31 @@ class ValuesTest(fixtures.TablesTest, AssertsCompiledSQL):
checkparams={},
)
+ def test_anon_alias(self):
+ people = self.tables.people
+ values = (
+ Values(
+ column("bookcase_id", Integer),
+ column("bookcase_owner_id", Integer),
+ )
+ .data([(1, 1), (2, 1), (3, 2), (3, 3)])
+ .alias()
+ )
+ stmt = select(people, values).select_from(
+ people.join(
+ values, values.c.bookcase_owner_id == people.c.people_id
+ )
+ )
+ self.assert_compile(
+ stmt,
+ "SELECT people.people_id, people.age, people.name, "
+ "anon_1.bookcase_id, anon_1.bookcase_owner_id FROM people "
+ "JOIN (VALUES (:param_1, :param_2), (:param_3, :param_4), "
+ "(:param_5, :param_6), (:param_7, :param_8)) AS anon_1 "
+ "(bookcase_id, bookcase_owner_id) "
+ "ON people.people_id = anon_1.bookcase_owner_id",
+ )
+
def test_with_join_unnamed(self):
people = self.tables.people
values = Values(
View Lorry Controller Status