[pypy-commit] pypy default: test and fix for _sqlite3.Cursor.close behavior

[bdkearns]

Author: Brian Kearns <bdkearns at gmail.com>
Branch: 
Changeset: r62034:38cd1a49ec1e
Date: 2013-03-04 20:12 -0500
http://bitbucket.org/pypy/pypy/changeset/38cd1a49ec1e/

Log:	test and fix for _sqlite3.Cursor.close behavior

diff --git a/lib_pypy/_sqlite3.py b/lib_pypy/_sqlite3.py
--- a/lib_pypy/_sqlite3.py
+++ b/lib_pypy/_sqlite3.py
@@ -758,12 +758,13 @@
         self.statement = None
         self.reset = False
         self.locked = False
+        self.closed = False
         self.initialized = True
 
     def _check_closed(self):
         if not self.initialized:
             raise ProgrammingError("Base Cursor.__init__ not called.")
-        if not getattr(self, 'connection', None):
+        if self.closed:
             raise ProgrammingError("Cannot operate on a closed cursor.")
         self.connection._check_thread()
         self.connection._check_closed()
@@ -925,14 +926,16 @@
         return sqlite.sqlite3_last_insert_rowid(self.connection.db)
 
     def close(self):
-        if not self.connection:
-            return
-        self._check_closed()
+        self.connection._check_thread()
+        self.connection._check_closed()
         if self.statement:
             self.statement.reset()
             self.statement = None
-        self.connection.cursors.remove(weakref.ref(self))
-        self.connection = None
+        self.closed = True
+        try:
+            self.connection.cursors.remove(weakref.ref(self))
+        except ValueError:
+            pass
 
     def setinputsizes(self, *args):
         pass
diff --git a/pypy/module/test_lib_pypy/test_sqlite3.py b/pypy/module/test_lib_pypy/test_sqlite3.py
--- a/pypy/module/test_lib_pypy/test_sqlite3.py
+++ b/pypy/module/test_lib_pypy/test_sqlite3.py
@@ -45,6 +45,13 @@
     e = pytest.raises(_sqlite3.ProgrammingError, "cur.execute('select 1')")
     assert '__init__' in e.value.message
 
+def test_cursor_after_close():
+     con = _sqlite3.connect(':memory:')
+     cur = con.execute('select 1')
+     cur.close()
+     con.close()
+     pytest.raises(_sqlite3.ProgrammingError, "cur.close()")
+
 @pytest.mark.skipif("not hasattr(sys, 'pypy_translation_info')")
 def test_connection_del(tmpdir):
     """For issue1325."""