Updated to pysqlite 2.4.1. Documentation additions will come later.

Lib/sqlite3/test/hooks.py

@@ -1,7 +1,7 @@
 #-*- coding: ISO-8859-1 -*-
 # pysqlite2/test/hooks.py: tests for various SQLite-specific hooks
 #
-# Copyright (C) 2006 Gerhard Häring <gh@ghaering.de>
+# Copyright (C) 2006-2007 Gerhard Häring <gh@ghaering.de>
 #
 # This file is part of pysqlite.
 #
@@ -21,7 +21,7 @@
 #    misrepresented as being the original software.
 # 3. This notice may not be removed or altered from any source distribution.
 
-import unittest
+import os, unittest
 import sqlite3 as sqlite
 
 class CollationTests(unittest.TestCase):
@@ -105,9 +105,80 @@ class CollationTests(unittest.TestCase):
             if not e.args[0].startswith("no such collation sequence"):
                 self.fail("wrong OperationalError raised")
 
+class ProgressTests(unittest.TestCase):
+    def CheckProgressHandlerUsed(self):
+        """
+        Test that the progress handler is invoked once it is set.
+        """
+        con = sqlite.connect(":memory:")
+        progress_calls = []
+        def progress():
+            progress_calls.append(None)
+            return 0
+        con.set_progress_handler(progress, 1)
+        con.execute("""
+            create table foo(a, b)
+            """)
+        self.failUnless(progress_calls)
+
+
+    def CheckOpcodeCount(self):
+        """
+        Test that the opcode argument is respected.
+        """
+        con = sqlite.connect(":memory:")
+        progress_calls = []
+        def progress():
+            progress_calls.append(None)
+            return 0
+        con.set_progress_handler(progress, 1)
+        curs = con.cursor()
+        curs.execute("""
+            create table foo (a, b)
+            """)
+        first_count = len(progress_calls)
+        progress_calls = []
+        con.set_progress_handler(progress, 2)
+        curs.execute("""
+            create table bar (a, b)
+            """)
+        second_count = len(progress_calls)
+        self.failUnless(first_count > second_count)
+
+    def CheckCancelOperation(self):
+        """
+        Test that returning a non-zero value stops the operation in progress.
+        """
+        con = sqlite.connect(":memory:")
+        progress_calls = []
+        def progress():
+            progress_calls.append(None)
+            return 1
+        con.set_progress_handler(progress, 1)
+        curs = con.cursor()
+        self.assertRaises(
+            sqlite.OperationalError,
+            curs.execute,
+            "create table bar (a, b)")
+
+    def CheckClearHandler(self):
+        """
+        Test that setting the progress handler to None clears the previously set handler.
+        """
+        con = sqlite.connect(":memory:")
+        action = 0
+        def progress():
+            action = 1
+            return 0
+        con.set_progress_handler(progress, 1)
+        con.set_progress_handler(None, 1)
+        con.execute("select 1 union select 2 union select 3").fetchall()
+        self.failUnlessEqual(action, 0, "progress handler was not cleared")
+
 def suite():
     collation_suite = unittest.makeSuite(CollationTests, "Check")
-    return unittest.TestSuite((collation_suite,))
+    progress_suite = unittest.makeSuite(ProgressTests, "Check")
+    return unittest.TestSuite((collation_suite, progress_suite))
 
 def test():
     runner = unittest.TextTestRunner()