mirror of
https://github.com/python/cpython.git
synced 2025-07-23 11:15:24 +00:00
Modernize sqlite3 tests
Update current tests that use old pattern with assertRaises to make them more maintainable.
This commit is contained in:
parent
0e1d6802ff
commit
1003b34c71
6 changed files with 62 additions and 257 deletions
|
@ -122,11 +122,8 @@ class ConnectionTests(unittest.TestCase):
|
||||||
|
|
||||||
def CheckFailedOpen(self):
|
def CheckFailedOpen(self):
|
||||||
YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db"
|
YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db"
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
con = sqlite.connect(YOU_CANNOT_OPEN_THIS)
|
con = sqlite.connect(YOU_CANNOT_OPEN_THIS)
|
||||||
except sqlite.OperationalError:
|
|
||||||
return
|
|
||||||
self.fail("should have raised an OperationalError")
|
|
||||||
|
|
||||||
def CheckClose(self):
|
def CheckClose(self):
|
||||||
self.cx.close()
|
self.cx.close()
|
||||||
|
@ -202,22 +199,12 @@ class CursorTests(unittest.TestCase):
|
||||||
self.cu.execute("delete from test")
|
self.cu.execute("delete from test")
|
||||||
|
|
||||||
def CheckExecuteIllegalSql(self):
|
def CheckExecuteIllegalSql(self):
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
self.cu.execute("select asdf")
|
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):
|
def CheckExecuteTooMuchSql(self):
|
||||||
try:
|
with self.assertRaises(sqlite.Warning):
|
||||||
self.cu.execute("select 5+4; select 4+5")
|
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):
|
def CheckExecuteTooMuchSql2(self):
|
||||||
self.cu.execute("select 5+4; -- foo bar")
|
self.cu.execute("select 5+4; -- foo bar")
|
||||||
|
@ -232,13 +219,8 @@ class CursorTests(unittest.TestCase):
|
||||||
""")
|
""")
|
||||||
|
|
||||||
def CheckExecuteWrongSqlArg(self):
|
def CheckExecuteWrongSqlArg(self):
|
||||||
try:
|
with self.assertRaises(ValueError):
|
||||||
self.cu.execute(42)
|
self.cu.execute(42)
|
||||||
self.fail("should have raised a ValueError")
|
|
||||||
except ValueError:
|
|
||||||
return
|
|
||||||
except:
|
|
||||||
self.fail("raised wrong exception.")
|
|
||||||
|
|
||||||
def CheckExecuteArgInt(self):
|
def CheckExecuteArgInt(self):
|
||||||
self.cu.execute("insert into test(id) values (?)", (42,))
|
self.cu.execute("insert into test(id) values (?)", (42,))
|
||||||
|
@ -263,27 +245,18 @@ class CursorTests(unittest.TestCase):
|
||||||
|
|
||||||
def CheckExecuteWrongNoOfArgs1(self):
|
def CheckExecuteWrongNoOfArgs1(self):
|
||||||
# too many parameters
|
# too many parameters
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.execute("insert into test(id) values (?)", (17, "Egon"))
|
self.cu.execute("insert into test(id) values (?)", (17, "Egon"))
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def CheckExecuteWrongNoOfArgs2(self):
|
def CheckExecuteWrongNoOfArgs2(self):
|
||||||
# too little parameters
|
# too little parameters
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.execute("insert into test(id) values (?)")
|
self.cu.execute("insert into test(id) values (?)")
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def CheckExecuteWrongNoOfArgs3(self):
|
def CheckExecuteWrongNoOfArgs3(self):
|
||||||
# no parameters, parameters are needed
|
# no parameters, parameters are needed
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.execute("insert into test(id) values (?)")
|
self.cu.execute("insert into test(id) values (?)")
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def CheckExecuteParamList(self):
|
def CheckExecuteParamList(self):
|
||||||
self.cu.execute("insert into test(name) values ('foo')")
|
self.cu.execute("insert into test(name) values ('foo')")
|
||||||
|
@ -322,27 +295,18 @@ class CursorTests(unittest.TestCase):
|
||||||
|
|
||||||
def CheckExecuteDictMappingTooLittleArgs(self):
|
def CheckExecuteDictMappingTooLittleArgs(self):
|
||||||
self.cu.execute("insert into test(name) values ('foo')")
|
self.cu.execute("insert into test(name) values ('foo')")
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.execute("select name from test where name=:name and id=:id", {"name": "foo"})
|
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):
|
def CheckExecuteDictMappingNoArgs(self):
|
||||||
self.cu.execute("insert into test(name) values ('foo')")
|
self.cu.execute("insert into test(name) values ('foo')")
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.execute("select name from test where name=:name")
|
self.cu.execute("select name from test where name=:name")
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def CheckExecuteDictMappingUnnamed(self):
|
def CheckExecuteDictMappingUnnamed(self):
|
||||||
self.cu.execute("insert into test(name) values ('foo')")
|
self.cu.execute("insert into test(name) values ('foo')")
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.execute("select name from test where name=?", {"name": "foo"})
|
self.cu.execute("select name from test where name=?", {"name": "foo"})
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def CheckClose(self):
|
def CheckClose(self):
|
||||||
self.cu.close()
|
self.cu.close()
|
||||||
|
@ -403,32 +367,16 @@ class CursorTests(unittest.TestCase):
|
||||||
self.cu.executemany("insert into test(income) values (?)", mygen())
|
self.cu.executemany("insert into test(income) values (?)", mygen())
|
||||||
|
|
||||||
def CheckExecuteManyWrongSqlArg(self):
|
def CheckExecuteManyWrongSqlArg(self):
|
||||||
try:
|
with self.assertRaises(ValueError):
|
||||||
self.cu.executemany(42, [(3,)])
|
self.cu.executemany(42, [(3,)])
|
||||||
self.fail("should have raised a ValueError")
|
|
||||||
except ValueError:
|
|
||||||
return
|
|
||||||
except:
|
|
||||||
self.fail("raised wrong exception.")
|
|
||||||
|
|
||||||
def CheckExecuteManySelect(self):
|
def CheckExecuteManySelect(self):
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
self.cu.executemany("select ?", [(3,)])
|
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):
|
def CheckExecuteManyNotIterable(self):
|
||||||
try:
|
with self.assertRaises(TypeError):
|
||||||
self.cu.executemany("insert into test(income) values (?)", 42)
|
self.cu.executemany("insert into test(income) values (?)", 42)
|
||||||
self.fail("should have raised a TypeError")
|
|
||||||
except TypeError:
|
|
||||||
return
|
|
||||||
except Exception as e:
|
|
||||||
print("raised", e.__class__)
|
|
||||||
self.fail("raised wrong exception.")
|
|
||||||
|
|
||||||
def CheckFetchIter(self):
|
def CheckFetchIter(self):
|
||||||
# Optional DB-API extension.
|
# Optional DB-API extension.
|
||||||
|
@ -505,22 +453,15 @@ class CursorTests(unittest.TestCase):
|
||||||
self.assertEqual(self.cu.connection, self.cx)
|
self.assertEqual(self.cu.connection, self.cx)
|
||||||
|
|
||||||
def CheckWrongCursorCallable(self):
|
def CheckWrongCursorCallable(self):
|
||||||
try:
|
with self.assertRaises(TypeError):
|
||||||
def f(): pass
|
def f(): pass
|
||||||
cur = self.cx.cursor(f)
|
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):
|
def CheckCursorWrongClass(self):
|
||||||
class Foo: pass
|
class Foo: pass
|
||||||
foo = Foo()
|
foo = Foo()
|
||||||
try:
|
with self.assertRaises(TypeError):
|
||||||
cur = sqlite.Cursor(foo)
|
cur = sqlite.Cursor(foo)
|
||||||
self.fail("should have raised a ValueError")
|
|
||||||
except TypeError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
@unittest.skipUnless(threading, 'This test requires threading.')
|
@unittest.skipUnless(threading, 'This test requires threading.')
|
||||||
class ThreadTests(unittest.TestCase):
|
class ThreadTests(unittest.TestCase):
|
||||||
|
@ -719,22 +660,14 @@ class ExtensionTests(unittest.TestCase):
|
||||||
def CheckScriptSyntaxError(self):
|
def CheckScriptSyntaxError(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
cur = con.cursor()
|
cur = con.cursor()
|
||||||
raised = False
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
try:
|
|
||||||
cur.executescript("create table test(x); asdf; create table test2(x)")
|
cur.executescript("create table test(x); asdf; create table test2(x)")
|
||||||
except sqlite.OperationalError:
|
|
||||||
raised = True
|
|
||||||
self.assertEqual(raised, True, "should have raised an exception")
|
|
||||||
|
|
||||||
def CheckScriptErrorNormal(self):
|
def CheckScriptErrorNormal(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
cur = con.cursor()
|
cur = con.cursor()
|
||||||
raised = False
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
try:
|
|
||||||
cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")
|
cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")
|
||||||
except sqlite.OperationalError:
|
|
||||||
raised = True
|
|
||||||
self.assertEqual(raised, True, "should have raised an exception")
|
|
||||||
|
|
||||||
def CheckCursorExecutescriptAsBytes(self):
|
def CheckCursorExecutescriptAsBytes(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
|
@ -772,59 +705,34 @@ class ClosedConTests(unittest.TestCase):
|
||||||
def CheckClosedConCursor(self):
|
def CheckClosedConCursor(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
cur = con.cursor()
|
cur = con.cursor()
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedConCommit(self):
|
def CheckClosedConCommit(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.commit()
|
con.commit()
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedConRollback(self):
|
def CheckClosedConRollback(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.rollback()
|
con.rollback()
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedCurExecute(self):
|
def CheckClosedCurExecute(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
cur = con.cursor()
|
cur = con.cursor()
|
||||||
con.close()
|
con.close()
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
cur.execute("select 4")
|
cur.execute("select 4")
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedCreateFunction(self):
|
def CheckClosedCreateFunction(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
def f(x): return 17
|
def f(x): return 17
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.create_function("foo", 1, f)
|
con.create_function("foo", 1, f)
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedCreateAggregate(self):
|
def CheckClosedCreateAggregate(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
|
@ -836,49 +744,29 @@ class ClosedConTests(unittest.TestCase):
|
||||||
pass
|
pass
|
||||||
def finalize(self):
|
def finalize(self):
|
||||||
return 17
|
return 17
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.create_aggregate("foo", 1, Agg)
|
con.create_aggregate("foo", 1, Agg)
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedSetAuthorizer(self):
|
def CheckClosedSetAuthorizer(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
def authorizer(*args):
|
def authorizer(*args):
|
||||||
return sqlite.DENY
|
return sqlite.DENY
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.set_authorizer(authorizer)
|
con.set_authorizer(authorizer)
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedSetProgressCallback(self):
|
def CheckClosedSetProgressCallback(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
def progress(): pass
|
def progress(): pass
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.set_progress_handler(progress, 100)
|
con.set_progress_handler(progress, 100)
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
def CheckClosedCall(self):
|
def CheckClosedCall(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.close()
|
con.close()
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con()
|
con()
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError")
|
|
||||||
|
|
||||||
class ClosedCurTests(unittest.TestCase):
|
class ClosedCurTests(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
@ -900,15 +788,9 @@ class ClosedCurTests(unittest.TestCase):
|
||||||
else:
|
else:
|
||||||
params = []
|
params = []
|
||||||
|
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
method = getattr(cur, method_name)
|
method = getattr(cur, method_name)
|
||||||
|
|
||||||
method(*params)
|
method(*params)
|
||||||
self.fail("Should have raised a ProgrammingError: method " + method_name)
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("Should have raised a ProgrammingError: " + method_name)
|
|
||||||
|
|
||||||
def suite():
|
def suite():
|
||||||
module_suite = unittest.makeSuite(ModuleTests, "Check")
|
module_suite = unittest.makeSuite(ModuleTests, "Check")
|
||||||
|
|
|
@ -33,19 +33,14 @@ class CollationTests(unittest.TestCase):
|
||||||
|
|
||||||
def CheckCreateCollationNotCallable(self):
|
def CheckCreateCollationNotCallable(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
try:
|
with self.assertRaises(TypeError) as cm:
|
||||||
con.create_collation("X", 42)
|
con.create_collation("X", 42)
|
||||||
self.fail("should have raised a TypeError")
|
self.assertEqual(str(cm.exception), 'parameter must be callable')
|
||||||
except TypeError as e:
|
|
||||||
self.assertEqual(e.args[0], "parameter must be callable")
|
|
||||||
|
|
||||||
def CheckCreateCollationNotAscii(self):
|
def CheckCreateCollationNotAscii(self):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
con.create_collation("collä", lambda x, y: (x > y) - (x < y))
|
con.create_collation("collä", lambda x, y: (x > y) - (x < y))
|
||||||
self.fail("should have raised a ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError as e:
|
|
||||||
pass
|
|
||||||
|
|
||||||
@unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 1),
|
@unittest.skipIf(sqlite.sqlite_version_info < (3, 2, 1),
|
||||||
'old SQLite versions crash on this test')
|
'old SQLite versions crash on this test')
|
||||||
|
@ -70,11 +65,9 @@ class CollationTests(unittest.TestCase):
|
||||||
self.fail("the expected order was not returned")
|
self.fail("the expected order was not returned")
|
||||||
|
|
||||||
con.create_collation("mycoll", None)
|
con.create_collation("mycoll", None)
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
result = con.execute(sql).fetchall()
|
result = con.execute(sql).fetchall()
|
||||||
self.fail("should have raised an OperationalError")
|
self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll')
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
self.assertEqual(e.args[0].lower(), "no such collation sequence: mycoll")
|
|
||||||
|
|
||||||
def CheckCollationReturnsLargeInteger(self):
|
def CheckCollationReturnsLargeInteger(self):
|
||||||
def mycoll(x, y):
|
def mycoll(x, y):
|
||||||
|
@ -106,8 +99,8 @@ class CollationTests(unittest.TestCase):
|
||||||
result = con.execute("""
|
result = con.execute("""
|
||||||
select x from (select 'a' as x union select 'b' as x) order by x collate mycoll
|
select x from (select 'a' as x union select 'b' as x) order by x collate mycoll
|
||||||
""").fetchall()
|
""").fetchall()
|
||||||
if result[0][0] != 'b' or result[1][0] != 'a':
|
self.assertEqual(result[0][0], 'b')
|
||||||
self.fail("wrong collation function is used")
|
self.assertEqual(result[1][0], 'a')
|
||||||
|
|
||||||
def CheckDeregisterCollation(self):
|
def CheckDeregisterCollation(self):
|
||||||
"""
|
"""
|
||||||
|
@ -117,12 +110,9 @@ class CollationTests(unittest.TestCase):
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
con.create_collation("mycoll", lambda x, y: (x > y) - (x < y))
|
con.create_collation("mycoll", lambda x, y: (x > y) - (x < y))
|
||||||
con.create_collation("mycoll", None)
|
con.create_collation("mycoll", None)
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
con.execute("select 'a' as x union select 'b' as x order by x collate mycoll")
|
con.execute("select 'a' as x union select 'b' as x order by x collate mycoll")
|
||||||
self.fail("should have raised an OperationalError")
|
self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll')
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
if not e.args[0].startswith("no such collation sequence"):
|
|
||||||
self.fail("wrong OperationalError raised")
|
|
||||||
|
|
||||||
class ProgressTests(unittest.TestCase):
|
class ProgressTests(unittest.TestCase):
|
||||||
def CheckProgressHandlerUsed(self):
|
def CheckProgressHandlerUsed(self):
|
||||||
|
|
|
@ -170,14 +170,8 @@ class RegressionTests(unittest.TestCase):
|
||||||
|
|
||||||
con = sqlite.connect(":memory:")
|
con = sqlite.connect(":memory:")
|
||||||
cur = Cursor(con)
|
cur = Cursor(con)
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
cur.execute("select 4+5").fetchall()
|
cur.execute("select 4+5").fetchall()
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
|
|
||||||
|
|
||||||
def CheckStrSubclass(self):
|
def CheckStrSubclass(self):
|
||||||
"""
|
"""
|
||||||
|
@ -196,13 +190,8 @@ class RegressionTests(unittest.TestCase):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
con = Connection(":memory:")
|
con = Connection(":memory:")
|
||||||
try:
|
with self.assertRaises(sqlite.ProgrammingError):
|
||||||
cur = con.cursor()
|
cur = con.cursor()
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
except sqlite.ProgrammingError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised ProgrammingError")
|
|
||||||
|
|
||||||
def CheckCursorRegistration(self):
|
def CheckCursorRegistration(self):
|
||||||
"""
|
"""
|
||||||
|
@ -223,13 +212,8 @@ class RegressionTests(unittest.TestCase):
|
||||||
cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)])
|
cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)])
|
||||||
cur.execute("select x from foo")
|
cur.execute("select x from foo")
|
||||||
con.rollback()
|
con.rollback()
|
||||||
try:
|
with self.assertRaises(sqlite.InterfaceError):
|
||||||
cur.fetchall()
|
cur.fetchall()
|
||||||
self.fail("should have raised InterfaceError")
|
|
||||||
except sqlite.InterfaceError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised InterfaceError")
|
|
||||||
|
|
||||||
def CheckAutoCommit(self):
|
def CheckAutoCommit(self):
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -118,13 +118,8 @@ class TransactionTests(unittest.TestCase):
|
||||||
return
|
return
|
||||||
self.cur1.execute("create table test(i)")
|
self.cur1.execute("create table test(i)")
|
||||||
self.cur1.execute("insert into test(i) values (5)")
|
self.cur1.execute("insert into test(i) values (5)")
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
self.cur2.execute("insert into test(i) values (5)")
|
self.cur2.execute("insert into test(i) values (5)")
|
||||||
self.fail("should have raised an OperationalError")
|
|
||||||
except sqlite.OperationalError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised an OperationalError")
|
|
||||||
|
|
||||||
def CheckLocking(self):
|
def CheckLocking(self):
|
||||||
"""
|
"""
|
||||||
|
@ -137,13 +132,8 @@ class TransactionTests(unittest.TestCase):
|
||||||
return
|
return
|
||||||
self.cur1.execute("create table test(i)")
|
self.cur1.execute("create table test(i)")
|
||||||
self.cur1.execute("insert into test(i) values (5)")
|
self.cur1.execute("insert into test(i) values (5)")
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
self.cur2.execute("insert into test(i) values (5)")
|
self.cur2.execute("insert into test(i) values (5)")
|
||||||
self.fail("should have raised an OperationalError")
|
|
||||||
except sqlite.OperationalError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised an OperationalError")
|
|
||||||
# NO self.con2.rollback() HERE!!!
|
# NO self.con2.rollback() HERE!!!
|
||||||
self.con1.commit()
|
self.con1.commit()
|
||||||
|
|
||||||
|
@ -159,13 +149,8 @@ class TransactionTests(unittest.TestCase):
|
||||||
cur.execute("select 1 union select 2 union select 3")
|
cur.execute("select 1 union select 2 union select 3")
|
||||||
|
|
||||||
con.rollback()
|
con.rollback()
|
||||||
try:
|
with self.assertRaises(sqlite.InterfaceError):
|
||||||
cur.fetchall()
|
cur.fetchall()
|
||||||
self.fail("InterfaceError should have been raised")
|
|
||||||
except sqlite.InterfaceError as e:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("InterfaceError should have been raised")
|
|
||||||
|
|
||||||
class SpecialCommandTests(unittest.TestCase):
|
class SpecialCommandTests(unittest.TestCase):
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
|
|
|
@ -185,24 +185,14 @@ class DeclTypesTests(unittest.TestCase):
|
||||||
def CheckUnsupportedSeq(self):
|
def CheckUnsupportedSeq(self):
|
||||||
class Bar: pass
|
class Bar: pass
|
||||||
val = Bar()
|
val = Bar()
|
||||||
try:
|
with self.assertRaises(sqlite.InterfaceError):
|
||||||
self.cur.execute("insert into test(f) values (?)", (val,))
|
self.cur.execute("insert into test(f) values (?)", (val,))
|
||||||
self.fail("should have raised an InterfaceError")
|
|
||||||
except sqlite.InterfaceError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised an InterfaceError")
|
|
||||||
|
|
||||||
def CheckUnsupportedDict(self):
|
def CheckUnsupportedDict(self):
|
||||||
class Bar: pass
|
class Bar: pass
|
||||||
val = Bar()
|
val = Bar()
|
||||||
try:
|
with self.assertRaises(sqlite.InterfaceError):
|
||||||
self.cur.execute("insert into test(f) values (:val)", {"val": val})
|
self.cur.execute("insert into test(f) values (:val)", {"val": val})
|
||||||
self.fail("should have raised an InterfaceError")
|
|
||||||
except sqlite.InterfaceError:
|
|
||||||
pass
|
|
||||||
except:
|
|
||||||
self.fail("should have raised an InterfaceError")
|
|
||||||
|
|
||||||
def CheckBlob(self):
|
def CheckBlob(self):
|
||||||
# default
|
# default
|
||||||
|
|
|
@ -162,11 +162,8 @@ class FunctionTests(unittest.TestCase):
|
||||||
self.con.close()
|
self.con.close()
|
||||||
|
|
||||||
def CheckFuncErrorOnCreate(self):
|
def CheckFuncErrorOnCreate(self):
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
self.con.create_function("bla", -100, lambda x: 2*x)
|
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 CheckFuncRefCount(self):
|
||||||
def getfunc():
|
def getfunc():
|
||||||
|
@ -231,12 +228,10 @@ class FunctionTests(unittest.TestCase):
|
||||||
|
|
||||||
def CheckFuncException(self):
|
def CheckFuncException(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
cur.execute("select raiseexception()")
|
cur.execute("select raiseexception()")
|
||||||
cur.fetchone()
|
cur.fetchone()
|
||||||
self.fail("should have raised OperationalError")
|
self.assertEqual(str(cm.exception), 'user-defined function raised exception')
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
self.assertEqual(e.args[0], 'user-defined function raised exception')
|
|
||||||
|
|
||||||
def CheckParamString(self):
|
def CheckParamString(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
|
@ -312,55 +307,42 @@ class AggregateTests(unittest.TestCase):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def CheckAggrErrorOnCreate(self):
|
def CheckAggrErrorOnCreate(self):
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError):
|
||||||
self.con.create_function("bla", -100, AggrSum)
|
self.con.create_function("bla", -100, AggrSum)
|
||||||
self.fail("should have raised an OperationalError")
|
|
||||||
except sqlite.OperationalError:
|
|
||||||
pass
|
|
||||||
|
|
||||||
def CheckAggrNoStep(self):
|
def CheckAggrNoStep(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
try:
|
with self.assertRaises(AttributeError) as cm:
|
||||||
cur.execute("select nostep(t) from test")
|
cur.execute("select nostep(t) from test")
|
||||||
self.fail("should have raised an AttributeError")
|
self.assertEqual(str(cm.exception), "'AggrNoStep' object has no attribute 'step'")
|
||||||
except AttributeError as e:
|
|
||||||
self.assertEqual(e.args[0], "'AggrNoStep' object has no attribute 'step'")
|
|
||||||
|
|
||||||
def CheckAggrNoFinalize(self):
|
def CheckAggrNoFinalize(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
cur.execute("select nofinalize(t) from test")
|
cur.execute("select nofinalize(t) from test")
|
||||||
val = cur.fetchone()[0]
|
val = cur.fetchone()[0]
|
||||||
self.fail("should have raised an OperationalError")
|
self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error")
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error")
|
|
||||||
|
|
||||||
def CheckAggrExceptionInInit(self):
|
def CheckAggrExceptionInInit(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
cur.execute("select excInit(t) from test")
|
cur.execute("select excInit(t) from test")
|
||||||
val = cur.fetchone()[0]
|
val = cur.fetchone()[0]
|
||||||
self.fail("should have raised an OperationalError")
|
self.assertEqual(str(cm.exception), "user-defined aggregate's '__init__' method raised error")
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
self.assertEqual(e.args[0], "user-defined aggregate's '__init__' method raised error")
|
|
||||||
|
|
||||||
def CheckAggrExceptionInStep(self):
|
def CheckAggrExceptionInStep(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
cur.execute("select excStep(t) from test")
|
cur.execute("select excStep(t) from test")
|
||||||
val = cur.fetchone()[0]
|
val = cur.fetchone()[0]
|
||||||
self.fail("should have raised an OperationalError")
|
self.assertEqual(str(cm.exception), "user-defined aggregate's 'step' method raised error")
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
self.assertEqual(e.args[0], "user-defined aggregate's 'step' method raised error")
|
|
||||||
|
|
||||||
def CheckAggrExceptionInFinalize(self):
|
def CheckAggrExceptionInFinalize(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
try:
|
with self.assertRaises(sqlite.OperationalError) as cm:
|
||||||
cur.execute("select excFinalize(t) from test")
|
cur.execute("select excFinalize(t) from test")
|
||||||
val = cur.fetchone()[0]
|
val = cur.fetchone()[0]
|
||||||
self.fail("should have raised an OperationalError")
|
self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error")
|
||||||
except sqlite.OperationalError as e:
|
|
||||||
self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error")
|
|
||||||
|
|
||||||
def CheckAggrCheckParamStr(self):
|
def CheckAggrCheckParamStr(self):
|
||||||
cur = self.con.cursor()
|
cur = self.con.cursor()
|
||||||
|
@ -433,22 +415,14 @@ class AuthorizerTests(unittest.TestCase):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def test_table_access(self):
|
def test_table_access(self):
|
||||||
try:
|
with self.assertRaises(sqlite.DatabaseError) as cm:
|
||||||
self.con.execute("select * from t2")
|
self.con.execute("select * from t2")
|
||||||
except sqlite.DatabaseError as e:
|
self.assertIn('prohibited', str(cm.exception))
|
||||||
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):
|
def test_column_access(self):
|
||||||
try:
|
with self.assertRaises(sqlite.DatabaseError) as cm:
|
||||||
self.con.execute("select c2 from t1")
|
self.con.execute("select c2 from t1")
|
||||||
except sqlite.DatabaseError as e:
|
self.assertIn('prohibited', str(cm.exception))
|
||||||
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):
|
class AuthorizerRaiseExceptionTests(AuthorizerTests):
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue