#-*- coding: iso-8859-1 -*-
# pysqlite2/test/dbapi.py: tests for DB-API compliance
#
# Copyright (C) 2004-2010 Gerhard Hring <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 threading
import unittest
import sqlite3 as sqlite

from test.support import TESTFN, unlink


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.assertTrue(issubclass(sqlite.Warning, Exception),
 "Warning is not a subclass of Exception")

 def CheckError(self):
 self.assertTrue(issubclass(sqlite.Error, Exception),
 "Error is not a subclass of Exception")

 def CheckInterfaceError(self):
 self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error),
 "InterfaceError is not a subclass of Error")

 def CheckDatabaseError(self):
 self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error),
 "DatabaseError is not a subclass of Error")

 def CheckDataError(self):
 self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError),
 "DataError is not a subclass of DatabaseError")

 def CheckOperationalError(self):
 self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError),
 "OperationalError is not a subclass of DatabaseError")

 def CheckIntegrityError(self):
 self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),
 "IntegrityError is not a subclass of DatabaseError")

 def CheckInternalError(self):
 self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError),
 "InternalError is not a subclass of DatabaseError")

 def CheckProgrammingError(self):
 self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),
 "ProgrammingError is not a subclass of DatabaseError")

 def CheckNotSupportedError(self):
 self.assertTrue(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"
 with self.assertRaises(sqlite.OperationalError):
 con = sqlite.connect(YOU_CANNOT_OPEN_THIS)

 def CheckClose(self):
 self.cx.close()

 def CheckExceptions(self):
 # Optional DB-API extension.
 self.assertEqual(self.cx.Warning, sqlite.Warning)
 self.assertEqual(self.cx.Error, sqlite.Error)
 self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError)
 self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError)
 self.assertEqual(self.cx.DataError, sqlite.DataError)
 self.assertEqual(self.cx.OperationalError, sqlite.OperationalError)
 self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError)
 self.assertEqual(self.cx.InternalError, sqlite.InternalError)
 self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError)
 self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError)

 def CheckInTransaction(self):
 # Can't use db from setUp because we want to test initial state.
 cx = sqlite.connect(":memory:")
 cu = cx.cursor()
 self.assertEqual(cx.in_transaction, False)
 cu.execute("create table transactiontest(id integer primary key, name text)")
 self.assertEqual(cx.in_transaction, False)
 cu.execute("insert into transactiontest(name) values (?)", ("foo",))
 self.assertEqual(cx.in_transaction, True)
 cu.execute("select name from transactiontest where name=?", ["foo"])
 row = cu.fetchone()
 self.assertEqual(cx.in_transaction, True)
 cx.commit()
 self.assertEqual(cx.in_transaction, False)
 cu.execute("select name from transactiontest where name=?", ["foo"])
 row = cu.fetchone()
 self.assertEqual(cx.in_transaction, False)

 def CheckInTransactionRO(self):
 with self.assertRaises(AttributeError):
 self.cx.in_transaction = True

 def CheckOpenWithPathLikeObject(self):
 """ Checks that we can successfully connect to a database using an object that
 is PathLike, i.e. has __fspath__(). """
 self.addCleanup(unlink, TESTFN)
 class Path:
 def __fspath__(self):
 return TESTFN
 path = Path()
 with sqlite.connect(path) as cx:
 cx.execute('create table test(id integer)')

 def CheckOpenUri(self):
 if sqlite.sqlite_version_info < (3, 7, 7):
 with self.assertRaises(sqlite.NotSupportedError):
 sqlite.connect(':memory:', uri=True)
 return
 self.addCleanup(unlink, TESTFN)
 with sqlite.connect(TESTFN) as cx:
 cx.execute('create table test(id integer)')
 with sqlite.connect('file:' + TESTFN, uri=True) as cx:
 cx.execute('insert into test(id) values(0)')
 with sqlite.connect('file:' + TESTFN + '?mode=ro', uri=True) as cx:
 with self.assertRaises(sqlite.OperationalError):
 cx.execute('insert into test(id) values(1)')

 @unittest.skipIf(sqlite.sqlite_version_info >= (3, 3, 1),
 'needs sqlite versions older than 3.3.1')
 def CheckSameThreadErrorOnOldVersion(self):
 with self.assertRaises(sqlite.NotSupportedError) as cm:
 sqlite.connect(':memory:', check_same_thread=False)
 self.assertEqual(str(cm.exception), 'shared connections not available')

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, unique_test text unique)"
 )
 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):
 with self.assertRaises(sqlite.OperationalError):
 self.cu.execute("select asdf")

 def CheckExecuteTooMuchSql(self):
 with self.assertRaises(sqlite.Warning):
 self.cu.execute("select 5+4; select 4+5")

 def CheckExecuteTooMuchSql2(self):
 self.cu.execute("select 5+4; -- foo bar")

 def CheckExecuteTooMuchSql3(self):
 self.cu.execute("""
 select 5+4;

 /*
 foo
 */
 """)

 def CheckExecuteWrongSqlArg(self):
 with self.assertRaises(ValueError):
 self.cu.execute(42)

 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 CheckExecuteArgStringWithZeroByte(self):
 self.cu.execute("insert into test(name) values (?)", ("Hu\x00go",))

 self.cu.execute("select name from test where id=?", (self.cu.lastrowid,))
 row = self.cu.fetchone()
 self.assertEqual(row[0], "Hu\x00go")

 def CheckExecuteNonIterable(self):
 with self.assertRaises(ValueError) as cm:
 self.cu.execute("insert into test(id) values (?)", 42)
 self.assertEqual(str(cm.exception), 'parameters are of unsupported type')

 def CheckExecuteWrongNoOfArgs1(self):
 # too many parameters
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.execute("insert into test(id) values (?)", (17, "Egon"))

 def CheckExecuteWrongNoOfArgs2(self):
 # too little parameters
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.execute("insert into test(id) values (?)")

 def CheckExecuteWrongNoOfArgs3(self):
 # no parameters, parameters are needed
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.execute("insert into test(id) values (?)")

 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.assertEqual(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.assertEqual(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.assertEqual(row[0], "foo")

 def CheckExecuteDictMapping_Mapping(self):
 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.assertEqual(row[0], "foo")

 def CheckExecuteDictMappingTooLittleArgs(self):
 self.cu.execute("insert into test(name) values ('foo')")
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"})

 def CheckExecuteDictMappingNoArgs(self):
 self.cu.execute("insert into test(name) values ('foo')")
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.execute("select name from test where name=:name")

 def CheckExecuteDictMappingUnnamed(self):
 self.cu.execute("insert into test(name) values ('foo')")
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.execute("select name from test where name=?", {"name": "foo"})

 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.assertEqual(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.assertEqual(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.assertEqual(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')")
 self.assertLess(2, self.cx.total_changes, msg='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):
 with self.assertRaises(ValueError):
 self.cu.executemany(42, [(3,)])

 def CheckExecuteManySelect(self):
 with self.assertRaises(sqlite.ProgrammingError):
 self.cu.executemany("select ?", [(3,)])

 def CheckExecuteManyNotIterable(self):
 with self.assertRaises(TypeError):
 self.cu.executemany("insert into test(income) values (?)", 42)

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

 def CheckFetchone(self):
 self.cu.execute("select name from test")
 row = self.cu.fetchone()
 self.assertEqual(row[0], "foo")
 row = self.cu.fetchone()
 self.assertEqual(row, None)

 def CheckFetchoneNoStatement(self):
 cur = self.cx.cursor()
 row = cur.fetchone()
 self.assertEqual(row, None)

 def CheckArraySize(self):
 # must default ot 1
 self.assertEqual(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.assertEqual(len(res), 2)

 def CheckFetchmany(self):
 self.cu.execute("select name from test")
 res = self.cu.fetchmany(100)
 self.assertEqual(len(res), 1)
 res = self.cu.fetchmany(100)
 self.assertEqual(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.assertEqual(len(res), 1)

 def CheckFetchall(self):
 self.cu.execute("select name from test")
 res = self.cu.fetchall()
 self.assertEqual(len(res), 1)
 res = self.cu.fetchall()
 self.assertEqual(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.assertEqual(self.cu.connection, self.cx)

 def CheckWrongCursorCallable(self):
 with self.assertRaises(TypeError):
 def f(): pass
 cur = self.cx.cursor(f)

 def CheckCursorWrongClass(self):
 class Foo: pass
 foo = Foo()
 with self.assertRaises(TypeError):
 cur = sqlite.Cursor(foo)

 def CheckLastRowIDOnReplace(self):
 """
 INSERT OR REPLACE and REPLACE INTO should produce the same behavior.
 """
 sql = '{} INTO test(id, unique_test) VALUES (?, ?)'
 for statement in ('INSERT OR REPLACE', 'REPLACE'):
 with self.subTest(statement=statement):
 self.cu.execute(sql.format(statement), (1, 'foo'))
 self.assertEqual(self.cu.lastrowid, 1)

 def CheckLastRowIDOnIgnore(self):
 self.cu.execute(
 "insert or ignore into test(unique_test) values (?)",
 ('test',))
 self.assertEqual(self.cu.lastrowid, 2)
 self.cu.execute(
 "insert or ignore into test(unique_test) values (?)",
 ('test',))
 self.assertEqual(self.cu.lastrowid, 2)

 def CheckLastRowIDInsertOR(self):
 results = []
 for statement in ('FAIL', 'ABORT', 'ROLLBACK'):
 sql = 'INSERT OR {} INTO test(unique_test) VALUES (?)'
 with self.subTest(statement='INSERT OR {}'.format(statement)):
 self.cu.execute(sql.format(statement), (statement,))
 results.append((statement, self.cu.lastrowid))
 with self.assertRaises(sqlite.IntegrityError):
 self.cu.execute(sql.format(statement), (statement,))
 results.append((statement, self.cu.lastrowid))
 expected = [
 ('FAIL', 2), ('FAIL', 2),
 ('ABORT', 3), ('ABORT', 3),
 ('ROLLBACK', 4), ('ROLLBACK', 4),
 ]
 self.assertEqual(results, expected)


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()
 if len(errors) > 0:
 self.fail("\n".join(errors))

 def CheckConClose(self):
 def run(con, errors):
 try:
 con.close()
 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 CheckCurImplicitBegin(self):
 def run(cur, errors):
 try:
 cur.execute("insert into test(name) values ('a')")
 errors.append("did not raise ProgrammingError")
 return
 except sqlite.ProgrammingError:
 return
 except:
 errors.append("raised wrong exception")

 errors = []
 t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
 t.start()
 t.join()
 if len(errors) > 0:
 self.fail("\n".join(errors))

 def CheckCurClose(self):
 def run(cur, errors):
 try:
 cur.close()
 errors.append("did not raise ProgrammingError")
 return
 except sqlite.ProgrammingError:
 return
 except:
 errors.append("raised wrong exception")

 errors = []
 t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
 t.start()
 t.join()
 if len(errors) > 0:
 self.fail("\n".join(errors))

 def CheckCurExecute(self):
 def run(cur, errors):
 try:
 cur.execute("select name from test")
 errors.append("did not raise ProgrammingError")
 return
 except sqlite.ProgrammingError:
 return
 except:
 errors.append("raised wrong exception")

 errors = []
 self.cur.execute("insert into test(name) values ('a')")
 t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
 t.start()
 t.join()
 if len(errors) > 0:
 self.fail("\n".join(errors))

 def CheckCurIterNext(self):
 def run(cur, errors):
 try:
 row = cur.fetchone()
 errors.append("did not raise ProgrammingError")
 return
 except sqlite.ProgrammingError:
 return
 except:
 errors.append("raised wrong exception")

 errors = []
 self.cur.execute("insert into test(name) values ('a')")
 self.cur.execute("select name from test")
 t = threading.Thread(target=run, kwargs={"cur": self.cur, "errors": errors})
 t.start()
 t.join()
 if len(errors) > 0:
 self.fail("\n".join(errors))

class ConstructorTests(unittest.TestCase):
 def CheckDate(self):
 d = sqlite.Date(2004, 10, 28)

 def CheckTime(self):
 t = sqlite.Time(12, 39, 35)

 def CheckTimestamp(self):
 ts = sqlite.Timestamp(2004, 10, 28, 12, 39, 35)

 def CheckDateFromTicks(self):
 d = sqlite.DateFromTicks(42)

 def CheckTimeFromTicks(self):
 t = sqlite.TimeFromTicks(42)

 def CheckTimestampFromTicks(self):
 ts = sqlite.TimestampFromTicks(42)

 def CheckBinary(self):
 b = sqlite.Binary(b"0円'")

class ExtensionTests(unittest.TestCase):
 def CheckScriptStringSql(self):
 con = sqlite.connect(":memory:")
 cur = con.cursor()
 cur.executescript("""
 -- bla bla
 /* a stupid comment */
 create table a(i);
 insert into a(i) values (5);
 """)
 cur.execute("select i from a")
 res = cur.fetchone()[0]
 self.assertEqual(res, 5)

 def CheckScriptSyntaxError(self):
 con = sqlite.connect(":memory:")
 cur = con.cursor()
 with self.assertRaises(sqlite.OperationalError):
 cur.executescript("create table test(x); asdf; create table test2(x)")

 def CheckScriptErrorNormal(self):
 con = sqlite.connect(":memory:")
 cur = con.cursor()
 with self.assertRaises(sqlite.OperationalError):
 cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")

 def CheckCursorExecutescriptAsBytes(self):
 con = sqlite.connect(":memory:")
 cur = con.cursor()
 with self.assertRaises(ValueError) as cm:
 cur.executescript(b"create table test(foo); insert into test(foo) values (5);")
 self.assertEqual(str(cm.exception), 'script argument must be unicode.')

 def CheckConnectionExecute(self):
 con = sqlite.connect(":memory:")
 result = con.execute("select 5").fetchone()[0]
 self.assertEqual(result, 5, "Basic test of Connection.execute")

 def CheckConnectionExecutemany(self):
 con = sqlite.connect(":memory:")
 con.execute("create table test(foo)")
 con.executemany("insert into test(foo) values (?)", [(3,), (4,)])
 result = con.execute("select foo from test order by foo").fetchall()
 self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany")
 self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany")

 def CheckConnectionExecutescript(self):
 con = sqlite.connect(":memory:")
 con.executescript("create table test(foo); insert into test(foo) values (5);")
 result = con.execute("select foo from test").fetchone()[0]
 self.assertEqual(result, 5, "Basic test of Connection.executescript")

class ClosedConTests(unittest.TestCase):
 def CheckClosedConCursor(self):
 con = sqlite.connect(":memory:")
 con.close()
 with self.assertRaises(sqlite.ProgrammingError):
 cur = con.cursor()

 def CheckClosedConCommit(self):
 con = sqlite.connect(":memory:")
 con.close()
 with self.assertRaises(sqlite.ProgrammingError):
 con.commit()

 def CheckClosedConRollback(self):
 con = sqlite.connect(":memory:")
 con.close()
 with self.assertRaises(sqlite.ProgrammingError):
 con.rollback()

 def CheckClosedCurExecute(self):
 con = sqlite.connect(":memory:")
 cur = con.cursor()
 con.close()
 with self.assertRaises(sqlite.ProgrammingError):
 cur.execute("select 4")

 def CheckClosedCreateFunction(self):
 con = sqlite.connect(":memory:")
 con.close()
 def f(x): return 17
 with self.assertRaises(sqlite.ProgrammingError):
 con.create_function("foo", 1, f)

 def CheckClosedCreateAggregate(self):
 con = sqlite.connect(":memory:")
 con.close()
 class Agg:
 def __init__(self):
 pass
 def step(self, x):
 pass
 def finalize(self):
 return 17
 with self.assertRaises(sqlite.ProgrammingError):
 con.create_aggregate("foo", 1, Agg)

 def CheckClosedSetAuthorizer(self):
 con = sqlite.connect(":memory:")
 con.close()
 def authorizer(*args):
 return sqlite.DENY
 with self.assertRaises(sqlite.ProgrammingError):
 con.set_authorizer(authorizer)

 def CheckClosedSetProgressCallback(self):
 con = sqlite.connect(":memory:")
 con.close()
 def progress(): pass
 with self.assertRaises(sqlite.ProgrammingError):
 con.set_progress_handler(progress, 100)

 def CheckClosedCall(self):
 con = sqlite.connect(":memory:")
 con.close()
 with self.assertRaises(sqlite.ProgrammingError):
 con()

class ClosedCurTests(unittest.TestCase):
 def CheckClosed(self):
 con = sqlite.connect(":memory:")
 cur = con.cursor()
 cur.close()

 for method_name in ("execute", "executemany", "executescript", "fetchall", "fetchmany", "fetchone"):
 if method_name in ("execute", "executescript"):
 params = ("select 4 union select 5",)
 elif method_name == "executemany":
 params = ("insert into foo(bar) values (?)", [(3,), (4,)])
 else:
 params = []

 with self.assertRaises(sqlite.ProgrammingError):
 method = getattr(cur, method_name)
 method(*params)


class SqliteOnConflictTests(unittest.TestCase):
 """
 Tests for SQLite's "insert on conflict" feature.

 See https://www.sqlite.org/lang_conflict.html for details.
 """

 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, unique_name TEXT UNIQUE
 );
 """)

 def tearDown(self):
 self.cu.close()
 self.cx.close()

 def CheckOnConflictRollbackWithExplicitTransaction(self):
 self.cx.isolation_level = None # autocommit mode
 self.cu = self.cx.cursor()
 # Start an explicit transaction.
 self.cu.execute("BEGIN")
 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
 with self.assertRaises(sqlite.IntegrityError):
 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
 # Use connection to commit.
 self.cx.commit()
 self.cu.execute("SELECT name, unique_name from test")
 # Transaction should have rolled back and nothing should be in table.
 self.assertEqual(self.cu.fetchall(), [])

 def CheckOnConflictAbortRaisesWithExplicitTransactions(self):
 # Abort cancels the current sql statement but doesn't change anything
 # about the current transaction.
 self.cx.isolation_level = None # autocommit mode
 self.cu = self.cx.cursor()
 # Start an explicit transaction.
 self.cu.execute("BEGIN")
 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
 with self.assertRaises(sqlite.IntegrityError):
 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
 self.cx.commit()
 self.cu.execute("SELECT name, unique_name FROM test")
 # Expect the first two inserts to work, third to do nothing.
 self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)])

 def CheckOnConflictRollbackWithoutTransaction(self):
 # Start of implicit transaction
 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
 with self.assertRaises(sqlite.IntegrityError):
 self.cu.execute("INSERT OR ROLLBACK INTO test(unique_name) VALUES ('foo')")
 self.cu.execute("SELECT name, unique_name FROM test")
 # Implicit transaction is rolled back on error.
 self.assertEqual(self.cu.fetchall(), [])

 def CheckOnConflictAbortRaisesWithoutTransactions(self):
 # Abort cancels the current sql statement but doesn't change anything
 # about the current transaction.
 self.cu.execute("INSERT INTO test(name) VALUES ('abort_test')")
 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
 with self.assertRaises(sqlite.IntegrityError):
 self.cu.execute("INSERT OR ABORT INTO test(unique_name) VALUES ('foo')")
 # Make sure all other values were inserted.
 self.cu.execute("SELECT name, unique_name FROM test")
 self.assertEqual(self.cu.fetchall(), [('abort_test', None), (None, 'foo',)])

 def CheckOnConflictFail(self):
 self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')")
 with self.assertRaises(sqlite.IntegrityError):
 self.cu.execute("INSERT OR FAIL INTO test(unique_name) VALUES ('foo')")
 self.assertEqual(self.cu.fetchall(), [])

 def CheckOnConflictIgnore(self):
 self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')")
 # Nothing should happen.
 self.cu.execute("INSERT OR IGNORE INTO test(unique_name) VALUES ('foo')")
 self.cu.execute("SELECT unique_name FROM test")
 self.assertEqual(self.cu.fetchall(), [('foo',)])

 def CheckOnConflictReplace(self):
 self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Data!', 'foo')")
 # There shouldn't be an IntegrityError exception.
 self.cu.execute("INSERT OR REPLACE INTO test(name, unique_name) VALUES ('Very different data!', 'foo')")
 self.cu.execute("SELECT name, unique_name FROM test")
 self.assertEqual(self.cu.fetchall(), [('Very different data!', 'foo')])


def suite():
 module_suite = unittest.makeSuite(ModuleTests, "Check")
 connection_suite = unittest.makeSuite(ConnectionTests, "Check")
 cursor_suite = unittest.makeSuite(CursorTests, "Check")
 thread_suite = unittest.makeSuite(ThreadTests, "Check")
 constructor_suite = unittest.makeSuite(ConstructorTests, "Check")
 ext_suite = unittest.makeSuite(ExtensionTests, "Check")
 closed_con_suite = unittest.makeSuite(ClosedConTests, "Check")
 closed_cur_suite = unittest.makeSuite(ClosedCurTests, "Check")
 on_conflict_suite = unittest.makeSuite(SqliteOnConflictTests, "Check")
 return unittest.TestSuite((
 module_suite, connection_suite, cursor_suite, thread_suite,
 constructor_suite, ext_suite, closed_con_suite, closed_cur_suite,
 on_conflict_suite,
 ))

def test():
 runner = unittest.TextTestRunner()
 runner.run(suite())

if __name__ == "__main__":
 test()