diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2009-06-10 21:18:24 +0000 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2009-06-10 21:18:24 +0000 |
| commit | 45cec095b4904ba71425d2fe18c143982dd08f43 (patch) | |
| tree | af5e540fdcbf1cb2a3337157d69d4b40be010fa8 /test/sql/test_labels.py | |
| parent | 698a3c1ac665e7cd2ef8d5ad3ebf51b7fe6661f4 (diff) | |
| download | sqlalchemy-45cec095b4904ba71425d2fe18c143982dd08f43.tar.gz | |
- unit tests have been migrated from unittest to nose.
See README.unittests for information on how to run
the tests. [ticket:970]
Diffstat (limited to 'test/sql/test_labels.py')
| -rw-r--r-- | test/sql/test_labels.py | 195 |
1 files changed, 195 insertions, 0 deletions
diff --git a/test/sql/test_labels.py b/test/sql/test_labels.py new file mode 100644 index 000000000..b946b0ae9 --- /dev/null +++ b/test/sql/test_labels.py @@ -0,0 +1,195 @@ +from sqlalchemy.test.testing import assert_raises, assert_raises_message +from sqlalchemy import * +from sqlalchemy import exc as exceptions +from sqlalchemy.test import * +from sqlalchemy.engine import default + +IDENT_LENGTH = 29 + +class LabelTypeTest(TestBase): + def test_type(self): + m = MetaData() + t = Table('sometable', m, + Column('col1', Integer), + Column('col2', Float)) + assert isinstance(t.c.col1.label('hi').type, Integer) + assert isinstance(select([t.c.col2]).as_scalar().label('lala').type, Float) + +class LongLabelsTest(TestBase, AssertsCompiledSQL): + @classmethod + def setup_class(cls): + global metadata, table1, table2, maxlen + metadata = MetaData(testing.db) + table1 = Table("some_large_named_table", metadata, + Column("this_is_the_primarykey_column", Integer, Sequence("this_is_some_large_seq"), primary_key=True), + Column("this_is_the_data_column", String(30)) + ) + + table2 = Table("table_with_exactly_29_characs", metadata, + Column("this_is_the_primarykey_column", Integer, Sequence("some_seq"), primary_key=True), + Column("this_is_the_data_column", String(30)) + ) + + metadata.create_all() + + maxlen = testing.db.dialect.max_identifier_length + testing.db.dialect.max_identifier_length = IDENT_LENGTH + + def teardown(self): + table1.delete().execute() + + @classmethod + def teardown_class(cls): + metadata.drop_all() + testing.db.dialect.max_identifier_length = maxlen + + def test_too_long_name_disallowed(self): + m = MetaData(testing.db) + t1 = Table("this_name_is_too_long_for_what_were_doing_in_this_test", m, Column('foo', Integer)) + assert_raises(exceptions.IdentifierError, m.create_all) + assert_raises(exceptions.IdentifierError, m.drop_all) + assert_raises(exceptions.IdentifierError, t1.create) + assert_raises(exceptions.IdentifierError, t1.drop) + + def test_result(self): + table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"}) + table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"}) + table1.insert().execute(**{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"}) + table1.insert().execute(**{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"}) + + s = table1.select(use_labels=True, order_by=[table1.c.this_is_the_primarykey_column]) + r = s.execute() + result = [] + for row in r: + result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column])) + assert result == [ + (1, "data1"), + (2, "data2"), + (3, "data3"), + (4, "data4"), + ], repr(result) + + # some dialects such as oracle (and possibly ms-sql in a future version) + # generate a subquery for limits/offsets. + # ensure that the generated result map corresponds to the selected table, not + # the select query + r = s.limit(2).execute() + result = [] + for row in r: + result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column])) + assert result == [ + (1, "data1"), + (2, "data2"), + ], repr(result) + + r = s.limit(2).offset(1).execute() + result = [] + for row in r: + result.append((row[table1.c.this_is_the_primarykey_column], row[table1.c.this_is_the_data_column])) + assert result == [ + (2, "data2"), + (3, "data3"), + ], repr(result) + + def test_table_alias_names(self): + self.assert_compile( + table2.alias().select(), + "SELECT table_with_exactly_29_c_1.this_is_the_primarykey_column, table_with_exactly_29_c_1.this_is_the_data_column FROM table_with_exactly_29_characs AS table_with_exactly_29_c_1" + ) + + ta = table2.alias() + dialect = default.DefaultDialect() + dialect.max_identifier_length = IDENT_LENGTH + self.assert_compile( + select([table1, ta]).select_from(table1.join(ta, table1.c.this_is_the_data_column==ta.c.this_is_the_data_column)).\ + where(ta.c.this_is_the_data_column=='data3'), + + "SELECT some_large_named_table.this_is_the_primarykey_column, some_large_named_table.this_is_the_data_column, " + "table_with_exactly_29_c_1.this_is_the_primarykey_column, table_with_exactly_29_c_1.this_is_the_data_column FROM " + "some_large_named_table JOIN table_with_exactly_29_characs AS table_with_exactly_29_c_1 ON " + "some_large_named_table.this_is_the_data_column = table_with_exactly_29_c_1.this_is_the_data_column " + "WHERE table_with_exactly_29_c_1.this_is_the_data_column = :this_is_the_data_column_1", + dialect=dialect + ) + + table2.insert().execute( + {"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"}, + {"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"}, + {"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"}, + {"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"}, + ) + + r = table2.alias().select().execute() + assert r.fetchall() == [(x, "data%d" % x) for x in range(1, 5)] + + def test_colbinds(self): + table1.insert().execute(**{"this_is_the_primarykey_column":1, "this_is_the_data_column":"data1"}) + table1.insert().execute(**{"this_is_the_primarykey_column":2, "this_is_the_data_column":"data2"}) + table1.insert().execute(**{"this_is_the_primarykey_column":3, "this_is_the_data_column":"data3"}) + table1.insert().execute(**{"this_is_the_primarykey_column":4, "this_is_the_data_column":"data4"}) + + r = table1.select(table1.c.this_is_the_primarykey_column == 4).execute() + assert r.fetchall() == [(4, "data4")] + + r = table1.select(or_( + table1.c.this_is_the_primarykey_column == 4, + table1.c.this_is_the_primarykey_column == 2 + )).execute() + assert r.fetchall() == [(2, "data2"), (4, "data4")] + + def test_insert_no_pk(self): + table1.insert().execute(**{"this_is_the_data_column":"data1"}) + table1.insert().execute(**{"this_is_the_data_column":"data2"}) + table1.insert().execute(**{"this_is_the_data_column":"data3"}) + table1.insert().execute(**{"this_is_the_data_column":"data4"}) + + @testing.requires.subqueries + def test_subquery(self): + q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo') + x = select([q]) + print x.execute().fetchall() + + @testing.requires.subqueries + def test_anon_alias(self): + compile_dialect = default.DefaultDialect() + compile_dialect.max_identifier_length = IDENT_LENGTH + + q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias() + x = select([q], use_labels=True) + + self.assert_compile(x, "SELECT anon_1.this_is_the_primarykey_column AS anon_1_this_is_the_prim_1, anon_1.this_is_the_data_column AS anon_1_this_is_the_data_2 " + "FROM (SELECT some_large_named_table.this_is_the_primarykey_column AS this_is_the_primarykey_column, some_large_named_table.this_is_the_data_column AS this_is_the_data_column " + "FROM some_large_named_table " + "WHERE some_large_named_table.this_is_the_primarykey_column = :this_is_the_primarykey__1) AS anon_1", dialect=compile_dialect) + + print x.execute().fetchall() + + def test_adjustable(self): + + q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias('foo') + x = select([q]) + + compile_dialect = default.DefaultDialect(label_length=10) + self.assert_compile(x, "SELECT foo.this_is_the_primarykey_column, foo.this_is_the_data_column FROM " + "(SELECT some_large_named_table.this_is_the_primarykey_column AS this_1, some_large_named_table.this_is_the_data_column " + "AS this_2 FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :this_1) AS foo", dialect=compile_dialect) + + compile_dialect = default.DefaultDialect(label_length=4) + self.assert_compile(x, "SELECT foo.this_is_the_primarykey_column, foo.this_is_the_data_column FROM " + "(SELECT some_large_named_table.this_is_the_primarykey_column AS _1, some_large_named_table.this_is_the_data_column AS _2 " + "FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :_1) AS foo", dialect=compile_dialect) + + q = table1.select(table1.c.this_is_the_primarykey_column == 4).alias() + x = select([q], use_labels=True) + + compile_dialect = default.DefaultDialect(label_length=10) + self.assert_compile(x, "SELECT anon_1.this_is_the_primarykey_column AS anon_1, anon_1.this_is_the_data_column AS anon_2 FROM " + "(SELECT some_large_named_table.this_is_the_primarykey_column AS this_3, some_large_named_table.this_is_the_data_column AS this_4 " + "FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :this_1) AS anon_1", dialect=compile_dialect) + + compile_dialect = default.DefaultDialect(label_length=4) + self.assert_compile(x, "SELECT _1.this_is_the_primarykey_column AS _1, _1.this_is_the_data_column AS _2 FROM " + "(SELECT some_large_named_table.this_is_the_primarykey_column AS _3, some_large_named_table.this_is_the_data_column AS _4 " + "FROM some_large_named_table WHERE some_large_named_table.this_is_the_primarykey_column = :_1) AS _1", dialect=compile_dialect) + + |
