diff options
| author | thegeorg <[email protected]> | 2024-03-15 23:28:02 +0300 |
|---|---|---|
| committer | thegeorg <[email protected]> | 2024-03-15 23:38:24 +0300 |
| commit | 6635e51dd2fc4c8fbc3e510cd31c9c9102b7cfda (patch) | |
| tree | d51d048d3eb5d7a24ec6e4302b589513a3b97de3 /contrib/tools/python/src/Lib/sqlite3 | |
| parent | e894914751eea93bb23102a533c8d481b80c2560 (diff) | |
Store generated list of frozen modules for contrib/tools/python
a94357bf8071b16879eeabac1f52f54e278c03ed
Diffstat (limited to 'contrib/tools/python/src/Lib/sqlite3')
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/__init__.py | 24 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/dbapi2.py | 90 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/dump.py | 70 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/__init__.py | 0 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/dbapi.py | 891 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/dump.py | 83 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/factory.py | 285 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/hooks.py | 229 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/py25tests.py | 80 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/regression.py | 442 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/transactions.py | 205 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/types.py | 421 | ||||
| -rw-r--r-- | contrib/tools/python/src/Lib/sqlite3/test/userfunctions.py | 473 |
13 files changed, 0 insertions, 3293 deletions
diff --git a/contrib/tools/python/src/Lib/sqlite3/__init__.py b/contrib/tools/python/src/Lib/sqlite3/__init__.py deleted file mode 100644 index 41ef2b76dfc..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/__init__.py +++ /dev/null @@ -1,24 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/__init__.py: the pysqlite2 package. -# -# Copyright (C) 2005 Gerhard H�ring <[email protected]> -# -# 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 dbapi2 import * diff --git a/contrib/tools/python/src/Lib/sqlite3/dbapi2.py b/contrib/tools/python/src/Lib/sqlite3/dbapi2.py deleted file mode 100644 index 0d4dcaf6207..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/dbapi2.py +++ /dev/null @@ -1,90 +0,0 @@ -# -*- coding: iso-8859-1 -*- -# pysqlite2/dbapi2.py: the DB-API 2.0 interface -# -# Copyright (C) 2004-2005 Gerhard H�ring <[email protected]> -# -# 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. - -import collections -import datetime -import time - -from _sqlite3 import * - -paramstyle = "qmark" - -threadsafety = 1 - -apilevel = "2.0" - -Date = datetime.date - -Time = datetime.time - -Timestamp = datetime.datetime - -def DateFromTicks(ticks): - return Date(*time.localtime(ticks)[:3]) - -def TimeFromTicks(ticks): - return Time(*time.localtime(ticks)[3:6]) - -def TimestampFromTicks(ticks): - return Timestamp(*time.localtime(ticks)[:6]) - -version_info = tuple([int(x) for x in version.split(".")]) -sqlite_version_info = tuple([int(x) for x in sqlite_version.split(".")]) - -Binary = buffer -collections.Sequence.register(Row) - -def register_adapters_and_converters(): - def adapt_date(val): - return val.isoformat() - - def adapt_datetime(val): - return val.isoformat(" ") - - def convert_date(val): - return datetime.date(*map(int, val.split("-"))) - - def convert_timestamp(val): - 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: - microseconds = 0 - - val = datetime.datetime(year, month, day, hours, minutes, seconds, microseconds) - return val - - - register_adapter(datetime.date, adapt_date) - register_adapter(datetime.datetime, adapt_datetime) - register_converter("date", convert_date) - register_converter("timestamp", convert_timestamp) - -register_adapters_and_converters() - -# Clean up namespace - -del(register_adapters_and_converters) diff --git a/contrib/tools/python/src/Lib/sqlite3/dump.py b/contrib/tools/python/src/Lib/sqlite3/dump.py deleted file mode 100644 index e5c5ef226db..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/dump.py +++ /dev/null @@ -1,70 +0,0 @@ -# Mimic the sqlite3 console shell's .dump command -# Author: Paul Kippes <[email protected]> - -# Every identifier in sql is quoted based on a comment in sqlite -# documentation "SQLite adds new keywords from time to time when it -# takes on new features. So to prevent your code from being broken by -# future enhancements, you should normally quote any identifier that -# is an English language word, even if you do not have to." - -def _iterdump(connection): - """ - Returns an iterator to the dump of the database in an SQL text format. - - Used to produce an SQL dump of the database. Useful to save an in-memory - database for later restoration. This function should not be called - directly but instead called from the Connection method, iterdump(). - """ - - cu = connection.cursor() - yield('BEGIN TRANSACTION;') - - # sqlite_master table contains the SQL CREATE statements for the database. - q = """ - SELECT "name", "type", "sql" - FROM "sqlite_master" - WHERE "sql" NOT NULL AND - "type" == 'table' - ORDER BY "name" - """ - schema_res = cu.execute(q) - for table_name, type, sql in schema_res.fetchall(): - if table_name == 'sqlite_sequence': - yield('DELETE FROM "sqlite_sequence";') - elif table_name == 'sqlite_stat1': - yield('ANALYZE "sqlite_master";') - elif table_name.startswith('sqlite_'): - continue - # NOTE: Virtual table support not implemented - #elif sql.startswith('CREATE VIRTUAL TABLE'): - # qtable = table_name.replace("'", "''") - # yield("INSERT INTO sqlite_master(type,name,tbl_name,rootpage,sql)"\ - # "VALUES('table','{0}','{0}',0,'{1}');".format( - # qtable, - # sql.replace("''"))) - else: - yield('%s;' % sql) - - # Build the insert statement for each row of the current table - table_name_ident = table_name.replace('"', '""') - res = cu.execute('PRAGMA table_info("{0}")'.format(table_name_ident)) - column_names = [str(table_info[1]) for table_info in res.fetchall()] - q = """SELECT 'INSERT INTO "{0}" VALUES({1})' FROM "{0}";""".format( - table_name_ident, - ",".join("""'||quote("{0}")||'""".format(col.replace('"', '""')) for col in column_names)) - query_res = cu.execute(q) - for row in query_res: - yield("%s;" % row[0]) - - # Now when the type is 'index', 'trigger', or 'view' - q = """ - SELECT "name", "type", "sql" - FROM "sqlite_master" - WHERE "sql" NOT NULL AND - "type" IN ('index', 'trigger', 'view') - """ - schema_res = cu.execute(q) - for name, type, sql in schema_res.fetchall(): - yield('%s;' % sql) - - yield('COMMIT;') diff --git a/contrib/tools/python/src/Lib/sqlite3/test/__init__.py b/contrib/tools/python/src/Lib/sqlite3/test/__init__.py deleted file mode 100644 index e69de29bb2d..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/__init__.py +++ /dev/null diff --git a/contrib/tools/python/src/Lib/sqlite3/test/dbapi.py b/contrib/tools/python/src/Lib/sqlite3/test/dbapi.py deleted file mode 100644 index 615ebf51069..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/dbapi.py +++ /dev/null @@ -1,891 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/dbapi.py: tests for DB-API compliance -# -# Copyright (C) 2004-2010 Gerhard H�ring <[email protected]> -# -# 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. - -import unittest -import sys -import sqlite3 as sqlite -from test import test_support -try: - import threading -except ImportError: - threading = None - -class ModuleTests(unittest.TestCase): - def CheckAPILevel(self): - self.assertEqual(sqlite.apilevel, "2.0", - "apilevel is %s, should be 2.0" % sqlite.apilevel) - - def CheckThreadSafety(self): - self.assertEqual(sqlite.threadsafety, 1, - "threadsafety is %d, should be 1" % sqlite.threadsafety) - - def CheckParamStyle(self): - self.assertEqual(sqlite.paramstyle, "qmark", - "paramstyle is '%s', should be 'qmark'" % - sqlite.paramstyle) - - def CheckWarning(self): - self.assertTrue(issubclass(sqlite.Warning, StandardError), - "Warning is not a subclass of StandardError") - - def CheckError(self): - self.assertTrue(issubclass(sqlite.Error, StandardError), - "Error is not a subclass of StandardError") - - def CheckInterfaceError(self): - self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error), - "InterfaceError is not a subclass of Error") - - def CheckDatabaseError(self): - self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error), - "DatabaseError is not a subclass of Error") - - def CheckDataError(self): - self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError), - "DataError is not a subclass of DatabaseError") - - def CheckOperationalError(self): - self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError), - "OperationalError is not a subclass of DatabaseError") - - def CheckIntegrityError(self): - self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError), - "IntegrityError is not a subclass of DatabaseError") - - def CheckInternalError(self): - self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError), - "InternalError is not a subclass of DatabaseError") - - def CheckProgrammingError(self): - self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError), - "ProgrammingError is not a subclass of DatabaseError") - - def CheckNotSupportedError(self): - self.assertTrue(issubclass(sqlite.NotSupportedError, - sqlite.DatabaseError), - "NotSupportedError is not a subclass of DatabaseError") - -class ConnectionTests(unittest.TestCase): - def setUp(self): - self.cx = sqlite.connect(":memory:") - cu = self.cx.cursor() - cu.execute("create table test(id integer primary key, name text)") - cu.execute("insert into test(name) values (?)", ("foo",)) - - def tearDown(self): - self.cx.close() - - def CheckCommit(self): - self.cx.commit() - - def CheckCommitAfterNoChanges(self): - """ - A commit should also work when no changes were made to the database. - """ - self.cx.commit() - self.cx.commit() - - def CheckRollback(self): - self.cx.rollback() - - def CheckRollbackAfterNoChanges(self): - """ - A rollback should also work when no changes were made to the database. - """ - self.cx.rollback() - self.cx.rollback() - - def CheckCursor(self): - cu = self.cx.cursor() - - def CheckFailedOpen(self): - YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db" - 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() - - def CheckExceptions(self): - # Optional DB-API extension. - self.assertEqual(self.cx.Warning, sqlite.Warning) - self.assertEqual(self.cx.Error, sqlite.Error) - self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError) - self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError) - self.assertEqual(self.cx.DataError, sqlite.DataError) - self.assertEqual(self.cx.OperationalError, sqlite.OperationalError) - self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError) - self.assertEqual(self.cx.InternalError, sqlite.InternalError) - self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError) - self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError) - -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)") - self.cu.execute("insert into test(name) values (?)", ("foo",)) - - def tearDown(self): - self.cu.close() - self.cx.close() - - def CheckExecuteNoArgs(self): - self.cu.execute("delete from test") - - def CheckExecuteIllegalSql(self): - 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): - 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") - - def CheckExecuteTooMuchSql3(self): - self.cu.execute(""" - select 5+4; - - /* - foo - */ - """) - - def CheckExecuteWrongSqlArg(self): - 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,)) - - def CheckExecuteArgFloat(self): - self.cu.execute("insert into test(income) values (?)", (2500.32,)) - - def CheckExecuteArgString(self): - self.cu.execute("insert into test(name) values (?)", ("Hugo",)) - - def CheckExecuteArgStringWithZeroByte(self): - self.cu.execute("insert into test(name) values (?)", ("Hu\x00go",)) - - self.cu.execute("select name from test where id=?", (self.cu.lastrowid,)) - row = self.cu.fetchone() - self.assertEqual(row[0], "Hu\x00go") - - def CheckExecuteWrongNoOfArgs1(self): - # too many parameters - 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 - 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 - 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')") - self.cu.execute("select name from test where name=?", ["foo"]) - row = self.cu.fetchone() - self.assertEqual(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.assertEqual(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.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" - - 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.assertEqual(row[0], "foo") - - def CheckExecuteDictMappingTooLittleArgs(self): - self.cu.execute("insert into test(name) values ('foo')") - 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')") - 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')") - 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() - - def CheckRowcountExecute(self): - self.cu.execute("delete from test") - self.cu.execute("insert into test(name) values ('foo')") - self.cu.execute("insert into test(name) values ('foo')") - self.cu.execute("update test set name='bar'") - self.assertEqual(self.cu.rowcount, 2) - - def CheckRowcountSelect(self): - """ - pysqlite does not know the rowcount of SELECT statements, because we - don't fetch all rows after executing the select statement. The rowcount - has thus to be -1. - """ - self.cu.execute("select 5 union select 6") - self.assertEqual(self.cu.rowcount, -1) - - def CheckRowcountExecutemany(self): - self.cu.execute("delete from test") - self.cu.executemany("insert into test(name) values (?)", [(1,), (2,), (3,)]) - self.assertEqual(self.cu.rowcount, 3) - - def CheckTotalChanges(self): - self.cu.execute("insert into test(name) values ('foo')") - self.cu.execute("insert into test(name) values ('foo')") - if self.cx.total_changes < 2: - self.fail("total changes reported wrong value") - - # Checks for executemany: - # Sequences are required by the DB-API, iterators - # enhancements in pysqlite. - - def CheckExecuteManySequence(self): - self.cu.executemany("insert into test(income) values (?)", [(x,) for x in range(100, 110)]) - - def CheckExecuteManyIterator(self): - class MyIter: - def __init__(self): - self.value = 5 - - def next(self): - if self.value == 10: - raise StopIteration - else: - self.value += 1 - return (self.value,) - - self.cu.executemany("insert into test(income) values (?)", MyIter()) - - def CheckExecuteManyGenerator(self): - def mygen(): - for i in range(5): - yield (i,) - - self.cu.executemany("insert into test(income) values (?)", mygen()) - - def CheckExecuteManyWrongSqlArg(self): - 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): - 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): - 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. - self.cu.execute("delete from test") - self.cu.execute("insert into test(id) values (?)", (5,)) - self.cu.execute("insert into test(id) values (?)", (6,)) - self.cu.execute("select id from test order by id") - lst = [] - for row in self.cu: - lst.append(row[0]) - self.assertEqual(lst[0], 5) - self.assertEqual(lst[1], 6) - - def CheckFetchone(self): - self.cu.execute("select name from test") - row = self.cu.fetchone() - self.assertEqual(row[0], "foo") - row = self.cu.fetchone() - self.assertEqual(row, None) - - def CheckFetchoneNoStatement(self): - cur = self.cx.cursor() - row = cur.fetchone() - self.assertEqual(row, None) - - def CheckArraySize(self): - # must default ot 1 - self.assertEqual(self.cu.arraysize, 1) - - # now set to 2 - self.cu.arraysize = 2 - - # now make the query return 3 rows - self.cu.execute("delete from test") - self.cu.execute("insert into test(name) values ('A')") - self.cu.execute("insert into test(name) values ('B')") - self.cu.execute("insert into test(name) values ('C')") - self.cu.execute("select name from test") - res = self.cu.fetchmany() - - self.assertEqual(len(res), 2) - - def CheckFetchmany(self): - self.cu.execute("select name from test") - res = self.cu.fetchmany(100) - self.assertEqual(len(res), 1) - res = self.cu.fetchmany(100) - self.assertEqual(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.assertEqual(len(res), 1) - - def CheckFetchall(self): - self.cu.execute("select name from test") - res = self.cu.fetchall() - self.assertEqual(len(res), 1) - res = self.cu.fetchall() - self.assertEqual(res, []) - - def CheckSetinputsizes(self): - self.cu.setinputsizes([3, 4, 5]) - - def CheckSetoutputsize(self): - self.cu.setoutputsize(5, 0) - - def CheckSetoutputsizeNoColumn(self): - self.cu.setoutputsize(42) - - def CheckCursorConnection(self): - # Optional DB-API extension. - self.assertEqual(self.cu.connection, self.cx) - - def CheckWrongCursorCallable(self): - 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() - try: - cur = sqlite.Cursor(foo) - self.fail("should have raised a ValueError") - except TypeError: - pass - [email protected](threading, 'This test requires threading.') -class ThreadTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - self.cur = self.con.cursor() - self.cur.execute("create table test(id integer primary key, name text, bin binary, ratio number, ts timestamp)") - - def tearDown(self): - self.cur.close() - self.con.close() - - def CheckConCursor(self): - def run(con, errors): - try: - cur = con.cursor() - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckConCommit(self): - def run(con, errors): - try: - con.commit() - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckConRollback(self): - def run(con, errors): - try: - con.rollback() - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckConClose(self): - def run(con, errors): - try: - con.close() - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - t = threading.Thread(target=run, kwargs={"con": self.con, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckCurImplicitBegin(self): - def run(cur, errors): - try: - cur.execute("insert into test(name) values ('a')") - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckCurClose(self): - def run(cur, errors): - try: - cur.close() - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckCurExecute(self): - def run(cur, errors): - try: - cur.execute("select name from test") - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - self.cur.execute("insert into test(name) values ('a')") - t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - - def CheckCurIterNext(self): - def run(cur, errors): - try: - row = cur.fetchone() - errors.append("did not raise ProgrammingError") - return - except sqlite.ProgrammingError: - return - except: - errors.append("raised wrong exception") - - errors = [] - self.cur.execute("insert into test(name) values ('a')") - self.cur.execute("select name from test") - t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors}) - t.start() - t.join() - if len(errors) > 0: - self.fail("\n".join(errors)) - -class ConstructorTests(unittest.TestCase): - def CheckDate(self): - d = sqlite.Date(2004, 10, 28) - - def CheckTime(self): - t = sqlite.Time(12, 39, 35) - - def CheckTimestamp(self): - ts = sqlite.Timestamp(2004, 10, 28, 12, 39, 35) - - def CheckDateFromTicks(self): - d = sqlite.DateFromTicks(42) - - def CheckTimeFromTicks(self): - t = sqlite.TimeFromTicks(42) - - def CheckTimestampFromTicks(self): - ts = sqlite.TimestampFromTicks(42) - - def CheckBinary(self): - with test_support.check_py3k_warnings(): - b = sqlite.Binary(chr(0) + "'") - -class ExtensionTests(unittest.TestCase): - def CheckScriptStringSql(self): - con = sqlite.connect(":memory:") - cur = con.cursor() - cur.executescript(""" - -- bla bla - /* a stupid comment */ - create table a(i); - insert into a(i) values (5); - """) - cur.execute("select i from a") - 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() - 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() - raised = False - try: - cur.executescript("create table test(sadfsadfdsa); select foo from hurz;") - except sqlite.OperationalError: - raised = True - self.assertEqual(raised, True, "should have raised an exception") - - def CheckConnectionExecute(self): - con = sqlite.connect(":memory:") - result = con.execute("select 5").fetchone()[0] - self.assertEqual(result, 5, "Basic test of Connection.execute") - - def CheckConnectionExecutemany(self): - con = sqlite.connect(":memory:") - con.execute("create table test(foo)") - con.executemany("insert into test(foo) values (?)", [(3,), (4,)]) - result = con.execute("select foo from test order by foo").fetchall() - self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany") - self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany") - - def CheckConnectionExecutescript(self): - con = sqlite.connect(":memory:") - con.executescript("create table test(foo); insert into test(foo) values (5);") - result = con.execute("select foo from test").fetchone()[0] - 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() - 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() - 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() - 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() - 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 - 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:") - con.close() - class Agg: - def __init__(self): - pass - def step(self, x): - pass - def finalize(self): - return 17 - 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 - 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 - 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() - 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() - cur.close() - - for method_name in ("execute", "executemany", "executescript", "fetchall", "fetchmany", "fetchone"): - if method_name in ("execute", "executescript"): - params = ("select 4 union select 5",) - elif method_name == "executemany": - params = ("insert into foo(bar) values (?)", [(3,), (4,)]) - else: - params = [] - - try: - method = getattr(cur, method_name) - - 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") - connection_suite = unittest.makeSuite(ConnectionTests, "Check") - cursor_suite = unittest.makeSuite(CursorTests, "Check") - thread_suite = unittest.makeSuite(ThreadTests, "Check") - constructor_suite = unittest.makeSuite(ConstructorTests, "Check") - ext_suite = unittest.makeSuite(ExtensionTests, "Check") - closed_con_suite = unittest.makeSuite(ClosedConTests, "Check") - closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check") - 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() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/dump.py b/contrib/tools/python/src/Lib/sqlite3/test/dump.py deleted file mode 100644 index b7de810c7a5..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/dump.py +++ /dev/null @@ -1,83 +0,0 @@ -# Author: Paul Kippes <[email protected]> - -import unittest -import sqlite3 as sqlite - -class DumpTests(unittest.TestCase): - def setUp(self): - self.cx = sqlite.connect(":memory:") - self.cu = self.cx.cursor() - - def tearDown(self): - self.cx.close() - - def CheckTableDump(self): - expected_sqls = [ - """CREATE TABLE "index"("index" blob);""" - , - """INSERT INTO "index" VALUES(X'01');""" - , - """CREATE TABLE "quoted""table"("quoted""field" text);""" - , - """INSERT INTO "quoted""table" VALUES('quoted''value');""" - , - "CREATE TABLE t1(id integer primary key, s1 text, " \ - "t1_i1 integer not null, i2 integer, unique (s1), " \ - "constraint t1_idx1 unique (i2));" - , - "INSERT INTO \"t1\" VALUES(1,'foo',10,20);" - , - "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));" - , - "CREATE TRIGGER trigger_1 update of t1_i1 on t1 " \ - "begin " \ - "update t2 set t2_i1 = new.t1_i1 where t2_i1 = old.t1_i1; " \ - "end;" - , - "CREATE VIEW v1 as select * from t1 left join t2 " \ - "using (id);" - ] - [self.cu.execute(s) for s in expected_sqls] - i = self.cx.iterdump() - actual_sqls = [s for s in i] - expected_sqls = ['BEGIN TRANSACTION;'] + expected_sqls + \ - ['COMMIT;'] - [self.assertEqual(expected_sqls[i], actual_sqls[i]) - for i in xrange(len(expected_sqls))] - - def CheckUnorderableRow(self): - # iterdump() should be able to cope with unorderable row types (issue #15545) - class UnorderableRow: - def __init__(self, cursor, row): - self.row = row - def __getitem__(self, index): - return self.row[index] - self.cx.row_factory = UnorderableRow - CREATE_ALPHA = """CREATE TABLE "alpha" ("one");""" - CREATE_BETA = """CREATE TABLE "beta" ("two");""" - expected = [ - "BEGIN TRANSACTION;", - CREATE_ALPHA, - CREATE_BETA, - "COMMIT;" - ] - self.cu.execute(CREATE_BETA) - self.cu.execute(CREATE_ALPHA) - got = list(self.cx.iterdump()) - self.assertEqual(expected, got) - -def suite(): - return unittest.TestSuite(unittest.makeSuite(DumpTests, "Check")) - -def test(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/factory.py b/contrib/tools/python/src/Lib/sqlite3/test/factory.py deleted file mode 100644 index b9e9cd7fd7f..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/factory.py +++ /dev/null @@ -1,285 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/factory.py: tests for the various factories in pysqlite -# -# Copyright (C) 2005-2007 Gerhard H�ring <[email protected]> -# -# 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. - -import unittest -import sqlite3 as sqlite -from collections import Sequence - -class MyConnection(sqlite.Connection): - def __init__(self, *args, **kwargs): - sqlite.Connection.__init__(self, *args, **kwargs) - -def dict_factory(cursor, row): - d = {} - for idx, col in enumerate(cursor.description): - d[col[0]] = row[idx] - return d - -class MyCursor(sqlite.Cursor): - def __init__(self, *args, **kwargs): - sqlite.Cursor.__init__(self, *args, **kwargs) - self.row_factory = dict_factory - -class ConnectionFactoryTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:", factory=MyConnection) - - def tearDown(self): - self.con.close() - - def CheckIsInstance(self): - self.assertIsInstance(self.con, MyConnection) - -class CursorFactoryTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - def tearDown(self): - self.con.close() - - def CheckIsInstance(self): - cur = self.con.cursor() - self.assertIsInstance(cur, sqlite.Cursor) - cur = self.con.cursor(MyCursor) - self.assertIsInstance(cur, MyCursor) - cur = self.con.cursor(factory=lambda con: MyCursor(con)) - self.assertIsInstance(cur, MyCursor) - - def CheckInvalidFactory(self): - # not a callable at all - self.assertRaises(TypeError, self.con.cursor, None) - # invalid callable with not exact one argument - self.assertRaises(TypeError, self.con.cursor, lambda: None) - # invalid callable returning non-cursor - self.assertRaises(TypeError, self.con.cursor, lambda con: None) - -class RowFactoryTestsBackwardsCompat(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - def CheckIsProducedByFactory(self): - cur = self.con.cursor(factory=MyCursor) - cur.execute("select 4+5 as foo") - row = cur.fetchone() - self.assertIsInstance(row, dict) - cur.close() - - def tearDown(self): - self.con.close() - -class RowFactoryTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - def CheckCustomFactory(self): - self.con.row_factory = lambda cur, row: list(row) - row = self.con.execute("select 1, 2").fetchone() - self.assertIsInstance(row, list) - - def CheckSqliteRowIndex(self): - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as a, 2 as b").fetchone() - self.assertIsInstance(row, sqlite.Row) - - 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'") - - 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[2] - with self.assertRaises(IndexError): - row[2L] - with self.assertRaises(IndexError): - row[-3] - with self.assertRaises(IndexError): - row[-3L] - with self.assertRaises(IndexError): - row[2**1000] - - def CheckSqliteRowIter(self): - """Checks if the row object is iterable""" - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as a, 2 as b").fetchone() - for col in row: - pass - - def CheckSqliteRowAsTuple(self): - """Checks if the row object can be converted to a tuple""" - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as a, 2 as b").fetchone() - t = tuple(row) - self.assertEqual(t, (row['a'], row['b'])) - - def CheckSqliteRowAsDict(self): - """Checks if the row object can be correctly converted to a dictionary""" - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as a, 2 as b").fetchone() - d = dict(row) - self.assertEqual(d["a"], row["a"]) - self.assertEqual(d["b"], row["b"]) - - def CheckSqliteRowHashCmp(self): - """Checks if the row object compares and hashes correctly""" - self.con.row_factory = sqlite.Row - row_1 = self.con.execute("select 1 as a, 2 as b").fetchone() - row_2 = self.con.execute("select 1 as a, 2 as b").fetchone() - row_3 = self.con.execute("select 1 as a, 3 as b").fetchone() - - self.assertEqual(row_1, row_1) - self.assertEqual(row_1, row_2) - self.assertTrue(row_2 != row_3) - - self.assertFalse(row_1 != row_1) - self.assertFalse(row_1 != row_2) - self.assertFalse(row_2 == row_3) - - self.assertEqual(row_1, row_2) - self.assertEqual(hash(row_1), hash(row_2)) - self.assertNotEqual(row_1, row_3) - self.assertNotEqual(hash(row_1), hash(row_3)) - - def CheckSqliteRowAsSequence(self): - """ Checks if the row object can act like a sequence """ - self.con.row_factory = sqlite.Row - row = self.con.execute("select 1 as a, 2 as b").fetchone() - - as_tuple = tuple(row) - self.assertEqual(list(reversed(row)), list(reversed(as_tuple))) - self.assertIsInstance(row, Sequence) - - def CheckFakeCursorClass(self): - # Issue #24257: Incorrect use of PyObject_IsInstance() caused - # segmentation fault. - # Issue #27861: Also applies for cursor factory. - class FakeCursor(str): - __class__ = sqlite.Cursor - self.con.row_factory = sqlite.Row - self.assertRaises(TypeError, self.con.cursor, FakeCursor) - self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ()) - - def tearDown(self): - self.con.close() - -class TextFactoryTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - def CheckUnicode(self): - austria = unicode("�sterreich", "latin1") - row = self.con.execute("select ?", (austria,)).fetchone() - self.assertEqual(type(row[0]), unicode, "type of row[0] must be unicode") - - def CheckString(self): - self.con.text_factory = str - austria = unicode("�sterreich", "latin1") - row = self.con.execute("select ?", (austria,)).fetchone() - 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: 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): - self.con.text_factory = sqlite.OptimizedUnicode - 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]), 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): - self.con.close() - -class TextFactoryTestsWithEmbeddedZeroBytes(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - self.con.execute("create table test (value text)") - self.con.execute("insert into test (value) values (?)", ("a\x00b",)) - - def CheckString(self): - # text_factory defaults to unicode - row = self.con.execute("select value from test").fetchone() - self.assertIs(type(row[0]), unicode) - self.assertEqual(row[0], "a\x00b") - - 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]), str) - self.assertEqual(row[0], "a\x00b") - - 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]), str) - self.assertEqual(row[0], "a\x00b") - - 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]), unicode) - self.assertEqual(row[0], u"�\x00�") - - def tearDown(self): - self.con.close() - -def suite(): - connection_suite = unittest.makeSuite(ConnectionFactoryTests, "Check") - cursor_suite = unittest.makeSuite(CursorFactoryTests, "Check") - row_suite_compat = unittest.makeSuite(RowFactoryTestsBackwardsCompat, "Check") - row_suite = unittest.makeSuite(RowFactoryTests, "Check") - text_suite = unittest.makeSuite(TextFactoryTests, "Check") - text_zero_bytes_suite = unittest.makeSuite(TextFactoryTestsWithEmbeddedZeroBytes, "Check") - return unittest.TestSuite((connection_suite, cursor_suite, row_suite_compat, row_suite, text_suite, text_zero_bytes_suite)) - -def test(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/hooks.py b/contrib/tools/python/src/Lib/sqlite3/test/hooks.py deleted file mode 100644 index a5b1632a181..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/hooks.py +++ /dev/null @@ -1,229 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/hooks.py: tests for various SQLite-specific hooks -# -# Copyright (C) 2006-2007 Gerhard H�ring <[email protected]> -# -# 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. - -import os, unittest -import sqlite3 as sqlite - -class CollationTests(unittest.TestCase): - def setUp(self): - pass - - def tearDown(self): - pass - - def CheckCreateCollationNotString(self): - con = sqlite.connect(":memory:") - with self.assertRaises(TypeError): - con.create_collation(None, lambda x, y: (x > y) - (x < y)) - - def CheckCreateCollationNotCallable(self): - con = sqlite.connect(":memory:") - try: - con.create_collation("X", 42) - 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:") - try: - con.create_collation("coll�", cmp) - self.fail("should have raised a ProgrammingError") - except sqlite.ProgrammingError, e: - pass - - def CheckCreateCollationBadUpper(self): - class BadUpperStr(str): - def upper(self): - return None - con = sqlite.connect(":memory:") - mycoll = lambda x, y: -((x > y) - (x < y)) - con.create_collation(BadUpperStr("mycoll"), mycoll) - 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') - - 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 -cmp(x, y) - - con = sqlite.connect(":memory:") - con.create_collation("mycoll", mycoll) - sql = """ - select x from ( - select 'a' as x - union - select 'b' as x - union - select 'c' as x - ) order by x collate mycoll - """ - result = con.execute(sql).fetchall() - 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) - try: - result = con.execute(sql).fetchall() - 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): - # reverse order - return -((x > y) - (x < y)) * 2**32 - con = sqlite.connect(":memory:") - con.create_collation("mycoll", mycoll) - sql = """ - select x from ( - select 'a' as x - union - select 'b' as x - union - select 'c' as x - ) order by x collate mycoll - """ - result = con.execute(sql).fetchall() - self.assertEqual(result, [('c',), ('b',), ('a',)], - msg="the expected order was not returned") - - def CheckCollationRegisterTwice(self): - """ - Register two different collation functions under the same name. - Verify that the last one is actually used. - """ - con = sqlite.connect(":memory:") - 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() - if result[0][0] != 'b' or result[1][0] != 'a': - self.fail("wrong collation function is used") - - def CheckDeregisterCollation(self): - """ - Register a collation, then deregister it. Make sure an error is raised if we try - to use it. - """ - con = sqlite.connect(":memory:") - con.create_collation("mycoll", cmp) - con.create_collation("mycoll", None) - try: - con.execute("select 'a' as x union select 'b' as x order by x collate 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): - """ - 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.assertTrue(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.assertGreaterEqual(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 = [] - def progress(): - 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(len(action), 0, "progress handler was not cleared") - -def suite(): - collation_suite = unittest.makeSuite(CollationTests, "Check") - progress_suite = unittest.makeSuite(ProgressTests, "Check") - return unittest.TestSuite((collation_suite, progress_suite)) - -def test(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/py25tests.py b/contrib/tools/python/src/Lib/sqlite3/test/py25tests.py deleted file mode 100644 index 7fd3c0e0cf8..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/py25tests.py +++ /dev/null @@ -1,80 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/regression.py: pysqlite regression tests -# -# Copyright (C) 2007 Gerhard H�ring <[email protected]> -# -# 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/contrib/tools/python/src/Lib/sqlite3/test/regression.py b/contrib/tools/python/src/Lib/sqlite3/test/regression.py deleted file mode 100644 index 42fc7c278b8..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/regression.py +++ /dev/null @@ -1,442 +0,0 @@ -#-*- coding: iso-8859-1 -*- -# pysqlite2/test/regression.py: pysqlite regression tests -# -# Copyright (C) 2006-2007 Gerhard H�ring <[email protected]> -# -# 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. - -import datetime -import unittest -import sqlite3 as sqlite -import weakref -from test import support - -class RegressionTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - def tearDown(self): - self.con.close() - - def CheckPragmaUserVersion(self): - # This used to crash pysqlite because this pragma command returns NULL for the column name - cur = self.con.cursor() - cur.execute("pragma user_version") - - def CheckPragmaSchemaVersion(self): - # This still crashed pysqlite <= 2.2.1 - con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES) - try: - cur = self.con.cursor() - cur.execute("pragma schema_version") - finally: - cur.close() - con.close() - - def CheckStatementReset(self): - # pysqlite 2.1.0 to 2.2.0 have the problem that not all statements are - # 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 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 xrange(10)]) - - for i in range(5): - cursors[i].execute(" " * i + "select x from test") - - con.rollback() - - def CheckColumnNameWithSpaces(self): - cur = self.con.cursor() - cur.execute('select 1 as "foo bar [datetime]"') - self.assertEqual(cur.description[0][0], "foo bar") - - cur.execute('select 1 as "foo baz"') - self.assertEqual(cur.description[0][0], "foo baz") - - 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 and 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 foo(bar)") - self.con.execute("drop table foo") - self.con.execute("create table 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 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 - 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 CheckRegisterAdapter(self): - """ - See issue 3312. - """ - self.assertRaises(TypeError, sqlite.register_adapter, {}, None) - - def CheckSetIsolationLevel(self): - """ - See issue 3312. - """ - con = sqlite.connect(":memory:") - self.assertRaises(UnicodeEncodeError, setattr, con, - "isolation_level", u"\xe9") - - def CheckCursorConstructorCallCheck(self): - """ - Verifies that cursor methods check whether base class __init__ was - called. - """ - class Cursor(sqlite.Cursor): - def __init__(self, con): - pass - - con = sqlite.connect(":memory:") - cur = Cursor(con) - try: - cur.execute("select 4+5").fetchall() - 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 CheckConnectionConstructorCallCheck(self): - """ - Verifies that connection methods check whether base class __init__ was - called. - """ - class Connection(sqlite.Connection): - def __init__(self, name): - pass - - con = Connection(":memory:") - try: - cur = con.cursor() - self.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("should have raised ProgrammingError") - - def CheckCursorRegistration(self): - """ - Verifies that subclassed cursor classes are correctly registered with - the connection object, too. (fetch-across-rollback problem) - """ - class Connection(sqlite.Connection): - def cursor(self): - return Cursor(self) - - class Cursor(sqlite.Cursor): - def __init__(self, con): - sqlite.Cursor.__init__(self, con) - - con = Connection(":memory:") - cur = con.cursor() - cur.execute("create table foo(x)") - cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)]) - cur.execute("select x from foo") - con.rollback() - try: - cur.fetchall() - self.fail("should have raised InterfaceError") - except sqlite.InterfaceError: - pass - except: - self.fail("should have raised InterfaceError") - - def CheckAutoCommit(self): - """ - Verifies that creating a connection in autocommit mode works. - 2.5.3 introduced a regression so that these could no longer - be created. - """ - con = sqlite.connect(":memory:", isolation_level=None) - - def CheckPragmaAutocommit(self): - """ - Verifies that running a PRAGMA statement that does an autocommit does - work. This did not work in 2.5.3/2.5.4. - """ - cur = self.con.cursor() - cur.execute("create table foo(bar)") - cur.execute("insert into foo(bar) values (5)") - - cur.execute("pragma page_size") - row = cur.fetchone() - - def CheckConnectionCall(self): - """ - Call a connection with a non-string SQL request: check error handling - of the statement constructor. - """ - self.assertRaises(sqlite.Warning, self.con, 1) - - def CheckRecursiveCursorUse(self): - """ - http://bugs.python.org/issue10811 - - Recursively using a cursor, such as when reusing it from a generator led to segfaults. - Now we catch recursive cursor usage and raise a ProgrammingError. - """ - con = sqlite.connect(":memory:") - - cur = con.cursor() - cur.execute("create table a (bar)") - cur.execute("create table b (baz)") - - def foo(): - cur.execute("insert into a (bar) values (?)", (1,)) - yield 1 - - with self.assertRaises(sqlite.ProgrammingError): - cur.executemany("insert into b (baz) values (?)", - ((i,) for i in foo())) - - def CheckConvertTimestampMicrosecondPadding(self): - """ - http://bugs.python.org/issue14720 - - The microsecond parsing of convert_timestamp() should pad with zeros, - since the microsecond string "456" actually represents "456000". - """ - - con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES) - cur = con.cursor() - cur.execute("CREATE TABLE t (x TIMESTAMP)") - - # Microseconds should be 456000 - cur.execute("INSERT INTO t (x) VALUES ('2012-04-04 15:06:00.456')") - - # Microseconds should be truncated to 123456 - cur.execute("INSERT INTO t (x) VALUES ('2012-04-04 15:06:00.123456789')") - - cur.execute("SELECT * FROM t") - values = [x[0] for x in cur.fetchall()] - - self.assertEqual(values, [ - datetime.datetime(2012, 4, 4, 15, 6, 0, 456000), - datetime.datetime(2012, 4, 4, 15, 6, 0, 123456), - ]) - - def CheckInvalidIsolationLevelType(self): - # isolation level is a string, not an integer - self.assertRaises(TypeError, - sqlite.connect, ":memory:", isolation_level=123) - - - def CheckNullCharacter(self): - # Issue #21147 - con = sqlite.connect(":memory:") - self.assertRaises(ValueError, con, "\0select 1") - self.assertRaises(ValueError, con, "select 1\0") - cur = con.cursor() - self.assertRaises(ValueError, cur.execute, " \0select 2") - self.assertRaises(ValueError, cur.execute, "select 2\0") - - def CheckCommitCursorReset(self): - """ - Connection.commit() did reset cursors, which made sqlite3 - to return rows multiple times when fetched from cursors - after commit. See issues 10513 and 23129 for details. - """ - con = sqlite.connect(":memory:") - con.executescript(""" - create table t(c); - create table t2(c); - insert into t values(0); - insert into t values(1); - insert into t values(2); - """) - - self.assertEqual(con.isolation_level, "") - - counter = 0 - for i, row in enumerate(con.execute("select c from t")): - 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): - """ - The interpreter shouldn't crash in case Cursor.__init__() is called - more than once. - """ - def callback(*args): - pass - con = sqlite.connect(":memory:") - cur = sqlite.Cursor(con) - ref = weakref.ref(cur, callback) - cur.__init__(con) - del cur - # The interpreter shouldn't crash when ref is collected. - del ref - support.gc_collect() - - def CheckDelIsolation_levelSegfault(self): - with self.assertRaises(AttributeError): - del self.con.isolation_level - - -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(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/transactions.py b/contrib/tools/python/src/Lib/sqlite3/test/transactions.py deleted file mode 100644 index a732251b2d4..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/transactions.py +++ /dev/null @@ -1,205 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/transactions.py: tests transactions -# -# Copyright (C) 2005-2007 Gerhard H�ring <[email protected]> -# -# 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. - -import sys -import os, unittest -import sqlite3 as sqlite - -def get_db_path(): - return "sqlite_testdb" - -class TransactionTests(unittest.TestCase): - def setUp(self): - try: - os.remove(get_db_path()) - except OSError: - pass - - self.con1 = sqlite.connect(get_db_path(), timeout=0.1) - self.cur1 = self.con1.cursor() - - self.con2 = sqlite.connect(get_db_path(), timeout=0.1) - self.cur2 = self.con2.cursor() - - def tearDown(self): - self.cur1.close() - self.con1.close() - - self.cur2.close() - self.con2.close() - - try: - os.unlink(get_db_path()) - except OSError: - pass - - 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), 1) - - def CheckInsertStartsTransaction(self): - self.cur1.execute("create table test(i)") - self.cur1.execute("insert into test(i) values (5)") - self.cur2.execute("select i from test") - res = self.cur2.fetchall() - self.assertEqual(len(res), 0) - - def CheckUpdateStartsTransaction(self): - self.cur1.execute("create table test(i)") - self.cur1.execute("insert into test(i) values (5)") - self.con1.commit() - self.cur1.execute("update test set i=6") - self.cur2.execute("select i from test") - res = self.cur2.fetchone()[0] - self.assertEqual(res, 5) - - def CheckDeleteStartsTransaction(self): - self.cur1.execute("create table test(i)") - self.cur1.execute("insert into test(i) values (5)") - self.con1.commit() - self.cur1.execute("delete from test") - self.cur2.execute("select i from test") - res = self.cur2.fetchall() - self.assertEqual(len(res), 1) - - def CheckReplaceStartsTransaction(self): - self.cur1.execute("create table test(i)") - self.cur1.execute("insert into test(i) values (5)") - self.con1.commit() - self.cur1.execute("replace into test(i) values (6)") - self.cur2.execute("select i from test") - res = self.cur2.fetchall() - self.assertEqual(len(res), 1) - self.assertEqual(res[0][0], 5) - - def CheckToggleAutoCommit(self): - self.cur1.execute("create table test(i)") - self.cur1.execute("insert into test(i) values (5)") - self.con1.isolation_level = None - self.assertEqual(self.con1.isolation_level, None) - self.cur2.execute("select i from test") - res = self.cur2.fetchall() - self.assertEqual(len(res), 1) - - self.con1.isolation_level = "DEFERRED" - self.assertEqual(self.con1.isolation_level , "DEFERRED") - self.cur1.execute("insert into test(i) values (5)") - self.cur2.execute("select i from test") - res = self.cur2.fetchall() - self.assertEqual(len(res), 1) - - 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)") - 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") - - 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)") - 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() - - def CheckRollbackCursorConsistency(self): - """ - Checks if cursors on the connection are set into a "reset" state - when a rollback is done on the connection. - """ - con = sqlite.connect(":memory:") - cur = con.cursor() - cur.execute("create table test(x)") - cur.execute("insert into test(x) values (5)") - cur.execute("select 1 union select 2 union select 3") - - con.rollback() - 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)") - self.cur.execute("drop table test") - - def CheckPragma(self): - self.cur.execute("create table test(i)") - self.cur.execute("insert into test(i) values (5)") - self.cur.execute("pragma count_changes=1") - - def tearDown(self): - self.cur.close() - self.con.close() - -def suite(): - default_suite = unittest.makeSuite(TransactionTests, "Check") - special_command_suite = unittest.makeSuite(SpecialCommandTests, "Check") - return unittest.TestSuite((default_suite, special_command_suite)) - -def test(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/types.py b/contrib/tools/python/src/Lib/sqlite3/test/types.py deleted file mode 100644 index fdc21aee72a..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/types.py +++ /dev/null @@ -1,421 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/types.py: tests for type conversion and detection -# -# Copyright (C) 2005-2007 Gerhard H�ring <[email protected]> -# -# 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. - -import datetime -import unittest -import sqlite3 as sqlite -from test import test_support -try: - import zlib -except ImportError: - zlib = None - - -class SqliteTypeTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - self.cur = self.con.cursor() - self.cur.execute("create table test(i integer, s varchar, f number, b blob)") - - def tearDown(self): - self.cur.close() - self.con.close() - - def CheckString(self): - 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], u"�sterreich") - - def CheckSmallInt(self): - self.cur.execute("insert into test(i) values (?)", (42,)) - self.cur.execute("select i from test") - row = self.cur.fetchone() - self.assertEqual(row[0], 42) - - def CheckLargeInt(self): - num = 2**40 - self.cur.execute("insert into test(i) values (?)", (num,)) - self.cur.execute("select i from test") - row = self.cur.fetchone() - self.assertEqual(row[0], num) - - def CheckFloat(self): - val = 3.14 - self.cur.execute("insert into test(f) values (?)", (val,)) - self.cur.execute("select f from test") - row = self.cur.fetchone() - self.assertEqual(row[0], val) - - def CheckBlob(self): - 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], val) - - def CheckUnicodeExecute(self): - self.cur.execute(u"select '�sterreich'") - row = self.cur.fetchone() - 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): - self.val = _val - - def __cmp__(self, other): - if not isinstance(other, DeclTypesTests.Foo): - raise ValueError - if self.val == other.val: - return 0 - else: - return 1 - - def __conform__(self, protocol): - if protocol is sqlite.PrepareProtocol: - return self.val - else: - return None - - def __str__(self): - return "<%s>" % self.val - - 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))") - - # override float, make them always return the same number - sqlite.converters["FLOAT"] = lambda x: 47.2 - - # and implement two custom ones - sqlite.converters["BOOL"] = lambda x: bool(int(x)) - sqlite.converters["FOO"] = DeclTypesTests.Foo - sqlite.converters["WRONG"] = lambda x: "WRONG" - sqlite.converters["NUMBER"] = float - - def tearDown(self): - del sqlite.converters["FLOAT"] - del sqlite.converters["BOOL"] - del sqlite.converters["FOO"] - del sqlite.converters["NUMBER"] - self.cur.close() - self.con.close() - - def CheckString(self): - # default - self.cur.execute("insert into test(s) values (?)", ("foo",)) - self.cur.execute('select s as "s [WRONG]" from test') - row = self.cur.fetchone() - self.assertEqual(row[0], "foo") - - def CheckSmallInt(self): - # default - self.cur.execute("insert into test(i) values (?)", (42,)) - self.cur.execute("select i from test") - row = self.cur.fetchone() - self.assertEqual(row[0], 42) - - def CheckLargeInt(self): - # default - num = 2**40 - self.cur.execute("insert into test(i) values (?)", (num,)) - self.cur.execute("select i from test") - row = self.cur.fetchone() - self.assertEqual(row[0], num) - - def CheckFloat(self): - # custom - val = 3.14 - self.cur.execute("insert into test(f) values (?)", (val,)) - self.cur.execute("select f from test") - row = self.cur.fetchone() - self.assertEqual(row[0], 47.2) - - def CheckBool(self): - # custom - self.cur.execute("insert into test(b) values (?)", (False,)) - self.cur.execute("select b from test") - row = self.cur.fetchone() - 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.assertEqual(row[0], True) - - def CheckUnicode(self): - # default - val = u"\xd6sterreich" - self.cur.execute("insert into test(u) values (?)", (val,)) - self.cur.execute("select u from test") - row = self.cur.fetchone() - self.assertEqual(row[0], val) - - def CheckFoo(self): - val = DeclTypesTests.Foo("bla") - self.cur.execute("insert into test(foo) values (?)", (val,)) - self.cur.execute("select foo from test") - row = self.cur.fetchone() - self.assertEqual(row[0], val) - - def CheckUnsupportedSeq(self): - class Bar: pass - val = Bar() - 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() - 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 - 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], val) - - def CheckNumber1(self): - self.cur.execute("insert into test(n1) values (5)") - value = self.cur.execute("select n1 from test").fetchone()[0] - # if the converter is not used, it's an int instead of a float - self.assertEqual(type(value), float) - - def CheckNumber2(self): - """Checks whether converter names are cut off at '(' characters""" - self.cur.execute("insert into test(n2) values (5)") - value = self.cur.execute("select n2 from test").fetchone()[0] - # if the converter is not used, it's an int instead of a float - self.assertEqual(type(value), float) - -class ColNamesTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES) - self.cur = self.con.cursor() - self.cur.execute("create table test(x foo)") - - 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): - del sqlite.converters["FOO"] - del sqlite.converters["BAR"] - del sqlite.converters["EXC"] - del sqlite.converters["B1B1"] - self.cur.close() - self.con.close() - - def CheckDeclTypeNotUsed(self): - """ - Assures that the declared type is not used when PARSE_DECLTYPES - is not set. - """ - self.cur.execute("insert into test(x) values (?)", ("xxx",)) - self.cur.execute("select x from test") - val = self.cur.fetchone()[0] - self.assertEqual(val, "xxx") - - def CheckNone(self): - self.cur.execute("insert into test(x) values (?)", (None,)) - self.cur.execute("select x from test") - val = self.cur.fetchone()[0] - self.assertEqual(val, None) - - def CheckColName(self): - self.cur.execute("insert into test(x) values (?)", ("xxx",)) - self.cur.execute('select x as "x [bar]" from test') - val = self.cur.fetchone()[0] - self.assertEqual(val, "<xxx>") - - # Check if the stripping of colnames works. Everything after the first - # whitespace should be stripped. - self.assertEqual(self.cur.description[0][0], "x") - - def CheckCaseInConverterName(self): - self.cur.execute("""select 'other' as "x [b1b1]\"""") - val = self.cur.fetchone()[0] - self.assertEqual(val, "MARKER") - - def CheckCursorDescriptionNoRow(self): - """ - cursor.description should at least provide the column name(s), even if - no row returned. - """ - self.cur.execute("select * from test where 0 = 1") - self.assertEqual(self.cur.description[0][0], "x") - -class ObjectAdaptationTests(unittest.TestCase): - def cast(obj): - return float(obj) - cast = staticmethod(cast) - - def setUp(self): - self.con = sqlite.connect(":memory:") - try: - del sqlite.adapters[int] - except: - pass - sqlite.register_adapter(int, ObjectAdaptationTests.cast) - self.cur = self.con.cursor() - - def tearDown(self): - del sqlite.adapters[(int, sqlite.PrepareProtocol)] - self.cur.close() - self.con.close() - - def CheckCasterIsUsed(self): - self.cur.execute("select ?", (4,)) - val = self.cur.fetchone()[0] - self.assertEqual(type(val), float) - [email protected](zlib, "requires zlib") -class BinaryConverterTests(unittest.TestCase): - def convert(s): - return zlib.decompress(s) - convert = staticmethod(convert) - - def setUp(self): - self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES) - sqlite.register_converter("bin", BinaryConverterTests.convert) - - def tearDown(self): - self.con.close() - - def CheckBinaryInputForConverter(self): - 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): - def setUp(self): - self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES) - self.cur = self.con.cursor() - self.cur.execute("create table test(d date, ts timestamp)") - - def tearDown(self): - self.cur.close() - self.con.close() - - def CheckSqliteDate(self): - d = sqlite.Date(2004, 2, 14) - self.cur.execute("insert into test(d) values (?)", (d,)) - self.cur.execute("select d from test") - d2 = self.cur.fetchone()[0] - self.assertEqual(d, d2) - - def CheckSqliteTimestamp(self): - ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0) - self.cur.execute("insert into test(ts) values (?)", (ts,)) - self.cur.execute("select ts from test") - ts2 = self.cur.fetchone()[0] - self.assertEqual(ts, ts2) - - 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") - ts = self.cur.fetchone()[0] - self.assertEqual(type(ts), datetime.datetime) - self.assertEqual(ts.year, now.year) - - def CheckDateTimeSubSeconds(self): - ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 500000) - self.cur.execute("insert into test(ts) values (?)", (ts,)) - self.cur.execute("select ts from test") - ts2 = self.cur.fetchone()[0] - self.assertEqual(ts, ts2) - - def CheckDateTimeSubSecondsFloatingPoint(self): - ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 510241) - self.cur.execute("insert into test(ts) values (?)", (ts,)) - self.cur.execute("select ts from test") - ts2 = self.cur.fetchone()[0] - self.assertEqual(ts, ts2) - -def suite(): - sqlite_type_suite = unittest.makeSuite(SqliteTypeTests, "Check") - decltypes_type_suite = unittest.makeSuite(DeclTypesTests, "Check") - colnames_type_suite = unittest.makeSuite(ColNamesTests, "Check") - adaptation_suite = unittest.makeSuite(ObjectAdaptationTests, "Check") - bin_suite = unittest.makeSuite(BinaryConverterTests, "Check") - date_suite = unittest.makeSuite(DateTimeTests, "Check") - return unittest.TestSuite((sqlite_type_suite, decltypes_type_suite, colnames_type_suite, adaptation_suite, bin_suite, date_suite)) - -def test(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() diff --git a/contrib/tools/python/src/Lib/sqlite3/test/userfunctions.py b/contrib/tools/python/src/Lib/sqlite3/test/userfunctions.py deleted file mode 100644 index 1d191518796..00000000000 --- a/contrib/tools/python/src/Lib/sqlite3/test/userfunctions.py +++ /dev/null @@ -1,473 +0,0 @@ -#-*- coding: ISO-8859-1 -*- -# pysqlite2/test/userfunctions.py: tests for user-defined functions and -# aggregates. -# -# Copyright (C) 2005-2007 Gerhard H�ring <[email protected]> -# -# 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. - -import unittest -import sqlite3 as sqlite -from test import test_support - -def func_returntext(): - return "foo" -def func_returnunicode(): - return u"bar" -def func_returnint(): - return 42 -def func_returnfloat(): - return 3.14 -def func_returnnull(): - return None -def func_returnblob(): - with test_support.check_py3k_warnings(): - return buffer("blob") -def func_returnlonglong(): - return 1<<31 -def func_raiseexception(): - 5 // 0 - -def func_isstring(v): - return type(v) is unicode -def func_isint(v): - return type(v) is int -def func_isfloat(v): - return type(v) is float -def func_isnone(v): - return type(v) is type(None) -def func_isblob(v): - return type(v) is buffer -def func_islonglong(v): - return isinstance(v, (int, long)) and v >= 1<<31 - -class AggrNoStep: - def __init__(self): - pass - - def finalize(self): - return 1 - -class AggrNoFinalize: - def __init__(self): - pass - - def step(self, x): - pass - -class AggrExceptionInInit: - def __init__(self): - 5 // 0 - - def step(self, x): - pass - - def finalize(self): - pass - -class AggrExceptionInStep: - def __init__(self): - pass - - def step(self, x): - 5 // 0 - - def finalize(self): - return 42 - -class AggrExceptionInFinalize: - def __init__(self): - pass - - def step(self, x): - pass - - def finalize(self): - 5 // 0 - -class AggrCheckType: - def __init__(self): - self.val = None - - def step(self, whichType, val): - 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 AggrSum: - def __init__(self): - self.val = 0.0 - - def step(self, val): - self.val += val - - def finalize(self): - return self.val - -class FunctionTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - - self.con.create_function("returntext", 0, func_returntext) - self.con.create_function("returnunicode", 0, func_returnunicode) - self.con.create_function("returnint", 0, func_returnint) - self.con.create_function("returnfloat", 0, func_returnfloat) - self.con.create_function("returnnull", 0, func_returnnull) - self.con.create_function("returnblob", 0, func_returnblob) - self.con.create_function("returnlonglong", 0, func_returnlonglong) - self.con.create_function("raiseexception", 0, func_raiseexception) - - self.con.create_function("isstring", 1, func_isstring) - self.con.create_function("isint", 1, func_isint) - self.con.create_function("isfloat", 1, func_isfloat) - self.con.create_function("isnone", 1, func_isnone) - self.con.create_function("isblob", 1, func_isblob) - self.con.create_function("islonglong", 1, func_islonglong) - - def tearDown(self): - self.con.close() - - def CheckFuncErrorOnCreate(self): - 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(): - def f(): - return 1 - return f - f = getfunc() - globals()["foo"] = f - # self.con.create_function("reftest", 0, getfunc()) - self.con.create_function("reftest", 0, f) - cur = self.con.cursor() - cur.execute("select reftest()") - - def CheckFuncReturnText(self): - cur = self.con.cursor() - cur.execute("select returntext()") - val = cur.fetchone()[0] - 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), unicode) - self.assertEqual(val, u"bar") - - def CheckFuncReturnInt(self): - cur = self.con.cursor() - cur.execute("select returnint()") - val = cur.fetchone()[0] - self.assertEqual(type(val), int) - self.assertEqual(val, 42) - - def CheckFuncReturnFloat(self): - cur = self.con.cursor() - cur.execute("select returnfloat()") - val = cur.fetchone()[0] - self.assertEqual(type(val), float) - if val < 3.139 or val > 3.141: - self.fail("wrong value") - - def CheckFuncReturnNull(self): - cur = self.con.cursor() - cur.execute("select returnnull()") - val = cur.fetchone()[0] - self.assertEqual(type(val), type(None)) - self.assertEqual(val, None) - - def CheckFuncReturnBlob(self): - cur = self.con.cursor() - cur.execute("select returnblob()") - val = cur.fetchone()[0] - with test_support.check_py3k_warnings(): - self.assertEqual(type(val), buffer) - self.assertEqual(val, buffer("blob")) - - def CheckFuncReturnLongLong(self): - cur = self.con.cursor() - cur.execute("select returnlonglong()") - val = cur.fetchone()[0] - self.assertEqual(val, 1<<31) - - def CheckFuncException(self): - cur = self.con.cursor() - try: - cur.execute("select raiseexception()") - cur.fetchone() - 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() - cur.execute("select isstring(?)", ("foo",)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckParamInt(self): - cur = self.con.cursor() - cur.execute("select isint(?)", (42,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckParamFloat(self): - cur = self.con.cursor() - cur.execute("select isfloat(?)", (3.14,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckParamNone(self): - cur = self.con.cursor() - cur.execute("select isnone(?)", (None,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckParamBlob(self): - cur = self.con.cursor() - with test_support.check_py3k_warnings(): - cur.execute("select isblob(?)", (buffer("blob"),)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckParamLongLong(self): - cur = self.con.cursor() - cur.execute("select islonglong(?)", (1<<42,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - -class AggregateTests(unittest.TestCase): - def setUp(self): - self.con = sqlite.connect(":memory:") - cur = self.con.cursor() - cur.execute(""" - create table test( - t text, - i integer, - f float, - n, - 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) - self.con.create_aggregate("excInit", 1, AggrExceptionInInit) - 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("mysum", 1, AggrSum) - - def tearDown(self): - #self.cur.close() - #self.con.close() - pass - - def CheckAggrErrorOnCreate(self): - 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() - try: - cur.execute("select nostep(t) from test") - 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() - try: - cur.execute("select nofinalize(t) from test") - val = cur.fetchone()[0] - 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() - try: - cur.execute("select excInit(t) from test") - val = cur.fetchone()[0] - 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() - try: - cur.execute("select excStep(t) from test") - val = cur.fetchone()[0] - 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() - try: - cur.execute("select excFinalize(t) from test") - val = cur.fetchone()[0] - 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() - cur.execute("select checkType('str', ?)", ("foo",)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckAggrCheckParamInt(self): - cur = self.con.cursor() - cur.execute("select checkType('int', ?)", (42,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckAggrCheckParamFloat(self): - cur = self.con.cursor() - cur.execute("select checkType('float', ?)", (3.14,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckAggrCheckParamNone(self): - cur = self.con.cursor() - cur.execute("select checkType('None', ?)", (None,)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckAggrCheckParamBlob(self): - cur = self.con.cursor() - with test_support.check_py3k_warnings(): - cur.execute("select checkType('blob', ?)", (buffer("blob"),)) - val = cur.fetchone()[0] - self.assertEqual(val, 1) - - def CheckAggrCheckAggrSum(self): - cur = self.con.cursor() - cur.execute("delete from test") - cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)]) - cur.execute("select mysum(i) from test") - val = cur.fetchone()[0] - self.assertEqual(val, 60) - -class AuthorizerTests(unittest.TestCase): - @staticmethod - def authorizer_cb(action, arg1, arg2, dbname, source): - if action != sqlite.SQLITE_SELECT: - return sqlite.SQLITE_DENY - if arg2 == 'c2' or arg1 == 't2': - return sqlite.SQLITE_DENY - return sqlite.SQLITE_OK - - def setUp(self): - self.con = sqlite.connect(":memory:") - self.con.executescript(""" - create table t1 (c1, c2); - create table t2 (c1, c2); - insert into t1 (c1, c2) values (1, 2); - insert into t2 (c1, c2) values (4, 5); - """) - - # For our security test: - self.con.execute("select c2 from t2") - - self.con.set_authorizer(self.authorizer_cb) - - def tearDown(self): - pass - - def test_table_access(self): - try: - self.con.execute("select * from t2") - 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): - try: - self.con.execute("select c2 from t1") - 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 - def authorizer_cb(action, arg1, arg2, dbname, source): - if action != sqlite.SQLITE_SELECT: - raise ValueError - if arg2 == 'c2' or arg1 == 't2': - raise ValueError - return sqlite.SQLITE_OK - -class AuthorizerIllegalTypeTests(AuthorizerTests): - @staticmethod - def authorizer_cb(action, arg1, arg2, dbname, source): - if action != sqlite.SQLITE_SELECT: - return 0.0 - if arg2 == 'c2' or arg1 == 't2': - return 0.0 - return sqlite.SQLITE_OK - -class AuthorizerLargeIntegerTests(AuthorizerTests): - @staticmethod - def authorizer_cb(action, arg1, arg2, dbname, source): - if action != sqlite.SQLITE_SELECT: - return 2**32 - if arg2 == 'c2' or arg1 == 't2': - return 2**32 - return sqlite.SQLITE_OK - - -def suite(): - function_suite = unittest.makeSuite(FunctionTests, "Check") - aggregate_suite = unittest.makeSuite(AggregateTests, "Check") - authorizer_suite = unittest.makeSuite(AuthorizerTests) - return unittest.TestSuite(( - function_suite, - aggregate_suite, - authorizer_suite, - unittest.makeSuite(AuthorizerRaiseExceptionTests), - unittest.makeSuite(AuthorizerIllegalTypeTests), - unittest.makeSuite(AuthorizerLargeIntegerTests), - )) - -def test(): - runner = unittest.TextTestRunner() - runner.run(suite()) - -if __name__ == "__main__": - test() |