Changeset 175

Timestamp:

09/24/07 11:13:01

Author:

fumanchu

Message:

Better error handling for pypgsql DDL, plus a new connection test.

Files:

• trunk/geniusql/providers/pypgsql.py (modified) (2 diffs)
• trunk/geniusql/test/zoo_fixture.py (modified) (2 diffs)

trunk/geniusql/providers/pypgsql.py

@@ -28 +28 @@
-            return libpq.PQconnectdb(connstr)
-        except libpq.DatabaseError, x:
-
-
+
+
+
-            raise
-    
-    def _del_conn(self, conn):
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
@@ -60 +86 @@
-                raise
-        return dbinfo
+    
+    def execute_ddl(self, sql, conn=None):
+        """Return a native response for the given DDL statement.
+        
+        In general, DDL statements should lock out other statements
+        (especially those isolated in other transactions). Use this
+        method to perform a locked DDL statement.
+        """
+        try:
+            postgres.PgDatabase.execute_ddl(self, sql, conn)
+        except libpq.OperationalError, x:
+            if x.args:
+                msg = x.args[0]
+                if "already exists" in msg or "does not exist" in msg:
+                    raise errors.MappingError(*x.args)
+            raise
-    
-    def fetch(self, sql, conn=None):

trunk/geniusql/test/zoo_fixture.py

@@ -1312 +1312 @@
-
-
+class ConnectionTests(unittest.TestCase):
+    
+    def test_ConnClose(self):
+        Animal = schema['Animal']
+        
+        # Start a sticky transaction so we use the same conn.
+        connmgr = db.connections
+        connmgr.start()
+        # Test a normal query.
+        self.assertEqual(len(Animal.select_all(lambda x: x.Legs == 4)), 4)
+        
+        # Take the connection down on our side.
+        txkey = connmgr.id()
+        conn = connmgr.transactions[txkey]
+        connmgr._del_conn(conn)
+        
+        # Try another query.
+        try:
+            fourlegs = len(Animal.select_all(lambda x: x.Legs == 4))
+            self.assertEqual(fourlegs, 4)
+        except:
+            del connmgr.transactions[txkey]
+        else:
+            raise AssertionError("Connection did not fail.")
+    
+    def test_ConnUnreachable(self):
+        Animal = schema['Animal']
+        
+        # Simulate an unreachable host during the grabbing of a connection.
+        # This is easy to simulate by just munging the hostname for the conn.
+        def callback():
+            fourlegs = len(Animal.select_all(lambda x: x.Legs == 4))
+            self.assertEqual(fourlegs, 4)
+        
+        db.connections.shutdown()
+        db.connections._simulate_unreachable(callback)
+
+
-##class ZooSchema(dejavu.Schema):
-##    
@@ -1401 +1439 @@
-                tools.TestRunner.run(loader(ConcurrencyTests))
-            
+            if 'connection' not in skiptests:
+                tools.TestRunner.run(loader(ConnectionTests))
+            
-            # Each thread opens a new SQLite :memory: database,
-            # so the concept of "isolation" is pretty meaningless.