diff options
Diffstat (limited to 'Lib/sqlite3')
| -rw-r--r-- | Lib/sqlite3/__init__.py | 5 | ||||
| -rw-r--r-- | Lib/sqlite3/dbapi2.py | 19 | ||||
| -rw-r--r-- | Lib/sqlite3/dump.py | 6 | ||||
| -rw-r--r-- | Lib/sqlite3/test/backup.py | 162 | ||||
| -rw-r--r-- | Lib/sqlite3/test/dbapi.py | 442 | ||||
| -rw-r--r-- | Lib/sqlite3/test/dump.py | 4 | ||||
| -rw-r--r-- | Lib/sqlite3/test/factory.py | 126 | ||||
| -rw-r--r-- | Lib/sqlite3/test/hooks.py | 137 | ||||
| -rw-r--r-- | Lib/sqlite3/test/py25tests.py | 80 | ||||
| -rw-r--r-- | Lib/sqlite3/test/regression.py | 194 | ||||
| -rw-r--r-- | Lib/sqlite3/test/transactions.py | 81 | ||||
| -rw-r--r-- | Lib/sqlite3/test/types.py | 170 | ||||
| -rw-r--r-- | Lib/sqlite3/test/userfunctions.py | 163 |
13 files changed, 675 insertions, 914 deletions
diff --git a/Lib/sqlite3/__init__.py b/Lib/sqlite3/__init__.py index 6c91df2..41ef2b7 100644 --- a/Lib/sqlite3/__init__.py +++ b/Lib/sqlite3/__init__.py @@ -1,6 +1,7 @@ +#-*- coding: ISO-8859-1 -*- # pysqlite2/__init__.py: the pysqlite2 package. # -# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de> +# Copyright (C) 2005 Gerhard Häring <gh@ghaering.de> # # This file is part of pysqlite. # @@ -20,4 +21,4 @@ # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. -from sqlite3.dbapi2 import * +from dbapi2 import * diff --git a/Lib/sqlite3/dbapi2.py b/Lib/sqlite3/dbapi2.py index 991682c..0d4dcaf 100644 --- a/Lib/sqlite3/dbapi2.py +++ b/Lib/sqlite3/dbapi2.py @@ -1,6 +1,7 @@ +# -*- coding: iso-8859-1 -*- # pysqlite2/dbapi2.py: the DB-API 2.0 interface # -# Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de> +# Copyright (C) 2004-2005 Gerhard Häring <gh@ghaering.de> # # This file is part of pysqlite. # @@ -20,9 +21,9 @@ # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. +import collections import datetime import time -import collections.abc from _sqlite3 import * @@ -50,8 +51,8 @@ def TimestampFromTicks(ticks): version_info = tuple([int(x) for x in version.split(".")]) sqlite_version_info = tuple([int(x) for x in sqlite_version.split(".")]) -Binary = memoryview -collections.abc.Sequence.register(Row) +Binary = buffer +collections.Sequence.register(Row) def register_adapters_and_converters(): def adapt_date(val): @@ -61,13 +62,13 @@ def register_adapters_and_converters(): return val.isoformat(" ") def convert_date(val): - return datetime.date(*map(int, val.split(b"-"))) + return datetime.date(*map(int, val.split("-"))) def convert_timestamp(val): - datepart, timepart = val.split(b" ") - year, month, day = map(int, datepart.split(b"-")) - timepart_full = timepart.split(b".") - hours, minutes, seconds = map(int, timepart_full[0].split(b":")) + datepart, timepart = val.split(" ") + year, month, day = map(int, datepart.split("-")) + timepart_full = timepart.split(".") + hours, minutes, seconds = map(int, timepart_full[0].split(":")) if len(timepart_full) == 2: microseconds = int('{:0<6.6}'.format(timepart_full[1].decode())) else: diff --git a/Lib/sqlite3/dump.py b/Lib/sqlite3/dump.py index de9c368..e5c5ef2 100644 --- a/Lib/sqlite3/dump.py +++ b/Lib/sqlite3/dump.py @@ -43,7 +43,7 @@ def _iterdump(connection): # qtable, # sql.replace("''"))) else: - yield('{0};'.format(sql)) + yield('%s;' % sql) # Build the insert statement for each row of the current table table_name_ident = table_name.replace('"', '""') @@ -54,7 +54,7 @@ def _iterdump(connection): ",".join("""'||quote("{0}")||'""".format(col.replace('"', '""')) for col in column_names)) query_res = cu.execute(q) for row in query_res: - yield("{0};".format(row[0])) + yield("%s;" % row[0]) # Now when the type is 'index', 'trigger', or 'view' q = """ @@ -65,6 +65,6 @@ def _iterdump(connection): """ schema_res = cu.execute(q) for name, type, sql in schema_res.fetchall(): - yield('{0};'.format(sql)) + yield('%s;' % sql) yield('COMMIT;') diff --git a/Lib/sqlite3/test/backup.py b/Lib/sqlite3/test/backup.py deleted file mode 100644 index 903bacf..0000000 --- a/Lib/sqlite3/test/backup.py +++ /dev/null @@ -1,162 +0,0 @@ -import sqlite3 as sqlite -import unittest - - -@unittest.skipIf(sqlite.sqlite_version_info < (3, 6, 11), "Backup API not supported") -class BackupTests(unittest.TestCase): - def setUp(self): - cx = self.cx = sqlite.connect(":memory:") - cx.execute('CREATE TABLE foo (key INTEGER)') - cx.executemany('INSERT INTO foo (key) VALUES (?)', [(3,), (4,)]) - cx.commit() - - def tearDown(self): - self.cx.close() - - def verify_backup(self, bckcx): - result = bckcx.execute("SELECT key FROM foo ORDER BY key").fetchall() - self.assertEqual(result[0][0], 3) - self.assertEqual(result[1][0], 4) - - def test_bad_target_none(self): - with self.assertRaises(TypeError): - self.cx.backup(None) - - def test_bad_target_filename(self): - with self.assertRaises(TypeError): - self.cx.backup('some_file_name.db') - - def test_bad_target_same_connection(self): - with self.assertRaises(ValueError): - self.cx.backup(self.cx) - - def test_bad_target_closed_connection(self): - bck = sqlite.connect(':memory:') - bck.close() - with self.assertRaises(sqlite.ProgrammingError): - self.cx.backup(bck) - - def test_bad_target_in_transaction(self): - bck = sqlite.connect(':memory:') - bck.execute('CREATE TABLE bar (key INTEGER)') - bck.executemany('INSERT INTO bar (key) VALUES (?)', [(3,), (4,)]) - with self.assertRaises(sqlite.OperationalError) as cm: - self.cx.backup(bck) - if sqlite.sqlite_version_info < (3, 8, 8): - self.assertEqual(str(cm.exception), 'target is in transaction') - - def test_keyword_only_args(self): - with self.assertRaises(TypeError): - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, 1) - - def test_simple(self): - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck) - self.verify_backup(bck) - - def test_progress(self): - journal = [] - - def progress(status, remaining, total): - journal.append(status) - - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, pages=1, progress=progress) - self.verify_backup(bck) - - self.assertEqual(len(journal), 2) - self.assertEqual(journal[0], sqlite.SQLITE_OK) - self.assertEqual(journal[1], sqlite.SQLITE_DONE) - - def test_progress_all_pages_at_once_1(self): - journal = [] - - def progress(status, remaining, total): - journal.append(remaining) - - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, progress=progress) - self.verify_backup(bck) - - self.assertEqual(len(journal), 1) - self.assertEqual(journal[0], 0) - - def test_progress_all_pages_at_once_2(self): - journal = [] - - def progress(status, remaining, total): - journal.append(remaining) - - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, pages=-1, progress=progress) - self.verify_backup(bck) - - self.assertEqual(len(journal), 1) - self.assertEqual(journal[0], 0) - - def test_non_callable_progress(self): - with self.assertRaises(TypeError) as cm: - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, pages=1, progress='bar') - self.assertEqual(str(cm.exception), 'progress argument must be a callable') - - def test_modifying_progress(self): - journal = [] - - def progress(status, remaining, total): - if not journal: - self.cx.execute('INSERT INTO foo (key) VALUES (?)', (remaining+1000,)) - self.cx.commit() - journal.append(remaining) - - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, pages=1, progress=progress) - self.verify_backup(bck) - - result = bck.execute("SELECT key FROM foo" - " WHERE key >= 1000" - " ORDER BY key").fetchall() - self.assertEqual(result[0][0], 1001) - - self.assertEqual(len(journal), 3) - self.assertEqual(journal[0], 1) - self.assertEqual(journal[1], 1) - self.assertEqual(journal[2], 0) - - def test_failing_progress(self): - def progress(status, remaining, total): - raise SystemError('nearly out of space') - - with self.assertRaises(SystemError) as err: - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, progress=progress) - self.assertEqual(str(err.exception), 'nearly out of space') - - def test_database_source_name(self): - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, name='main') - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, name='temp') - with self.assertRaises(sqlite.OperationalError) as cm: - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, name='non-existing') - self.assertIn( - str(cm.exception), - ['SQL logic error', 'SQL logic error or missing database'] - ) - - self.cx.execute("ATTACH DATABASE ':memory:' AS attached_db") - self.cx.execute('CREATE TABLE attached_db.foo (key INTEGER)') - self.cx.executemany('INSERT INTO attached_db.foo (key) VALUES (?)', [(3,), (4,)]) - self.cx.commit() - with sqlite.connect(':memory:') as bck: - self.cx.backup(bck, name='attached_db') - self.verify_backup(bck) - - -def suite(): - return unittest.makeSuite(BackupTests) - -if __name__ == "__main__": - unittest.main() diff --git a/Lib/sqlite3/test/dbapi.py b/Lib/sqlite3/test/dbapi.py index be11337..615ebf5 100644 --- a/Lib/sqlite3/test/dbapi.py +++ b/Lib/sqlite3/test/dbapi.py @@ -1,4 +1,4 @@ -#-*- coding: iso-8859-1 -*- +#-*- coding: ISO-8859-1 -*- # pysqlite2/test/dbapi.py: tests for DB-API compliance # # Copyright (C) 2004-2010 Gerhard Häring <gh@ghaering.de> @@ -21,12 +21,14 @@ # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. -import threading import unittest +import sys import sqlite3 as sqlite - -from test.support import TESTFN, unlink - +from test import test_support +try: + import threading +except ImportError: + threading = None class ModuleTests(unittest.TestCase): def CheckAPILevel(self): @@ -43,12 +45,12 @@ class ModuleTests(unittest.TestCase): sqlite.paramstyle) def CheckWarning(self): - self.assertTrue(issubclass(sqlite.Warning, Exception), - "Warning is not a subclass of Exception") + self.assertTrue(issubclass(sqlite.Warning, StandardError), + "Warning is not a subclass of StandardError") def CheckError(self): - self.assertTrue(issubclass(sqlite.Error, Exception), - "Error is not a subclass of Exception") + self.assertTrue(issubclass(sqlite.Error, StandardError), + "Error is not a subclass of StandardError") def CheckInterfaceError(self): self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error), @@ -84,7 +86,6 @@ class ModuleTests(unittest.TestCase): "NotSupportedError is not a subclass of DatabaseError") class ConnectionTests(unittest.TestCase): - def setUp(self): self.cx = sqlite.connect(":memory:") cu = self.cx.cursor() @@ -119,8 +120,11 @@ class ConnectionTests(unittest.TestCase): def CheckFailedOpen(self): YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db" - with self.assertRaises(sqlite.OperationalError): + try: con = sqlite.connect(YOU_CANNOT_OPEN_THIS) + except sqlite.OperationalError: + return + self.fail("should have raised an OperationalError") def CheckClose(self): self.cx.close() @@ -138,68 +142,11 @@ class ConnectionTests(unittest.TestCase): self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError) self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError) - def CheckInTransaction(self): - # Can't use db from setUp because we want to test initial state. - cx = sqlite.connect(":memory:") - cu = cx.cursor() - self.assertEqual(cx.in_transaction, False) - cu.execute("create table transactiontest(id integer primary key, name text)") - self.assertEqual(cx.in_transaction, False) - cu.execute("insert into transactiontest(name) values (?)", ("foo",)) - self.assertEqual(cx.in_transaction, True) - cu.execute("select name from transactiontest where name=?", ["foo"]) - row = cu.fetchone() - self.assertEqual(cx.in_transaction, True) - cx.commit() - self.assertEqual(cx.in_transaction, False) - cu.execute("select name from transactiontest where name=?", ["foo"]) - row = cu.fetchone() - self.assertEqual(cx.in_transaction, False) - - def CheckInTransactionRO(self): - with self.assertRaises(AttributeError): - self.cx.in_transaction = True - - def CheckOpenWithPathLikeObject(self): - """ Checks that we can successfully connect to a database using an object that - is PathLike, i.e. has __fspath__(). """ - self.addCleanup(unlink, TESTFN) - class Path: - def __fspath__(self): - return TESTFN - path = Path() - with sqlite.connect(path) as cx: - cx.execute('create table test(id integer)') - - def CheckOpenUri(self): - if sqlite.sqlite_version_info < (3, 7, 7): - with self.assertRaises(sqlite.NotSupportedError): - sqlite.connect(':memory:', uri=True) - return - self.addCleanup(unlink, TESTFN) - with sqlite.connect(TESTFN) as cx: - cx.execute('create table test(id integer)') - with sqlite.connect('file:' + TESTFN, uri=True) as cx: - cx.execute('insert into test(id) values(0)') - with sqlite.connect('file:' + TESTFN + '?mode=ro', uri=True) as cx: - with self.assertRaises(sqlite.OperationalError): - cx.execute('insert into test(id) values(1)') - - @unittest.skipIf(sqlite.sqlite_version_info >= (3, 3, 1), - 'needs sqlite versions older than 3.3.1') - def CheckSameThreadErrorOnOldVersion(self): - with self.assertRaises(sqlite.NotSupportedError) as cm: - sqlite.connect(':memory:', check_same_thread=False) - self.assertEqual(str(cm.exception), 'shared connections not available') - class CursorTests(unittest.TestCase): def setUp(self): self.cx = sqlite.connect(":memory:") self.cu = self.cx.cursor() - self.cu.execute( - "create table test(id integer primary key, name text, " - "income number, unique_test text unique)" - ) + self.cu.execute("create table test(id integer primary key, name text, income number)") self.cu.execute("insert into test(name) values (?)", ("foo",)) def tearDown(self): @@ -210,12 +157,22 @@ class CursorTests(unittest.TestCase): self.cu.execute("delete from test") def CheckExecuteIllegalSql(self): - with self.assertRaises(sqlite.OperationalError): + try: self.cu.execute("select asdf") + self.fail("should have raised an OperationalError") + except sqlite.OperationalError: + return + except: + self.fail("raised wrong exception") def CheckExecuteTooMuchSql(self): - with self.assertRaises(sqlite.Warning): + try: self.cu.execute("select 5+4; select 4+5") + self.fail("should have raised a Warning") + except sqlite.Warning: + return + except: + self.fail("raised wrong exception") def CheckExecuteTooMuchSql2(self): self.cu.execute("select 5+4; -- foo bar") @@ -230,8 +187,13 @@ class CursorTests(unittest.TestCase): """) def CheckExecuteWrongSqlArg(self): - with self.assertRaises(TypeError): + try: self.cu.execute(42) + self.fail("should have raised a ValueError") + except ValueError: + return + except: + self.fail("raised wrong exception.") def CheckExecuteArgInt(self): self.cu.execute("insert into test(id) values (?)", (42,)) @@ -249,25 +211,29 @@ class CursorTests(unittest.TestCase): row = self.cu.fetchone() self.assertEqual(row[0], "Hu\x00go") - def CheckExecuteNonIterable(self): - with self.assertRaises(ValueError) as cm: - self.cu.execute("insert into test(id) values (?)", 42) - self.assertEqual(str(cm.exception), 'parameters are of unsupported type') - def CheckExecuteWrongNoOfArgs1(self): # too many parameters - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.execute("insert into test(id) values (?)", (17, "Egon")) + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass def CheckExecuteWrongNoOfArgs2(self): # too little parameters - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.execute("insert into test(id) values (?)") + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass def CheckExecuteWrongNoOfArgs3(self): # no parameters, parameters are needed - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.execute("insert into test(id) values (?)") + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass def CheckExecuteParamList(self): self.cu.execute("insert into test(name) values ('foo')") @@ -295,6 +261,10 @@ class CursorTests(unittest.TestCase): self.assertEqual(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" @@ -306,18 +276,27 @@ class CursorTests(unittest.TestCase): def CheckExecuteDictMappingTooLittleArgs(self): self.cu.execute("insert into test(name) values ('foo')") - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"}) + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass def CheckExecuteDictMappingNoArgs(self): self.cu.execute("insert into test(name) values ('foo')") - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.execute("select name from test where name=:name") + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass def CheckExecuteDictMappingUnnamed(self): self.cu.execute("insert into test(name) values ('foo')") - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.execute("select name from test where name=?", {"name": "foo"}) + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass def CheckClose(self): self.cu.close() @@ -346,7 +325,8 @@ class CursorTests(unittest.TestCase): def CheckTotalChanges(self): self.cu.execute("insert into test(name) values ('foo')") self.cu.execute("insert into test(name) values ('foo')") - self.assertLess(2, self.cx.total_changes, msg='total changes reported wrong value') + if self.cx.total_changes < 2: + self.fail("total changes reported wrong value") # Checks for executemany: # Sequences are required by the DB-API, iterators @@ -360,7 +340,7 @@ class CursorTests(unittest.TestCase): def __init__(self): self.value = 5 - def __next__(self): + def next(self): if self.value == 10: raise StopIteration else: @@ -377,16 +357,32 @@ class CursorTests(unittest.TestCase): self.cu.executemany("insert into test(income) values (?)", mygen()) def CheckExecuteManyWrongSqlArg(self): - with self.assertRaises(TypeError): + try: self.cu.executemany(42, [(3,)]) + self.fail("should have raised a ValueError") + except ValueError: + return + except: + self.fail("raised wrong exception.") def CheckExecuteManySelect(self): - with self.assertRaises(sqlite.ProgrammingError): + try: self.cu.executemany("select ?", [(3,)]) + self.fail("should have raised a ProgrammingError") + except sqlite.ProgrammingError: + return + except: + self.fail("raised wrong exception.") def CheckExecuteManyNotIterable(self): - with self.assertRaises(TypeError): + try: self.cu.executemany("insert into test(income) values (?)", 42) + self.fail("should have raised a TypeError") + except TypeError: + return + except Exception, e: + print "raised", e.__class__ + self.fail("raised wrong exception.") def CheckFetchIter(self): # Optional DB-API extension. @@ -463,54 +459,24 @@ class CursorTests(unittest.TestCase): self.assertEqual(self.cu.connection, self.cx) def CheckWrongCursorCallable(self): - with self.assertRaises(TypeError): + try: def f(): pass cur = self.cx.cursor(f) + self.fail("should have raised a TypeError") + except TypeError: + return + self.fail("should have raised a ValueError") def CheckCursorWrongClass(self): class Foo: pass foo = Foo() - with self.assertRaises(TypeError): + try: cur = sqlite.Cursor(foo) + self.fail("should have raised a ValueError") + except TypeError: + pass - def CheckLastRowIDOnReplace(self): - """ - INSERT OR REPLACE and REPLACE INTO should produce the same behavior. - """ - sql = '{} INTO test(id, unique_test) VALUES (?, ?)' - for statement in ('INSERT OR REPLACE', 'REPLACE'): - with self.subTest(statement=statement): - self.cu.execute(sql.format(statement), (1, 'foo')) - self.assertEqual(self.cu.lastrowid, 1) - - def CheckLastRowIDOnIgnore(self): - self.cu.execute( - "insert or ignore into test(unique_test) values (?)", - ('test',)) - self.assertEqual(self.cu.lastrowid, 2) - self.cu.execute( - "insert or ignore into test(unique_test) values (?)", - ('test',)) - self.assertEqual(self.cu.lastrowid, 2) - - def CheckLastRowIDInsertOR(self): - results = [] - for statement in ('FAIL', 'ABORT', 'ROLLBACK'): - sql = 'INSERT OR {} INTO test(unique_test) VALUES (?)' - with self.subTest(statement='INSERT OR {}'.format(statement)): - self.cu.execute(sql.format(statement), (statement,)) - results.append((statement, self.cu.lastrowid)) - with self.assertRaises(sqlite.IntegrityError): - self.cu.execute(sql.format(statement), (statement,)) - results.append((statement, self.cu.lastrowid)) - expected = [ - ('FAIL', 2), ('FAIL', 2), - ('ABORT', 3), ('ABORT', 3), - ('ROLLBACK', 4), ('ROLLBACK', 4), - ] - self.assertEqual(results, expected) - - +@unittest.skipUnless(threading, 'This test requires threading.') class ThreadTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") @@ -688,7 +654,8 @@ class ConstructorTests(unittest.TestCase): ts = sqlite.TimestampFromTicks(42) def CheckBinary(self): - b = sqlite.Binary(b"\0'") + with test_support.check_py3k_warnings(): + b = sqlite.Binary(chr(0) + "'") class ExtensionTests(unittest.TestCase): def CheckScriptStringSql(self): @@ -704,24 +671,39 @@ class ExtensionTests(unittest.TestCase): res = cur.fetchone()[0] self.assertEqual(res, 5) + def CheckScriptStringUnicode(self): + con = sqlite.connect(":memory:") + cur = con.cursor() + cur.executescript(u""" + create table a(i); + insert into a(i) values (5); + select i from a; + delete from a; + insert into a(i) values (6); + """) + cur.execute("select i from a") + res = cur.fetchone()[0] + self.assertEqual(res, 6) + def CheckScriptSyntaxError(self): con = sqlite.connect(":memory:") cur = con.cursor() - with self.assertRaises(sqlite.OperationalError): + raised = False + try: cur.executescript("create table test(x); asdf; create table test2(x)") + except sqlite.OperationalError: + raised = True + self.assertEqual(raised, True, "should have raised an exception") def CheckScriptErrorNormal(self): con = sqlite.connect(":memory:") cur = con.cursor() - with self.assertRaises(sqlite.OperationalError): + raised = False + try: cur.executescript("create table test(sadfsadfdsa); select foo from hurz;") - - def CheckCursorExecutescriptAsBytes(self): - con = sqlite.connect(":memory:") - cur = con.cursor() - with self.assertRaises(ValueError) as cm: - cur.executescript(b"create table test(foo); insert into test(foo) values (5);") - self.assertEqual(str(cm.exception), 'script argument must be unicode.') + except sqlite.OperationalError: + raised = True + self.assertEqual(raised, True, "should have raised an exception") def CheckConnectionExecute(self): con = sqlite.connect(":memory:") @@ -743,37 +725,68 @@ class ExtensionTests(unittest.TestCase): self.assertEqual(result, 5, "Basic test of Connection.executescript") class ClosedConTests(unittest.TestCase): + def setUp(self): + pass + + def tearDown(self): + pass + def CheckClosedConCursor(self): con = sqlite.connect(":memory:") con.close() - with self.assertRaises(sqlite.ProgrammingError): + try: cur = con.cursor() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedConCommit(self): con = sqlite.connect(":memory:") con.close() - with self.assertRaises(sqlite.ProgrammingError): + try: con.commit() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedConRollback(self): con = sqlite.connect(":memory:") con.close() - with self.assertRaises(sqlite.ProgrammingError): + try: con.rollback() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedCurExecute(self): con = sqlite.connect(":memory:") cur = con.cursor() con.close() - with self.assertRaises(sqlite.ProgrammingError): + try: cur.execute("select 4") + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedCreateFunction(self): con = sqlite.connect(":memory:") con.close() def f(x): return 17 - with self.assertRaises(sqlite.ProgrammingError): + try: con.create_function("foo", 1, f) + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedCreateAggregate(self): con = sqlite.connect(":memory:") @@ -785,31 +798,57 @@ class ClosedConTests(unittest.TestCase): pass def finalize(self): return 17 - with self.assertRaises(sqlite.ProgrammingError): + try: con.create_aggregate("foo", 1, Agg) + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedSetAuthorizer(self): con = sqlite.connect(":memory:") con.close() def authorizer(*args): return sqlite.DENY - with self.assertRaises(sqlite.ProgrammingError): + try: con.set_authorizer(authorizer) + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedSetProgressCallback(self): con = sqlite.connect(":memory:") con.close() def progress(): pass - with self.assertRaises(sqlite.ProgrammingError): + try: con.set_progress_handler(progress, 100) + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") def CheckClosedCall(self): con = sqlite.connect(":memory:") con.close() - with self.assertRaises(sqlite.ProgrammingError): + try: con() + self.fail("Should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError") class ClosedCurTests(unittest.TestCase): + def setUp(self): + pass + + def tearDown(self): + pass + def CheckClosed(self): con = sqlite.connect(":memory:") cur = con.cursor() @@ -823,103 +862,15 @@ class ClosedCurTests(unittest.TestCase): else: params = [] - with self.assertRaises(sqlite.ProgrammingError): + try: method = getattr(cur, method_name) - method(*params) - - -class SqliteOnConflictTests(unittest.TestCase): - """ - Tests for SQLite's "insert on conflict" feature. - - See https://www.sqlite.org/lang_conflict.html for details. - """ - - def setUp(self): - self.cx = sqlite.connect(":memory:") - self.cu = self.cx.cursor() - self.cu.execute(""" - CREATE TABLE test( - id INTEGER PRIMARY KEY, name TEXT, unique_name TEXT UNIQUE - ); - """) - - def tearDown(self): - self.cu.close() - self.cx.close() - - def CheckOnConflictRollbackWithExplicitTransaction(self): - self.cx.isolation_level = None # autocommit mode - self.cu = self.cx.cursor() - # Start an explicit transaction. - self.cu.execute("BEGIN") - self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") - self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") - with self.assertRaises(sqlite.IntegrityError): - self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") - # Use connection to commit. - self.cx.commit() - self.cu.execute("SELECT name, unique_name from test") - # Transaction should have rolled back and nothing should be in table. - self.assertEqual(self.cu.fetchall(), []) - - def CheckOnConflictAbortRaisesWithExplicitTransactions(self): - # Abort cancels the current sql statement but doesn't change anything - # about the current transaction. - self.cx.isolation_level = None # autocommit mode - self.cu = self.cx.cursor() - # Start an explicit transaction. - self.cu.execute("BEGIN") - self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") - self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") - with self.assertRaises(sqlite.IntegrityError): - self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") - self.cx.commit() - self.cu.execute("SELECT name, unique_name FROM test") - # Expect the first two inserts to work, third to do nothing. - self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)]) - - def CheckOnConflictRollbackWithoutTransaction(self): - # Start of implicit transaction - self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") - self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") - with self.assertRaises(sqlite.IntegrityError): - self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')") - self.cu.execute("SELECT name, unique_name FROM test") - # Implicit transaction is rolled back on error. - self.assertEqual(self.cu.fetchall(), []) - - def CheckOnConflictAbortRaisesWithoutTransactions(self): - # Abort cancels the current sql statement but doesn't change anything - # about the current transaction. - self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')") - self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") - with self.assertRaises(sqlite.IntegrityError): - self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')") - # Make sure all other values were inserted. - self.cu.execute("SELECT name, unique_name FROM test") - self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)]) - - def CheckOnConflictFail(self): - self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')") - with self.assertRaises(sqlite.IntegrityError): - self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')") - self.assertEqual(self.cu.fetchall(), []) - - def CheckOnConflictIgnore(self): - self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')") - # Nothing should happen. - self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')") - self.cu.execute("SELECT unique_name FROM test") - self.assertEqual(self.cu.fetchall(), [('foo',)]) - - def CheckOnConflictReplace(self): - self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Data!', 'foo')") - # There shouldn't be an IntegrityError exception. - self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Very different data!', 'foo')") - self.cu.execute("SELECT name, unique_name FROM test") - self.assertEqual(self.cu.fetchall(), [('Very different data!', 'foo')]) + method(*params) + self.fail("Should have raised a ProgrammingError: method " + method_name) + except sqlite.ProgrammingError: + pass + except: + self.fail("Should have raised a ProgrammingError: " + method_name) def suite(): module_suite = unittest.makeSuite(ModuleTests, "Check") @@ -930,12 +881,7 @@ def suite(): ext_suite = unittest.makeSuite(ExtensionTests, "Check") closed_con_suite = unittest.makeSuite(ClosedConTests, "Check") closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check") - on_conflict_suite = unittest.makeSuite(SqliteOnConflictTests, "Check") - return unittest.TestSuite(( - module_suite, connection_suite, cursor_suite, thread_suite, - constructor_suite, ext_suite, closed_con_suite, closed_cur_suite, - on_conflict_suite, - )) + return unittest.TestSuite((module_suite, connection_suite, cursor_suite, thread_suite, constructor_suite, ext_suite, closed_con_suite, closed_cur_suite)) def test(): runner = unittest.TextTestRunner() diff --git a/Lib/sqlite3/test/dump.py b/Lib/sqlite3/test/dump.py index a1f45a4..b7de810 100644 --- a/Lib/sqlite3/test/dump.py +++ b/Lib/sqlite3/test/dump.py @@ -29,6 +29,8 @@ class DumpTests(unittest.TestCase): , "INSERT INTO \"t1\" VALUES(2,'foo2',30,30);" , + u"INSERT INTO \"t1\" VALUES(3,'f\xc3\xb6',40,10);" + , "CREATE TABLE t2(id integer, t2_i1 integer, " \ "t2_i2 integer, primary key (id)," \ "foreign key(t2_i1) references t1(t1_i1));" @@ -47,7 +49,7 @@ class DumpTests(unittest.TestCase): expected_sqls = ['BEGIN TRANSACTION;'] + expected_sqls + \ ['COMMIT;'] [self.assertEqual(expected_sqls[i], actual_sqls[i]) - for i in range(len(expected_sqls))] + for i in xrange(len(expected_sqls))] def CheckUnorderableRow(self): # iterdump() should be able to cope with unorderable row types (issue #15545) diff --git a/Lib/sqlite3/test/factory.py b/Lib/sqlite3/test/factory.py index 95dd24b..b8e0f64 100644 --- a/Lib/sqlite3/test/factory.py +++ b/Lib/sqlite3/test/factory.py @@ -1,4 +1,4 @@ -#-*- coding: iso-8859-1 -*- +#-*- coding: ISO-8859-1 -*- # pysqlite2/test/factory.py: tests for the various factories in pysqlite # # Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de> @@ -23,7 +23,7 @@ import unittest import sqlite3 as sqlite -from collections.abc import Sequence +from collections import Sequence class MyConnection(sqlite.Connection): def __init__(self, *args, **kwargs): @@ -98,59 +98,38 @@ class RowFactoryTests(unittest.TestCase): def CheckSqliteRowIndex(self): self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as a_1, 2 as b").fetchone() + row = self.con.execute("select 1 as a, 2 as b").fetchone() self.assertIsInstance(row, sqlite.Row) - self.assertEqual(row["a_1"], 1, "by name: wrong result for column 'a_1'") - self.assertEqual(row["b"], 2, "by name: wrong result for column 'b'") + col1, col2 = row["a"], row["b"] + self.assertEqual(col1, 1, "by name: wrong result for column 'a'") + self.assertEqual(col2, 2, "by name: wrong result for column 'a'") - self.assertEqual(row["A_1"], 1, "by name: wrong result for column 'A_1'") - self.assertEqual(row["B"], 2, "by name: wrong result for column 'B'") + col1, col2 = row["A"], row["B"] + self.assertEqual(col1, 1, "by name: wrong result for column 'A'") + self.assertEqual(col2, 2, "by name: wrong result for column 'B'") self.assertEqual(row[0], 1, "by index: wrong result for column 0") + self.assertEqual(row[0L], 1, "by index: wrong result for column 0") self.assertEqual(row[1], 2, "by index: wrong result for column 1") + self.assertEqual(row[1L], 2, "by index: wrong result for column 1") self.assertEqual(row[-1], 2, "by index: wrong result for column -1") + self.assertEqual(row[-1L], 2, "by index: wrong result for column -1") self.assertEqual(row[-2], 1, "by index: wrong result for column -2") + self.assertEqual(row[-2L], 1, "by index: wrong result for column -2") with self.assertRaises(IndexError): row['c'] with self.assertRaises(IndexError): - row['a_\x11'] - with self.assertRaises(IndexError): - row['a\x7f1'] - with self.assertRaises(IndexError): row[2] with self.assertRaises(IndexError): - row[-3] + row[2L] with self.assertRaises(IndexError): - row[2**1000] - - def CheckSqliteRowIndexUnicode(self): - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as \xff").fetchone() - self.assertEqual(row["\xff"], 1) + row[-3] with self.assertRaises(IndexError): - row['\u0178'] + row[-3L] with self.assertRaises(IndexError): - row['\xdf'] - - def CheckSqliteRowSlice(self): - # A sqlite.Row can be sliced like a list. - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1, 2, 3, 4").fetchone() - self.assertEqual(row[0:0], ()) - self.assertEqual(row[0:1], (1,)) - self.assertEqual(row[1:3], (2, 3)) - self.assertEqual(row[3:1], ()) - # Explicit bounds are optional. - self.assertEqual(row[1:], (2, 3, 4)) - self.assertEqual(row[:3], (1, 2, 3)) - # Slices can use negative indices. - self.assertEqual(row[-2:-1], (3,)) - self.assertEqual(row[-2:], (3, 4)) - # Slicing supports steps. - self.assertEqual(row[0:4:2], (1, 3)) - self.assertEqual(row[3:0:-2], (4, 2)) + row[2**1000] def CheckSqliteRowIter(self): """Checks if the row object is iterable""" @@ -197,15 +176,6 @@ class RowFactoryTests(unittest.TestCase): self.assertTrue(row_1 != row_5) self.assertTrue(row_1 != object()) - with self.assertRaises(TypeError): - row_1 > row_2 - with self.assertRaises(TypeError): - row_1 < row_2 - with self.assertRaises(TypeError): - row_1 >= row_2 - with self.assertRaises(TypeError): - row_1 <= row_2 - self.assertEqual(hash(row_1), hash(row_2)) def CheckSqliteRowAsSequence(self): @@ -235,33 +205,31 @@ class TextFactoryTests(unittest.TestCase): self.con = sqlite.connect(":memory:") def CheckUnicode(self): - austria = "Österreich" + austria = unicode("Österreich", "latin1") row = self.con.execute("select ?", (austria,)).fetchone() - self.assertEqual(type(row[0]), str, "type of row[0] must be unicode") + self.assertEqual(type(row[0]), unicode, "type of row[0] must be unicode") def CheckString(self): - self.con.text_factory = bytes - austria = "Österreich" + self.con.text_factory = str + austria = unicode("Österreich", "latin1") row = self.con.execute("select ?", (austria,)).fetchone() - self.assertEqual(type(row[0]), bytes, "type of row[0] must be bytes") + self.assertEqual(type(row[0]), str, "type of row[0] must be str") self.assertEqual(row[0], austria.encode("utf-8"), "column must equal original data in UTF-8") def CheckCustom(self): - self.con.text_factory = lambda x: str(x, "utf-8", "ignore") - austria = "Österreich" - row = self.con.execute("select ?", (austria,)).fetchone() - self.assertEqual(type(row[0]), str, "type of row[0] must be unicode") - self.assertTrue(row[0].endswith("reich"), "column must contain original data") + self.con.text_factory = lambda x: unicode(x, "utf-8", "ignore") + austria = unicode("Österreich", "latin1") + row = self.con.execute("select ?", (austria.encode("latin1"),)).fetchone() + self.assertEqual(type(row[0]), unicode, "type of row[0] must be unicode") + self.assertTrue(row[0].endswith(u"reich"), "column must contain original data") def CheckOptimizedUnicode(self): - # In py3k, str objects are always returned when text_factory - # is OptimizedUnicode self.con.text_factory = sqlite.OptimizedUnicode - austria = "Österreich" - germany = "Deutchland" + austria = unicode("Österreich", "latin1") + germany = unicode("Deutchland") a_row = self.con.execute("select ?", (austria,)).fetchone() d_row = self.con.execute("select ?", (germany,)).fetchone() - self.assertEqual(type(a_row[0]), str, "type of non-ASCII row must be str") + self.assertEqual(type(a_row[0]), unicode, "type of non-ASCII row must be unicode") self.assertEqual(type(d_row[0]), str, "type of ASCII-only row must be str") def tearDown(self): @@ -274,29 +242,33 @@ class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase): self.con.execute("insert into test (value) values (?)", ("a\x00b",)) def CheckString(self): - # text_factory defaults to str + # text_factory defaults to unicode row = self.con.execute("select value from test").fetchone() - self.assertIs(type(row[0]), str) + self.assertIs(type(row[0]), unicode) self.assertEqual(row[0], "a\x00b") - def CheckBytes(self): - self.con.text_factory = bytes + def CheckCustom(self): + # A custom factory should receive a str argument + self.con.text_factory = lambda x: x row = self.con.execute("select value from test").fetchone() - self.assertIs(type(row[0]), bytes) - self.assertEqual(row[0], b"a\x00b") + self.assertIs(type(row[0]), str) + self.assertEqual(row[0], "a\x00b") - def CheckBytearray(self): - self.con.text_factory = bytearray + def CheckOptimizedUnicodeAsString(self): + # ASCII -> str argument + self.con.text_factory = sqlite.OptimizedUnicode row = self.con.execute("select value from test").fetchone() - self.assertIs(type(row[0]), bytearray) - self.assertEqual(row[0], b"a\x00b") + self.assertIs(type(row[0]), str) + self.assertEqual(row[0], "a\x00b") - def CheckCustom(self): - # A custom factory should receive a bytes argument - self.con.text_factory = lambda x: x + def CheckOptimizedUnicodeAsUnicode(self): + # Non-ASCII -> unicode argument + self.con.text_factory = sqlite.OptimizedUnicode + self.con.execute("delete from test") + self.con.execute("insert into test (value) values (?)", (u'ä\0ö',)) row = self.con.execute("select value from test").fetchone() - self.assertIs(type(row[0]), bytes) - self.assertEqual(row[0], b"a\x00b") + self.assertIs(type(row[0]), unicode) + self.assertEqual(row[0], u"ä\x00ö") def tearDown(self): self.con.close() diff --git a/Lib/sqlite3/test/hooks.py b/Lib/sqlite3/test/hooks.py index d74e74b..a5b1632 100644 --- a/Lib/sqlite3/test/hooks.py +++ b/Lib/sqlite3/test/hooks.py @@ -1,4 +1,4 @@ -#-*- coding: iso-8859-1 -*- +#-*- coding: ISO-8859-1 -*- # pysqlite2/test/hooks.py: tests for various SQLite-specific hooks # # Copyright (C) 2006-2007 Gerhard Häring <gh@ghaering.de> @@ -21,12 +21,16 @@ # 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 -from test.support import TESTFN, unlink - class CollationTests(unittest.TestCase): + def setUp(self): + pass + + def tearDown(self): + pass + def CheckCreateCollationNotString(self): con = sqlite.connect(":memory:") with self.assertRaises(TypeError): @@ -34,14 +38,19 @@ class CollationTests(unittest.TestCase): def CheckCreateCollationNotCallable(self): con = sqlite.connect(":memory:") - with self.assertRaises(TypeError) as cm: + try: con.create_collation("X", 42) - self.assertEqual(str(cm.exception), 'parameter must be callable') + self.fail("should have raised a TypeError") + except TypeError, e: + self.assertEqual(e.args[0], "parameter must be callable") def CheckCreateCollationNotAscii(self): con = sqlite.connect(":memory:") - with self.assertRaises(sqlite.ProgrammingError): - con.create_collation("collä", lambda x, y: (x > y) - (x < y)) + try: + con.create_collation("collä", cmp) + self.fail("should have raised a ProgrammingError") + except sqlite.ProgrammingError, e: + pass def CheckCreateCollationBadUpper(self): class BadUpperStr(str): @@ -60,12 +69,12 @@ class CollationTests(unittest.TestCase): self.assertEqual(result[0][0], 'b') self.assertEqual(result[1][0], 'a') - @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 1), - 'old SQLite versions crash on this test') def CheckCollationIsUsed(self): + if sqlite.version_info < (3, 2, 1): # old SQLite versions crash on this test + return def mycoll(x, y): # reverse order - return -((x > y) - (x < y)) + return -cmp(x, y) con = sqlite.connect(":memory:") con.create_collation("mycoll", mycoll) @@ -79,13 +88,15 @@ class CollationTests(unittest.TestCase): ) order by x collate mycoll """ result = con.execute(sql).fetchall() - self.assertEqual(result, [('c',), ('b',), ('a',)], - msg='the expected order was not returned') + if result[0][0] != "c" or result[1][0] != "b" or result[2][0] != "a": + self.fail("the expected order was not returned") con.create_collation("mycoll", None) - with self.assertRaises(sqlite.OperationalError) as cm: + try: result = con.execute(sql).fetchall() - self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll') + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.assertEqual(e.args[0].lower(), "no such collation sequence: mycoll") def CheckCollationReturnsLargeInteger(self): def mycoll(x, y): @@ -112,13 +123,13 @@ class CollationTests(unittest.TestCase): Verify that the last one is actually used. """ con = sqlite.connect(":memory:") - con.create_collation("mycoll", lambda x, y: (x > y) - (x < y)) - con.create_collation("mycoll", lambda x, y: -((x > y) - (x < y))) + con.create_collation("mycoll", cmp) + con.create_collation("mycoll", lambda x, y: -cmp(x, y)) result = con.execute(""" select x from (select 'a' as x union select 'b' as x) order by x collate mycoll """).fetchall() - self.assertEqual(result[0][0], 'b') - self.assertEqual(result[1][0], 'a') + if result[0][0] != 'b' or result[1][0] != 'a': + self.fail("wrong collation function is used") def CheckDeregisterCollation(self): """ @@ -126,11 +137,14 @@ class CollationTests(unittest.TestCase): to use it. """ con = sqlite.connect(":memory:") - con.create_collation("mycoll", lambda x, y: (x > y) - (x < y)) + con.create_collation("mycoll", cmp) con.create_collation("mycoll", None) - with self.assertRaises(sqlite.OperationalError) as cm: + try: con.execute("select 'a' as x union select 'b' as x order by x collate mycoll") - self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll') + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + if not e.args[0].startswith("no such collation sequence"): + self.fail("wrong OperationalError raised") class ProgressTests(unittest.TestCase): def CheckProgressHandlerUsed(self): @@ -177,7 +191,9 @@ class ProgressTests(unittest.TestCase): 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() @@ -191,88 +207,19 @@ class ProgressTests(unittest.TestCase): Test that setting the progress handler to None clears the previously set handler. """ con = sqlite.connect(":memory:") - action = 0 + action = [] def progress(): - nonlocal action - action = 1 + action.append(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.assertEqual(action, 0, "progress handler was not cleared") - -class TraceCallbackTests(unittest.TestCase): - def CheckTraceCallbackUsed(self): - """ - Test that the trace callback is invoked once it is set. - """ - con = sqlite.connect(":memory:") - traced_statements = [] - def trace(statement): - traced_statements.append(statement) - con.set_trace_callback(trace) - con.execute("create table foo(a, b)") - self.assertTrue(traced_statements) - self.assertTrue(any("create table foo" in stmt for stmt in traced_statements)) - - def CheckClearTraceCallback(self): - """ - Test that setting the trace callback to None clears the previously set callback. - """ - con = sqlite.connect(":memory:") - traced_statements = [] - def trace(statement): - traced_statements.append(statement) - con.set_trace_callback(trace) - con.set_trace_callback(None) - con.execute("create table foo(a, b)") - self.assertFalse(traced_statements, "trace callback was not cleared") - - def CheckUnicodeContent(self): - """ - Test that the statement can contain unicode literals. - """ - unicode_value = '\xf6\xe4\xfc\xd6\xc4\xdc\xdf\u20ac' - con = sqlite.connect(":memory:") - traced_statements = [] - def trace(statement): - traced_statements.append(statement) - con.set_trace_callback(trace) - con.execute("create table foo(x)") - # Can't execute bound parameters as their values don't appear - # in traced statements before SQLite 3.6.21 - # (cf. http://www.sqlite.org/draft/releaselog/3_6_21.html) - con.execute('insert into foo(x) values ("%s")' % unicode_value) - con.commit() - self.assertTrue(any(unicode_value in stmt for stmt in traced_statements), - "Unicode data %s garbled in trace callback: %s" - % (ascii(unicode_value), ', '.join(map(ascii, traced_statements)))) - - @unittest.skipIf(sqlite.sqlite_version_info < (3, 3, 9), "sqlite3_prepare_v2 is not available") - def CheckTraceCallbackContent(self): - # set_trace_callback() shouldn't produce duplicate content (bpo-26187) - traced_statements = [] - def trace(statement): - traced_statements.append(statement) - - queries = ["create table foo(x)", - "insert into foo(x) values(1)"] - self.addCleanup(unlink, TESTFN) - con1 = sqlite.connect(TESTFN, isolation_level=None) - con2 = sqlite.connect(TESTFN) - con1.set_trace_callback(trace) - cur = con1.cursor() - cur.execute(queries[0]) - con2.execute("create table bar(x)") - cur.execute(queries[1]) - self.assertEqual(traced_statements, queries) - + self.assertEqual(len(action), 0, "progress handler was not cleared") def suite(): collation_suite = unittest.makeSuite(CollationTests, "Check") progress_suite = unittest.makeSuite(ProgressTests, "Check") - trace_suite = unittest.makeSuite(TraceCallbackTests, "Check") - return unittest.TestSuite((collation_suite, progress_suite, trace_suite)) + 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..7fd3c0e --- /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.assertEqual(count, 1) + + def CheckContextManagerRollback(self): + """Is a rollback called in the context manager?""" + global did_rollback + self.assertEqual(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.assertEqual(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 c714116..42fc7c2 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-2010 Gerhard Häring <gh@ghaering.de> +# Copyright (C) 2006-2007 Gerhard Häring <gh@ghaering.de> # # This file is part of pysqlite. # @@ -25,7 +25,6 @@ import datetime import unittest import sqlite3 as sqlite import weakref -import functools from test import support class RegressionTests(unittest.TestCase): @@ -55,10 +54,10 @@ class RegressionTests(unittest.TestCase): # reset before a rollback, but only those that are still in the # statement cache. The others are not accessible from the connection object. con = sqlite.connect(":memory:", cached_statements=5) - cursors = [con.cursor() for x in range(5)] + cursors = [con.cursor() for x in xrange(5)] cursors[0].execute("create table test(x)") for i in range(10): - cursors[0].executemany("insert into test(x) values (?)", [(x,) for x in range(10)]) + cursors[0].executemany("insert into test(x) values (?)", [(x,) for x in xrange(10)]) for i in range(5): cursors[i].execute(" " * i + "select x from test") @@ -87,8 +86,9 @@ class RegressionTests(unittest.TestCase): cur.execute("select 1 x union select " + str(i)) con.close() - @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 2), 'needs sqlite 3.2.2 or newer') 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)") @@ -118,6 +118,18 @@ class RegressionTests(unittest.TestCase): """ self.con.execute("") + def CheckUnicodeConnect(self): + """ + With pysqlite 2.4.0 you needed to use a string or an 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 @@ -133,15 +145,6 @@ class RegressionTests(unittest.TestCase): con.execute("insert into foo(bar) values (5)") con.execute(SELECT) - def CheckErrorMsgDecodeError(self): - # When porting the module to Python 3.0, the error message about - # decoding errors disappeared. This verifies they're back again. - with self.assertRaises(sqlite.OperationalError) as cm: - self.con.execute("select 'xxx' || ? || 'yyy' colname", - (bytes(bytearray([250])),)).fetchone() - msg = "Could not decode to UTF-8 column 'colname' with text 'xxx" - self.assertIn(msg, str(cm.exception)) - def CheckRegisterAdapter(self): """ See issue 3312. @@ -149,34 +152,12 @@ class RegressionTests(unittest.TestCase): self.assertRaises(TypeError, sqlite.register_adapter, {}, None) def CheckSetIsolationLevel(self): - # See issue 27881. - class CustomStr(str): - def upper(self): - return None - def __del__(self): - con.isolation_level = "" - + """ + See issue 3312. + """ con = sqlite.connect(":memory:") - con.isolation_level = None - for level in "", "DEFERRED", "IMMEDIATE", "EXCLUSIVE": - with self.subTest(level=level): - con.isolation_level = level - con.isolation_level = level.lower() - con.isolation_level = level.capitalize() - con.isolation_level = CustomStr(level) - - # setting isolation_level failure should not alter previous state - con.isolation_level = None - con.isolation_level = "DEFERRED" - pairs = [ - (1, TypeError), (b'', TypeError), ("abc", ValueError), - ("IMMEDIATE\0EXCLUSIVE", ValueError), ("\xe9", ValueError), - ] - for value, exc in pairs: - with self.subTest(level=value): - with self.assertRaises(exc): - con.isolation_level = value - self.assertEqual(con.isolation_level, "DEFERRED") + self.assertRaises(UnicodeEncodeError, setattr, con, + "isolation_level", u"\xe9") def CheckCursorConstructorCallCheck(self): """ @@ -189,19 +170,17 @@ class RegressionTests(unittest.TestCase): con = sqlite.connect(":memory:") cur = Cursor(con) - with self.assertRaises(sqlite.ProgrammingError): + try: cur.execute("select 4+5").fetchall() - with self.assertRaisesRegex(sqlite.ProgrammingError, - r'^Base Cursor\.__init__ not called\.$'): + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("should have raised ProgrammingError") + with self.assertRaisesRegexp(sqlite.ProgrammingError, + r'^Base Cursor\.__init__ not called\.$'): cur.close() - def CheckStrSubclass(self): - """ - The Python 3.0 port of the module didn't cope with values of subclasses of str. - """ - class MyStr(str): pass - self.con.execute("select ?", (MyStr("abc"),)) - def CheckConnectionConstructorCallCheck(self): """ Verifies that connection methods check whether base class __init__ was @@ -212,8 +191,13 @@ class RegressionTests(unittest.TestCase): pass con = Connection(":memory:") - with self.assertRaises(sqlite.ProgrammingError): + try: cur = con.cursor() + self.fail("should have raised ProgrammingError") + except sqlite.ProgrammingError: + pass + except: + self.fail("should have raised ProgrammingError") def CheckCursorRegistration(self): """ @@ -234,8 +218,13 @@ class RegressionTests(unittest.TestCase): cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)]) cur.execute("select x from foo") con.rollback() - with self.assertRaises(sqlite.InterfaceError): + try: cur.fetchall() + self.fail("should have raised InterfaceError") + except sqlite.InterfaceError: + pass + except: + self.fail("should have raised InterfaceError") def CheckAutoCommit(self): """ @@ -262,14 +251,7 @@ class RegressionTests(unittest.TestCase): Call a connection with a non-string SQL request: check error handling of the statement constructor. """ - self.assertRaises(TypeError, self.con, 1) - - def CheckCollation(self): - def collation_cb(a, b): - return 1 - self.assertRaises(sqlite.ProgrammingError, self.con.create_collation, - # Lone surrogate cannot be encoded to the default encoding (utf8) - "\uDC80", collation_cb) + self.assertRaises(sqlite.Warning, self.con, 1) def CheckRecursiveCursorUse(self): """ @@ -352,16 +334,15 @@ class RegressionTests(unittest.TestCase): counter = 0 for i, row in enumerate(con.execute("select c from t")): - with self.subTest(i=i, row=row): - con.execute("insert into t2(c) values (?)", (i,)) - con.commit() - if counter == 0: - self.assertEqual(row[0], 0) - elif counter == 1: - self.assertEqual(row[0], 1) - elif counter == 2: - self.assertEqual(row[0], 2) - counter += 1 + con.execute("insert into t2(c) values (?)", (i,)) + con.commit() + if counter == 0: + self.assertEqual(row[0], 0) + elif counter == 1: + self.assertEqual(row[0], 1) + elif counter == 2: + self.assertEqual(row[0], 2) + counter += 1 self.assertEqual(counter, 3, "should have returned exactly three rows") def CheckBpo31770(self): @@ -384,26 +365,73 @@ class RegressionTests(unittest.TestCase): with self.assertRaises(AttributeError): del self.con.isolation_level - def CheckBpo37347(self): - class Printer: - def log(self, *args): - return sqlite.SQLITE_OK - for method in [self.con.set_trace_callback, - functools.partial(self.con.set_progress_handler, n=1), - self.con.set_authorizer]: - printer_instance = Printer() - method(printer_instance.log) - method(printer_instance.log) - self.con.execute("select 1") # trigger seg fault - method(None) +class UnhashableFunc: + def __hash__(self): + raise TypeError('unhashable type') + + def __init__(self, return_value=None): + self.calls = 0 + self.return_value = return_value + + def __call__(self, *args, **kwargs): + self.calls += 1 + return self.return_value + + +class UnhashableCallbacksTestCase(unittest.TestCase): + """ + https://bugs.python.org/issue34052 + + Registering unhashable callbacks raises TypeError, callbacks are not + registered in SQLite after such registration attempt. + """ + def setUp(self): + self.con = sqlite.connect(':memory:') + + def tearDown(self): + self.con.close() + def test_progress_handler(self): + f = UnhashableFunc(return_value=0) + with self.assertRaisesRegexp(TypeError, 'unhashable type'): + self.con.set_progress_handler(f, 1) + self.con.execute('SELECT 1') + self.assertFalse(f.calls) + + def test_func(self): + func_name = 'func_name' + f = UnhashableFunc() + with self.assertRaisesRegexp(TypeError, 'unhashable type'): + self.con.create_function(func_name, 0, f) + msg = 'no such function: %s' % func_name + with self.assertRaisesRegexp(sqlite.OperationalError, msg): + self.con.execute('SELECT %s()' % func_name) + self.assertFalse(f.calls) + + def test_authorizer(self): + f = UnhashableFunc(return_value=sqlite.SQLITE_DENY) + with self.assertRaisesRegexp(TypeError, 'unhashable type'): + self.con.set_authorizer(f) + self.con.execute('SELECT 1') + self.assertFalse(f.calls) + + def test_aggr(self): + class UnhashableType(type): + __hash__ = None + aggr_name = 'aggr_name' + with self.assertRaisesRegexp(TypeError, 'unhashable type'): + self.con.create_aggregate(aggr_name, 0, UnhashableType('Aggr', (), {})) + msg = 'no such function: %s' % aggr_name + with self.assertRaisesRegexp(sqlite.OperationalError, msg): + self.con.execute('SELECT %s()' % aggr_name) def suite(): regression_suite = unittest.makeSuite(RegressionTests, "Check") return unittest.TestSuite(( regression_suite, + unittest.makeSuite(UnhashableCallbacksTestCase), )) def test(): diff --git a/Lib/sqlite3/test/transactions.py b/Lib/sqlite3/test/transactions.py index b8a13de..a732251 100644 --- a/Lib/sqlite3/test/transactions.py +++ b/Lib/sqlite3/test/transactions.py @@ -1,4 +1,4 @@ -#-*- coding: iso-8859-1 -*- +#-*- coding: ISO-8859-1 -*- # pysqlite2/test/transactions.py: tests transactions # # Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de> @@ -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 @@ -52,13 +53,13 @@ class TransactionTests(unittest.TestCase): except OSError: pass - def CheckDMLDoesNotAutoCommitBefore(self): + def CheckDMLdoesAutoCommitBefore(self): self.cur1.execute("create table test(i)") self.cur1.execute("insert into test(i) values (5)") self.cur1.execute("create table test2(j)") self.cur2.execute("select i from test") res = self.cur2.fetchall() - self.assertEqual(len(res), 0) + self.assertEqual(len(res), 1) def CheckInsertStartsTransaction(self): self.cur1.execute("create table test(i)") @@ -111,25 +112,39 @@ class TransactionTests(unittest.TestCase): res = self.cur2.fetchall() self.assertEqual(len(res), 1) - @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 2), - 'test hangs on sqlite versions older than 3.2.2') def CheckRaiseTimeout(self): + if sqlite.sqlite_version_info < (3, 2, 2): + # This will fail (hang) on earlier versions of sqlite. + # Determine exact version it was fixed. 3.2.1 hangs. + return self.cur1.execute("create table test(i)") self.cur1.execute("insert into test(i) values (5)") - with self.assertRaises(sqlite.OperationalError): + 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") - @unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 2), - 'test hangs on sqlite versions older than 3.2.2') def CheckLocking(self): """ This tests the improved concurrency with pysqlite 2.3.4. You needed to roll back con2 before you could commit con1. """ + if sqlite.sqlite_version_info < (3, 2, 2): + # This will fail (hang) on earlier versions of sqlite. + # Determine exact version it was fixed. 3.2.1 hangs. + return self.cur1.execute("create table test(i)") self.cur1.execute("insert into test(i) values (5)") - with self.assertRaises(sqlite.OperationalError): + 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() @@ -145,14 +160,24 @@ class TransactionTests(unittest.TestCase): cur.execute("select 1 union select 2 union select 3") con.rollback() - with self.assertRaises(sqlite.InterfaceError): + try: cur.fetchall() + self.fail("InterfaceError should have been raised") + except sqlite.InterfaceError, e: + pass + except: + self.fail("InterfaceError should have been raised") class SpecialCommandTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") self.cur = self.con.cursor() + def CheckVacuum(self): + self.cur.execute("create table test(i)") + self.cur.execute("insert into test(i) values (5)") + self.cur.execute("vacuum") + def CheckDropTable(self): self.cur.execute("create table test(i)") self.cur.execute("insert into test(i) values (5)") @@ -167,44 +192,10 @@ class SpecialCommandTests(unittest.TestCase): self.cur.close() self.con.close() -class TransactionalDDL(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - def CheckDdlDoesNotAutostartTransaction(self): - # For backwards compatibility reasons, DDL statements should not - # implicitly start a transaction. - self.con.execute("create table test(i)") - self.con.rollback() - result = self.con.execute("select * from test").fetchall() - self.assertEqual(result, []) - - def CheckImmediateTransactionalDDL(self): - # You can achieve transactional DDL by issuing a BEGIN - # statement manually. - self.con.execute("begin immediate") - self.con.execute("create table test(i)") - self.con.rollback() - with self.assertRaises(sqlite.OperationalError): - self.con.execute("select * from test") - - def CheckTransactionalDDL(self): - # You can achieve transactional DDL by issuing a BEGIN - # statement manually. - self.con.execute("begin") - self.con.execute("create table test(i)") - self.con.rollback() - with self.assertRaises(sqlite.OperationalError): - self.con.execute("select * from test") - - def tearDown(self): - self.con.close() - def suite(): default_suite = unittest.makeSuite(TransactionTests, "Check") special_command_suite = unittest.makeSuite(SpecialCommandTests, "Check") - ddl_suite = unittest.makeSuite(TransactionalDDL, "Check") - return unittest.TestSuite((default_suite, special_command_suite, ddl_suite)) + return unittest.TestSuite((default_suite, special_command_suite)) def test(): runner = unittest.TextTestRunner() diff --git a/Lib/sqlite3/test/types.py b/Lib/sqlite3/test/types.py index 19ecd07..fdc21ae 100644 --- a/Lib/sqlite3/test/types.py +++ b/Lib/sqlite3/test/types.py @@ -1,7 +1,7 @@ -#-*- coding: iso-8859-1 -*- +#-*- 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. # @@ -24,6 +24,7 @@ import datetime import unittest import sqlite3 as sqlite +from test import test_support try: import zlib except ImportError: @@ -41,10 +42,10 @@ class SqliteTypeTests(unittest.TestCase): self.con.close() def CheckString(self): - self.cur.execute("insert into test(s) values (?)", ("Österreich",)) + self.cur.execute("insert into test(s) values (?)", (u"Österreich",)) self.cur.execute("select s from test") row = self.cur.fetchone() - self.assertEqual(row[0], "Österreich") + self.assertEqual(row[0], u"Österreich") def CheckSmallInt(self): self.cur.execute("insert into test(i) values (?)", (42,)) @@ -67,31 +68,57 @@ class SqliteTypeTests(unittest.TestCase): self.assertEqual(row[0], val) def CheckBlob(self): - sample = b"Guglhupf" - val = memoryview(sample) + with test_support.check_py3k_warnings(): + val = buffer("Guglhupf") self.cur.execute("insert into test(b) values (?)", (val,)) self.cur.execute("select b from test") row = self.cur.fetchone() - self.assertEqual(row[0], sample) + self.assertEqual(row[0], val) def CheckUnicodeExecute(self): - self.cur.execute("select 'Österreich'") + self.cur.execute(u"select 'Österreich'") row = self.cur.fetchone() - self.assertEqual(row[0], "Österreich") + self.assertEqual(row[0], u"Österreich") + + def CheckNonUtf8_Default(self): + try: + self.cur.execute("select ?", (chr(150),)) + self.fail("should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + + def CheckNonUtf8_TextFactoryString(self): + orig_text_factory = self.con.text_factory + try: + self.con.text_factory = str + self.cur.execute("select ?", (chr(150),)) + finally: + self.con.text_factory = orig_text_factory + + def CheckNonUtf8_TextFactoryOptimizedUnicode(self): + orig_text_factory = self.con.text_factory + try: + try: + self.con.text_factory = sqlite.OptimizedUnicode + self.cur.execute("select ?", (chr(150),)) + self.fail("should have raised a ProgrammingError") + except sqlite.ProgrammingError: + pass + finally: + self.con.text_factory = orig_text_factory class DeclTypesTests(unittest.TestCase): class Foo: def __init__(self, _val): - if isinstance(_val, bytes): - # sqlite3 always calls __init__ with a bytes created from a - # UTF-8 string when __conform__ was used to store the object. - _val = _val.decode('utf-8') self.val = _val - def __eq__(self, other): + def __cmp__(self, other): if not isinstance(other, DeclTypesTests.Foo): - return NotImplemented - return self.val == other.val + raise ValueError + if self.val == other.val: + return 0 + else: + return 1 def __conform__(self, protocol): if protocol is sqlite.PrepareProtocol: @@ -102,16 +129,10 @@ class DeclTypesTests(unittest.TestCase): def __str__(self): return "<%s>" % self.val - class BadConform: - def __init__(self, exc): - self.exc = exc - def __conform__(self, protocol): - raise self.exc - def setUp(self): self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES) self.cur = self.con.cursor() - self.cur.execute("create table test(i int, s str, f float, b bool, u unicode, foo foo, bin blob, n1 number, n2 number(5), bad bad)") + self.cur.execute("create table test(i int, s str, f float, b bool, u unicode, foo foo, bin blob, n1 number, n2 number(5))") # override float, make them always return the same number sqlite.converters["FLOAT"] = lambda x: 47.2 @@ -119,7 +140,6 @@ class DeclTypesTests(unittest.TestCase): # and implement two custom ones sqlite.converters["BOOL"] = lambda x: bool(int(x)) sqlite.converters["FOO"] = DeclTypesTests.Foo - sqlite.converters["BAD"] = DeclTypesTests.BadConform sqlite.converters["WRONG"] = lambda x: "WRONG" sqlite.converters["NUMBER"] = float @@ -127,8 +147,6 @@ class DeclTypesTests(unittest.TestCase): del sqlite.converters["FLOAT"] del sqlite.converters["BOOL"] del sqlite.converters["FOO"] - del sqlite.converters["BAD"] - del sqlite.converters["WRONG"] del sqlite.converters["NUMBER"] self.cur.close() self.con.close() @@ -168,17 +186,17 @@ class DeclTypesTests(unittest.TestCase): self.cur.execute("insert into test(b) values (?)", (False,)) self.cur.execute("select b from test") row = self.cur.fetchone() - self.assertIs(row[0], False) + self.assertEqual(row[0], False) self.cur.execute("delete from test") self.cur.execute("insert into test(b) values (?)", (True,)) self.cur.execute("select b from test") row = self.cur.fetchone() - self.assertIs(row[0], True) + self.assertEqual(row[0], True) def CheckUnicode(self): # default - val = "\xd6sterreich" + val = u"\xd6sterreich" self.cur.execute("insert into test(u) values (?)", (val,)) self.cur.execute("select u from test") row = self.cur.fetchone() @@ -191,39 +209,36 @@ class DeclTypesTests(unittest.TestCase): row = self.cur.fetchone() self.assertEqual(row[0], val) - def CheckErrorInConform(self): - val = DeclTypesTests.BadConform(TypeError) - with self.assertRaises(sqlite.InterfaceError): - self.cur.execute("insert into test(bad) values (?)", (val,)) - with self.assertRaises(sqlite.InterfaceError): - self.cur.execute("insert into test(bad) values (:val)", {"val": val}) - - val = DeclTypesTests.BadConform(KeyboardInterrupt) - with self.assertRaises(KeyboardInterrupt): - self.cur.execute("insert into test(bad) values (?)", (val,)) - with self.assertRaises(KeyboardInterrupt): - self.cur.execute("insert into test(bad) values (:val)", {"val": val}) - def CheckUnsupportedSeq(self): class Bar: pass val = Bar() - with self.assertRaises(sqlite.InterfaceError): + try: self.cur.execute("insert into test(f) values (?)", (val,)) + self.fail("should have raised an InterfaceError") + except sqlite.InterfaceError: + pass + except: + self.fail("should have raised an InterfaceError") def CheckUnsupportedDict(self): class Bar: pass val = Bar() - with self.assertRaises(sqlite.InterfaceError): + try: self.cur.execute("insert into test(f) values (:val)", {"val": val}) + self.fail("should have raised an InterfaceError") + except sqlite.InterfaceError: + pass + except: + self.fail("should have raised an InterfaceError") def CheckBlob(self): # default - sample = b"Guglhupf" - val = memoryview(sample) + with test_support.check_py3k_warnings(): + val = buffer("Guglhupf") self.cur.execute("insert into test(bin) values (?)", (val,)) self.cur.execute("select bin from test") row = self.cur.fetchone() - self.assertEqual(row[0], sample) + self.assertEqual(row[0], val) def CheckNumber1(self): self.cur.execute("insert into test(n1) values (5)") @@ -244,9 +259,9 @@ class ColNamesTests(unittest.TestCase): self.cur = self.con.cursor() self.cur.execute("create table test(x foo)") - sqlite.converters["FOO"] = lambda x: "[%s]" % x.decode("ascii") - sqlite.converters["BAR"] = lambda x: "<%s>" % x.decode("ascii") - sqlite.converters["EXC"] = lambda x: 5/0 + sqlite.converters["FOO"] = lambda x: "[%s]" % x + sqlite.converters["BAR"] = lambda x: "<%s>" % x + sqlite.converters["EXC"] = lambda x: 5 // 0 sqlite.converters["B1B1"] = lambda x: "MARKER" def tearDown(self): @@ -284,7 +299,7 @@ class ColNamesTests(unittest.TestCase): self.assertEqual(self.cur.description[0][0], "x") def CheckCaseInConverterName(self): - self.cur.execute("select 'other' as \"x [b1b1]\"") + self.cur.execute("""select 'other' as "x [b1b1]\"""") val = self.cur.fetchone()[0] self.assertEqual(val, "MARKER") @@ -296,45 +311,6 @@ class ColNamesTests(unittest.TestCase): self.cur.execute("select * from test where 0 = 1") self.assertEqual(self.cur.description[0][0], "x") - def CheckCursorDescriptionInsert(self): - self.cur.execute("insert into test values (1)") - self.assertIsNone(self.cur.description) - - -@unittest.skipIf(sqlite.sqlite_version_info < (3, 8, 3), "CTEs not supported") -class CommonTableExpressionTests(unittest.TestCase): - - def setUp(self): - self.con = sqlite.connect(":memory:") - self.cur = self.con.cursor() - self.cur.execute("create table test(x foo)") - - def tearDown(self): - self.cur.close() - self.con.close() - - def CheckCursorDescriptionCTESimple(self): - self.cur.execute("with one as (select 1) select * from one") - self.assertIsNotNone(self.cur.description) - self.assertEqual(self.cur.description[0][0], "1") - - def CheckCursorDescriptionCTESMultipleColumns(self): - self.cur.execute("insert into test values(1)") - self.cur.execute("insert into test values(2)") - self.cur.execute("with testCTE as (select * from test) select * from testCTE") - self.assertIsNotNone(self.cur.description) - self.assertEqual(self.cur.description[0][0], "x") - - def CheckCursorDescriptionCTE(self): - self.cur.execute("insert into test values (1)") - self.cur.execute("with bar as (select * from test) select * from test where x = 1") - self.assertIsNotNone(self.cur.description) - self.assertEqual(self.cur.description[0][0], "x") - self.cur.execute("with bar as (select * from test) select * from test where x = 2") - self.assertIsNotNone(self.cur.description) - self.assertEqual(self.cur.description[0][0], "x") - - class ObjectAdaptationTests(unittest.TestCase): def cast(obj): return float(obj) @@ -373,8 +349,9 @@ class BinaryConverterTests(unittest.TestCase): self.con.close() def CheckBinaryInputForConverter(self): - testdata = b"abcdefg" * 10 - result = self.con.execute('select ? as "x [bin]"', (memoryview(zlib.compress(testdata)),)).fetchone()[0] + testdata = "abcdefg" * 10 + with test_support.check_py3k_warnings(): + result = self.con.execute('select ? as "x [bin]"', (buffer(zlib.compress(testdata)),)).fetchone()[0] self.assertEqual(testdata, result) class DateTimeTests(unittest.TestCase): @@ -401,9 +378,11 @@ class DateTimeTests(unittest.TestCase): ts2 = self.cur.fetchone()[0] self.assertEqual(ts, ts2) - @unittest.skipIf(sqlite.sqlite_version_info < (3, 1), - 'the date functions are available on 3.1 or later') def CheckSqlTimestamp(self): + # The date functions are only available in SQLite version 3.1 or later + if sqlite.sqlite_version_info < (3, 1): + return + now = datetime.datetime.utcnow() self.cur.execute("insert into test(ts) values (current_timestamp)") self.cur.execute("select ts from test") @@ -432,8 +411,7 @@ def suite(): adaptation_suite = unittest.makeSuite(ObjectAdaptationTests, "Check") bin_suite = unittest.makeSuite(BinaryConverterTests, "Check") date_suite = unittest.makeSuite(DateTimeTests, "Check") - cte_suite = unittest.makeSuite(CommonTableExpressionTests, "Check") - return unittest.TestSuite((sqlite_type_suite, decltypes_type_suite, colnames_type_suite, adaptation_suite, bin_suite, date_suite, cte_suite)) + return unittest.TestSuite((sqlite_type_suite, decltypes_type_suite, colnames_type_suite, adaptation_suite, bin_suite, date_suite)) def test(): runner = unittest.TextTestRunner() diff --git a/Lib/sqlite3/test/userfunctions.py b/Lib/sqlite3/test/userfunctions.py index 9501f53..1d19151 100644 --- a/Lib/sqlite3/test/userfunctions.py +++ b/Lib/sqlite3/test/userfunctions.py @@ -1,4 +1,4 @@ -#-*- coding: iso-8859-1 -*- +#-*- coding: ISO-8859-1 -*- # pysqlite2/test/userfunctions.py: tests for user-defined functions and # aggregates. # @@ -23,13 +23,13 @@ # 3. This notice may not be removed or altered from any source distribution. import unittest -import unittest.mock import sqlite3 as sqlite +from test import test_support def func_returntext(): return "foo" def func_returnunicode(): - return "bar" + return u"bar" def func_returnint(): return 42 def func_returnfloat(): @@ -37,14 +37,15 @@ def func_returnfloat(): def func_returnnull(): return None def func_returnblob(): - return b"blob" + with test_support.check_py3k_warnings(): + return buffer("blob") def func_returnlonglong(): return 1<<31 def func_raiseexception(): - 5/0 + 5 // 0 def func_isstring(v): - return type(v) is str + return type(v) is unicode def func_isint(v): return type(v) is int def func_isfloat(v): @@ -52,12 +53,9 @@ def func_isfloat(v): def func_isnone(v): return type(v) is type(None) def func_isblob(v): - return isinstance(v, (bytes, memoryview)) + return type(v) is buffer def func_islonglong(v): - return isinstance(v, int) and v >= 1<<31 - -def func(*args): - return len(args) + return isinstance(v, (int, long)) and v >= 1<<31 class AggrNoStep: def __init__(self): @@ -75,7 +73,7 @@ class AggrNoFinalize: class AggrExceptionInInit: def __init__(self): - 5/0 + 5 // 0 def step(self, x): pass @@ -88,7 +86,7 @@ class AggrExceptionInStep: pass def step(self, x): - 5/0 + 5 // 0 def finalize(self): return 42 @@ -101,33 +99,19 @@ class AggrExceptionInFinalize: pass def finalize(self): - 5/0 + 5 // 0 class AggrCheckType: def __init__(self): self.val = None def step(self, whichType, val): - theType = {"str": str, "int": int, "float": float, "None": type(None), - "blob": bytes} + theType = {"str": unicode, "int": int, "float": float, "None": type(None), "blob": buffer} self.val = int(theType[whichType] is type(val)) def finalize(self): return self.val -class AggrCheckTypes: - def __init__(self): - self.val = 0 - - def step(self, whichType, *vals): - theType = {"str": str, "int": int, "float": float, "None": type(None), - "blob": bytes} - for val in vals: - self.val += int(theType[whichType] is type(val)) - - def finalize(self): - return self.val - class AggrSum: def __init__(self): self.val = 0.0 @@ -157,14 +141,16 @@ class FunctionTests(unittest.TestCase): self.con.create_function("isnone", 1, func_isnone) self.con.create_function("isblob", 1, func_isblob) self.con.create_function("islonglong", 1, func_islonglong) - self.con.create_function("spam", -1, func) def tearDown(self): self.con.close() def CheckFuncErrorOnCreate(self): - with self.assertRaises(sqlite.OperationalError): + try: self.con.create_function("bla", -100, lambda x: 2*x) + self.fail("should have raised an OperationalError") + except sqlite.OperationalError: + pass def CheckFuncRefCount(self): def getfunc(): @@ -182,15 +168,15 @@ class FunctionTests(unittest.TestCase): cur = self.con.cursor() cur.execute("select returntext()") val = cur.fetchone()[0] - self.assertEqual(type(val), str) + self.assertEqual(type(val), unicode) self.assertEqual(val, "foo") def CheckFuncReturnUnicode(self): cur = self.con.cursor() cur.execute("select returnunicode()") val = cur.fetchone()[0] - self.assertEqual(type(val), str) - self.assertEqual(val, "bar") + self.assertEqual(type(val), unicode) + self.assertEqual(val, u"bar") def CheckFuncReturnInt(self): cur = self.con.cursor() @@ -218,8 +204,9 @@ class FunctionTests(unittest.TestCase): cur = self.con.cursor() cur.execute("select returnblob()") val = cur.fetchone()[0] - self.assertEqual(type(val), bytes) - self.assertEqual(val, b"blob") + with test_support.check_py3k_warnings(): + self.assertEqual(type(val), buffer) + self.assertEqual(val, buffer("blob")) def CheckFuncReturnLongLong(self): cur = self.con.cursor() @@ -229,10 +216,12 @@ class FunctionTests(unittest.TestCase): def CheckFuncException(self): cur = self.con.cursor() - with self.assertRaises(sqlite.OperationalError) as cm: + try: cur.execute("select raiseexception()") cur.fetchone() - self.assertEqual(str(cm.exception), 'user-defined function raised exception') + self.fail("should have raised OperationalError") + except sqlite.OperationalError, e: + self.assertEqual(e.args[0], 'user-defined function raised exception') def CheckParamString(self): cur = self.con.cursor() @@ -260,7 +249,8 @@ class FunctionTests(unittest.TestCase): def CheckParamBlob(self): cur = self.con.cursor() - cur.execute("select isblob(?)", (memoryview(b"blob"),)) + with test_support.check_py3k_warnings(): + cur.execute("select isblob(?)", (buffer("blob"),)) val = cur.fetchone()[0] self.assertEqual(val, 1) @@ -270,35 +260,6 @@ class FunctionTests(unittest.TestCase): val = cur.fetchone()[0] self.assertEqual(val, 1) - def CheckAnyArguments(self): - cur = self.con.cursor() - cur.execute("select spam(?, ?)", (1, 2)) - val = cur.fetchone()[0] - self.assertEqual(val, 2) - - def CheckFuncNonDeterministic(self): - mock = unittest.mock.Mock(return_value=None) - self.con.create_function("deterministic", 0, mock, deterministic=False) - self.con.execute("select deterministic() = deterministic()") - self.assertEqual(mock.call_count, 2) - - @unittest.skipIf(sqlite.sqlite_version_info < (3, 8, 3), "deterministic parameter not supported") - def CheckFuncDeterministic(self): - mock = unittest.mock.Mock(return_value=None) - self.con.create_function("deterministic", 0, mock, deterministic=True) - self.con.execute("select deterministic() = deterministic()") - self.assertEqual(mock.call_count, 1) - - @unittest.skipIf(sqlite.sqlite_version_info >= (3, 8, 3), "SQLite < 3.8.3 needed") - def CheckFuncDeterministicNotSupported(self): - with self.assertRaises(sqlite.NotSupportedError): - self.con.create_function("deterministic", 0, int, deterministic=True) - - def CheckFuncDeterministicKeywordOnly(self): - with self.assertRaises(TypeError): - self.con.create_function("deterministic", 0, int, True) - - class AggregateTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") @@ -312,8 +273,9 @@ class AggregateTests(unittest.TestCase): b blob ) """) - cur.execute("insert into test(t, i, f, n, b) values (?, ?, ?, ?, ?)", - ("foo", 5, 3.14, None, memoryview(b"blob"),)) + with test_support.check_py3k_warnings(): + cur.execute("insert into test(t, i, f, n, b) values (?, ?, ?, ?, ?)", + ("foo", 5, 3.14, None, buffer("blob"),)) self.con.create_aggregate("nostep", 1, AggrNoStep) self.con.create_aggregate("nofinalize", 1, AggrNoFinalize) @@ -321,7 +283,6 @@ class AggregateTests(unittest.TestCase): self.con.create_aggregate("excStep", 1, AggrExceptionInStep) self.con.create_aggregate("excFinalize", 1, AggrExceptionInFinalize) self.con.create_aggregate("checkType", 2, AggrCheckType) - self.con.create_aggregate("checkTypes", -1, AggrCheckTypes) self.con.create_aggregate("mysum", 1, AggrSum) def tearDown(self): @@ -330,42 +291,55 @@ class AggregateTests(unittest.TestCase): pass def CheckAggrErrorOnCreate(self): - with self.assertRaises(sqlite.OperationalError): + try: self.con.create_function("bla", -100, AggrSum) + self.fail("should have raised an OperationalError") + except sqlite.OperationalError: + pass def CheckAggrNoStep(self): cur = self.con.cursor() - with self.assertRaises(AttributeError) as cm: + try: cur.execute("select nostep(t) from test") - self.assertEqual(str(cm.exception), "'AggrNoStep' object has no attribute 'step'") + self.fail("should have raised an AttributeError") + except AttributeError, e: + self.assertEqual(e.args[0], "AggrNoStep instance has no attribute 'step'") def CheckAggrNoFinalize(self): cur = self.con.cursor() - with self.assertRaises(sqlite.OperationalError) as cm: + try: cur.execute("select nofinalize(t) from test") val = cur.fetchone()[0] - self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error") + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error") def CheckAggrExceptionInInit(self): cur = self.con.cursor() - with self.assertRaises(sqlite.OperationalError) as cm: + try: cur.execute("select excInit(t) from test") val = cur.fetchone()[0] - self.assertEqual(str(cm.exception), "user-defined aggregate's '__init__' method raised error") + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.assertEqual(e.args[0], "user-defined aggregate's '__init__' method raised error") def CheckAggrExceptionInStep(self): cur = self.con.cursor() - with self.assertRaises(sqlite.OperationalError) as cm: + try: cur.execute("select excStep(t) from test") val = cur.fetchone()[0] - self.assertEqual(str(cm.exception), "user-defined aggregate's 'step' method raised error") + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.assertEqual(e.args[0], "user-defined aggregate's 'step' method raised error") def CheckAggrExceptionInFinalize(self): cur = self.con.cursor() - with self.assertRaises(sqlite.OperationalError) as cm: + try: cur.execute("select excFinalize(t) from test") val = cur.fetchone()[0] - self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error") + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error") def CheckAggrCheckParamStr(self): cur = self.con.cursor() @@ -379,12 +353,6 @@ class AggregateTests(unittest.TestCase): val = cur.fetchone()[0] self.assertEqual(val, 1) - def CheckAggrCheckParamsInt(self): - cur = self.con.cursor() - cur.execute("select checkTypes('int', ?, ?)", (42, 24)) - val = cur.fetchone()[0] - self.assertEqual(val, 2) - def CheckAggrCheckParamFloat(self): cur = self.con.cursor() cur.execute("select checkType('float', ?)", (3.14,)) @@ -399,7 +367,8 @@ class AggregateTests(unittest.TestCase): def CheckAggrCheckParamBlob(self): cur = self.con.cursor() - cur.execute("select checkType('blob', ?)", (memoryview(b"blob"),)) + with test_support.check_py3k_warnings(): + cur.execute("select checkType('blob', ?)", (buffer("blob"),)) val = cur.fetchone()[0] self.assertEqual(val, 1) @@ -438,14 +407,22 @@ class AuthorizerTests(unittest.TestCase): pass def test_table_access(self): - with self.assertRaises(sqlite.DatabaseError) as cm: + try: self.con.execute("select * from t2") - self.assertIn('prohibited', str(cm.exception)) + except sqlite.DatabaseError, e: + if not e.args[0].endswith("prohibited"): + self.fail("wrong exception text: %s" % e.args[0]) + return + self.fail("should have raised an exception due to missing privileges") def test_column_access(self): - with self.assertRaises(sqlite.DatabaseError) as cm: + try: self.con.execute("select c2 from t1") - self.assertIn('prohibited', str(cm.exception)) + except sqlite.DatabaseError, e: + if not e.args[0].endswith("prohibited"): + self.fail("wrong exception text: %s" % e.args[0]) + return + self.fail("should have raised an exception due to missing privileges") class AuthorizerRaiseExceptionTests(AuthorizerTests): @staticmethod |