| #-*- coding: ISO-8859-1 -*- |
| # pysqlite2/test/userfunctions.py: tests for user-defined functions and |
| # aggregates. |
| # |
| # Copyright (C) 2005-2007 Gerhard Häring <gh@ghaering.de> |
| # |
| # This file is part of pysqlite. |
| # |
| # This software is provided 'as-is', without any express or implied |
| # warranty. In no event will the authors be held liable for any damages |
| # arising from the use of this software. |
| # |
| # Permission is granted to anyone to use this software for any purpose, |
| # including commercial applications, and to alter it and redistribute it |
| # freely, subject to the following restrictions: |
| # |
| # 1. The origin of this software must not be misrepresented; you must not |
| # claim that you wrote the original software. If you use this software |
| # in a product, an acknowledgment in the product documentation would be |
| # appreciated but is not required. |
| # 2. Altered source versions must be plainly marked as such, and must not be |
| # misrepresented as being the original software. |
| # 3. This notice may not be removed or altered from any source distribution. |
| |
| 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() |
