#-*- coding: ISO-8859-1 -*-

# pysqlite2/test/dbapi.py: tests for DB-API compliance

#

# Copyright (C) 2004-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 sys

import threading

import sqlite3 as sqlite

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.assert_(issubclass(sqlite.Warning, StandardError),

                     "Warning is not a subclass of StandardError")

    def CheckError(self):

        self.failUnless(issubclass(sqlite.Error, StandardError),

                        "Error is not a subclass of StandardError")

    def CheckInterfaceError(self):

        self.failUnless(issubclass(sqlite.InterfaceError, sqlite.Error),

                        "InterfaceError is not a subclass of Error")

    def CheckDatabaseError(self):

        self.failUnless(issubclass(sqlite.DatabaseError, sqlite.Error),

                        "DatabaseError is not a subclass of Error")

    def CheckDataError(self):

        self.failUnless(issubclass(sqlite.DataError, sqlite.DatabaseError),

                        "DataError is not a subclass of DatabaseError")

    def CheckOperationalError(self):

        self.failUnless(issubclass(sqlite.OperationalError, sqlite.DatabaseError),

                        "OperationalError is not a subclass of DatabaseError")

    def CheckIntegrityError(self):

        self.failUnless(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),

                        "IntegrityError is not a subclass of DatabaseError")

    def CheckInternalError(self):

        self.failUnless(issubclass(sqlite.InternalError, sqlite.DatabaseError),

                        "InternalError is not a subclass of DatabaseError")

    def CheckProgrammingError(self):

        self.failUnless(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),

                        "ProgrammingError is not a subclass of DatabaseError")

    def CheckNotSupportedError(self):

        self.failUnless(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.failUnlessEqual(self.cx.Warning, sqlite.Warning)

        self.failUnlessEqual(self.cx.Error, sqlite.Error)

        self.failUnlessEqual(self.cx.InterfaceError, sqlite.InterfaceError)

        self.failUnlessEqual(self.cx.DatabaseError, sqlite.DatabaseError)

        self.failUnlessEqual(self.cx.DataError, sqlite.DataError)

        self.failUnlessEqual(self.cx.OperationalError, sqlite.OperationalError)

        self.failUnlessEqual(self.cx.IntegrityError, sqlite.IntegrityError)

        self.failUnlessEqual(self.cx.InternalError, sqlite.InternalError)

        self.failUnlessEqual(self.cx.ProgrammingError, sqlite.ProgrammingError)

        self.failUnlessEqual(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 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.failUnlessEqual(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.failUnlessEqual(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.failUnlessEqual(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.failUnlessEqual(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.failUnlessEqual(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.failUnlessEqual(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.failUnlessEqual(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.failUnlessEqual(lst[0], 5)

        self.failUnlessEqual(lst[1], 6)

    def CheckFetchone(self):

        self.cu.execute("select name from test")

        row = self.cu.fetchone()

        self.failUnlessEqual(row[0], "foo")

        row = self.cu.fetchone()

        self.failUnlessEqual(row, None)

    def CheckFetchoneNoStatement(self):

        cur = self.cx.cursor()

        row = cur.fetchone()

        self.failUnlessEqual(row, None)

    def CheckArraySize(self):

        # must default ot 1

        self.failUnlessEqual(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.failUnlessEqual(len(res), 2)

    def CheckFetchmany(self):

        self.cu.execute("select name from test")

        res = self.cu.fetchmany(100)

        self.failUnlessEqual(len(res), 1)

        res = self.cu.fetchmany(100)

        self.failUnlessEqual(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.failUnlessEqual(len(res), 1)

    def CheckFetchall(self):

        self.cu.execute("select name from test")

        res = self.cu.fetchall()

        self.failUnlessEqual(len(res), 1)

        res = self.cu.fetchall()

        self.failUnlessEqual(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.failUnlessEqual(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

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