diff options
Diffstat (limited to 'Lib/sqlite3/test')
| -rw-r--r-- | Lib/sqlite3/test/hooks.py | 2 | ||||
| -rw-r--r-- | Lib/sqlite3/test/regression.py | 8 | ||||
| -rw-r--r-- | Lib/sqlite3/test/types.py | 52 | ||||
| -rw-r--r-- | Lib/sqlite3/test/userfunctions.py | 107 |
4 files changed, 130 insertions, 39 deletions
diff --git a/Lib/sqlite3/test/hooks.py b/Lib/sqlite3/test/hooks.py index b10b3ef..761bdaa 100644 --- a/Lib/sqlite3/test/hooks.py +++ b/Lib/sqlite3/test/hooks.py @@ -48,6 +48,8 @@ class CollationTests(unittest.TestCase): pass 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) diff --git a/Lib/sqlite3/test/regression.py b/Lib/sqlite3/test/regression.py index 25e4b63..c8733b9 100644 --- a/Lib/sqlite3/test/regression.py +++ b/Lib/sqlite3/test/regression.py @@ -61,6 +61,14 @@ class RegressionTests(unittest.TestCase): con.rollback() + def CheckColumnNameWithSpaces(self): + cur = self.con.cursor() + cur.execute('select 1 as "foo bar [datetime]"') + self.failUnlessEqual(cur.description[0][0], "foo bar") + + cur.execute('select 1 as "foo baz"') + self.failUnlessEqual(cur.description[0][0], "foo baz") + def suite(): regression_suite = unittest.makeSuite(RegressionTests, "Check") return unittest.TestSuite((regression_suite,)) diff --git a/Lib/sqlite3/test/types.py b/Lib/sqlite3/test/types.py index e49f7dd..8da5722 100644 --- a/Lib/sqlite3/test/types.py +++ b/Lib/sqlite3/test/types.py @@ -21,7 +21,7 @@ # misrepresented as being the original software. # 3. This notice may not be removed or altered from any source distribution. -import datetime +import bz2, datetime import unittest import sqlite3 as sqlite @@ -101,16 +101,16 @@ class DeclTypesTests(unittest.TestCase): self.cur.execute("create table test(i int, s str, f float, b bool, u unicode, foo foo, bin blob)") # override float, make them always return the same number - sqlite.converters["float"] = lambda x: 47.2 + 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["BOOL"] = lambda x: bool(int(x)) + sqlite.converters["FOO"] = DeclTypesTests.Foo def tearDown(self): - del sqlite.converters["float"] - del sqlite.converters["bool"] - del sqlite.converters["foo"] + del sqlite.converters["FLOAT"] + del sqlite.converters["BOOL"] + del sqlite.converters["FOO"] self.cur.close() self.con.close() @@ -208,14 +208,14 @@ class ColNamesTests(unittest.TestCase): 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["FOO"] = lambda x: "[%s]" % x + sqlite.converters["BAR"] = lambda x: "<%s>" % x + sqlite.converters["EXC"] = lambda x: 5/0 def tearDown(self): - del sqlite.converters["foo"] - del sqlite.converters["bar"] - del sqlite.converters["exc"] + del sqlite.converters["FOO"] + del sqlite.converters["BAR"] + del sqlite.converters["EXC"] self.cur.close() self.con.close() @@ -231,12 +231,6 @@ class ColNamesTests(unittest.TestCase): val = self.cur.fetchone()[0] self.failUnlessEqual(val, None) - def CheckExc(self): - # Exceptions in type converters result in returned Nones - self.cur.execute('select 5 as "x [exc]"') - val = self.cur.fetchone()[0] - self.failUnlessEqual(val, None) - def CheckColName(self): self.cur.execute("insert into test(x) values (?)", ("xxx",)) self.cur.execute('select x as "x [bar]" from test') @@ -279,6 +273,23 @@ class ObjectAdaptationTests(unittest.TestCase): val = self.cur.fetchone()[0] self.failUnlessEqual(type(val), float) +class BinaryConverterTests(unittest.TestCase): + def convert(s): + return bz2.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 + result = self.con.execute('select ? as "x [bin]"', (buffer(bz2.compress(testdata)),)).fetchone()[0] + self.failUnlessEqual(testdata, result) + class DateTimeTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_DECLTYPES) @@ -328,8 +339,9 @@ def suite(): 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, date_suite)) + return unittest.TestSuite((sqlite_type_suite, decltypes_type_suite, colnames_type_suite, adaptation_suite, bin_suite, date_suite)) def test(): runner = unittest.TextTestRunner() diff --git a/Lib/sqlite3/test/userfunctions.py b/Lib/sqlite3/test/userfunctions.py index 78656e7..31bf289 100644 --- a/Lib/sqlite3/test/userfunctions.py +++ b/Lib/sqlite3/test/userfunctions.py @@ -55,6 +55,9 @@ class AggrNoStep: def __init__(self): pass + def finalize(self): + return 1 + class AggrNoFinalize: def __init__(self): pass @@ -144,9 +147,12 @@ class FunctionTests(unittest.TestCase): def CheckFuncRefCount(self): def getfunc(): def f(): - return val + return 1 return f - self.con.create_function("reftest", 0, getfunc()) + 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()") @@ -195,9 +201,12 @@ class FunctionTests(unittest.TestCase): def CheckFuncException(self): cur = self.con.cursor() - cur.execute("select raiseexception()") - val = cur.fetchone()[0] - self.failUnlessEqual(val, None) + try: + cur.execute("select raiseexception()") + cur.fetchone() + self.fail("should have raised OperationalError") + except sqlite.OperationalError, e: + self.failUnlessEqual(e.args[0], 'user-defined function raised exception') def CheckParamString(self): cur = self.con.cursor() @@ -267,31 +276,47 @@ class AggregateTests(unittest.TestCase): def CheckAggrNoStep(self): cur = self.con.cursor() - cur.execute("select nostep(t) from test") + try: + cur.execute("select nostep(t) from test") + self.fail("should have raised an AttributeError") + except AttributeError, e: + self.failUnlessEqual(e.args[0], "AggrNoStep instance has no attribute 'step'") def CheckAggrNoFinalize(self): cur = self.con.cursor() - cur.execute("select nofinalize(t) from test") - val = cur.fetchone()[0] - self.failUnlessEqual(val, None) + try: + cur.execute("select nofinalize(t) from test") + val = cur.fetchone()[0] + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.failUnlessEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error") def CheckAggrExceptionInInit(self): cur = self.con.cursor() - cur.execute("select excInit(t) from test") - val = cur.fetchone()[0] - self.failUnlessEqual(val, None) + try: + cur.execute("select excInit(t) from test") + val = cur.fetchone()[0] + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.failUnlessEqual(e.args[0], "user-defined aggregate's '__init__' method raised error") def CheckAggrExceptionInStep(self): cur = self.con.cursor() - cur.execute("select excStep(t) from test") - val = cur.fetchone()[0] - self.failUnlessEqual(val, 42) + try: + cur.execute("select excStep(t) from test") + val = cur.fetchone()[0] + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.failUnlessEqual(e.args[0], "user-defined aggregate's 'step' method raised error") def CheckAggrExceptionInFinalize(self): cur = self.con.cursor() - cur.execute("select excFinalize(t) from test") - val = cur.fetchone()[0] - self.failUnlessEqual(val, None) + try: + cur.execute("select excFinalize(t) from test") + val = cur.fetchone()[0] + self.fail("should have raised an OperationalError") + except sqlite.OperationalError, e: + self.failUnlessEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error") def CheckAggrCheckParamStr(self): cur = self.con.cursor() @@ -331,10 +356,54 @@ class AggregateTests(unittest.TestCase): val = cur.fetchone()[0] self.failUnlessEqual(val, 60) +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 + +class AuthorizerTests(unittest.TestCase): + 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(authorizer_cb) + + def tearDown(self): + pass + + def CheckTableAccess(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 CheckColumnAccess(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") + def suite(): function_suite = unittest.makeSuite(FunctionTests, "Check") aggregate_suite = unittest.makeSuite(AggregateTests, "Check") - return unittest.TestSuite((function_suite, aggregate_suite)) + authorizer_suite = unittest.makeSuite(AuthorizerTests, "Check") + return unittest.TestSuite((function_suite, aggregate_suite, authorizer_suite)) def test(): runner = unittest.TextTestRunner() |