- 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 990140c891413d643b0d00acba15b6a942296a69..74b9e6f43707da3ed525fec9a8ee540a6ffd23ae 100644 (file)
--- 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 0535da6f86f0b9e6cf8484435cfc724481021dca..24fa9170588279eb3f26b3c2c26dffd54e302c47 100644 (file)
--- 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 b4afbbade012d2600d90f42c70352bfa6be0744b..3fbf33e60ade8d24ed01e42aec00597f4eb484a9 100644 (file)
--- 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 4cdac97d87917cafedf62c36f185bdf06afac09f..8096b0c9717e8be9ba4fa2ad246421868daa0b6d 100644 (file)
--- 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")