diff options
Diffstat (limited to 'test/dialect/postgres.py')
| -rw-r--r-- | test/dialect/postgres.py | 191 |
1 files changed, 86 insertions, 105 deletions
diff --git a/test/dialect/postgres.py b/test/dialect/postgres.py index 4affabb6c..11e2c139e 100644 --- a/test/dialect/postgres.py +++ b/test/dialect/postgres.py @@ -22,46 +22,47 @@ class SequenceTest(SQLCompileTest): class CompileTest(SQLCompileTest): def test_update_returning(self): dialect = postgres.dialect() - table1 = table('mytable', + table1 = table('mytable', column('myid', Integer), column('name', String), column('description', String), ) - + u = update(table1, values=dict(name='foo'), postgres_returning=[table1.c.myid, table1.c.name]) self.assert_compile(u, "UPDATE mytable SET name=%(name)s RETURNING mytable.myid, mytable.name", dialect=dialect) - + u = update(table1, values=dict(name='foo'), postgres_returning=[table1]) self.assert_compile(u, "UPDATE mytable SET name=%(name)s "\ "RETURNING mytable.myid, mytable.name, mytable.description", dialect=dialect) - + u = update(table1, values=dict(name='foo'), postgres_returning=[func.length(table1.c.name)]) self.assert_compile(u, "UPDATE mytable SET name=%(name)s RETURNING length(mytable.name)", dialect=dialect) - + def test_insert_returning(self): dialect = postgres.dialect() - table1 = table('mytable', + table1 = table('mytable', column('myid', Integer), column('name', String), column('description', String), ) - + i = insert(table1, values=dict(name='foo'), postgres_returning=[table1.c.myid, table1.c.name]) self.assert_compile(i, "INSERT INTO mytable (name) VALUES (%(name)s) RETURNING mytable.myid, mytable.name", dialect=dialect) - + i = insert(table1, values=dict(name='foo'), postgres_returning=[table1]) self.assert_compile(i, "INSERT INTO mytable (name) VALUES (%(name)s) "\ "RETURNING mytable.myid, mytable.name, mytable.description", dialect=dialect) - + i = insert(table1, values=dict(name='foo'), postgres_returning=[func.length(table1.c.name)]) self.assert_compile(i, "INSERT INTO mytable (name) VALUES (%(name)s) RETURNING length(mytable.name)", dialect=dialect) class ReturningTest(AssertMixin): - @testing.supported('postgres') + __only_on__ = 'postgres' + @testing.exclude('postgres', '<', (8, 2)) def test_update_returning(self): meta = MetaData(testbase.db) - table = Table('tables', meta, + table = Table('tables', meta, Column('id', Integer, primary_key=True), Column('persons', Integer), Column('full', Boolean) @@ -69,20 +70,19 @@ class ReturningTest(AssertMixin): table.create() try: table.insert().execute([{'persons': 5, 'full': False}, {'persons': 3, 'full': False}]) - + result = table.update(table.c.persons > 4, dict(full=True), postgres_returning=[table.c.id]).execute() self.assertEqual(result.fetchall(), [(1,)]) - + result2 = select([table.c.id, table.c.full]).order_by(table.c.id).execute() self.assertEqual(result2.fetchall(), [(1,True),(2,False)]) finally: table.drop() - @testing.supported('postgres') @testing.exclude('postgres', '<', (8, 2)) def test_insert_returning(self): meta = MetaData(testbase.db) - table = Table('tables', meta, + table = Table('tables', meta, Column('id', Integer, primary_key=True), Column('persons', Integer), Column('full', Boolean) @@ -90,104 +90,99 @@ class ReturningTest(AssertMixin): table.create() try: result = table.insert(postgres_returning=[table.c.id]).execute({'persons': 1, 'full': False}) - + self.assertEqual(result.fetchall(), [(1,)]) - + # Multiple inserts only return the last row result2 = table.insert(postgres_returning=[table]).execute( [{'persons': 2, 'full': False}, {'persons': 3, 'full': True}]) - + self.assertEqual(result2.fetchall(), [(3,3,True)]) - + result3 = table.insert(postgres_returning=[(table.c.id*2).label('double_id')]).execute({'persons': 4, 'full': False}) self.assertEqual([dict(row) for row in result3], [{'double_id':8}]) - + result4 = testbase.db.execute('insert into tables (id, persons, "full") values (5, 10, true) returning persons') self.assertEqual([dict(row) for row in result4], [{'persons': 10}]) finally: table.drop() - - + + class InsertTest(AssertMixin): - @testing.supported('postgres') + __only_on__ = 'postgres' + def setUpAll(self): global metadata metadata = MetaData(testbase.db) - - @testing.supported('postgres') + def tearDown(self): metadata.drop_all() metadata.tables.clear() - - @testing.supported('postgres') + def test_compiled_insert(self): - table = Table('testtable', metadata, + table = Table('testtable', metadata, Column('id', Integer, primary_key=True), Column('data', String(30))) - + metadata.create_all() ins = table.insert(values={'data':bindparam('x')}).compile() ins.execute({'x':"five"}, {'x':"seven"}) assert table.select().execute().fetchall() == [(1, 'five'), (2, 'seven')] - - @testing.supported('postgres') + def test_sequence_insert(self): - table = Table('testtable', metadata, + table = Table('testtable', metadata, Column('id', Integer, Sequence('my_seq'), primary_key=True), Column('data', String(30))) metadata.create_all() self._assert_data_with_sequence(table, "my_seq") - @testing.supported('postgres') def test_opt_sequence_insert(self): - table = Table('testtable', metadata, + table = Table('testtable', metadata, Column('id', Integer, Sequence('my_seq', optional=True), primary_key=True), Column('data', String(30))) metadata.create_all() self._assert_data_autoincrement(table) - @testing.supported('postgres') def test_autoincrement_insert(self): - table = Table('testtable', metadata, + table = Table('testtable', metadata, Column('id', Integer, primary_key=True), Column('data', String(30))) metadata.create_all() self._assert_data_autoincrement(table) - @testing.supported('postgres') def test_noautoincrement_insert(self): - table = Table('testtable', metadata, + table = Table('testtable', metadata, Column('id', Integer, primary_key=True, autoincrement=False), Column('data', String(30))) metadata.create_all() self._assert_data_noautoincrement(table) - + def _assert_data_autoincrement(self, table): def go(): # execute with explicit id r = table.insert().execute({'id':30, 'data':'d1'}) assert r.last_inserted_ids() == [30] - + # execute with prefetch id r = table.insert().execute({'data':'d2'}) assert r.last_inserted_ids() == [1] - + # executemany with explicit ids table.insert().execute({'id':31, 'data':'d3'}, {'id':32, 'data':'d4'}) - + # executemany, uses SERIAL table.insert().execute({'data':'d5'}, {'data':'d6'}) - + # single execute, explicit id, inline table.insert(inline=True).execute({'id':33, 'data':'d7'}) - + # single execute, inline, uses SERIAL table.insert(inline=True).execute({'data':'d8'}) - + # note that the test framework doesnt capture the "preexecute" of a seqeuence # or default. we just see it in the bind params. - + self.assert_sql(testbase.db, go, [], with_sequences=[ ( "INSERT INTO testtable (id, data) VALUES (:id, :data)", @@ -214,7 +209,7 @@ class InsertTest(AssertMixin): [{'data':'d8'}] ), ]) - + assert table.select().execute().fetchall() == [ (30, 'd1'), (1, 'd2'), @@ -227,7 +222,7 @@ class InsertTest(AssertMixin): ] table.delete().execute() - # test the same series of events using a reflected + # test the same series of events using a reflected # version of the table m2 = MetaData(testbase.db) table = Table(table.name, m2, autoload=True) @@ -240,7 +235,7 @@ class InsertTest(AssertMixin): table.insert().execute({'data':'d5'}, {'data':'d6'}) table.insert(inline=True).execute({'id':33, 'data':'d7'}) table.insert(inline=True).execute({'data':'d8'}) - + self.assert_sql(testbase.db, go, [], with_sequences=[ ( "INSERT INTO testtable (id, data) VALUES (:id, :data)", @@ -267,7 +262,7 @@ class InsertTest(AssertMixin): [{'data':'d8'}] ), ]) - + assert table.select().execute().fetchall() == [ (30, 'd1'), (5, 'd2'), @@ -279,7 +274,7 @@ class InsertTest(AssertMixin): (8, 'd8'), ] table.delete().execute() - + def _assert_data_with_sequence(self, table, seqname): def go(): table.insert().execute({'id':30, 'data':'d1'}) @@ -326,10 +321,10 @@ class InsertTest(AssertMixin): (33, 'd7'), (4, 'd8'), ] - - # cant test reflection here since the Sequence must be + + # cant test reflection here since the Sequence must be # explicitly specified - + def _assert_data_noautoincrement(self, table): table.insert().execute({'id':30, 'data':'d1'}) try: @@ -342,10 +337,10 @@ class InsertTest(AssertMixin): assert False except exceptions.IntegrityError, e: assert "violates not-null constraint" in str(e) - + table.insert().execute({'id':31, 'data':'d2'}, {'id':32, 'data':'d3'}) table.insert(inline=True).execute({'id':33, 'data':'d4'}) - + assert table.select().execute().fetchall() == [ (30, 'd1'), (31, 'd2'), @@ -354,7 +349,7 @@ class InsertTest(AssertMixin): ] table.delete().execute() - # test the same series of events using a reflected + # test the same series of events using a reflected # version of the table m2 = MetaData(testbase.db) table = Table(table.name, m2, autoload=True) @@ -369,21 +364,22 @@ class InsertTest(AssertMixin): assert False except exceptions.IntegrityError, e: assert "violates not-null constraint" in str(e) - + table.insert().execute({'id':31, 'data':'d2'}, {'id':32, 'data':'d3'}) table.insert(inline=True).execute({'id':33, 'data':'d4'}) - + assert table.select().execute().fetchall() == [ (30, 'd1'), (31, 'd2'), (32, 'd3'), (33, 'd4'), ] - + class DomainReflectionTest(AssertMixin): "Test PostgreSQL domains" - @testing.supported('postgres') + __only_on__ = 'postgres' + def setUpAll(self): con = testbase.db.connect() try: @@ -396,7 +392,6 @@ class DomainReflectionTest(AssertMixin): con.execute('CREATE TABLE alt_schema.testtable(question integer, answer alt_schema.testdomain, anything integer)') con.execute('CREATE TABLE crosschema (question integer, answer alt_schema.testdomain)') - @testing.supported('postgres') def tearDownAll(self): con = testbase.db.connect() con.execute('DROP TABLE testtable') @@ -405,35 +400,30 @@ class DomainReflectionTest(AssertMixin): con.execute('DROP DOMAIN testdomain') con.execute('DROP DOMAIN alt_schema.testdomain') - @testing.supported('postgres') def test_table_is_reflected(self): metadata = MetaData(testbase.db) table = Table('testtable', metadata, autoload=True) self.assertEquals(set(table.columns.keys()), set(['question', 'answer']), "Columns of reflected table didn't equal expected columns") self.assertEquals(table.c.answer.type.__class__, postgres.PGInteger) - - @testing.supported('postgres') + def test_domain_is_reflected(self): metadata = MetaData(testbase.db) table = Table('testtable', metadata, autoload=True) self.assertEquals(str(table.columns.answer.default.arg), '42', "Reflected default value didn't equal expected value") self.assertFalse(table.columns.answer.nullable, "Expected reflected column to not be nullable.") - @testing.supported('postgres') def test_table_is_reflected_alt_schema(self): metadata = MetaData(testbase.db) table = Table('testtable', metadata, autoload=True, schema='alt_schema') self.assertEquals(set(table.columns.keys()), set(['question', 'answer', 'anything']), "Columns of reflected table didn't equal expected columns") self.assertEquals(table.c.anything.type.__class__, postgres.PGInteger) - @testing.supported('postgres') def test_schema_domain_is_reflected(self): metadata = MetaData(testbase.db) table = Table('testtable', metadata, autoload=True, schema='alt_schema') self.assertEquals(str(table.columns.answer.default.arg), '0', "Reflected default value didn't equal expected value") self.assertTrue(table.columns.answer.nullable, "Expected reflected column to be nullable.") - @testing.supported('postgres') def test_crosschema_domain_is_reflected(self): metadata = MetaData(testbase.db) table = Table('crosschema', metadata, autoload=True) @@ -441,10 +431,11 @@ class DomainReflectionTest(AssertMixin): self.assertTrue(table.columns.answer.nullable, "Expected reflected column to be nullable.") class MiscTest(AssertMixin): - @testing.supported('postgres') + __only_on__ = 'postgres' + def test_date_reflection(self): m1 = MetaData(testbase.db) - t1 = Table('pgdate', m1, + t1 = Table('pgdate', m1, Column('date1', DateTime(timezone=True)), Column('date2', DateTime(timezone=False)) ) @@ -457,7 +448,6 @@ class MiscTest(AssertMixin): finally: m1.drop_all() - @testing.supported('postgres') def test_pg_weirdchar_reflection(self): meta1 = MetaData(testbase.db) subject = Table("subject", meta1, @@ -477,11 +467,10 @@ class MiscTest(AssertMixin): self.assert_((subject.c['id$']==referer.c.ref).compare(subject.join(referer).onclause)) finally: meta1.drop_all() - - @testing.supported('postgres') + def test_checksfor_sequence(self): meta1 = MetaData(testbase.db) - t = Table('mytable', meta1, + t = Table('mytable', meta1, Column('col1', Integer, Sequence('fooseq'))) try: testbase.db.execute("CREATE SEQUENCE fooseq") @@ -489,7 +478,6 @@ class MiscTest(AssertMixin): finally: t.drop(checkfirst=True) - @testing.supported('postgres') def test_distinct_on(self): t = Table('mytable', MetaData(testbase.db), Column('id', Integer, primary_key=True), @@ -507,7 +495,6 @@ class MiscTest(AssertMixin): 'SELECT DISTINCT ON (mytable.id, mytable.a) mytable.id, mytable.a \n' 'FROM mytable') - @testing.supported('postgres') def test_schema_reflection(self): """note: this test requires that the 'alt_schema' schema be separate and accessible by the test user""" @@ -538,7 +525,6 @@ class MiscTest(AssertMixin): finally: meta1.drop_all() - @testing.supported('postgres') def test_schema_reflection_2(self): meta1 = MetaData(testbase.db) subject = Table("subject", meta1, @@ -558,8 +544,7 @@ class MiscTest(AssertMixin): self.assert_((subject.c.id==referer.c.ref).compare(subject.join(referer).onclause)) finally: meta1.drop_all() - - @testing.supported('postgres') + def test_schema_reflection_3(self): meta1 = MetaData(testbase.db) subject = Table("subject", meta1, @@ -581,13 +566,12 @@ class MiscTest(AssertMixin): self.assert_((subject.c.id==referer.c.ref).compare(subject.join(referer).onclause)) finally: meta1.drop_all() - - @testing.supported('postgres') + def test_preexecute_passivedefault(self): - """test that when we get a primary key column back + """test that when we get a primary key column back from reflecting a table which has a default value on it, we pre-execute that PassiveDefault upon insert.""" - + try: meta = MetaData(testbase.db) testbase.db.execute(""" @@ -608,25 +592,29 @@ class MiscTest(AssertMixin): finally: testbase.db.execute("drop table speedy_users", None) - @testing.supported('postgres') def test_create_partial_index(self): tbl = Table('testtbl', MetaData(), Column('data',Integer)) idx = Index('test_idx1', tbl.c.data, postgres_where=and_(tbl.c.data > 5, tbl.c.data < 10)) - + executed_sql = [] mock_strategy = MockEngineStrategy() mock_conn = mock_strategy.create('postgres://', executed_sql.append) - + idx.create(mock_conn) - + assert executed_sql == ['CREATE INDEX test_idx1 ON testtbl (data) WHERE testtbl.data > 5 AND testtbl.data < 10'] class TimezoneTest(AssertMixin): - """test timezone-aware datetimes. psycopg will return a datetime with a tzinfo attached to it, - if postgres returns it. python then will not let you compare a datetime with a tzinfo to a datetime - that doesnt have one. this test illustrates two ways to have datetime types with and without timezone - info. """ - @testing.supported('postgres') + """Test timezone-aware datetimes. + + psycopg will return a datetime with a tzinfo attached to it, if postgres + returns it. python then will not let you compare a datetime with a tzinfo + to a datetime that doesnt have one. this test illustrates two ways to + have datetime types with and without timezone info. + """ + + __only_on__ = 'postgres' + def setUpAll(self): global tztable, notztable, metadata metadata = MetaData(testbase.db) @@ -643,11 +631,9 @@ class TimezoneTest(AssertMixin): Column("name", String(20)), ) metadata.create_all() - @testing.supported('postgres') def tearDownAll(self): metadata.drop_all() - @testing.supported('postgres') def test_with_timezone(self): # get a date with a tzinfo somedate = testbase.db.connect().scalar(func.current_timestamp().select()) @@ -655,7 +641,6 @@ class TimezoneTest(AssertMixin): c = tztable.update(tztable.c.id==1).execute(name='newname') print tztable.select(tztable.c.id==1).execute().fetchone() - @testing.supported('postgres') def test_without_timezone(self): # get a date without a tzinfo somedate = datetime.datetime(2005, 10,20, 11, 52, 00) @@ -664,22 +649,21 @@ class TimezoneTest(AssertMixin): print notztable.select(tztable.c.id==1).execute().fetchone() class ArrayTest(AssertMixin): - @testing.supported('postgres') + __only_on__ = 'postgres' + def setUpAll(self): global metadata, arrtable metadata = MetaData(testbase.db) - + arrtable = Table('arrtable', metadata, Column('id', Integer, primary_key=True), Column('intarr', postgres.PGArray(Integer)), Column('strarr', postgres.PGArray(String), nullable=False) ) metadata.create_all() - @testing.supported('postgres') def tearDownAll(self): metadata.drop_all() - - @testing.supported('postgres') + def test_reflect_array_column(self): metadata2 = MetaData(testbase.db) tbl = Table('arrtable', metadata2, autoload=True) @@ -687,8 +671,7 @@ class ArrayTest(AssertMixin): self.assertTrue(isinstance(tbl.c.strarr.type, postgres.PGArray)) self.assertTrue(isinstance(tbl.c.intarr.type.item_type, Integer)) self.assertTrue(isinstance(tbl.c.strarr.type.item_type, String)) - - @testing.supported('postgres') + def test_insert_array(self): arrtable.insert().execute(intarr=[1,2,3], strarr=['abc', 'def']) results = arrtable.select().execute().fetchall() @@ -697,7 +680,6 @@ class ArrayTest(AssertMixin): self.assertEquals(results[0]['strarr'], ['abc','def']) arrtable.delete().execute() - @testing.supported('postgres') def test_array_where(self): arrtable.insert().execute(intarr=[1,2,3], strarr=['abc', 'def']) arrtable.insert().execute(intarr=[4,5,6], strarr='ABC') @@ -705,8 +687,7 @@ class ArrayTest(AssertMixin): self.assertEquals(len(results), 1) self.assertEquals(results[0]['intarr'], [1,2,3]) arrtable.delete().execute() - - @testing.supported('postgres') + def test_array_concat(self): arrtable.insert().execute(intarr=[1,2,3], strarr=['abc', 'def']) results = select([arrtable.c.intarr + [4,5,6]]).execute().fetchall() |