userfunctions.py - Android社区 - https://www.androidos.net.cn/

#-*- 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()

普通文本 | 474行 | 14.95 KB

登录后可以享受更多权益