diff options
Diffstat (limited to 'test/sql')
| -rw-r--r-- | test/sql/test_resultset.py | 226 | ||||
| -rw-r--r-- | test/sql/test_type_expressions.py | 7 |
2 files changed, 197 insertions, 36 deletions
diff --git a/test/sql/test_resultset.py b/test/sql/test_resultset.py index 8461996ea..d7dc9edc3 100644 --- a/test/sql/test_resultset.py +++ b/test/sql/test_resultset.py @@ -10,6 +10,7 @@ from sqlalchemy import ( from sqlalchemy.engine import result as _result from sqlalchemy.testing.schema import Table, Column import operator +from sqlalchemy.testing import assertions class ResultProxyTest(fixtures.TablesTest): @@ -604,17 +605,11 @@ class ResultProxyTest(fixtures.TablesTest): lambda: r['user_id'] ) - assert_raises_message( - exc.InvalidRequestError, - "Ambiguous column name", - lambda: r[users.c.user_id] - ) - - assert_raises_message( - exc.InvalidRequestError, - "Ambiguous column name", - lambda: r[addresses.c.user_id] - ) + # pure positional targeting; users.c.user_id + # and addresses.c.user_id are known! + # works as of 1.1 issue #3501 + eq_(r[users.c.user_id], 1) + eq_(r[addresses.c.user_id], None) # try to trick it - fake_table isn't in the result! # we get the correct error @@ -652,31 +647,17 @@ class ResultProxyTest(fixtures.TablesTest): result = select([users.c.user_id, ua.c.user_id]).execute() row = result.first() - assert_raises_message( - exc.InvalidRequestError, - "Ambiguous column name", - lambda: row[users.c.user_id] - ) + # as of 1.1 issue #3501, we use pure positional + # targeting for the column objects here + eq_(row[users.c.user_id], 1) - assert_raises_message( - exc.InvalidRequestError, - "Ambiguous column name", - lambda: row[ua.c.user_id] - ) + eq_(row[ua.c.user_id], 1) - # Unfortunately, this fails - - # we'd like - # "Could not locate column in row" - # to be raised here, but the check for - # "common column" in _compare_name_for_result() - # has other requirements to be more liberal. - # Ultimately the - # expression system would need a way to determine - # if given two columns in a "proxy" relationship, if they - # refer to a different parent table + # this now works as of 1.1 issue #3501; + # previously this was stuck on "ambiguous column name" assert_raises_message( exc.InvalidRequestError, - "Ambiguous column name", + "Could not locate column in row", lambda: row[u2.c.user_id] ) @@ -1012,7 +993,7 @@ class KeyTargetingTest(fixtures.TablesTest): eq_(row.q, "c1") assert_raises_message( exc.InvalidRequestError, - "Ambiguous column name 'b'", + "Ambiguous column name 'a'", getattr, row, "b" ) assert_raises_message( @@ -1134,3 +1115,182 @@ class KeyTargetingTest(fixtures.TablesTest): in_(keyed2.c.b, row) in_(stmt.c.keyed2_a, row) in_(stmt.c.keyed2_b, row) + + +class PositionalTextTest(fixtures.TablesTest): + run_inserts = 'once' + run_deletes = None + __backend__ = True + + @classmethod + def define_tables(cls, metadata): + Table( + 'text1', + metadata, + Column("a", CHAR(2)), + Column("b", CHAR(2)), + Column("c", CHAR(2)), + Column("d", CHAR(2)) + ) + + @classmethod + def insert_data(cls): + cls.tables.text1.insert().execute([ + dict(a="a1", b="b1", c="c1", d="d1"), + ]) + + def test_via_column(self): + c1, c2, c3, c4 = column('q'), column('p'), column('r'), column('d') + stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c4) + + result = testing.db.execute(stmt) + row = result.first() + + eq_(row[c2], "b1") + eq_(row[c4], "d1") + eq_(row[1], "b1") + eq_(row["b"], "b1") + eq_(row.keys(), ["a", "b", "c", "d"]) + eq_(row["r"], "c1") + eq_(row["d"], "d1") + + def test_fewer_cols_than_sql_positional(self): + c1, c2 = column('q'), column('p') + stmt = text("select a, b, c, d from text1").columns(c1, c2) + + # no warning as this can be similar for non-positional + result = testing.db.execute(stmt) + row = result.first() + + eq_(row[c1], "a1") + eq_(row["c"], "c1") + + def test_fewer_cols_than_sql_non_positional(self): + c1, c2 = column('a'), column('p') + stmt = text("select a, b, c, d from text1").columns(c2, c1, d=CHAR) + + # no warning as this can be similar for non-positional + result = testing.db.execute(stmt) + row = result.first() + + # c1 name matches, locates + eq_(row[c1], "a1") + eq_(row["c"], "c1") + + # c2 name does not match, doesn't locate + assert_raises_message( + exc.NoSuchColumnError, + "in row for column 'p'", + lambda: row[c2] + ) + + def test_more_cols_than_sql(self): + c1, c2, c3, c4 = column('q'), column('p'), column('r'), column('d') + stmt = text("select a, b from text1").columns(c1, c2, c3, c4) + + with assertions.expect_warnings( + r"Number of columns in textual SQL \(4\) is " + "smaller than number of columns requested \(2\)"): + result = testing.db.execute(stmt) + + row = result.first() + eq_(row[c2], "b1") + + assert_raises_message( + exc.NoSuchColumnError, + "in row for column 'r'", + lambda: row[c3] + ) + + def test_dupe_col_obj(self): + c1, c2, c3 = column('q'), column('p'), column('r') + stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c2) + + assert_raises_message( + exc.InvalidRequestError, + "Duplicate column expression requested in " + "textual SQL: <.*.ColumnClause.*; p>", + testing.db.execute, stmt + ) + + def test_anon_aliased_unique(self): + text1 = self.tables.text1 + + c1 = text1.c.a.label(None) + c2 = text1.alias().c.c + c3 = text1.alias().c.b + c4 = text1.alias().c.d.label(None) + + stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c4) + result = testing.db.execute(stmt) + row = result.first() + + eq_(row[c1], "a1") + eq_(row[c2], "b1") + eq_(row[c3], "c1") + eq_(row[c4], "d1") + + # key fallback rules still match this to a column + # unambiguously based on its name + eq_(row[text1.c.a], "a1") + + # key fallback rules still match this to a column + # unambiguously based on its name + eq_(row[text1.c.d], "d1") + + # text1.c.b goes nowhere....because we hit key fallback + # but the text1.c.b doesn't derive from text1.c.c + assert_raises_message( + exc.NoSuchColumnError, + "Could not locate column in row for column 'text1.b'", + lambda: row[text1.c.b] + ) + + def test_anon_aliased_overlapping(self): + text1 = self.tables.text1 + + c1 = text1.c.a.label(None) + c2 = text1.alias().c.a + c3 = text1.alias().c.a.label(None) + c4 = text1.c.a.label(None) + + stmt = text("select a, b, c, d from text1").columns(c1, c2, c3, c4) + result = testing.db.execute(stmt) + row = result.first() + + eq_(row[c1], "a1") + eq_(row[c2], "b1") + eq_(row[c3], "c1") + eq_(row[c4], "d1") + + # key fallback rules still match this to a column + # unambiguously based on its name + eq_(row[text1.c.a], "a1") + + def test_anon_aliased_name_conflict(self): + text1 = self.tables.text1 + + c1 = text1.c.a.label("a") + c2 = text1.alias().c.a + c3 = text1.alias().c.a.label("a") + c4 = text1.c.a.label("a") + + # all cols are named "a". if we are positional, we don't care. + # this is new logic in 1.1 + stmt = text("select a, b as a, c as a, d as a from text1").columns( + c1, c2, c3, c4) + result = testing.db.execute(stmt) + row = result.first() + + eq_(row[c1], "a1") + eq_(row[c2], "b1") + eq_(row[c3], "c1") + eq_(row[c4], "d1") + + # fails, because we hit key fallback and find conflicts + # in columns that are presnet + assert_raises_message( + exc.NoSuchColumnError, + "Could not locate column in row for column 'text1.a'", + lambda: row[text1.c.a] + ) diff --git a/test/sql/test_type_expressions.py b/test/sql/test_type_expressions.py index 574edfe9e..0ef3a3e16 100644 --- a/test/sql/test_type_expressions.py +++ b/test/sql/test_type_expressions.py @@ -59,13 +59,14 @@ class SelectTest(_ExprFixture, fixtures.TestBase, AssertsCompiledSQL): # the lower() function goes into the result_map, we don't really # need this but it's fine self.assert_compile( - compiled._create_result_map()['test_table_y'][1][2], + compiled._create_result_map()['test_table_y'][1][3], "lower(test_table.y)" ) # then the original column gets put in there as well. - # it's not important that it's the last value. + # as of 1.1 it's important that it is first as this is + # taken as significant by the result processor. self.assert_compile( - compiled._create_result_map()['test_table_y'][1][-1], + compiled._create_result_map()['test_table_y'][1][0], "test_table.y" ) |