diff options
Diffstat (limited to 'test/sql')
| -rw-r--r-- | test/sql/test_compare.py | 81 | ||||
| -rw-r--r-- | test/sql/test_compiler.py | 18 | ||||
| -rw-r--r-- | test/sql/test_external_traversal.py | 125 | ||||
| -rw-r--r-- | test/sql/test_select.py | 163 |
4 files changed, 372 insertions, 15 deletions
diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py index 8cc7b7fb6..247332d8c 100644 --- a/test/sql/test_compare.py +++ b/test/sql/test_compare.py @@ -28,6 +28,7 @@ from sqlalchemy import util from sqlalchemy import values from sqlalchemy.dialects import mysql from sqlalchemy.dialects import postgresql +from sqlalchemy.future import select as future_select from sqlalchemy.schema import Sequence from sqlalchemy.sql import bindparam from sqlalchemy.sql import ColumnElement @@ -91,6 +92,18 @@ table_c = Table("c", meta, Column("x", Integer), Column("y", Integer)) table_d = Table("d", meta, Column("y", Integer), Column("z", Integer)) +def opt1(ctx): + pass + + +def opt2(ctx): + pass + + +def opt3(ctx): + pass + + class MyEntity(HasCacheKey): def __init__(self, name, element): self.name = name @@ -326,6 +339,28 @@ class CoreFixtures(object): .correlate_except(table_b), ), lambda: ( + future_select(table_a.c.a), + future_select(table_a.c.a).join( + table_b, table_a.c.a == table_b.c.a + ), + future_select(table_a.c.a).join_from( + table_a, table_b, table_a.c.a == table_b.c.a + ), + future_select(table_a.c.a).join_from(table_a, table_b), + future_select(table_a.c.a).join_from(table_c, table_b), + future_select(table_a.c.a) + .join(table_b, table_a.c.a == table_b.c.a) + .join(table_c, table_b.c.b == table_c.c.x), + future_select(table_a.c.a).join(table_b), + future_select(table_a.c.a).join(table_c), + future_select(table_a.c.a).join( + table_b, table_a.c.a == table_b.c.b + ), + future_select(table_a.c.a).join( + table_c, table_a.c.a == table_c.c.x + ), + ), + lambda: ( select([table_a.c.a]).cte(), select([table_a.c.a]).cte(recursive=True), select([table_a.c.a]).cte(name="some_cte", recursive=True), @@ -610,6 +645,22 @@ class CoreFixtures(object): fixtures.append(_complex_fixtures) + def _statements_w_context_options_fixtures(): + + return [ + select([table_a])._add_context_option(opt1, True), + select([table_a])._add_context_option(opt1, 5), + select([table_a]) + ._add_context_option(opt1, True) + ._add_context_option(opt2, True), + select([table_a]) + ._add_context_option(opt1, True) + ._add_context_option(opt2, 5), + select([table_a])._add_context_option(opt3, True), + ] + + fixtures.append(_statements_w_context_options_fixtures) + class CacheKeyFixture(object): def _run_cache_key_fixture(self, fixture, compare_values): @@ -986,30 +1037,34 @@ class CompareAndCopyTest(CoreFixtures, fixtures.TestBase): class CompareClausesTest(fixtures.TestBase): - def test_compare_metadata_tables(self): - # metadata Table objects cache on their own identity, not their - # structure. This is mainly to reduce the size of cache keys - # as well as reduce computational overhead, as Table objects have - # very large internal state and they are also generally global - # objects. + def test_compare_metadata_tables_annotations_one(self): + # test that cache keys from annotated version of tables refresh + # properly t1 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer)) t2 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer)) ne_(t1._generate_cache_key(), t2._generate_cache_key()) - eq_(t1._generate_cache_key().key, (t1, "_annotations", ())) + eq_(t1._generate_cache_key().key, (t1,)) + + t2 = t1._annotate({"foo": "bar"}) + eq_( + t2._generate_cache_key().key, + (t1, "_annotations", (("foo", "bar"),)), + ) + eq_( + t2._annotate({"bat": "bar"})._generate_cache_key().key, + (t1, "_annotations", (("bat", "bar"), ("foo", "bar"))), + ) - def test_compare_metadata_tables_annotations(self): - # metadata Table objects cache on their own identity, not their - # structure. This is mainly to reduce the size of cache keys - # as well as reduce computational overhead, as Table objects have - # very large internal state and they are also generally global - # objects. + def test_compare_metadata_tables_annotations_two(self): t1 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer)) t2 = Table("a", MetaData(), Column("q", Integer), Column("p", Integer)) + eq_(t2._generate_cache_key().key, (t2,)) + t1 = t1._annotate({"orm": True}) t2 = t2._annotate({"orm": True}) diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py index b3ae7e12d..c580e972d 100644 --- a/test/sql/test_compiler.py +++ b/test/sql/test_compiler.py @@ -29,6 +29,7 @@ from sqlalchemy import exc from sqlalchemy import except_ from sqlalchemy import exists from sqlalchemy import Float +from sqlalchemy import ForeignKey from sqlalchemy import func from sqlalchemy import Integer from sqlalchemy import intersect @@ -85,7 +86,6 @@ from sqlalchemy.testing import fixtures from sqlalchemy.testing import is_ from sqlalchemy.util import u - table1 = table( "mytable", column("myid", Integer), @@ -123,6 +123,13 @@ table5 = Table( schema="dbo.remote_owner", ) +parent = Table("parent", metadata, Column("id", Integer, primary_key=True)) +child = Table( + "child", + metadata, + Column("id", Integer, primary_key=True), + Column("parent_id", ForeignKey("parent.id")), +) users = table( "users", column("user_id"), column("user_name"), column("password") ) @@ -5218,7 +5225,14 @@ class ResultMapTest(fixtures.TestBase): eq_(len(stmt.subquery().c), 7) # will render 7 as well - eq_(len(stmt._compile_state_factory(stmt, None).columns_plus_names), 7) + eq_( + len( + stmt._compile_state_factory( + stmt, stmt.compile() + ).columns_plus_names + ), + 7, + ) wrapped = stmt._generate() wrapped = wrapped.add_columns( diff --git a/test/sql/test_external_traversal.py b/test/sql/test_external_traversal.py index 37fb752fe..fb2501667 100644 --- a/test/sql/test_external_traversal.py +++ b/test/sql/test_external_traversal.py @@ -17,6 +17,7 @@ from sqlalchemy import testing from sqlalchemy import text from sqlalchemy import tuple_ from sqlalchemy import union +from sqlalchemy.future import select as future_select from sqlalchemy.sql import ClauseElement from sqlalchemy.sql import column from sqlalchemy.sql import operators @@ -754,6 +755,59 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL): ":col1_1) AS anon_1", ) + def test_this_thing_using_setup_joins_one(self): + s = ( + future_select(t1) + .join_from(t1, t2, t1.c.col1 == t2.c.col2) + .subquery() + ) + s2 = future_select(s.c.col1).join_from(t3, s, t3.c.col2 == s.c.col1) + + self.assert_compile( + s2, + "SELECT anon_1.col1 FROM table3 JOIN (SELECT table1.col1 AS " + "col1, table1.col2 AS col2, table1.col3 AS col3 FROM table1 " + "JOIN table2 ON table1.col1 = table2.col2) AS anon_1 " + "ON table3.col2 = anon_1.col1", + ) + t1a = t1.alias() + s2 = sql_util.ClauseAdapter(t1a).traverse(s2) + self.assert_compile( + s2, + "SELECT anon_1.col1 FROM table3 JOIN (SELECT table1_1.col1 AS " + "col1, table1_1.col2 AS col2, table1_1.col3 AS col3 " + "FROM table1 AS table1_1 JOIN table2 ON table1_1.col1 = " + "table2.col2) AS anon_1 ON table3.col2 = anon_1.col1", + ) + + def test_this_thing_using_setup_joins_two(self): + s = ( + future_select(t1.c.col1) + .join(t2, t1.c.col1 == t2.c.col2) + .subquery() + ) + s2 = future_select(s.c.col1) + + self.assert_compile( + s2, + "SELECT anon_1.col1 FROM (SELECT table1.col1 AS col1 " + "FROM table1 JOIN table2 ON table1.col1 = table2.col2) AS anon_1", + ) + + t1alias = t1.alias("t1alias") + j = t1.join(t1alias, t1.c.col1 == t1alias.c.col2) + + vis = sql_util.ClauseAdapter(j) + + s2 = vis.traverse(s2) + self.assert_compile( + s2, + "SELECT anon_1.col1 FROM (SELECT table1.col1 AS col1 " + "FROM table1 JOIN table1 AS t1alias " + "ON table1.col1 = t1alias.col2 " + "JOIN table2 ON table1.col1 = table2.col2) AS anon_1", + ) + def test_select_fromtwice_one(self): t1a = t1.alias() @@ -803,6 +857,77 @@ class ClauseTest(fixtures.TestBase, AssertsCompiledSQL): "AS anon_1 WHERE table1.col1 = anon_1.col1)", ) + def test_select_setup_joins_adapt_element_one(self): + s = future_select(t1).join(t2, t1.c.col1 == t2.c.col2) + + t1a = t1.alias() + + s2 = sql_util.ClauseAdapter(t1a).traverse(s) + + self.assert_compile( + s, + "SELECT table1.col1, table1.col2, table1.col3 " + "FROM table1 JOIN table2 ON table1.col1 = table2.col2", + ) + self.assert_compile( + s2, + "SELECT table1_1.col1, table1_1.col2, table1_1.col3 " + "FROM table1 AS table1_1 JOIN table2 " + "ON table1_1.col1 = table2.col2", + ) + + def test_select_setup_joins_adapt_element_two(self): + s = future_select(literal_column("1")).join_from( + t1, t2, t1.c.col1 == t2.c.col2 + ) + + t1a = t1.alias() + + s2 = sql_util.ClauseAdapter(t1a).traverse(s) + + self.assert_compile( + s, "SELECT 1 FROM table1 JOIN table2 ON table1.col1 = table2.col2" + ) + self.assert_compile( + s2, + "SELECT 1 FROM table1 AS table1_1 " + "JOIN table2 ON table1_1.col1 = table2.col2", + ) + + def test_select_setup_joins_adapt_element_three(self): + s = future_select(literal_column("1")).join_from( + t1, t2, t1.c.col1 == t2.c.col2 + ) + + t2a = t2.alias() + + s2 = sql_util.ClauseAdapter(t2a).traverse(s) + + self.assert_compile( + s, "SELECT 1 FROM table1 JOIN table2 ON table1.col1 = table2.col2" + ) + self.assert_compile( + s2, + "SELECT 1 FROM table1 " + "JOIN table2 AS table2_1 ON table1.col1 = table2_1.col2", + ) + + def test_select_setup_joins_straight_clone(self): + s = future_select(t1).join(t2, t1.c.col1 == t2.c.col2) + + s2 = CloningVisitor().traverse(s) + + self.assert_compile( + s, + "SELECT table1.col1, table1.col2, table1.col3 " + "FROM table1 JOIN table2 ON table1.col1 = table2.col2", + ) + self.assert_compile( + s2, + "SELECT table1.col1, table1.col2, table1.col3 " + "FROM table1 JOIN table2 ON table1.col1 = table2.col2", + ) + class ColumnAdapterTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = "default" diff --git a/test/sql/test_select.py b/test/sql/test_select.py new file mode 100644 index 000000000..7bac921a1 --- /dev/null +++ b/test/sql/test_select.py @@ -0,0 +1,163 @@ +from sqlalchemy import Column +from sqlalchemy import exc +from sqlalchemy import ForeignKey +from sqlalchemy import Integer +from sqlalchemy import MetaData +from sqlalchemy import String +from sqlalchemy import Table +from sqlalchemy.future import select as future_select +from sqlalchemy.sql import column +from sqlalchemy.sql import table +from sqlalchemy.testing import assert_raises_message +from sqlalchemy.testing import AssertsCompiledSQL +from sqlalchemy.testing import fixtures + +table1 = table( + "mytable", + column("myid", Integer), + column("name", String), + column("description", String), +) + +table2 = table( + "myothertable", column("otherid", Integer), column("othername", String) +) + +metadata = MetaData() + + +parent = Table( + "parent", + metadata, + Column("id", Integer, primary_key=True), + Column("data", String(50)), +) +child = Table( + "child", + metadata, + Column("id", Integer, primary_key=True), + Column("parent_id", ForeignKey("parent.id")), + Column("data", String(50)), +) + + +class FutureSelectTest(fixtures.TestBase, AssertsCompiledSQL): + __dialect__ = "default" + + def test_join_nofrom_implicit_left_side_explicit_onclause(self): + stmt = future_select(table1).join( + table2, table1.c.myid == table2.c.otherid + ) + + self.assert_compile( + stmt, + "SELECT mytable.myid, mytable.name, mytable.description " + "FROM mytable JOIN myothertable " + "ON mytable.myid = myothertable.otherid", + ) + + def test_join_nofrom_explicit_left_side_explicit_onclause(self): + stmt = future_select(table1).join_from( + table1, table2, table1.c.myid == table2.c.otherid + ) + + self.assert_compile( + stmt, + "SELECT mytable.myid, mytable.name, mytable.description " + "FROM mytable JOIN myothertable " + "ON mytable.myid = myothertable.otherid", + ) + + def test_join_nofrom_implicit_left_side_implicit_onclause(self): + stmt = future_select(parent).join(child) + + self.assert_compile( + stmt, + "SELECT parent.id, parent.data FROM parent JOIN child " + "ON parent.id = child.parent_id", + ) + + def test_join_nofrom_explicit_left_side_implicit_onclause(self): + stmt = future_select(parent).join_from(parent, child) + + self.assert_compile( + stmt, + "SELECT parent.id, parent.data FROM parent JOIN child " + "ON parent.id = child.parent_id", + ) + + def test_join_froms_implicit_left_side_explicit_onclause(self): + stmt = ( + future_select(table1) + .select_from(table1) + .join(table2, table1.c.myid == table2.c.otherid) + ) + + self.assert_compile( + stmt, + "SELECT mytable.myid, mytable.name, mytable.description " + "FROM mytable JOIN myothertable " + "ON mytable.myid = myothertable.otherid", + ) + + def test_join_froms_explicit_left_side_explicit_onclause(self): + stmt = ( + future_select(table1) + .select_from(table1) + .join_from(table1, table2, table1.c.myid == table2.c.otherid) + ) + + self.assert_compile( + stmt, + "SELECT mytable.myid, mytable.name, mytable.description " + "FROM mytable JOIN myothertable " + "ON mytable.myid = myothertable.otherid", + ) + + def test_join_froms_implicit_left_side_implicit_onclause(self): + stmt = future_select(parent).select_from(parent).join(child) + + self.assert_compile( + stmt, + "SELECT parent.id, parent.data FROM parent JOIN child " + "ON parent.id = child.parent_id", + ) + + def test_join_froms_explicit_left_side_implicit_onclause(self): + stmt = ( + future_select(parent).select_from(parent).join_from(parent, child) + ) + + self.assert_compile( + stmt, + "SELECT parent.id, parent.data FROM parent JOIN child " + "ON parent.id = child.parent_id", + ) + + def test_joins_w_filter_by(self): + stmt = ( + future_select(parent) + .filter_by(data="p1") + .join(child) + .filter_by(data="c1") + .join_from(table1, table2, table1.c.myid == table2.c.otherid) + .filter_by(otherid=5) + ) + + self.assert_compile( + stmt, + "SELECT parent.id, parent.data FROM parent JOIN child " + "ON parent.id = child.parent_id, mytable JOIN myothertable " + "ON mytable.myid = myothertable.otherid " + "WHERE parent.data = :data_1 AND child.data = :data_2 " + "AND myothertable.otherid = :otherid_1", + checkparams={"data_1": "p1", "data_2": "c1", "otherid_1": 5}, + ) + + def test_filter_by_no_property(self): + assert_raises_message( + exc.InvalidRequestError, + 'Entity namespace for "mytable" has no property "foo"', + future_select(table1).filter_by, + foo="bar", + ) |