From: Mike Bayer Date: Thu, 4 Oct 2007 03:19:38 +0000 (+0000) Subject: - move PG RETURNING tests to postgres dialect test X-Git-Tag: rel_0_4_0~76 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=f67fcd182277e6347ad0b97e265ccb24a4e5bd49;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git - move PG RETURNING tests to postgres dialect test - added server_version_info() support for PG dialect - exclude PG versions < 8.4 for RETURNING tests --- diff --git a/lib/sqlalchemy/databases/postgres.py b/lib/sqlalchemy/databases/postgres.py index 990140c891..74b9e6f437 100644 --- a/lib/sqlalchemy/databases/postgres.py +++ b/lib/sqlalchemy/databases/postgres.py @@ -382,6 +382,13 @@ class PGDialect(default.DefaultDialect): """ % locals() return [row[0].decode(self.encoding) for row in connection.execute(s)] + def server_version_info(self, connection): + v = connection.execute("select version()").scalar() + m = re.match('PostgreSQL (\d+)\.(\d+)\.(\d+)', v) + if not m: + raise exceptions.AssertionError("Could not determine version from string '%s'" % v) + return tuple([int(x) for x in m.group(1, 2, 3)]) + def reflecttable(self, connection, table, include_columns): preparer = self.identifier_preparer if table.schema is not None: diff --git a/test/dialect/postgres.py b/test/dialect/postgres.py index 0535da6f86..24fa917058 100644 --- a/test/dialect/postgres.py +++ b/test/dialect/postgres.py @@ -5,6 +5,7 @@ from sqlalchemy import exceptions from sqlalchemy.databases import postgres from sqlalchemy.engine.strategies import MockEngineStrategy from testlib import * +from sqlalchemy.sql import table, column class SequenceTest(SQLCompileTest): def test_basic(self): @@ -18,6 +19,92 @@ class SequenceTest(SQLCompileTest): seq = Sequence("My_Seq", schema="Some_Schema") assert dialect.identifier_preparer.format_sequence(seq) == '"Some_Schema"."My_Seq"' +class CompileTest(SQLCompileTest): + def test_update_returning(self): + dialect = postgres.dialect() + 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', + 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') + @testing.exclude('postgres', '<', (8, 4)) + def test_update_returning(self): + meta = MetaData(testbase.db) + table = Table('tables', meta, + Column('id', Integer, primary_key=True), + Column('persons', Integer), + Column('full', Boolean) + ) + 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, 4)) + def test_insert_returning(self): + meta = MetaData(testbase.db) + table = Table('tables', meta, + Column('id', Integer, primary_key=True), + Column('persons', Integer), + Column('full', Boolean) + ) + 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}]) + finally: + table.drop() + + class InsertTest(AssertMixin): @testing.supported('postgres') def setUpAll(self): diff --git a/test/sql/query.py b/test/sql/query.py index b4afbbade0..3fbf33e60a 100644 --- a/test/sql/query.py +++ b/test/sql/query.py @@ -564,50 +564,6 @@ class QueryTest(PersistTest): r = s.execute().fetchall() assert len(r) == 1 - @testing.supported('postgres') - def test_update_returning(self): - meta = MetaData(testbase.db) - table = Table('tables', meta, - Column('id', Integer, primary_key=True), - Column('persons', Integer), - Column('full', Boolean) - ) - 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') - def test_insert_returning(self): - meta = MetaData(testbase.db) - table = Table('tables', meta, - Column('id', Integer, primary_key=True), - Column('persons', Integer), - Column('full', Boolean) - ) - 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}]) - finally: - table.drop() class CompoundTest(PersistTest): """test compound statements like UNION, INTERSECT, particularly their ability to nest on diff --git a/test/sql/select.py b/test/sql/select.py index 4cdac97d87..8096b0c971 100644 --- a/test/sql/select.py +++ b/test/sql/select.py @@ -1185,34 +1185,6 @@ class CRUDTest(SQLCompileTest): u = table1.update(table1.c.name==s) self.assert_compile(u, "UPDATE mytable SET myid=:myid, name=:name, description=:description WHERE mytable.name = (SELECT myothertable.othername FROM myothertable WHERE myothertable.otherid = mytable.myid)") - @testing.supported('postgres') - def testupdatereturning(self): - dialect = postgres.dialect() - - 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) - - @testing.supported('postgres') - def testinsertreturning(self): - dialect = postgres.dialect() - - 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) - def testdelete(self): self.assert_compile(delete(table1, table1.c.myid == 7), "DELETE FROM mytable WHERE mytable.myid = :mytable_myid")