diff options
Diffstat (limited to 'Lib/sqlite3/test')
| -rw-r--r-- | Lib/sqlite3/test/dbapi.py | 193 | ||||
| -rw-r--r-- | Lib/sqlite3/test/factory.py | 18 | ||||
| -rw-r--r-- | Lib/sqlite3/test/hooks.py | 32 | ||||
| -rw-r--r-- | Lib/sqlite3/test/regression.py | 36 | ||||
| -rw-r--r-- | Lib/sqlite3/test/transactions.py | 21 | ||||
| -rw-r--r-- | Lib/sqlite3/test/types.py | 14 | ||||
| -rw-r--r-- | Lib/sqlite3/test/userfunctions.py | 93 |
7 files changed, 136 insertions, 271 deletions
diff --git a/Lib/sqlite3/test/dbapi.py b/Lib/sqlite3/test/dbapi.py index 04649fc5492..903e5990316 100644 --- a/Lib/sqlite3/test/dbapi.py +++ b/Lib/sqlite3/test/dbapi.py @@ -122,11 +122,8 @@ class ConnectionTests(unittest.TestCase): def CheckFailedOpen(self): YOU_CANNOT_OPEN_THIS = "/foo/bar/bla/23534/mydb.db" - try: + with self.assertRaises(sqlite.OperationalError): con = sqlite.connect(YOU_CANNOT_OPEN_THIS) - except sqlite.OperationalError: - return - self.fail("should have raised an OperationalError") def CheckClose(self): self.cx.close() @@ -180,6 +177,12 @@ class ConnectionTests(unittest.TestCase): with self.assertRaises(sqlite.OperationalError): cx.execute('insert into test(id) values(1)') + def CheckSameThreadErrorOnOldVersion(self): + if sqlite.sqlite_version_info >= (3, 3, 1): + self.skipTest('test needs sqlite3 versions older than 3.3.1') + 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): @@ -196,22 +199,12 @@ class CursorTests(unittest.TestCase): self.cu.execute("delete from test") def CheckExecuteIllegalSql(self): - try: + with self.assertRaises(sqlite.OperationalError): 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: + with self.assertRaises(sqlite.Warning): 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") @@ -226,13 +219,8 @@ class CursorTests(unittest.TestCase): """) def CheckExecuteWrongSqlArg(self): - try: + with self.assertRaises(ValueError): 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,)) @@ -250,29 +238,25 @@ class CursorTests(unittest.TestCase): 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 - try: + with self.assertRaises(sqlite.ProgrammingError): 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: + with self.assertRaises(sqlite.ProgrammingError): 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: + with self.assertRaises(sqlite.ProgrammingError): 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')") @@ -311,27 +295,18 @@ class CursorTests(unittest.TestCase): def CheckExecuteDictMappingTooLittleArgs(self): 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.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass def CheckExecuteDictMappingNoArgs(self): 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.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass def CheckExecuteDictMappingUnnamed(self): 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.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass def CheckClose(self): self.cu.close() @@ -360,8 +335,7 @@ class CursorTests(unittest.TestCase): 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") + self.assertLess(2, self.cx.total_changes, msg='total changes reported wrong value') # Checks for executemany: # Sequences are required by the DB-API, iterators @@ -392,32 +366,16 @@ class CursorTests(unittest.TestCase): self.cu.executemany("insert into test(income) values (?)", mygen()) def CheckExecuteManyWrongSqlArg(self): - try: + with self.assertRaises(ValueError): 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: + with self.assertRaises(sqlite.ProgrammingError): 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: + with self.assertRaises(TypeError): 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): # Optional DB-API extension. @@ -494,22 +452,15 @@ class CursorTests(unittest.TestCase): self.assertEqual(self.cu.connection, self.cx) def CheckWrongCursorCallable(self): - try: + with self.assertRaises(TypeError): 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: + with self.assertRaises(TypeError): cur = sqlite.Cursor(foo) - self.fail("should have raised a ValueError") - except TypeError: - pass @unittest.skipUnless(threading, 'This test requires threading.') class ThreadTests(unittest.TestCase): @@ -708,22 +659,21 @@ class ExtensionTests(unittest.TestCase): def CheckScriptSyntaxError(self): con = sqlite.connect(":memory:") cur = con.cursor() - raised = False - try: + with self.assertRaises(sqlite.OperationalError): 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): con = sqlite.connect(":memory:") cur = con.cursor() - raised = False - try: + with self.assertRaises(sqlite.OperationalError): 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): + 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:") @@ -754,59 +704,34 @@ class ClosedConTests(unittest.TestCase): def CheckClosedConCursor(self): con = sqlite.connect(":memory:") con.close() - try: + with self.assertRaises(sqlite.ProgrammingError): 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): con = sqlite.connect(":memory:") con.close() - try: + with self.assertRaises(sqlite.ProgrammingError): con.commit() - self.fail("Should have raised a ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("Should have raised a ProgrammingError") def CheckClosedConRollback(self): con = sqlite.connect(":memory:") con.close() - try: + with self.assertRaises(sqlite.ProgrammingError): con.rollback() - self.fail("Should have raised a ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("Should have raised a ProgrammingError") def CheckClosedCurExecute(self): con = sqlite.connect(":memory:") cur = con.cursor() con.close() - try: + with self.assertRaises(sqlite.ProgrammingError): 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): con = sqlite.connect(":memory:") con.close() def f(x): return 17 - try: + with self.assertRaises(sqlite.ProgrammingError): 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): con = sqlite.connect(":memory:") @@ -818,49 +743,29 @@ class ClosedConTests(unittest.TestCase): pass def finalize(self): return 17 - try: + with self.assertRaises(sqlite.ProgrammingError): 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): con = sqlite.connect(":memory:") con.close() def authorizer(*args): return sqlite.DENY - try: + with self.assertRaises(sqlite.ProgrammingError): 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): con = sqlite.connect(":memory:") con.close() def progress(): pass - try: + with self.assertRaises(sqlite.ProgrammingError): 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): con = sqlite.connect(":memory:") con.close() - try: + with self.assertRaises(sqlite.ProgrammingError): con() - self.fail("Should have raised a ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("Should have raised a ProgrammingError") class ClosedCurTests(unittest.TestCase): def setUp(self): @@ -882,15 +787,9 @@ class ClosedCurTests(unittest.TestCase): else: params = [] - try: + with self.assertRaises(sqlite.ProgrammingError): method = getattr(cur, method_name) - 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(): module_suite = unittest.makeSuite(ModuleTests, "Check") diff --git a/Lib/sqlite3/test/factory.py b/Lib/sqlite3/test/factory.py index a8348b4626f..f587596f644 100644 --- a/Lib/sqlite3/test/factory.py +++ b/Lib/sqlite3/test/factory.py @@ -111,6 +111,24 @@ class RowFactoryTests(unittest.TestCase): with self.assertRaises(IndexError): row[2**1000] + def CheckSqliteRowSlice(self): + # A sqlite.Row can be sliced like a list. + self.con.row_factory = sqlite.Row + row = self.con.execute("select 1, 2, 3, 4").fetchone() + self.assertEqual(row[0:0], ()) + self.assertEqual(row[0:1], (1,)) + self.assertEqual(row[1:3], (2, 3)) + self.assertEqual(row[3:1], ()) + # Explicit bounds are optional. + self.assertEqual(row[1:], (2, 3, 4)) + self.assertEqual(row[:3], (1, 2, 3)) + # Slices can use negative indices. + self.assertEqual(row[-2:-1], (3,)) + self.assertEqual(row[-2:], (3, 4)) + # Slicing supports steps. + self.assertEqual(row[0:4:2], (1, 3)) + self.assertEqual(row[3:0:-2], (4, 2)) + def CheckSqliteRowIter(self): """Checks if the row object is iterable""" self.con.row_factory = sqlite.Row diff --git a/Lib/sqlite3/test/hooks.py b/Lib/sqlite3/test/hooks.py index ede0becad4e..de69569d1c0 100644 --- a/Lib/sqlite3/test/hooks.py +++ b/Lib/sqlite3/test/hooks.py @@ -33,19 +33,14 @@ class CollationTests(unittest.TestCase): def CheckCreateCollationNotCallable(self): con = sqlite.connect(":memory:") - try: + with self.assertRaises(TypeError) as cm: con.create_collation("X", 42) - self.fail("should have raised a TypeError") - except TypeError as e: - self.assertEqual(e.args[0], "parameter must be callable") + self.assertEqual(str(cm.exception), 'parameter must be callable') def CheckCreateCollationNotAscii(self): con = sqlite.connect(":memory:") - try: + with self.assertRaises(sqlite.ProgrammingError): 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), 'old SQLite versions crash on this test') @@ -66,15 +61,13 @@ class CollationTests(unittest.TestCase): ) order by x collate mycoll """ result = con.execute(sql).fetchall() - if result[0][0] != "c" or result[1][0] != "b" or result[2][0] != "a": - self.fail("the expected order was not returned") + self.assertEqual(result, [('c',), ('b',), ('a',)], + msg='the expected order was not returned') con.create_collation("mycoll", None) - try: + with self.assertRaises(sqlite.OperationalError) as cm: result = con.execute(sql).fetchall() - self.fail("should have raised an OperationalError") - except sqlite.OperationalError as e: - self.assertEqual(e.args[0].lower(), "no such collation sequence: mycoll") + self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll') def CheckCollationReturnsLargeInteger(self): def mycoll(x, y): @@ -106,8 +99,8 @@ class CollationTests(unittest.TestCase): result = con.execute(""" select x from (select 'a' as x union select 'b' as x) order by x collate mycoll """).fetchall() - if result[0][0] != 'b' or result[1][0] != 'a': - self.fail("wrong collation function is used") + self.assertEqual(result[0][0], 'b') + self.assertEqual(result[1][0], 'a') def CheckDeregisterCollation(self): """ @@ -117,12 +110,9 @@ class CollationTests(unittest.TestCase): con = sqlite.connect(":memory:") con.create_collation("mycoll", lambda x, y: (x > y) - (x < y)) 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") - self.fail("should have raised an OperationalError") - except sqlite.OperationalError as e: - if not e.args[0].startswith("no such collation sequence"): - self.fail("wrong OperationalError raised") + self.assertEqual(str(cm.exception), 'no such collation sequence: mycoll') class ProgressTests(unittest.TestCase): def CheckProgressHandlerUsed(self): diff --git a/Lib/sqlite3/test/regression.py b/Lib/sqlite3/test/regression.py index eaaaa2c528e..0cf9002d8c0 100644 --- a/Lib/sqlite3/test/regression.py +++ b/Lib/sqlite3/test/regression.py @@ -73,7 +73,7 @@ class RegressionTests(unittest.TestCase): def CheckStatementFinalizationOnCloseDb(self): # pysqlite versions <= 2.3.3 only finalized statements in the statement # cache when closing the database. statements that were still - # referenced in cursors weren't closed an could provoke " + # referenced in cursors weren't closed and could provoke " # "OperationalError: Unable to close due to unfinalised statements". con = sqlite.connect(":memory:") cursors = [] @@ -134,17 +134,11 @@ class RegressionTests(unittest.TestCase): def CheckErrorMsgDecodeError(self): # When porting the module to Python 3.0, the error message about # decoding errors disappeared. This verifies they're back again. - failure = None - try: + with self.assertRaises(sqlite.OperationalError) as cm: self.con.execute("select 'xxx' || ? || 'yyy' colname", (bytes(bytearray([250])),)).fetchone() - failure = "should have raised an OperationalError with detailed description" - except sqlite.OperationalError as e: - msg = e.args[0] - if not msg.startswith("Could not decode to UTF-8 column 'colname' with text 'xxx"): - failure = "OperationalError did not have expected description text" - if failure: - self.fail(failure) + msg = "Could not decode to UTF-8 column 'colname' with text 'xxx" + self.assertIn(msg, str(cm.exception)) def CheckRegisterAdapter(self): """ @@ -170,14 +164,8 @@ class RegressionTests(unittest.TestCase): con = sqlite.connect(":memory:") cur = Cursor(con) - try: + with self.assertRaises(sqlite.ProgrammingError): 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): """ @@ -196,13 +184,8 @@ class RegressionTests(unittest.TestCase): pass con = Connection(":memory:") - try: + with self.assertRaises(sqlite.ProgrammingError): cur = con.cursor() - self.fail("should have raised ProgrammingError") - except sqlite.ProgrammingError: - pass - except: - self.fail("should have raised ProgrammingError") def CheckCursorRegistration(self): """ @@ -223,13 +206,8 @@ class RegressionTests(unittest.TestCase): cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)]) cur.execute("select x from foo") con.rollback() - try: + with self.assertRaises(sqlite.InterfaceError): cur.fetchall() - self.fail("should have raised InterfaceError") - except sqlite.InterfaceError: - pass - except: - self.fail("should have raised InterfaceError") def CheckAutoCommit(self): """ diff --git a/Lib/sqlite3/test/transactions.py b/Lib/sqlite3/test/transactions.py index feb4fa16969..eae26c4f960 100644 --- a/Lib/sqlite3/test/transactions.py +++ b/Lib/sqlite3/test/transactions.py @@ -118,13 +118,8 @@ class TransactionTests(unittest.TestCase): return self.cur1.execute("create table test(i)") 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.fail("should have raised an OperationalError") - except sqlite.OperationalError: - pass - except: - self.fail("should have raised an OperationalError") def CheckLocking(self): """ @@ -137,13 +132,8 @@ class TransactionTests(unittest.TestCase): return self.cur1.execute("create table test(i)") 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.fail("should have raised an OperationalError") - except sqlite.OperationalError: - pass - except: - self.fail("should have raised an OperationalError") # NO self.con2.rollback() HERE!!! self.con1.commit() @@ -159,13 +149,8 @@ class TransactionTests(unittest.TestCase): cur.execute("select 1 union select 2 union select 3") con.rollback() - try: + with self.assertRaises(sqlite.InterfaceError): 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): def setUp(self): diff --git a/Lib/sqlite3/test/types.py b/Lib/sqlite3/test/types.py index adad571b9a0..f7c8f9cee29 100644 --- a/Lib/sqlite3/test/types.py +++ b/Lib/sqlite3/test/types.py @@ -185,24 +185,14 @@ class DeclTypesTests(unittest.TestCase): def CheckUnsupportedSeq(self): class Bar: pass val = Bar() - try: + with self.assertRaises(sqlite.InterfaceError): 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): class Bar: pass val = Bar() - try: + with self.assertRaises(sqlite.InterfaceError): 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): # default diff --git a/Lib/sqlite3/test/userfunctions.py b/Lib/sqlite3/test/userfunctions.py index 69e2ec2991c..4075045b727 100644 --- a/Lib/sqlite3/test/userfunctions.py +++ b/Lib/sqlite3/test/userfunctions.py @@ -55,6 +55,9 @@ def func_isblob(v): def func_islonglong(v): return isinstance(v, int) and v >= 1<<31 +def func(*args): + return len(args) + class AggrNoStep: def __init__(self): pass @@ -111,6 +114,19 @@ class AggrCheckType: def finalize(self): return self.val +class AggrCheckTypes: + def __init__(self): + self.val = 0 + + def step(self, whichType, *vals): + theType = {"str": str, "int": int, "float": float, "None": type(None), + "blob": bytes} + for val in vals: + self.val += int(theType[whichType] is type(val)) + + def finalize(self): + return self.val + class AggrSum: def __init__(self): self.val = 0.0 @@ -140,16 +156,14 @@ class FunctionTests(unittest.TestCase): self.con.create_function("isnone", 1, func_isnone) self.con.create_function("isblob", 1, func_isblob) self.con.create_function("islonglong", 1, func_islonglong) + self.con.create_function("spam", -1, func) def tearDown(self): self.con.close() def CheckFuncErrorOnCreate(self): - try: + with self.assertRaises(sqlite.OperationalError): self.con.create_function("bla", -100, lambda x: 2*x) - self.fail("should have raised an OperationalError") - except sqlite.OperationalError: - pass def CheckFuncRefCount(self): def getfunc(): @@ -214,12 +228,10 @@ class FunctionTests(unittest.TestCase): def CheckFuncException(self): cur = self.con.cursor() - try: + with self.assertRaises(sqlite.OperationalError) as cm: cur.execute("select raiseexception()") cur.fetchone() - self.fail("should have raised OperationalError") - except sqlite.OperationalError as e: - self.assertEqual(e.args[0], 'user-defined function raised exception') + self.assertEqual(str(cm.exception), 'user-defined function raised exception') def CheckParamString(self): cur = self.con.cursor() @@ -257,6 +269,13 @@ class FunctionTests(unittest.TestCase): val = cur.fetchone()[0] self.assertEqual(val, 1) + def CheckAnyArguments(self): + cur = self.con.cursor() + cur.execute("select spam(?, ?)", (1, 2)) + val = cur.fetchone()[0] + self.assertEqual(val, 2) + + class AggregateTests(unittest.TestCase): def setUp(self): self.con = sqlite.connect(":memory:") @@ -279,6 +298,7 @@ class AggregateTests(unittest.TestCase): self.con.create_aggregate("excStep", 1, AggrExceptionInStep) self.con.create_aggregate("excFinalize", 1, AggrExceptionInFinalize) self.con.create_aggregate("checkType", 2, AggrCheckType) + self.con.create_aggregate("checkTypes", -1, AggrCheckTypes) self.con.create_aggregate("mysum", 1, AggrSum) def tearDown(self): @@ -287,55 +307,42 @@ class AggregateTests(unittest.TestCase): pass def CheckAggrErrorOnCreate(self): - try: + with self.assertRaises(sqlite.OperationalError): self.con.create_function("bla", -100, AggrSum) - self.fail("should have raised an OperationalError") - except sqlite.OperationalError: - pass def CheckAggrNoStep(self): cur = self.con.cursor() - try: + with self.assertRaises(AttributeError) as cm: cur.execute("select nostep(t) from test") - self.fail("should have raised an AttributeError") - except AttributeError as e: - self.assertEqual(e.args[0], "'AggrNoStep' object has no attribute 'step'") + self.assertEqual(str(cm.exception), "'AggrNoStep' object has no attribute 'step'") def CheckAggrNoFinalize(self): cur = self.con.cursor() - try: + with self.assertRaises(sqlite.OperationalError) as cm: cur.execute("select nofinalize(t) from test") val = cur.fetchone()[0] - self.fail("should have raised an OperationalError") - except sqlite.OperationalError as e: - self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error") + self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error") def CheckAggrExceptionInInit(self): cur = self.con.cursor() - try: + with self.assertRaises(sqlite.OperationalError) as cm: cur.execute("select excInit(t) from test") val = cur.fetchone()[0] - self.fail("should have raised an OperationalError") - except sqlite.OperationalError as e: - self.assertEqual(e.args[0], "user-defined aggregate's '__init__' method raised error") + self.assertEqual(str(cm.exception), "user-defined aggregate's '__init__' method raised error") def CheckAggrExceptionInStep(self): cur = self.con.cursor() - try: + with self.assertRaises(sqlite.OperationalError) as cm: cur.execute("select excStep(t) from test") val = cur.fetchone()[0] - self.fail("should have raised an OperationalError") - except sqlite.OperationalError as e: - self.assertEqual(e.args[0], "user-defined aggregate's 'step' method raised error") + self.assertEqual(str(cm.exception), "user-defined aggregate's 'step' method raised error") def CheckAggrExceptionInFinalize(self): cur = self.con.cursor() - try: + with self.assertRaises(sqlite.OperationalError) as cm: cur.execute("select excFinalize(t) from test") val = cur.fetchone()[0] - self.fail("should have raised an OperationalError") - except sqlite.OperationalError as e: - self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error") + self.assertEqual(str(cm.exception), "user-defined aggregate's 'finalize' method raised error") def CheckAggrCheckParamStr(self): cur = self.con.cursor() @@ -349,6 +356,12 @@ class AggregateTests(unittest.TestCase): val = cur.fetchone()[0] self.assertEqual(val, 1) + def CheckAggrCheckParamsInt(self): + cur = self.con.cursor() + cur.execute("select checkTypes('int', ?, ?)", (42, 24)) + val = cur.fetchone()[0] + self.assertEqual(val, 2) + def CheckAggrCheckParamFloat(self): cur = self.con.cursor() cur.execute("select checkType('float', ?)", (3.14,)) @@ -402,22 +415,14 @@ class AuthorizerTests(unittest.TestCase): pass def test_table_access(self): - try: + with self.assertRaises(sqlite.DatabaseError) as cm: self.con.execute("select * from t2") - except sqlite.DatabaseError as e: - if not e.args[0].endswith("prohibited"): - self.fail("wrong exception text: %s" % e.args[0]) - return - self.fail("should have raised an exception due to missing privileges") + self.assertIn('prohibited', str(cm.exception)) def test_column_access(self): - try: + with self.assertRaises(sqlite.DatabaseError) as cm: self.con.execute("select c2 from t1") - except sqlite.DatabaseError as e: - if not e.args[0].endswith("prohibited"): - self.fail("wrong exception text: %s" % e.args[0]) - return - self.fail("should have raised an exception due to missing privileges") + self.assertIn('prohibited', str(cm.exception)) class AuthorizerRaiseExceptionTests(AuthorizerTests): @staticmethod |