diff options
Diffstat (limited to 'test/dialect/test_postgresql.py')
| -rw-r--r-- | test/dialect/test_postgresql.py | 96 |
1 files changed, 61 insertions, 35 deletions
diff --git a/test/dialect/test_postgresql.py b/test/dialect/test_postgresql.py index 4fd5bc9c1..00e5c07ab 100644 --- a/test/dialect/test_postgresql.py +++ b/test/dialect/test_postgresql.py @@ -1,6 +1,6 @@ # coding: utf-8 -from __future__ import with_statement + from sqlalchemy.testing.assertions import eq_, assert_raises, \ assert_raises_message, is_, AssertsExecutionResults, \ @@ -555,28 +555,29 @@ class EnumTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL): @testing.fails_on('postgresql+pg8000', 'zxjdbc fails on ENUM: column "XXX" is of type ' 'XXX but expression is of type text') + @testing.provide_metadata def test_unicode_labels(self): - metadata = MetaData(testing.db) + metadata = self.metadata t1 = Table('table', metadata, Column('id', Integer, primary_key=True), Column('value', - Enum(u'réveillé', u'drôle', u'S’il', + Enum(util.u('réveillé'), util.u('drôle'), util.u('S’il'), name='onetwothreetype')) ) - metadata.create_all() - try: - t1.insert().execute(value=u'drôle') - t1.insert().execute(value=u'réveillé') - t1.insert().execute(value=u'S’il') - eq_(t1.select().order_by(t1.c.id).execute().fetchall(), - [(1, u'drôle'), (2, u'réveillé'), (3, u'S’il')] - ) - m2 = MetaData(testing.db) - t2 = Table('table', m2, autoload=True) - assert t2.c.value.type.enums == (u'réveillé', u'drôle', u'S’il') - finally: - metadata.drop_all() + t1.insert().execute(value=util.u('drôle')) + t1.insert().execute(value=util.u('réveillé')) + t1.insert().execute(value=util.u('S’il')) + eq_(t1.select().order_by(t1.c.id).execute().fetchall(), + [(1, util.u('drôle')), (2, util.u('réveillé')), + (3, util.u('S’il'))] + ) + m2 = MetaData(testing.db) + t2 = Table('table', m2, autoload=True) + eq_( + t2.c.value.type.enums, + (util.u('réveillé'), util.u('drôle'), util.u('S’il')) + ) def test_non_native_type(self): metadata = MetaData() @@ -1239,7 +1240,7 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): : try: con.execute(ddl) - except exc.DBAPIError, e: + except exc.DBAPIError as e: if not 'already exists' in str(e): raise e con.execute('CREATE TABLE testtable (question integer, answer ' @@ -1477,7 +1478,7 @@ class ReflectionTest(fixtures.TestBase): meta1.create_all() meta2 = MetaData(testing.db) subject = Table('subject', meta2, autoload=True) - eq_(subject.primary_key.columns.keys(), [u'p2', u'p1']) + eq_(subject.primary_key.columns.keys(), ['p2', 'p1']) @testing.provide_metadata def test_pg_weirdchar_reflection(self): @@ -1750,7 +1751,7 @@ class ReflectionTest(fixtures.TestBase): conn.execute("ALTER TABLE t RENAME COLUMN x to y") ind = testing.db.dialect.get_indexes(conn, "t", None) - eq_(ind, [{'unique': False, 'column_names': [u'y'], 'name': u'idx1'}]) + eq_(ind, [{'unique': False, 'column_names': ['y'], 'name': 'idx1'}]) conn.close() class CustomTypeReflectionTest(fixtures.TestBase): @@ -2175,18 +2176,18 @@ class ArrayTest(fixtures.TablesTest, AssertsExecutionResults): def test_insert_array(self): arrtable = self.tables.arrtable - arrtable.insert().execute(intarr=[1, 2, 3], strarr=[u'abc', - u'def']) + arrtable.insert().execute(intarr=[1, 2, 3], strarr=[util.u('abc'), + util.u('def')]) results = arrtable.select().execute().fetchall() eq_(len(results), 1) eq_(results[0]['intarr'], [1, 2, 3]) - eq_(results[0]['strarr'], ['abc', 'def']) + eq_(results[0]['strarr'], [util.u('abc'), util.u('def')]) def test_array_where(self): arrtable = self.tables.arrtable - arrtable.insert().execute(intarr=[1, 2, 3], strarr=[u'abc', - u'def']) - arrtable.insert().execute(intarr=[4, 5, 6], strarr=u'ABC') + arrtable.insert().execute(intarr=[1, 2, 3], strarr=[util.u('abc'), + util.u('def')]) + arrtable.insert().execute(intarr=[4, 5, 6], strarr=util.u('ABC')) results = arrtable.select().where(arrtable.c.intarr == [1, 2, 3]).execute().fetchall() eq_(len(results), 1) @@ -2195,7 +2196,7 @@ class ArrayTest(fixtures.TablesTest, AssertsExecutionResults): def test_array_concat(self): arrtable = self.tables.arrtable arrtable.insert().execute(intarr=[1, 2, 3], - strarr=[u'abc', u'def']) + strarr=[util.u('abc'), util.u('def')]) results = select([arrtable.c.intarr + [4, 5, 6]]).execute().fetchall() eq_(len(results), 1) @@ -2204,15 +2205,15 @@ class ArrayTest(fixtures.TablesTest, AssertsExecutionResults): def test_array_subtype_resultprocessor(self): arrtable = self.tables.arrtable arrtable.insert().execute(intarr=[4, 5, 6], - strarr=[[u'm\xe4\xe4'], [u'm\xf6\xf6' - ]]) - arrtable.insert().execute(intarr=[1, 2, 3], strarr=[u'm\xe4\xe4' - , u'm\xf6\xf6']) + strarr=[[util.ue('m\xe4\xe4')], [ + util.ue('m\xf6\xf6')]]) + arrtable.insert().execute(intarr=[1, 2, 3], strarr=[ + util.ue('m\xe4\xe4'), util.ue('m\xf6\xf6')]) results = \ arrtable.select(order_by=[arrtable.c.intarr]).execute().fetchall() eq_(len(results), 2) - eq_(results[0]['strarr'], [u'm\xe4\xe4', u'm\xf6\xf6']) - eq_(results[1]['strarr'], [[u'm\xe4\xe4'], [u'm\xf6\xf6']]) + eq_(results[0]['strarr'], [util.ue('m\xe4\xe4'), util.ue('m\xf6\xf6')]) + eq_(results[1]['strarr'], [[util.ue('m\xe4\xe4')], [util.ue('m\xf6\xf6')]]) def test_array_literal(self): eq_( @@ -2264,7 +2265,7 @@ class ArrayTest(fixtures.TablesTest, AssertsExecutionResults): testing.db.execute( arrtable.insert(), intarr=[4, 5, 6], - strarr=[u'abc', u'def'] + strarr=[util.u('abc'), util.u('def')] ) eq_( testing.db.scalar(select([arrtable.c.intarr[2:3]])), @@ -2900,8 +2901,8 @@ class HStoreTest(fixtures.TestBase): dialect, None) assert_raises_message( ValueError, - r'''After '\[\.\.\.\], "key1"=>"value1", ', could not parse ''' - '''residual at position 36: 'crapcrapcrap, "key3"\[\.\.\.\]''', + r'''After u?'\[\.\.\.\], "key1"=>"value1", ', could not parse ''' + '''residual at position 36: u?'crapcrapcrap, "key3"\[\.\.\.\]''', proc, '"key2"=>"value2", "key1"=>"value1", ' 'crapcrapcrap, "key3"=>"value3"' @@ -3206,3 +3207,28 @@ class HStoreRoundTripTest(fixtures.TablesTest): def test_fixed_round_trip_native(self): engine = testing.db self._test_fixed_round_trip(engine) + + def _test_unicode_round_trip(self, engine): + s = select([ + hstore( + array([u'réveillé', u'drôle', u'S’il']), + array([u'réveillé', u'drôle', u'S’il']) + ) + ]) + eq_( + engine.scalar(s), + { + u'réveillé': u'réveillé', + u'drôle': u'drôle', + u'S’il': u'S’il' + } + ) + + def test_unicode_round_trip_python(self): + engine = self._non_native_engine() + self._test_unicode_round_trip(engine) + + @testing.only_on("postgresql+psycopg2") + def test_unicode_round_trip_native(self): + engine = testing.db + self._test_unicode_round_trip(engine) |