Updated to pysqlite 2.4.1. Documentation additions will come later.
diff --git a/Lib/sqlite3/test/dbapi.py b/Lib/sqlite3/test/dbapi.py
index b08da9c..b27486d 100644
--- a/Lib/sqlite3/test/dbapi.py
+++ b/Lib/sqlite3/test/dbapi.py
@@ -1,7 +1,7 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/dbapi.py: tests for DB-API compliance
#
-# Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de>
+# Copyright (C) 2004-2007 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
@@ -22,6 +22,7 @@
# 3. This notice may not be removed or altered from any source distribution.
import unittest
+import sys
import threading
import sqlite3 as sqlite
@@ -223,12 +224,45 @@
except sqlite.ProgrammingError:
pass
+ def CheckExecuteParamList(self):
+ self.cu.execute("insert into test(name) values ('foo')")
+ self.cu.execute("select name from test where name=?", ["foo"])
+ row = self.cu.fetchone()
+ self.failUnlessEqual(row[0], "foo")
+
+ def CheckExecuteParamSequence(self):
+ class L(object):
+ def __len__(self):
+ return 1
+ def __getitem__(self, x):
+ assert x == 0
+ return "foo"
+
+ self.cu.execute("insert into test(name) values ('foo')")
+ self.cu.execute("select name from test where name=?", L())
+ row = self.cu.fetchone()
+ self.failUnlessEqual(row[0], "foo")
+
def CheckExecuteDictMapping(self):
self.cu.execute("insert into test(name) values ('foo')")
self.cu.execute("select name from test where name=:name", {"name": "foo"})
row = self.cu.fetchone()
self.failUnlessEqual(row[0], "foo")
+ def CheckExecuteDictMapping_Mapping(self):
+ # Test only works with Python 2.5 or later
+ if sys.version_info < (2, 5, 0):
+ return
+
+ class D(dict):
+ def __missing__(self, key):
+ return "foo"
+
+ self.cu.execute("insert into test(name) values ('foo')")
+ self.cu.execute("select name from test where name=:name", D())
+ row = self.cu.fetchone()
+ self.failUnlessEqual(row[0], "foo")
+
def CheckExecuteDictMappingTooLittleArgs(self):
self.cu.execute("insert into test(name) values ('foo')")
try:
@@ -378,6 +412,12 @@
res = self.cu.fetchmany(100)
self.failUnlessEqual(res, [])
+ def CheckFetchmanyKwArg(self):
+ """Checks if fetchmany works with keyword arguments"""
+ self.cu.execute("select name from test")
+ res = self.cu.fetchmany(size=100)
+ self.failUnlessEqual(len(res), 1)
+
def CheckFetchall(self):
self.cu.execute("select name from test")
res = self.cu.fetchall()
diff --git a/Lib/sqlite3/test/hooks.py b/Lib/sqlite3/test/hooks.py
index cb0a621..547dc65 100644
--- a/Lib/sqlite3/test/hooks.py
+++ b/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 @@
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()
diff --git a/Lib/sqlite3/test/py25tests.py b/Lib/sqlite3/test/py25tests.py
new file mode 100644
index 0000000..bce26b9
--- /dev/null
+++ b/Lib/sqlite3/test/py25tests.py
@@ -0,0 +1,80 @@
+#-*- coding: ISO-8859-1 -*-
+# pysqlite2/test/regression.py: pysqlite regression tests
+#
+# Copyright (C) 2007 Gerhard Häring <gh@ghaering.de>
+#
+# This file is part of pysqlite.
+#
+# This software is provided 'as-is', without any express or implied
+# warranty. In no event will the authors be held liable for any damages
+# arising from the use of this software.
+#
+# Permission is granted to anyone to use this software for any purpose,
+# including commercial applications, and to alter it and redistribute it
+# freely, subject to the following restrictions:
+#
+# 1. The origin of this software must not be misrepresented; you must not
+# claim that you wrote the original software. If you use this software
+# in a product, an acknowledgment in the product documentation would be
+# appreciated but is not required.
+# 2. Altered source versions must be plainly marked as such, and must not be
+# misrepresented as being the original software.
+# 3. This notice may not be removed or altered from any source distribution.
+
+from __future__ import with_statement
+import unittest
+import sqlite3 as sqlite
+
+did_rollback = False
+
+class MyConnection(sqlite.Connection):
+ def rollback(self):
+ global did_rollback
+ did_rollback = True
+ sqlite.Connection.rollback(self)
+
+class ContextTests(unittest.TestCase):
+ def setUp(self):
+ global did_rollback
+ self.con = sqlite.connect(":memory:", factory=MyConnection)
+ self.con.execute("create table test(c unique)")
+ did_rollback = False
+
+ def tearDown(self):
+ self.con.close()
+
+ def CheckContextManager(self):
+ """Can the connection be used as a context manager at all?"""
+ with self.con:
+ pass
+
+ def CheckContextManagerCommit(self):
+ """Is a commit called in the context manager?"""
+ with self.con:
+ self.con.execute("insert into test(c) values ('foo')")
+ self.con.rollback()
+ count = self.con.execute("select count(*) from test").fetchone()[0]
+ self.failUnlessEqual(count, 1)
+
+ def CheckContextManagerRollback(self):
+ """Is a rollback called in the context manager?"""
+ global did_rollback
+ self.failUnlessEqual(did_rollback, False)
+ try:
+ with self.con:
+ self.con.execute("insert into test(c) values (4)")
+ self.con.execute("insert into test(c) values (4)")
+ except sqlite.IntegrityError:
+ pass
+ self.failUnlessEqual(did_rollback, True)
+
+def suite():
+ ctx_suite = unittest.makeSuite(ContextTests, "Check")
+ return unittest.TestSuite((ctx_suite,))
+
+def test():
+ runner = unittest.TextTestRunner()
+ runner.run(suite())
+
+if __name__ == "__main__":
+ test()
diff --git a/Lib/sqlite3/test/regression.py b/Lib/sqlite3/test/regression.py
index addedb1..45eae90 100644
--- a/Lib/sqlite3/test/regression.py
+++ b/Lib/sqlite3/test/regression.py
@@ -1,7 +1,7 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/regression.py: pysqlite regression tests
#
-# 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,6 +21,7 @@
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
+import datetime
import unittest
import sqlite3 as sqlite
@@ -79,6 +80,79 @@
cur.fetchone()
cur.fetchone()
+ def CheckStatementFinalizationOnCloseDb(self):
+ # pysqlite versions <= 2.3.3 only finalized statements in the statement
+ # cache when closing the database. statements that were still
+ # referenced in cursors weren't closed an could provoke "
+ # "OperationalError: Unable to close due to unfinalised statements".
+ con = sqlite.connect(":memory:")
+ cursors = []
+ # default statement cache size is 100
+ for i in range(105):
+ cur = con.cursor()
+ cursors.append(cur)
+ cur.execute("select 1 x union select " + str(i))
+ con.close()
+
+ def CheckOnConflictRollback(self):
+ if sqlite.sqlite_version_info < (3, 2, 2):
+ return
+ con = sqlite.connect(":memory:")
+ con.execute("create table foo(x, unique(x) on conflict rollback)")
+ con.execute("insert into foo(x) values (1)")
+ try:
+ con.execute("insert into foo(x) values (1)")
+ except sqlite.DatabaseError:
+ pass
+ con.execute("insert into foo(x) values (2)")
+ try:
+ con.commit()
+ except sqlite.OperationalError:
+ self.fail("pysqlite knew nothing about the implicit ROLLBACK")
+
+ def CheckWorkaroundForBuggySqliteTransferBindings(self):
+ """
+ pysqlite would crash with older SQLite versions unless
+ a workaround is implemented.
+ """
+ self.con.execute("create table if not exists foo(bar)")
+ self.con.execute("create table if not exists foo(bar)")
+
+ def CheckEmptyStatement(self):
+ """
+ pysqlite used to segfault with SQLite versions 3.5.x. These return NULL
+ for "no-operation" statements
+ """
+ self.con.execute("")
+
+ def CheckUnicodeConnect(self):
+ """
+ With pysqlite 2.4.0 you needed to use a string or a APSW connection
+ object for opening database connections.
+
+ Formerly, both bytestrings and unicode strings used to work.
+
+ Let's make sure unicode strings work in the future.
+ """
+ con = sqlite.connect(u":memory:")
+ con.close()
+
+ def CheckTypeMapUsage(self):
+ """
+ pysqlite until 2.4.1 did not rebuild the row_cast_map when recompiling
+ a statement. This test exhibits the problem.
+ """
+ SELECT = "select * from foo"
+ con = sqlite.connect(":memory:",detect_types=sqlite.PARSE_DECLTYPES)
+ con.execute("create table foo(bar timestamp)")
+ con.execute("insert into foo(bar) values (?)", (datetime.datetime.now(),))
+ con.execute(SELECT)
+ con.execute("drop table foo")
+ con.execute("create table foo(bar integer)")
+ con.execute("insert into foo(bar) values (5)")
+ con.execute(SELECT)
+
+
def suite():
regression_suite = unittest.makeSuite(RegressionTests, "Check")
return unittest.TestSuite((regression_suite,))
diff --git a/Lib/sqlite3/test/transactions.py b/Lib/sqlite3/test/transactions.py
index 1f0b19a..14cae25 100644
--- a/Lib/sqlite3/test/transactions.py
+++ b/Lib/sqlite3/test/transactions.py
@@ -1,7 +1,7 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/transactions.py: tests transactions
#
-# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
+# Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de>
#
# This file is part of pysqlite.
#
@@ -21,6 +21,7 @@
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
+import sys
import os, unittest
import sqlite3 as sqlite
@@ -119,6 +120,23 @@
except:
self.fail("should have raised an OperationalError")
+ def CheckLocking(self):
+ """
+ This tests the improved concurrency with pysqlite 2.3.4. You needed
+ to roll back con2 before you could commit con1.
+ """
+ self.cur1.execute("create table test(i)")
+ self.cur1.execute("insert into test(i) values (5)")
+ try:
+ self.cur2.execute("insert into test(i) values (5)")
+ self.fail("should have raised an OperationalError")
+ except sqlite.OperationalError:
+ pass
+ except:
+ self.fail("should have raised an OperationalError")
+ # NO self.con2.rollback() HERE!!!
+ self.con1.commit()
+
class SpecialCommandTests(unittest.TestCase):
def setUp(self):
self.con = sqlite.connect(":memory:")
diff --git a/Lib/sqlite3/test/types.py b/Lib/sqlite3/test/types.py
index 3cc9aff..1970401 100644
--- a/Lib/sqlite3/test/types.py
+++ b/Lib/sqlite3/test/types.py
@@ -1,7 +1,7 @@
#-*- coding: ISO-8859-1 -*-
# pysqlite2/test/types.py: tests for type conversion and detection
#
-# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de>
+# Copyright (C) 2005-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 bz2, datetime
+import zlib, datetime
import unittest
import sqlite3 as sqlite
@@ -287,7 +287,7 @@
class BinaryConverterTests(unittest.TestCase):
def convert(s):
- return bz2.decompress(s)
+ return zlib.decompress(s)
convert = staticmethod(convert)
def setUp(self):
@@ -299,7 +299,7 @@
def CheckBinaryInputForConverter(self):
testdata = "abcdefg" * 10
- result = self.con.execute('select ? as "x [bin]"', (buffer(bz2.compress(testdata)),)).fetchone()[0]
+ result = self.con.execute('select ? as "x [bin]"', (buffer(zlib.compress(testdata)),)).fetchone()[0]
self.failUnlessEqual(testdata, result)
class DateTimeTests(unittest.TestCase):
@@ -331,7 +331,8 @@
if sqlite.sqlite_version_info < (3, 1):
return
- now = datetime.datetime.utcnow()
+ # SQLite's current_timestamp uses UTC time, while datetime.datetime.now() uses local time.
+ now = datetime.datetime.now()
self.cur.execute("insert into test(ts) values (current_timestamp)")
self.cur.execute("select ts from test")
ts = self.cur.fetchone()[0]