~ubuntu-branches/ubuntu/lucid/python-apsw/lucid

# APSW test suite - runs under both Python 2 and Python 3 hence a lot

# of wierd constructs to be simultaneously compatible with both.

# (2to3 is not used).

import sys

 

write=sys.stdout.write

write("                Python "+sys.executable+" "+str(sys.version_info)+"\n")

write("Testing with APSW file "+apsw.__file__+"\n")

write("          APSW version "+apsw.apswversion()+"\n")

write("    SQLite lib version "+apsw.sqlitelibversion()+"\n")

write("SQLite headers version "+str(apsw.SQLITE_VERSION_NUMBER)+"\n")

 

if [int(x) for x in apsw.sqlitelibversion().split(".")]<[3,5,9]:

    write("You are using an earlier version of SQLite than recommended\n")

 

sys.stdout.flush()

 

py3=sys.version_info>=(3,0)

if py3:

    import queue as Queue

else:

    import Queue

import re

import gc

 

# Unicode string/bytes prefix

if py3:

    UPREFIX=""

    BPREFIX="b"

else:

    UPREFIX="u"

    BPREFIX=""

# Return a unicode string - x should have been raw

def u(x):

    return eval(UPREFIX+"'''"+x+"'''")

 

# Return a bytes (py3)/buffer (py2) - x should have been raw

def b(x):

    if py3:

        return eval(BPREFIX+"'''"+x+"'''")

    return eval("buffer('''"+x+"''')")

 

# return bytes (py3)/string (py2) - x should have been raw

# Use this instead of b for file i/o data as py2 uses str

def BYTES(x):

    if py3: return b(x)

    return eval("'''"+x+"'''")

 

def l(x):

    if py3: return eval(x)

    return eval(x+"L")

 

# Various py3 things

if py3:

    long=int

 

if not py3:

    # emulation of py3 next built-in.  In py2 the iternext method is exposed

    # as object.next() but in py3 it is object.__next__().

    def next(iterator, *args):

        if len(args)>1:

            raise TypeError("bad args")

        try:

            return iterator.next()

        except StopIteration:

            if len(args):

                return args[0]

            raise

            

    for i in range(howmany):

# An instance of this class is used to get the -1 return value to the

# C api PyObject_IsTrue

class BadIsTrue(int):

    # py2 does this

    def __nonzero__(self):

        1/0

    # py3 does this

    def __bool__(self):

        1/0

 

        if py3:

            self.set_daemon(True)

        else:

            self.setDaemon(True)

        self.started=False

 

    def start(self):

        if not self.started:

            self.started=True

            threading.Thread.start(self)

            if py3:

                exec("raise res[1].with_traceback(res[2])")

            else:

                exec("raise res[0], res[1], res[2]")

        'collationneeded': 1,

        'execute': 1,

        'executemany': 2,

        'setexectrace': 1,

        'setrowtrace': 1,

        }

 

    blob_nargs={

        'write': 1,

        'read': 1,

        'seek': 2

        }

        gc.collect()

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from ["+tablename+"]"))[0], 0)

        "Test connection opening"

        # bad keyword arg

        if not py3:

            self.assertRaises(UnicodeDecodeError, apsw.Connection, "\xef\x22\xd3\x9e")

        # bad open flags can't be tested as sqlite accepts them all - ticket #3037

        # self.assertRaises(apsw.CantOpenError, apsw.Connection, "<non-existent-file>", flags=65535)

 

        # bad vfs

        self.assertRaises(TypeError, apsw.Connection, "foo", vfs=3, flags=-1)

        self.assertRaises(apsw.SQLError, apsw.Connection, "foo", vfs="jhjkds", flags=-1)

        vals=[ [1], [None], [math.pi], ["kjkljkljl"], [u(r"\u1234\u345432432423423kjgjklhdfgkjhsdfjkghdfjskh")],

               [b(r"78696ghgjhgjhkgjkhgjhg\xfe\xdf")] ]

        c.executemany("insert into foo values(?)", vals)

        for i in range(MEMLEAKITERATIONS):

            db.collationneeded(lambda x: 4)

            for i in range(100):

            (u(r"($\N{LATIN SMALL LETTER E WITH CIRCUMFLEX},:\N{LATIN SMALL LETTER A WITH TILDE},$\N{LATIN SMALL LETTER O WITH DIAERESIS})"), (1,2,3)),

            (u(r"($\N{LATIN SMALL LETTER E WITH CIRCUMFLEX},:\N{LATIN SMALL LETTER A WITH TILDE},$\N{LATIN SMALL LETTER O WITH DIAERESIS})"),

             {u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}"): 1,

              u(r"\N{LATIN SMALL LETTER A WITH TILDE}"): 2,

              u(r"\N{LATIN SMALL LETTER O WITH DIAERESIS}"): 3,})

            )

            self.failUnlessEqual(next(c.execute("select * from foo")), (1,2,3))

        self.failUnlessEqual((1,None,3), next(c.execute("select * from foo")))

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from foo where x=99"))[0], 1)

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from foo where x=102"))[0], 1)

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from foo where x=100"))[0], 1)

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from foo where x=1000"))[0], 0)

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from foo where x=101"))[0], 1)

        self.failUnlessEqual(next(self.db.cursor().execute("select count(*) from foo where x=105"))[0], 0)

            self.failUnlessEqual(next(c.execute("select * from foo where x=?",(x,))), (x,y,z))

        self.failUnlessEqual(next(c.execute("select count(*) from foo"))[0], 2)

        self.failUnlessEqual(next(c.execute("select count(*) from foo"))[0], 2)

        # shouldn't be able to manually create

        self.assertRaises(TypeError, type(c))

        

        self.failUnlessEqual(3, next(c.execute(u("select 3")))[0])

        if not py3:

            self.assertRaises(UnicodeDecodeError, c.execute, "\x99\xaa\xbb\xcc")

            (u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}\N{LATIN SMALL LETTER A WITH TILDE}"),

             u(r"\N{LATIN SMALL LETTER O WITH DIAERESIS}\N{LATIN SMALL LETTER U WITH CIRCUMFLEX}")),

        self.assertRaises(StopIteration, lambda xx=0: next(c))

        self.assertRaises(StopIteration, lambda xx=0: next(c))

              u(r"a \u1234 unicode \ufe54 string \u0089"),  # simple unicode string

              u(r"\N{BLACK STAR} \N{WHITE STAR} \N{LIGHTNING} \N{COMET} "), # funky unicode

              u(r"\N{MUSICAL SYMBOL G CLEF}"), # http://www.cmlenz.net/archives/2008/07/the-truth-about-unicode-in-python

              long(2147483648),  # detail)

              long(-2147483648),

              long(2147483999),

              long(-2147483999),

              992147483999,

              -992147483999,

              9223372036854775807,

              -9223372036854775808,

              b("a set of bytes"),      # bag of bytes initialised from a string, but don't confuse it with a

              b("".join(["\\x%02x" % (x,) for x in range(256)])), # string

              b("".join(["\\x%02x" % (x,) for x in range(256)])*20000),  # non-trivial size

        self.assertRaises(OverflowError, c.execute, "insert into foo values(9999,?)", (922337203685477580799,))

        self.assertRaises(OverflowError, c.execute, "insert into foo values(9999,?)", (-922337203685477580799,))

        if not py3: # py3 - all strings are unicode so not a problem

            self.assertRaises(UnicodeDecodeError, c.execute, "insert into foo values(9999,?)", ("\xfe\xfb\x80\x92",))

        self.failUnlessEqual(0, next(c.execute("select count(*) from foo where row=9999"))[0])

 

        # playing with default encoding and non-ascii strings - py2 only

        if py3: return

        enc=sys.getdefaultencoding()

        reload(sys) # gets setdefaultencoding function back

        try:

            for v in vals:

                if type(v)!=unicode:

                    continue

                def encoding(*args):

                    return v.encode("utf8") # returns as str not unicode

                self.db.createscalarfunction("encoding", encoding)

                sys.setdefaultencoding("utf8")

                for row in c.execute("select encoding(3)"):

                    self.failUnlessEqual(v, row[0])

                c.execute("insert into foo values(1234,?)", (v.encode("utf8"),))

                for row in c.execute("select x from foo where rowid="+str(self.db.last_insert_rowid())):

                    self.failUnlessEqual(v, row[0])

        finally:

            sys.setdefaultencoding(enc)

        retval=apsw.SQLITE_DENY

                return retval

        for val in apsw.SQLITE_DENY, long(apsw.SQLITE_DENY), 0x800276889000212112:

            retval=val

            if val<100:

                self.assertRaises(apsw.AuthError, c.execute, "create table privatetwo(x)")

            else:

                self.assertRaises(OverflowError, c.execute, "create table privatetwo(x)")

        count=next(c.execute("select count(*) from one"))[0]

        self.failUnlessEqual(count, next(c.execute("select count(*) from one"))[0])

        self.failUnlessEqual(count, next(c.execute("select count(*) from one"))[0])

        self.failUnlessEqual(3, next(c.execute("select max(x) from two"))[0])

        def tracefunc(*args):

            return BadIsTrue()

        c.setexectrace(tracefunc)

        self.assertRaises(ZeroDivisionError, c.execute, "select max(x) from two")

        self.failUnlessEqual(next(c.execute("select * from foo")), vals)

        self.failUnlessEqual(next(c.execute("select * from foo")), tuple([7]*len(vals)))

        self.failUnlessEqual(next(c.execute("select * from foo")), (7,))

        self.failUnlessEqual(next(c.execute("select * from foo")), vals)

        self.failUnlessEqual(next(c.execute("select * from foo")), vals)

        self.assertRaises(TypeError, self.db.createscalarfunction, u(r"twelve\N{BLACK STAR}"), ilove7) # must be ascii

            self.failUnlessEqual( (7,), next(c.execute("select seven(x,y,z) from foo where x=?", (i,))))

        if not py3:

            self.assertRaises(UnicodeDecodeError, c.execute, "select ilove8bit(*) from foo")

        # coverage

        def bad(*args): 1/0

        self.db.createscalarfunction("bad", bad)

        self.assertRaises(ZeroDivisionError, c.execute, "select bad(3)+bad(4)")

        # turn a blob into a string to fail python utf8 conversion

        self.assertRaises(UnicodeDecodeError, c.execute, "select bad(cast (x'fffffcfb9208' as TEXT))")

        

        self.assertRaises(TypeError, self.db.createaggregatefunction, u(r"twelve\N{BLACK STAR}"), 12) # must be ascii

        v=next(c.execute("select longest(x,y,z) from foo"))[0]

 

        # create a whole bunch to check they are freed

        for i in range(1024):

            self.db.createcollation("x"*i, lambda x,y: i)

        for ii in range(1024):

            self.db.createcollation("x"*ii, lambda x,y: ii)

 

                return -1 # return an int

            if values[0]>values[1]: 

                return long(1) # and a long

        uni=u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}")

        valsrev=valsrev[1:]+valsrev[:1] # except one out of order

        try:

            c.execute("select x from foo order by x collate strnum")

        except (apsw.SQLError,apsw.SchemaChangeError):

            # exception returned could be either depending on if

            # statementcache is used.  See SQLite ticket 2158

            pass

 

        # collation needed testing

        self.assertRaises(TypeError, self.db.collationneeded, 12)

        def cn1(): pass

        def cn2(x, y): 1/0

        def cn3(x, y):

            self.assert_(x is self.db)

            self.failUnlessEqual(y, "strnum")

            self.db.createcollation("strnum", strnumcollate)

 

        self.db.collationneeded(cn1)

        try:

            for _ in c.execute("select x from foo order by x collate strnum"): pass

        except TypeError:

            pass

        self.db.collationneeded(cn2)

        try:

            for _ in c.execute("select x from foo order by x collate strnum"): pass

        except ZeroDivisionError:

            pass

        self.db.collationneeded(cn3)

        for _ in c.execute("select x from foo order by x collate strnum"): pass

        self.db.collationneeded(None)

        self.db.createcollation("strnum", None)

        

        # check it really has gone

        try:

            c.execute("select x from foo order by x collate strnum")

        except (apsw.SQLError,apsw.SchemaChangeError):

            # exception returned could be either depending on if

            # statementcache is used.  See SQLite ticket 2158

            pass

        

        next(c.execute("select max(x) from foo"))

        self.failUnlessEqual(0, next(c.execute("select count(*) from foo where x=-10"))[0])

        def ph():

            return BadIsTrue()

        self.db.setprogresshandler(ph, 1)

        self.assertRaises(ZeroDivisionError, c.execute, "update foo set x=-10")

        for i in range(100):

        for i in range(100):

        self.failUnlessEqual(True, self.db.complete(u(r"select '\u9494\ua7a7';")))

        if not py3:

            self.assertRaises(UnicodeDecodeError, self.db.complete, "select '\x94\xa7';")

                self.fail("Transaction wasn't exclusive")

        TIMEOUT=3 # seconds, must be integer as sqlite can round down to nearest second anyway

        def bh(*args):

            return BadIsTrue()

        db2=apsw.Connection("testdb")

        c=self.db.cursor()

        c2=db2.cursor()

        c2.execute("begin exclusive")

        self.db.setbusyhandler(bh)

        self.assertRaises(ZeroDivisionError, c.execute, "begin immediate ; select * from foo")

        del c

        del c2

        db2.close()        

 

        self.assertEqual(1, next(cur2.execute("select count(*) from test where x=123"))[0])

        def ch():

            return BadIsTrue()

        self.db.setcommithook(ch)

        self.assertRaises(ZeroDivisionError, c.execute, "insert into foo values(?)", (1,))

        

        # improve code coverage

        c.execute("create table bar(x,y); insert into bar values(1,2); insert into bar values(3,4)")

        def uh(*args):

            1/0

        self.db.setupdatehook(uh)

        self.assertRaises(ZeroDivisionError, c.execute, "insert into foo select * from bar")

        self.db.setupdatehook(None)

        

        self.assertEqual(1, next(c.execute("select count(*) from foo where x=1000"))[0])

        # coverage

        wasrun=[False]

        def profile(*args):

            wasrun[0]=True

        def uh(*args): 1/0

        self.db.setprofile(profile)

        self.db.setupdatehook(uh)

        self.assertRaises(ZeroDivisionError, c.execute, "insert into foo values(3)")

        self.failUnlessEqual(wasrun[0], False)

        self.db.setprofile(None)

        self.db.setupdatehook(None)

        # We used to require all operations on a connection happen in

        # the same thread.  Now they can happen in any thread, so we

        # ensure that inuse errors are detected by doing a long

        # running operation in one thread.

        c.execute("create table foo(x);begin;")

        c.executemany("insert into foo values(?)", randomintegers(100000))

        c.execute("commit")

 

        vals={"stop": False,

              "raised": False}

        def wt():

            try:

                while not vals["stop"]:

                    c.execute("select min(max(x-1+x),min(x-1+x)) from foo")

            except apsw.ThreadingViolationError:

                vals["raised"]=True

                vals["stop"]=True

                

        t=ThreadRunner(wt)

        t.start()

        # ensure thread t has started

        time.sleep(0.1)

        b4=time.time()

        # try to get a threadingviolation for 30 seconds

        try:

            try:

                while not vals["stop"] and time.time()-b4<30:

                    c.execute("select * from foo")

            except apsw.ThreadingViolationError:

                vals["stop"]=True

                vals["raised"]=True

        finally:

            vals["stop"]=True

        t.go()

        self.assertEqual(vals["raised"], True)

              u(r"a \u1234 unicode \ufe54 string \u0089"),

              u(r"a \u1234 unicode \ufe54 string \u0089\0and some text"),

              u(r"\N{BLACK STAR} \N{WHITE STAR} \N{LIGHTNING} \N{COMET}\0more\0than you\0can handle"),

              u(r"\N{BLACK STAR} \N{WHITE STAR} \N{LIGHTNING} \N{COMET}\0\0\0\0\0sequences\0\0\0of them"))

 

        # See http://www.sqlite.org/cvstrac/tktview?tn=3056

        if True: # [int(x) for x in apsw.sqlitelibversion().split(".")]<[3,5,8]:

            vals=vals+(

              "a simple string\0",

              u(r"a \u1234 unicode \ufe54 string \u0089\0"),

              )

        if not py3:

            self.assertRaises(UnicodeDecodeError, c.execute, "insert into foo values(9000,?)", ("a simple string\0with a null and \xfe\xfb\x80\x92",))

        # check execute

        for v in vals:

            self.failUnlessEqual(v, next(c.execute("select ?", (v,)))[0])

            # nulls not allowed in main query string, so lets check the other bits (unicode etc)

            v2=v.replace("\0", " zero ")

            self.failUnlessEqual(v2, next(c.execute("select '%s'" % (v2,)))[0])

 

        # unicode issues

        if not py3:

            self.assertRaises(UnicodeDecodeError, self.db.loadextension, "\xa7\x94")

        # they need to be enabled first (off by default)

        self.assertRaises(ZeroDivisionError, self.db.enableloadextension, BadIsTrue())

        self.failUnlessEqual(1, next(c.execute("select half(2)"))[0])

        self.failUnlessEqual(4, next(c.execute("select doubleup(2)"))[0])

        

 

    def testMakeSqliteMsgFromException(self):

        "Test C function that converts exception into SQLite error code"

        class Source:

            def Create1(self, *args):

                e=apsw.IOError()

                e.extendedresult=apsw.SQLITE_IOERR_BLOCKED

                raise e

 

            def Create2(self, *args):

                e=apsw.IOError()

                e.extendedresult=long(apsw.SQLITE_IOERR_BLOCKED)

                raise e

 

            def Create3(self, *args):

                e=apsw.IOError()

                e.extendedresult=(long("0x80")<<32)+apsw.SQLITE_IOERR_BLOCKED # bigger than 32 bits

                raise e

 

            

        self.db.createmodule("foo", Source())

        for i in "1", "2", "3":

            Source.Create=getattr(Source, "Create"+i)

            try:

                self.db.cursor().execute("create virtual table vt using foo()")

                1/0

            except:

                klass,value,tb=sys.exc_info()

            # check types and values

            if i=="3":

                self.failUnlessEqual(klass, ValueError)

                continue

            self.failUnlessEqual(klass, apsw.IOError)

            self.assert_(isinstance(value, apsw.IOError))

            # python 2.3 totally messes up on long<->int and signed conversions causing the test to fail

            # but the code is fine - so just ignore rest of test for py2.3

            if sys.version_info<(2,4):

                return

 

            self.failUnlessEqual(value.extendedresult& ((long(0xffff)<<16)|long(0xffff)), apsw.SQLITE_IOERR_BLOCKED)

            # walk the stack to verify that C level value is correct

            found=False

            while tb:

                frame=tb.tb_frame

                if frame.f_code.co_name=="resetcursor":

                    self.failUnlessEqual(frame.f_locals["res"], apsw.SQLITE_IOERR_BLOCKED|apsw.SQLITE_IOERR)

                    found=True

                    break

                tb=tb.tb_next

            self.assertEqual(found, True)

            ( 1,           "Joe Smith",    1.1, u(r"\u00f6\u1234"), "foo"),

            ( 6000000000,  "Road Runner", -7.3, u(r"\u00f6\u1235"), "foo"),

            ( 77,          "Fred",           0, u(r"\u00f6\u1236"), "foo"),

        dataschema="create table this_should_be_ignored"+str(data[0][1:])

        for i in range(20):

            self.db.createmodule("x"*i, lambda x: i)

        for ii in range(20):

            self.db.createmodule("x"*ii, lambda x: ii)

 

        # If shared cache is enabled then vtable creation is supposed to fail

        # See http://www.sqlite.org/cvstrac/tktview?tn=3144

        try:

            apsw.enablesharedcache(True)

            db=apsw.Connection("testdb2")

            db.createmodule("y", lambda x: 2)

        finally:

            apsw.enablesharedcache(False)

 

                    raise ValueError("Create arguments are not correct.  Expected "+str(self.expectargs)+" but got "+str(args[1:]))

                raise Exception(u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}\N{LATIN SMALL LETTER A WITH TILDE}\N{LATIN SMALL LETTER O WITH DIAERESIS}"))

        uni=u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}\N{LATIN SMALL LETTER A WITH TILDE}\N{LATIN SMALL LETTER O WITH DIAERESIS}")

        self.db.createmodule("testmod1dash1", Source("testmod1dash1", "main", uni, "1", '"'+uni+'"'))

        self.assertRaises(ZeroDivisionError, cur.execute, u('create virtual table %s using testmod1dash1(1,"%s")') % (uni, uni))

                return (None,12,u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}"), "anything", "bad")

 

            def BestIndex5(self, constraints, orderbys):

                # unicode error

                return (None, None, "\xde\xad\xbe\xef")

 

            def BestIndex6(self, constraints, orderbys):

                return ( (0, 1, (2, BadIsTrue()), 3, 4), )

 

            def BestIndex7(self, constraints, orderbys):

                return (None, long(77), "foo", BadIsTrue(), 99)

                retval=( [long(4),(3,True),[long(2),False],1, (0, False)], 997, u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}"), False, 99)[:self._bestindexreturn]

                return -922337203685477580799 # too big

                return 9223372036854775807 # ok

            def Rename1(self, too, many, args):

                1/0

 

            def Rename2(self, x):

                1/0

 

            def Rename3(self, x):

                return ["thisshouldbeignored"*25, [1]]

 

                assert idxstr==u(r"\N{LATIN SMALL LETTER E WITH CIRCUMFLEX}")

            def Eof3(self):

                return BadIsTrue()

 

        if not py3:

            VTable.BestIndex=VTable.BestIndex5

            self.assertRaises(UnicodeDecodeError, cur.execute, allconstraints)

        VTable.BestIndex=VTable.BestIndex6

        self.assertRaises(ZeroDivisionError, cur.execute, allconstraints)

        VTable.BestIndex=VTable.BestIndex7

        self.assertRaises(ZeroDivisionError, cur.execute, allconstraints)

        Cursor.Eof=Cursor.Eof3

        self.assertRaises(ZeroDivisionError,cur.execute, allconstraints)

        self.failUnlessEqual(self.db.last_insert_rowid(), 9223372036854775807)

        # rename

        sql="alter table foo rename to bar"

        VTable.Rename=VTable.Rename1

        self.assertRaises(TypeError, cur.execute, sql)

        VTable.Rename=VTable.Rename2

        self.assertRaises(ZeroDivisionError, cur.execute, sql)

        VTable.Rename=VTable.Rename3

        # this is to catch memory leaks

        cur.execute(sql)

        del VTable.Rename # method is optional

        cur.execute("alter table bar rename to foo") # put things back

 

        sql="delete from foo where name=='Fred'"

                columns,data=getfiledata([eval(UPREFIX+a) for a in args]) # eval strips off layer of quotes

        oldestmanual=(99999999999999999,"","")

        rowid=next(cur.execute("create table foo(x blob); insert into foo values(zeroblob(98765)); select rowid from foo"))[0]

        blob=self.db.blobopen("main", "foo", "x", rowid, True)

        blob.close()

        nargs=self.blob_nargs

        for func in [x for x in dir(blob) if not x.startswith("__") and not x in ("close",)]:

            args=("one", "two", "three")[:nargs.get(func,0)]

            try:

                getattr(blob, func)(*args)

                self.fail("blob method "+func+" didn't notice that the connection is closed")

            except ValueError: # we issue ValueError to be consistent with file objects

                pass

        

        # For binary/blobs I use an anonymous area slightly larger than 2GB chunk of memory, but don't touch any of it

        self.assertRaises(apsw.TooBigError,  c.execute, "insert into foo values(?)", (f,))

        c.execute("insert into foo values(?)", ("jkghjk"*1024,))

        b=self.db.blobopen("main", "foo", "theblob", self.db.last_insert_rowid(), True)

        b.read(1)

        self.assertRaises(ValueError, b.write, f)

        def func(): return f

        self.db.createscalarfunction("toobig", func)

        self.assertRaises(apsw.TooBigError, c.execute, "select toobig()")

        # Other testing by fault injection

        if not hasattr(apsw, "faultdict"):

            return

 

        ## SetContextResultLargeUnicode

        apsw.faultdict["SetContextResultLargeUnicode"]=True

        try:

            db=apsw.Connection(":memory:")

            db.createscalarfunction("foo", lambda x: u("a unicode string"))

            for row in db.cursor().execute("select foo(3)"):

                pass

            1/0

        except:

            klass, value, tb=sys.exc_info()

            self.assert_(klass is apsw.TooBigError)

 

        ## SetContextResultLargeString

        if sys.version_info<(3,0):

            apsw.faultdict["SetContextResultLargeString"]=True

            try:

                db=apsw.Connection(":memory:")

                def func(x):

                    return "an ordinary string"*10000

                db.createscalarfunction("foo", func)

                for row in db.cursor().execute("select foo(3)"):

                    pass

                1/0

            except:

                klass, value, tb=sys.exc_info()

                self.assert_(klass is apsw.TooBigError)

 

        open(fname, "wb").write(b("A")*8192)

        except:

            klass,e,tb=sys.exc_info()

            self.failUnless(isinstance(e, apsw.NotADBError))

    def testLimits(self):

        "Verify setting and getting limits"

        self.assertRaises(TypeError, self.db.limit, "apollo", 11)

        c=self.db.cursor()

        c.execute("create table foo(x)")

        c.execute("insert into foo values(?)", ("x"*1024,))

        old=self.db.limit(apsw.SQLITE_LIMIT_LENGTH)

        self.db.limit(apsw.SQLITE_LIMIT_LENGTH, 1023)

        self.assertRaises(apsw.TooBigError, c.execute, "insert into foo values(?)", ("y"*1024,))

        self.failUnlessEqual(1023, self.db.limit(apsw.SQLITE_LIMIT_LENGTH, 0))

        # bug in sqlite - see http://www.sqlite.org/cvstrac/tktview?tn=3085

        if False:

            c.execute("insert into foo values(?)", ("x"*1024,))

            self.failUnlessEqual(apsw.SQLITE_MAX_LENGTH, self.db.limit(apsw.SQLITE_LIMIT_LENGTH))

 

    def testIssue4(self):

        # http://code.google.com/p/apsw/issues/detail?id=4

        connection = apsw.Connection(":memory:")

        cursor = connection.cursor()

 

        cursor.execute("CREATE TABLE A_TABLE (ID ABC PRIMARY KEY NOT NULL)")

        try:

            cursor.execute("INSERT INTO A_TABLE VALUES (NULL)")

        except:

            klass,e,tb=sys.exc_info()

            assert "A_TABLE.ID" in str(e)

    

        try:

            cursor.execute("INSERT INTO A_TABLE VALUES (?)", (None,))

        except:

            klass,e,tb=sys.exc_info()

            assert "A_TABLE.ID" in str(e)

 

    def testIssue15(self):

        # http://code.google.com/p/apsw/issues/detail?id=15

        self.db.cursor().execute("create table foo(x)")

        self.db.cursor().execute("begin exclusive")

        db2=apsw.Connection("testdb")

        db2.setbusytimeout(30000)

        t=ThreadRunner(db2.cursor().execute, "select * from foo")

        t.start()

        time.sleep(1)

        self.db.cursor().execute("commit")

        t.go()

        

 

        def unraise():

            # We cause an unraiseable error to happen by writing to a

            # blob open for reading.  The close method called in the

            # destructor will then also give the error

            db=apsw.Connection(":memory:")

            rowid=next(db.cursor().execute("create table foo(x); insert into foo values(x'aabbccdd'); select rowid from foo"))[0]

            blob=db.blobopen("main", "foo", "x", rowid, False)

            try:

                blob.write(b("badd"))

            except apsw.ReadOnlyError:

                pass

            del db

            del blob

            gc.collect()

            

        unraise()

        unraise()

    def testStatementCache(self, scsize=100):

        "Verify statement cache integrity"

        l=[self.db.cursor().execute("select x from foo") for i in range(scsize+200)]

        db2=apsw.Connection("testdb", statementcachesize=scsize)

 

    def testStatementCacheZeroSize(self):

        self.db=apsw.Connection("testdb", statementcachesize=-1)

        self.testStatementCache(-1)

 

    def testWikipedia(self):

        "Use front page of wikipedia to check unicode handling"

        # the text also includes characters that can't be represented in 16 bits

        text=u(r"""WIKIPEDIA\nEnglish\nThe Free Encyclopedia\n2 386 000+ articles\nDeutsch\nDie freie Enzyklop\\u00e4die\n753 000+ Artikel\nFran\\u00e7ais\nL\\u2019encyclop\\u00e9die libre\n662 000+ articles\nPolski\nWolna encyklopedia\n503 000+ hase\\u0142\n\\u65e5\\u672c\\u8a9e\n\\u30d5\\u30ea\\u30fc\\u767e\\u79d1\\u4e8b\\u5178\n492 000+ \\u8a18\\u4e8b\nItaliano\nL\\u2019enciclopedia libera\n456 000+ voci\nNederlands\nDe vrije encyclopedie\n440 000+ artikelen\nPortugu\\u00eas\nA enciclop\\u00e9dia livre\n380 000+ artigos\nEspa\\u00f1ol\nLa enciclopedia libre\n363 000+ art\\u00edculos\n\\u0420\\u0443\\u0441\\u0441\\u043a\\u0438\\u0439\n\\u0421\\u0432\\u043e\\u0431\\u043e\\u0434\\u043d\\u0430\\u044f \\u044d\\u043d\\u0446\\u0438\\u043a\\u043b\\u043e\\u043f\\u0435\\u0434\\u0438\\u044f\n285 000+ \\u0441\\u0442\\u0430\\u0442\\u0435\\u0439\nSearch \\u00b7 Suche \\u00b7 Rechercher \\u00b7 Szukaj \\u00b7 \\u691c\\u7d22 \\u00b7 Ricerca \\u00b7 Zoeken \\u00b7 Busca \\u00b7 Buscar\n\\u041f\\u043e\\u0438\\u0441\\u043a \\u00b7 S\\u00f6k \\u00b7 \\u641c\\u7d22 \\u00b7 S\\u00f8k \\u00b7 Haku \\u00b7 Cerca \\u00b7 Suk \\u00b7 \\u041f\\u043e\\u0448\\u0443\\u043a \\u00b7 C\\u0103utare \\u00b7 Ara\n 100 000+ \nCatal\\u00e0 \\u00b7 Deutsch \\u00b7 English \\u00b7 Espa\\u00f1ol \\u00b7 Fran\\u00e7ais \\u00b7 Italiano \\u00b7 Nederlands \\u00b7 \\u65e5\\u672c\\u8a9e \\u00b7 Norsk (bokm\\u00e5l) \\u00b7 Polski \\u00b7 Portugu\\u00eas \\u00b7 \\u0420\\u0443\\u0441\\u0441\\u043a\\u0438\\u0439 \\u00b7 Rom\\u00e2n\\u0103 \\u00b7 Suomi \\u00b7 Svenska \\u00b7 T\\u00fcrk\\u00e7e \\u00b7 \\u0423\\u043a\\u0440\\u0430\\u0457\\u043d\\u0441\\u044c\\u043a\\u0430 \\u00b7 Volap\\u00fck \\u00b7 \\u4e2d\\u6587\n 10 000+ \n\\u0627\\u0644\\u0639\\u0631\\u0628\\u064a\\u0629 \\u00b7 Asturianu \\u00b7 Krey\\u00f2l Ayisyen \\u00b7 Az\\u0259rbaycan / \\u0622\\u0630\\u0631\\u0628\\u0627\\u064a\\u062c\\u0627\\u0646 \\u062f\\u064a\\u0644\\u06cc \\u00b7 \\u09ac\\u09be\\u0982\\u09b2\\u09be \\u00b7 \\u0411\\u0435\\u043b\\u0430\\u0440\\u0443\\u0441\\u043a\\u0430\\u044f (\\u0410\\u043a\\u0430\\u0434\\u044d\\u043c\\u0456\\u0447\\u043d\\u0430\\u044f) \\u00b7 \\u09ac\\u09bf\\u09b7\\u09cd\\u09a3\\u09c1\\u09aa\\u09cd\\u09b0\\u09bf\\u09af\\u09bc\\u09be \\u09ae\\u09a3\\u09bf\\u09aa\\u09c1\\u09b0\\u09c0 \\u00b7 Bosanski \\u00b7 Brezhoneg \\u00b7 \\u0411\\u044a\\u043b\\u0433\\u0430\\u0440\\u0441\\u043a\\u0438 \\u00b7 \\u010cesky \\u00b7 Cymraeg \\u00b7 Dansk \\u00b7 Eesti \\u00b7 \\u0395\\u03bb\\u03bb\\u03b7\\u03bd\\u03b9\\u03ba\\u03ac \\u00b7 Esperanto \\u00b7 Euskara \\u00b7 \\u0641\\u0627\\u0631\\u0633\\u06cc \\u00b7 Galego \\u00b7 \\ud55c\\uad6d\\uc5b4 \\u00b7 \\u0939\\u093f\\u0928\\u094d\\u0926\\u0940 \\u00b7 Hrvatski \\u00b7 Ido \\u00b7 Bahasa Indonesia \\u00b7 \\u00cdslenska \\u00b7 \\u05e2\\u05d1\\u05e8\\u05d9\\u05ea \\u00b7 Basa Jawa \\u00b7 \\u10e5\\u10d0\\u10e0\\u10d7\\u10e3\\u10da\\u10d8 \\u00b7 Kurd\\u00ee / \\u0643\\u0648\\u0631\\u062f\\u06cc \\u00b7 Latina \\u00b7 Lumbaart \\u00b7 Latvie\\u0161u \\u00b7 L\\u00ebtzebuergesch \\u00b7 Lietuvi\\u0173 \\u00b7 Magyar \\u00b7 \\u041c\\u0430\\u043a\\u0435\\u0434\\u043e\\u043d\\u0441\\u043a\\u0438 \\u00b7 \\u092e\\u0930\\u093e\\u0920\\u0940 \\u00b7 Bahasa Melayu \\u00b7 \\u0928\\u0947\\u092a\\u093e\\u0932 \\u092d\\u093e\\u0937\\u093e \\u00b7 Norsk (nynorsk) \\u00b7 Nnapulitano \\u00b7 Occitan \\u00b7 Piemont\\u00e8is \\u00b7 Plattd\\u00fc\\u00fctsch \\u00b7 Shqip \\u00b7 Sicilianu \\u00b7 Simple English \\u00b7 Sinugboanon \\u00b7 Sloven\\u010dina \\u00b7 Sloven\\u0161\\u010dina \\u00b7 \\u0421\\u0440\\u043f\\u0441\\u043a\\u0438 \\u00b7 Srpskohrvatski / \\u0421\\u0440\\u043f\\u0441\\u043a\\u043e\\u0445\\u0440\\u0432\\u0430\\u0442\\u0441\\u043a\\u0438 \\u00b7 Basa Sunda \\u00b7 Tagalog \\u00b7 \\u0ba4\\u0bae\\u0bbf\\u0bb4\\u0bcd \\u00b7 \\u0c24\\u0c46\\u0c32\\u0c41\\u0c17\\u0c41 \\u00b7 \\u0e44\\u0e17\\u0e22 \\u00b7 Ti\\u1ebfng Vi\\u1ec7t \\u00b7 Walon\n 1 000+ \nAfrikaans \\u00b7 Alemannisch \\u00b7 \\u12a0\\u121b\\u122d\\u129b \\u00b7 Aragon\\u00e9s \\u00b7 Arm\\u00e3neashce \\u00b7 Arpitan \\u00b7 B\\u00e2n-l\\u00e2m-g\\u00fa \\u00b7 Basa Banyumasan \\u00b7 \\u0411\\u0435\\u043b\\u0430\\u0440\\u0443\\u0441\\u043a\\u0430\\u044f (\\u0422\\u0430\\u0440\\u0430\\u0448\\u043a\\u0435\\u0432i\\u0446\\u0430) \\u00b7 \\u092d\\u094b\\u091c\\u092a\\u0941\\u0930\\u0940 \\u00b7 Boarisch \\u00b7 Corsu \\u00b7 \\u0427\\u0103\\u0432\\u0430\\u0448 \\u00b7 Deitsch \\u00b7 \\u078b\\u07a8\\u0788\\u07ac\\u0780\\u07a8 \\u00b7 Eald Englisc \\u00b7 F\\u00f8royskt \\u00b7 Frysk \\u00b7 Furlan \\u00b7 Gaeilge \\u00b7 Gaelg \\u00b7 G\\u00e0idhlig \\u00b7 \\u53e4\\u6587 / \\u6587\\u8a00\\u6587 \\u00b7 \\u02bb\\u014clelo Hawai\\u02bbi \\u00b7 \\u0540\\u0561\\u0575\\u0565\\u0580\\u0565\\u0576 \\u00b7 Hornjoserbsce \\u00b7 Ilokano \\u00b7 Interlingua \\u00b7 \\u0418\\u0440\\u043e\\u043d \\u00e6\\u0432\\u0437\\u0430\\u0433 \\u00b7 \\u0c95\\u0ca8\\u0ccd\\u0ca8\\u0ca1 \\u00b7 Kapampangan \\u00b7 Kasz\\u00ebbsczi \\u00b7 Kernewek \\u00b7 \\u1797\\u17b6\\u179f\\u17b6\\u1781\\u17d2\\u1798\\u17c2\\u179a \\u00b7 Ladino / \\u05dc\\u05d0\\u05d3\\u05d9\\u05e0\\u05d5 \\u00b7 Ligure \\u00b7 Limburgs \\u00b7 Ling\\u00e1la \\u00b7 \\u0d2e\\u0d32\\u0d2f\\u0d3e\\u0d33\\u0d02 \\u00b7 Malti \\u00b7 M\\u0101ori \\u00b7 \\u041c\\u043e\\u043d\\u0433\\u043e\\u043b \\u00b7 N\\u0101huatlaht\\u014dlli \\u00b7 Nedersaksisch \\u00b7 \\u0928\\u0947\\u092a\\u093e\\u0932\\u0940 \\u00b7 Nouormand \\u00b7 Novial \\u00b7 O\\u2018zbek \\u00b7 \\u092a\\u093e\\u0934\\u093f \\u00b7 Pangasin\\u00e1n \\u00b7 \\u067e\\u069a\\u062a\\u0648 \\u00b7 \\u049a\\u0430\\u0437\\u0430\\u049b\\u0448\\u0430 \\u00b7 Ripoarisch \\u00b7 Rumantsch \\u00b7 Runa Simi \\u00b7 \\u0938\\u0902\\u0938\\u094d\\u0915\\u0943\\u0924\\u092e\\u094d \\u00b7 S\\u00e1megiella \\u00b7 Scots \\u00b7 Kiswahili \\u00b7 Tarand\\u00edne \\u00b7 Tatar\\u00e7a \\u00b7 \\u0422\\u043e\\u04b7\\u0438\\u043a\\u04e3 \\u00b7 Lea faka-Tonga \\u00b7 T\\u00fcrkmen \\u00b7 \\u0627\\u0631\\u062f\\u0648 \\u00b7 V\\u00e8neto \\u00b7 V\\u00f5ro \\u00b7 West-Vlams \\u00b7 Winaray \\u00b7 \\u5434\\u8bed \\u00b7 \\u05d9\\u05d9\\u05b4\\u05d3\\u05d9\\u05e9 \\u00b7 \\u7cb5\\u8a9e \\u00b7 Yor\\u00f9b\\u00e1 \\u00b7 Zazaki \\u00b7 \\u017demait\\u0117\\u0161ka\n 100+ \n\\u0710\\u072a\\u0721\\u071d\\u0710 \\u00b7 Ava\\u00f1e\\u2019\\u1ebd \\u00b7 \\u0410\\u0432\\u0430\\u0440 \\u00b7 Aymara \\u00b7 Bamanankan \\u00b7 \\u0411\\u0430\\u0448\\u04a1\\u043e\\u0440\\u0442 \\u00b7 Bikol Central \\u00b7 \\u0f56\\u0f7c\\u0f51\\u0f0b\\u0f61\\u0f72\\u0f42 \\u00b7 Chamoru \\u00b7 Chavacano de Zamboanga \\u00b7 Bislama \\u00b7 Din\\u00e9 Bizaad \\u00b7 Dolnoserbski \\u00b7 Emigli\\u00e0n-Rumagn\\u00f2l \\u00b7 E\\u028begbe \\u00b7 \\u06af\\u06cc\\u0644\\u06a9\\u06cc \\u00b7 \\u0a97\\u0ac1\\u0a9c\\u0ab0\\u0abe\\u0aa4\\u0ac0 \\u00b7 \\U00010332\\U0001033f\\U00010344\\U00010339\\U00010343\\U0001033a \\u00b7 Hak-k\\u00e2-fa / \\u5ba2\\u5bb6\\u8a71 \\u00b7 Igbo \\u00b7 \\u1403\\u14c4\\u1483\\u144e\\u1450\\u1466 / Inuktitut \\u00b7 Interlingue \\u00b7 \\u0915\\u0936\\u094d\\u092e\\u0940\\u0930\\u0940 / \\u0643\\u0634\\u0645\\u064a\\u0631\\u064a \\u00b7 Kongo \\u00b7 \\u041a\\u044b\\u0440\\u0433\\u044b\\u0437\\u0447\\u0430 \\u00b7 \\u0e9e\\u0eb2\\u0eaa\\u0eb2\\u0ea5\\u0eb2\\u0ea7 \\u00b7 lojban \\u00b7 Malagasy \\u00b7 M\\u0101z\\u0259r\\u016bni / \\u0645\\u0627\\u0632\\u0650\\u0631\\u0648\\u0646\\u06cc \\u00b7 M\\u00ecng-d\\u0115\\u0324ng-ng\\u1e73\\u0304 \\u00b7 \\u041c\\u043e\\u043b\\u0434\\u043e\\u0432\\u0435\\u043d\\u044f\\u0441\\u043a\\u044d \\u00b7 \\u1017\\u1019\\u102c\\u1005\\u102c \\u00b7 Ekakair\\u0169 Naoero \\u00b7 N\\u0113hiyaw\\u0113win / \\u14c0\\u1426\\u1403\\u152d\\u140d\\u140f\\u1423 \\u00b7 Norfuk / Pitkern \\u00b7 \\u041d\\u043e\\u0445\\u0447\\u0438\\u0439\\u043d \\u00b7 \\u0b13\\u0b21\\u0b3c\\u0b3f\\u0b06 \\u00b7 Afaan Oromoo \\u00b7 \\u0985\\u09b8\\u09ae\\u09c0\\u09af\\u09bc\\u09be \\u00b7 \\u0a2a\\u0a70\\u0a1c\\u0a3e\\u0a2c\\u0a40 / \\u067e\\u0646\\u062c\\u0627\\u0628\\u06cc \\u00b7 Papiamentu \\u00b7 Q\\u0131r\\u0131mtatarca \\u00b7 Romani / \\u0930\\u094b\\u092e\\u093e\\u0928\\u0940 \\u00b7 Kinyarwanda \\u00b7 Gagana S\\u0101moa \\u00b7 Sardu \\u00b7 Seeltersk \\u00b7 \\u0dc3\\u0dd2\\u0d82\\u0dc4\\u0dbd \\u00b7 \\u0633\\u0646\\u068c\\u064a \\u00b7 \\u0421\\u043b\\u043e\\u0432\\u0463\\u043d\\u044c\\u0441\\u043a\\u044a \\u00b7 Af Soomaali \\u00b7 SiSwati \\u00b7 Reo Tahiti \\u00b7 Taqbaylit \\u00b7 Tetun \\u00b7 \\u1275\\u130d\\u122d\\u129b \\u00b7 Tok Pisin \\u00b7 \\u13e3\\u13b3\\u13a9 \\u00b7 \\u0423\\u0434\\u043c\\u0443\\u0440\\u0442 \\u00b7 Uyghur / \\u0626\\u06c7\\u064a\\u063a\\u06c7\\u0631\\u0686\\u0647 \\u00b7 Tshiven\\u1e13a \\u00b7 Wollof \\u00b7 isiXhosa \\u00b7 Ze\\u00eauws \\u00b7 isiZulu\nOther languages \\u00b7 Weitere Sprachen \\u00b7 \\u4ed6\\u306e\\u8a00\\u8a9e \\u00b7 Kompletna lista j\\u0119zyk\\u00f3w \\u00b7 \\u5176\\u4ed6\\u8bed\\u8a00 \\u00b7 \\u0414\\u0440\\u0443\\u0433\\u0438\\u0435 \\u044f\\u0437\\u044b\\u043a\\u0438 \\u00b7 Aliaj lingvoj \\u00b7 \\ub2e4\\ub978 \\uc5b8\\uc5b4 \\u00b7 Ng\\u00f4n ng\\u1eef kh\\u00e1c""")

 

        self.db.close()

 

        for encoding in "UTF-16", "UTF-16le", "UTF-16be", "UTF-8":

            if os.path.exists("testdb"):

                os.remove("testdb")

            db=apsw.Connection("testdb")

            c=db.cursor()

            c.execute("pragma encoding=\"%s\"" % (encoding,))

            for row in c.execute("pragma encoding"):

                # we use startswith as UTF-16 will be returned with le/be suffix

                self.assert_(row[0].startswith(encoding))

            c.execute("create table foo(x); insert into foo values(?)", (text,))

            for row in c.execute("select * from foo"):

                self.failUnlessEqual(row[0], text)

            db.close()

 

 

    def sourceCheckFunction(self, name, lines):

        # Checks an individual function does things right

        if name.startswith("ZeroBlobBind_"):

                return

 

        if name.startswith("APSWCursor_"):

            # these methods aren't publically exported so the checks

            # will already have been done by their callers

            if name[len("APSWCursor_"):] in ("dealloc", "init", "dobinding", "dobindings", "doexectrace", "dorowtrace", "step", "close"):

                return

            use=None

            closed=None

            for i,line in enumerate(lines):

                if "CHECK_USE" in line and use is None:

                    use=i

                if "CHECK_CLOSED" in line and closed is None:

                    closed=i

            if use is None:

                self.fail("CHECK_USE missing in "+name)

            if closed is None:

                self.fail("CHECK_CLOSED missing in "+name)

            if closed<use:

                self.fail("CHECK_CLOSED should be after CHECK_USE in "+name)

            return

 

        if name.startswith("Connection_"):

            # these methods aren't publically exported so the checks

            # will already have been done by their callers

            if name[len("Connection_"):] in ("internal_cleanup", "dealloc", "init", "close", "interrupt"):

                return

            use=None

            closed=None

            for i,line in enumerate(lines):

                if "CHECK_USE" in line and use is None:

                    use=i

                if "CHECK_CLOSED" in line and closed is None:

                    closed=i

            if use is None:

                self.fail("CHECK_USE missing in "+name)

            if closed is None:

                self.fail("CHECK_CLOSED missing in "+name)

            if closed<use:

                self.fail("CHECK_CLOSED should be after CHECK_USE in "+name)

            return

 

        if name.startswith("APSWBlob_"):

            # these methods aren't publically exported so the checks