- Added PGUuid and PGBit types to

sqlalchemy.databases.postgres. [ticket:1327]

- Refection of unknown PG types won't crash when those
types are specified within a domain.  [ticket:1327]

- executemany() in conjunction with INSERT..RETURNING is documented as undefined by psycopg2.

1 files changed, 58 insertions, 5 deletions

diff --git a/test/dialect/postgres.py b/test/dialect/postgres.py
index 4b1ab9178..3867d1b01 100644
--- a/test/dialect/postgres.py
+++ b/test/dialect/postgres.py
@@ -95,11 +95,15 @@ class ReturningTest(TestBase, AssertsExecutionResults):
 
             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)])
-
+            @testing.fails_on('postgres', 'Known limitation of psycopg2')
+            def test_executemany():
+                # return value is documented as failing with psycopg2/executemany
+                result2 = table.insert(postgres_returning=[table]).execute(
+                     [{'persons': 2, 'full': False}, {'persons': 3, 'full': True}])
+                self.assertEqual(result2.fetchall(), [(2, 2, False), (3,3,True)])
+            
+            test_executemany()
+            
             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}])
 
@@ -432,6 +436,25 @@ class DomainReflectionTest(TestBase, AssertsExecutionResults):
         self.assertEquals(str(table.columns.answer.server_default.arg), '0', "Reflected default value didn't equal expected value")
         self.assertTrue(table.columns.answer.nullable, "Expected reflected column to be nullable.")
 
+    def test_unknown_types(self):
+        from sqlalchemy.databases import postgres
+
+        ischema_names = postgres.ischema_names
+        postgres.ischema_names = {}
+        try:
+            m2 = MetaData(testing.db)
+            self.assertRaises(exc.SAWarning, Table, "testtable", m2, autoload=True)
+
+            @testing.emits_warning('Did not recognize type')
+            def warns():
+                m3 = MetaData(testing.db)
+                t3 = Table("testtable", m3, autoload=True)
+                assert t3.c.answer.type.__class__ == sa.types.NullType
+
+        finally:
+            postgres.ischema_names = ischema_names
+
+
 class MiscTest(TestBase, AssertsExecutionResults):
     __only_on__ = 'postgres'
 
@@ -878,6 +901,36 @@ class ServerSideCursorsTest(TestBase, AssertsExecutionResults):
         finally:
             test_table.drop(checkfirst=True)
 
+class SpecialTypesTest(TestBase, ComparesTables):
+    """test DDL and reflection of PG-specific types """
+    
+    __only_on__ = 'postgres'
+    __excluded_on__ = (('postgres', '<', (8, 3, 0)),)
+    
+    def setUpAll(self):
+        global metadata, table
+        metadata = MetaData(testing.db)
+        
+        table = Table('sometable', metadata,
+            Column('id', postgres.PGUuid, primary_key=True),
+            Column('flag', postgres.PGBit),
+            Column('addr', postgres.PGInet),
+            Column('addr2', postgres.PGMacAddr),
+            Column('addr3', postgres.PGCidr)
+        )
+        
+        metadata.create_all()
+    
+    def tearDownAll(self):
+        metadata.drop_all()
+    
+    def test_reflection(self):
+        m = MetaData(testing.db)
+        t = Table('sometable', m, autoload=True)
+        
+        self.assert_tables_equal(table, t)
+        
+
 class MatchTest(TestBase, AssertsCompiledSQL):
     __only_on__ = 'postgres'
     __excluded_on__ = (('postgres', '<', (8, 3, 0)),)