from zipimport import zipimporter
def iter_zipimport_modules(importer, prefix=''):
- dirlist = zipimport._zip_directory_cache[importer.archive].keys()
- dirlist.sort()
+ dirlist = sorted(zipimport._zip_directory_cache[importer.archive])
_prefix = importer.prefix
plen = len(_prefix)
yielded = {}
"Warning is not a subclass of Exception")
def CheckError(self):
- self.failUnless(issubclass(sqlite.Error, Exception),
+ self.assertTrue(issubclass(sqlite.Error, Exception),
"Error is not a subclass of Exception")
def CheckInterfaceError(self):
- self.failUnless(issubclass(sqlite.InterfaceError, sqlite.Error),
+ self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error),
"InterfaceError is not a subclass of Error")
def CheckDatabaseError(self):
- self.failUnless(issubclass(sqlite.DatabaseError, sqlite.Error),
+ self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error),
"DatabaseError is not a subclass of Error")
def CheckDataError(self):
- self.failUnless(issubclass(sqlite.DataError, sqlite.DatabaseError),
+ self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError),
"DataError is not a subclass of DatabaseError")
def CheckOperationalError(self):
- self.failUnless(issubclass(sqlite.OperationalError, sqlite.DatabaseError),
+ self.assertTrue(issubclass(sqlite.OperationalError, sqlite.DatabaseError),
"OperationalError is not a subclass of DatabaseError")
def CheckIntegrityError(self):
- self.failUnless(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),
+ self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError),
"IntegrityError is not a subclass of DatabaseError")
def CheckInternalError(self):
- self.failUnless(issubclass(sqlite.InternalError, sqlite.DatabaseError),
+ self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError),
"InternalError is not a subclass of DatabaseError")
def CheckProgrammingError(self):
- self.failUnless(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),
+ self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError),
"ProgrammingError is not a subclass of DatabaseError")
def CheckNotSupportedError(self):
- self.failUnless(issubclass(sqlite.NotSupportedError,
+ self.assertTrue(issubclass(sqlite.NotSupportedError,
sqlite.DatabaseError),
"NotSupportedError is not a subclass of DatabaseError")
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)
+ 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)
class CursorTests(unittest.TestCase):
def setUp(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")
+ self.assertEqual(row[0], "foo")
def CheckExecuteParamSequence(self):
class L(object):
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")
+ 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.failUnlessEqual(row[0], "foo")
+ self.assertEqual(row[0], "foo")
def CheckExecuteDictMapping_Mapping(self):
class D(dict):
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")
+ self.assertEqual(row[0], "foo")
def CheckExecuteDictMappingTooLittleArgs(self):
self.cu.execute("insert into test(name) values ('foo')")
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)
+ self.assertEqual(self.cu.rowcount, 2)
def CheckRowcountSelect(self):
"""
has thus to be -1.
"""
self.cu.execute("select 5 union select 6")
- self.failUnlessEqual(self.cu.rowcount, -1)
+ 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.failUnlessEqual(self.cu.rowcount, 3)
+ self.assertEqual(self.cu.rowcount, 3)
def CheckTotalChanges(self):
self.cu.execute("insert into test(name) values ('foo')")
lst = []
for row in self.cu:
lst.append(row[0])
- self.failUnlessEqual(lst[0], 5)
- self.failUnlessEqual(lst[1], 6)
+ 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.failUnlessEqual(row[0], "foo")
+ self.assertEqual(row[0], "foo")
row = self.cu.fetchone()
- self.failUnlessEqual(row, None)
+ self.assertEqual(row, None)
def CheckFetchoneNoStatement(self):
cur = self.cx.cursor()
row = cur.fetchone()
- self.failUnlessEqual(row, None)
+ self.assertEqual(row, None)
def CheckArraySize(self):
# must default ot 1
- self.failUnlessEqual(self.cu.arraysize, 1)
+ self.assertEqual(self.cu.arraysize, 1)
# now set to 2
self.cu.arraysize = 2
self.cu.execute("select name from test")
res = self.cu.fetchmany()
- self.failUnlessEqual(len(res), 2)
+ self.assertEqual(len(res), 2)
def CheckFetchmany(self):
self.cu.execute("select name from test")
res = self.cu.fetchmany(100)
- self.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
res = self.cu.fetchmany(100)
- self.failUnlessEqual(res, [])
+ 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.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
def CheckFetchall(self):
self.cu.execute("select name from test")
res = self.cu.fetchall()
- self.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
res = self.cu.fetchall()
- self.failUnlessEqual(res, [])
+ self.assertEqual(res, [])
def CheckSetinputsizes(self):
self.cu.setinputsizes([3, 4, 5])
def CheckCursorConnection(self):
# Optional DB-API extension.
- self.failUnlessEqual(self.cu.connection, self.cx)
+ self.assertEqual(self.cu.connection, self.cx)
def CheckWrongCursorCallable(self):
try:
""")
cur.execute("select i from a")
res = cur.fetchone()[0]
- self.failUnlessEqual(res, 5)
+ self.assertEqual(res, 5)
def CheckScriptErrorIncomplete(self):
con = sqlite.connect(":memory:")
cur.executescript("create table test(sadfsadfdsa")
except sqlite.ProgrammingError:
raised = True
- self.failUnlessEqual(raised, True, "should have raised an exception")
+ self.assertEqual(raised, True, "should have raised an exception")
def CheckScriptErrorNormal(self):
con = sqlite.connect(":memory:")
cur.executescript("create table test(sadfsadfdsa); select foo from hurz;")
except sqlite.OperationalError:
raised = True
- self.failUnlessEqual(raised, True, "should have raised an exception")
+ self.assertEqual(raised, True, "should have raised an exception")
def CheckConnectionExecute(self):
con = sqlite.connect(":memory:")
result = con.execute("select 5").fetchone()[0]
- self.failUnlessEqual(result, 5, "Basic test of Connection.execute")
+ 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.failUnlessEqual(result[0][0], 3, "Basic test of Connection.executemany")
- self.failUnlessEqual(result[1][0], 4, "Basic test of Connection.executemany")
+ 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.failUnlessEqual(result, 5, "Basic test of Connection.executescript")
+ self.assertEqual(result, 5, "Basic test of Connection.executescript")
class ClosedTests(unittest.TestCase):
def setUp(self):
self.con.close()
def CheckIsInstance(self):
- self.failUnless(isinstance(self.con,
+ self.assertTrue(isinstance(self.con,
MyConnection),
"connection is not instance of MyConnection")
def CheckIsInstance(self):
cur = self.con.cursor(factory=MyCursor)
- self.failUnless(isinstance(cur,
+ self.assertTrue(isinstance(cur,
MyCursor),
"cursor is not instance of MyCursor")
cur = self.con.cursor(factory=MyCursor)
cur.execute("select 4+5 as foo")
row = cur.fetchone()
- self.failUnless(isinstance(row,
+ self.assertTrue(isinstance(row,
dict),
"row is not instance of dict")
cur.close()
def CheckCustomFactory(self):
self.con.row_factory = lambda cur, row: list(row)
row = self.con.execute("select 1, 2").fetchone()
- self.failUnless(isinstance(row,
+ self.assertTrue(isinstance(row,
list),
"row is not instance of list")
def CheckSqliteRowIndex(self):
self.con.row_factory = sqlite.Row
row = self.con.execute("select 1 as a, 2 as b").fetchone()
- self.failUnless(isinstance(row,
+ self.assertTrue(isinstance(row,
sqlite.Row),
"row is not instance of sqlite.Row")
col1, col2 = row["a"], row["b"]
- self.failUnless(col1 == 1, "by name: wrong result for column 'a'")
- self.failUnless(col2 == 2, "by name: wrong result for column 'a'")
+ self.assertTrue(col1 == 1, "by name: wrong result for column 'a'")
+ self.assertTrue(col2 == 2, "by name: wrong result for column 'a'")
col1, col2 = row["A"], row["B"]
- self.failUnless(col1 == 1, "by name: wrong result for column 'A'")
- self.failUnless(col2 == 2, "by name: wrong result for column 'B'")
+ self.assertTrue(col1 == 1, "by name: wrong result for column 'A'")
+ self.assertTrue(col2 == 2, "by name: wrong result for column 'B'")
col1, col2 = row[0], row[1]
- self.failUnless(col1 == 1, "by index: wrong result for column 0")
- self.failUnless(col2 == 2, "by index: wrong result for column 1")
+ self.assertTrue(col1 == 1, "by index: wrong result for column 0")
+ self.assertTrue(col2 == 2, "by index: wrong result for column 1")
def CheckSqliteRowIter(self):
"""Checks if the row object is iterable"""
self.con.row_factory = sqlite.Row
row = self.con.execute("select 1 as a, 2 as b").fetchone()
d = dict(row)
- self.failUnlessEqual(d["a"], row["a"])
- self.failUnlessEqual(d["b"], row["b"])
+ self.assertEqual(d["a"], row["a"])
+ self.assertEqual(d["b"], row["b"])
def CheckSqliteRowHashCmp(self):
"""Checks if the row object compares and hashes correctly"""
row_2 = self.con.execute("select 1 as a, 2 as b").fetchone()
row_3 = self.con.execute("select 1 as a, 3 as b").fetchone()
- self.failUnless(row_1 == row_1)
- self.failUnless(row_1 == row_2)
- self.failUnless(row_2 != row_3)
+ self.assertTrue(row_1 == row_1)
+ self.assertTrue(row_1 == row_2)
+ self.assertTrue(row_2 != row_3)
- self.failIf(row_1 != row_1)
- self.failIf(row_1 != row_2)
- self.failIf(row_2 == row_3)
+ self.assertFalse(row_1 != row_1)
+ self.assertFalse(row_1 != row_2)
+ self.assertFalse(row_2 == row_3)
- self.failUnlessEqual(row_1, row_2)
- self.failUnlessEqual(hash(row_1), hash(row_2))
- self.failIfEqual(row_1, row_3)
- self.failIfEqual(hash(row_1), hash(row_3))
+ self.assertEqual(row_1, row_2)
+ self.assertEqual(hash(row_1), hash(row_2))
+ self.assertNotEqual(row_1, row_3)
+ self.assertNotEqual(hash(row_1), hash(row_3))
def tearDown(self):
self.con.close()
def CheckUnicode(self):
austria = "Österreich"
row = self.con.execute("select ?", (austria,)).fetchone()
- self.failUnless(type(row[0]) == str, "type of row[0] must be unicode")
+ self.assertTrue(type(row[0]) == str, "type of row[0] must be unicode")
def CheckString(self):
self.con.text_factory = bytes
austria = "Österreich"
row = self.con.execute("select ?", (austria,)).fetchone()
- self.failUnless(type(row[0]) == bytes, "type of row[0] must be bytes")
- self.failUnless(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8")
+ self.assertTrue(type(row[0]) == bytes, "type of row[0] must be bytes")
+ self.assertTrue(row[0] == austria.encode("utf-8"), "column must equal original data in UTF-8")
def CheckCustom(self):
self.con.text_factory = lambda x: str(x, "utf-8", "ignore")
austria = "Österreich"
row = self.con.execute("select ?", (austria,)).fetchone()
- self.failUnless(type(row[0]) == str, "type of row[0] must be unicode")
- self.failUnless(row[0].endswith("reich"), "column must contain original data")
+ self.assertTrue(type(row[0]) == str, "type of row[0] must be unicode")
+ self.assertTrue(row[0].endswith("reich"), "column must contain original data")
def CheckOptimizedUnicode(self):
self.con.text_factory = sqlite.OptimizedUnicode
germany = "Deutchland"
a_row = self.con.execute("select ?", (austria,)).fetchone()
d_row = self.con.execute("select ?", (germany,)).fetchone()
- self.failUnless(type(a_row[0]) == str, "type of non-ASCII row must be str")
- self.failUnless(type(d_row[0]) == str, "type of ASCII-only row must be str")
+ self.assertTrue(type(a_row[0]) == str, "type of non-ASCII row must be str")
+ self.assertTrue(type(d_row[0]) == str, "type of ASCII-only row must be str")
def tearDown(self):
self.con.close()
con.create_collation("X", 42)
self.fail("should have raised a TypeError")
except TypeError as e:
- self.failUnlessEqual(e.args[0], "parameter must be callable")
+ self.assertEqual(e.args[0], "parameter must be callable")
def CheckCreateCollationNotAscii(self):
con = sqlite.connect(":memory:")
result = con.execute(sql).fetchall()
self.fail("should have raised an OperationalError")
except sqlite.OperationalError as e:
- self.failUnlessEqual(e.args[0].lower(), "no such collation sequence: mycoll")
+ self.assertEqual(e.args[0].lower(), "no such collation sequence: mycoll")
def CheckCollationRegisterTwice(self):
"""
con.execute("""
create table foo(a, b)
""")
- self.failUnless(progress_calls)
+ self.assertTrue(progress_calls)
def CheckOpcodeCount(self):
create table bar (a, b)
""")
second_count = len(progress_calls)
- self.failUnless(first_count > second_count)
+ self.assertTrue(first_count > second_count)
def CheckCancelOperation(self):
"""
con.set_progress_handler(progress, 1)
con.set_progress_handler(None, 1)
con.execute("select 1 union select 2 union select 3").fetchall()
- self.failUnlessEqual(action, 0, "progress handler was not cleared")
+ self.assertEqual(action, 0, "progress handler was not cleared")
def suite():
collation_suite = unittest.makeSuite(CollationTests, "Check")
def CheckColumnNameWithSpaces(self):
cur = self.con.cursor()
cur.execute('select 1 as "foo bar [datetime]"')
- self.failUnlessEqual(cur.description[0][0], "foo bar")
+ self.assertEqual(cur.description[0][0], "foo bar")
cur.execute('select 1 as "foo baz"')
- self.failUnlessEqual(cur.description[0][0], "foo baz")
+ self.assertEqual(cur.description[0][0], "foo baz")
def CheckStatementAvailable(self):
# pysqlite up to 2.3.2 crashed on this, because the active statement handle was not checked
self.cur1.execute("create table test2(j)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
- self.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
def CheckInsertStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
- self.failUnlessEqual(len(res), 0)
+ self.assertEqual(len(res), 0)
def CheckUpdateStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("update test set i=6")
self.cur2.execute("select i from test")
res = self.cur2.fetchone()[0]
- self.failUnlessEqual(res, 5)
+ self.assertEqual(res, 5)
def CheckDeleteStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("delete from test")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
- self.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
def CheckReplaceStartsTransaction(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("replace into test(i) values (6)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
- self.failUnlessEqual(len(res), 1)
- self.failUnlessEqual(res[0][0], 5)
+ self.assertEqual(len(res), 1)
+ self.assertEqual(res[0][0], 5)
def CheckToggleAutoCommit(self):
self.cur1.execute("create table test(i)")
self.cur1.execute("insert into test(i) values (5)")
self.con1.isolation_level = None
- self.failUnlessEqual(self.con1.isolation_level, None)
+ self.assertEqual(self.con1.isolation_level, None)
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
- self.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
self.con1.isolation_level = "DEFERRED"
- self.failUnlessEqual(self.con1.isolation_level , "DEFERRED")
+ self.assertEqual(self.con1.isolation_level , "DEFERRED")
self.cur1.execute("insert into test(i) values (5)")
self.cur2.execute("select i from test")
res = self.cur2.fetchall()
- self.failUnlessEqual(len(res), 1)
+ self.assertEqual(len(res), 1)
def CheckRaiseTimeout(self):
if sqlite.sqlite_version_info < (3, 2, 2):
self.cur.execute("insert into test(s) values (?)", ("Österreich",))
self.cur.execute("select s from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], "Österreich")
+ self.assertEqual(row[0], "Österreich")
def CheckSmallInt(self):
self.cur.execute("insert into test(i) values (?)", (42,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], 42)
+ self.assertEqual(row[0], 42)
def CheckLargeInt(self):
num = 2**40
self.cur.execute("insert into test(i) values (?)", (num,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], num)
+ self.assertEqual(row[0], num)
def CheckFloat(self):
val = 3.14
self.cur.execute("insert into test(f) values (?)", (val,))
self.cur.execute("select f from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], val)
+ self.assertEqual(row[0], val)
def CheckBlob(self):
sample = b"Guglhupf"
self.cur.execute("insert into test(b) values (?)", (val,))
self.cur.execute("select b from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], sample)
+ self.assertEqual(row[0], sample)
def CheckUnicodeExecute(self):
self.cur.execute("select 'Österreich'")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], "Österreich")
+ self.assertEqual(row[0], "Österreich")
class DeclTypesTests(unittest.TestCase):
class Foo:
self.cur.execute("insert into test(s) values (?)", ("foo",))
self.cur.execute('select s as "s [WRONG]" from test')
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], "foo")
+ self.assertEqual(row[0], "foo")
def CheckSmallInt(self):
# default
self.cur.execute("insert into test(i) values (?)", (42,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], 42)
+ self.assertEqual(row[0], 42)
def CheckLargeInt(self):
# default
self.cur.execute("insert into test(i) values (?)", (num,))
self.cur.execute("select i from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], num)
+ self.assertEqual(row[0], num)
def CheckFloat(self):
# custom
self.cur.execute("insert into test(f) values (?)", (val,))
self.cur.execute("select f from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], 47.2)
+ self.assertEqual(row[0], 47.2)
def CheckBool(self):
# custom
self.cur.execute("insert into test(b) values (?)", (False,))
self.cur.execute("select b from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], False)
+ self.assertEqual(row[0], False)
self.cur.execute("delete from test")
self.cur.execute("insert into test(b) values (?)", (True,))
self.cur.execute("select b from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], True)
+ self.assertEqual(row[0], True)
def CheckUnicode(self):
# default
self.cur.execute("insert into test(u) values (?)", (val,))
self.cur.execute("select u from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], val)
+ self.assertEqual(row[0], val)
def CheckFoo(self):
val = DeclTypesTests.Foo("bla")
self.cur.execute("insert into test(foo) values (?)", (val,))
self.cur.execute("select foo from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], val)
+ self.assertEqual(row[0], val)
def CheckUnsupportedSeq(self):
class Bar: pass
self.cur.execute("insert into test(bin) values (?)", (val,))
self.cur.execute("select bin from test")
row = self.cur.fetchone()
- self.failUnlessEqual(row[0], sample)
+ self.assertEqual(row[0], sample)
def CheckNumber1(self):
self.cur.execute("insert into test(n1) values (5)")
value = self.cur.execute("select n1 from test").fetchone()[0]
# if the converter is not used, it's an int instead of a float
- self.failUnlessEqual(type(value), float)
+ self.assertEqual(type(value), float)
def CheckNumber2(self):
"""Checks wether converter names are cut off at '(' characters"""
self.cur.execute("insert into test(n2) values (5)")
value = self.cur.execute("select n2 from test").fetchone()[0]
# if the converter is not used, it's an int instead of a float
- self.failUnlessEqual(type(value), float)
+ self.assertEqual(type(value), float)
class ColNamesTests(unittest.TestCase):
def setUp(self):
self.cur.execute("insert into test(x) values (?)", ("xxx",))
self.cur.execute("select x from test")
val = self.cur.fetchone()[0]
- self.failUnlessEqual(val, "xxx")
+ self.assertEqual(val, "xxx")
def CheckNone(self):
self.cur.execute("insert into test(x) values (?)", (None,))
self.cur.execute("select x from test")
val = self.cur.fetchone()[0]
- self.failUnlessEqual(val, None)
+ self.assertEqual(val, None)
def CheckColName(self):
self.cur.execute("insert into test(x) values (?)", ("xxx",))
self.cur.execute('select x as "x [bar]" from test')
val = self.cur.fetchone()[0]
- self.failUnlessEqual(val, "<xxx>")
+ self.assertEqual(val, "<xxx>")
# Check if the stripping of colnames works. Everything after the first
# whitespace should be stripped.
- self.failUnlessEqual(self.cur.description[0][0], "x")
+ self.assertEqual(self.cur.description[0][0], "x")
def CheckCaseInConverterName(self):
self.cur.execute("select 'other' as \"x [b1b1]\"")
val = self.cur.fetchone()[0]
- self.failUnlessEqual(val, "MARKER")
+ self.assertEqual(val, "MARKER")
def CheckCursorDescriptionNoRow(self):
"""
def CheckCasterIsUsed(self):
self.cur.execute("select ?", (4,))
val = self.cur.fetchone()[0]
- self.failUnlessEqual(type(val), float)
+ self.assertEqual(type(val), float)
class BinaryConverterTests(unittest.TestCase):
def convert(s):
def CheckBinaryInputForConverter(self):
testdata = b"abcdefg" * 10
result = self.con.execute('select ? as "x [bin]"', (memoryview(zlib.compress(testdata)),)).fetchone()[0]
- self.failUnlessEqual(testdata, result)
+ self.assertEqual(testdata, result)
class DateTimeTests(unittest.TestCase):
def setUp(self):
self.cur.execute("insert into test(d) values (?)", (d,))
self.cur.execute("select d from test")
d2 = self.cur.fetchone()[0]
- self.failUnlessEqual(d, d2)
+ self.assertEqual(d, d2)
def CheckSqliteTimestamp(self):
ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0)
self.cur.execute("insert into test(ts) values (?)", (ts,))
self.cur.execute("select ts from test")
ts2 = self.cur.fetchone()[0]
- self.failUnlessEqual(ts, ts2)
+ self.assertEqual(ts, ts2)
def CheckSqlTimestamp(self):
# The date functions are only available in SQLite version 3.1 or later
self.cur.execute("insert into test(ts) values (current_timestamp)")
self.cur.execute("select ts from test")
ts = self.cur.fetchone()[0]
- self.failUnlessEqual(type(ts), datetime.datetime)
- self.failUnlessEqual(ts.year, now.year)
+ self.assertEqual(type(ts), datetime.datetime)
+ self.assertEqual(ts.year, now.year)
def CheckDateTimeSubSeconds(self):
ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 500000)
self.cur.execute("insert into test(ts) values (?)", (ts,))
self.cur.execute("select ts from test")
ts2 = self.cur.fetchone()[0]
- self.failUnlessEqual(ts, ts2)
+ self.assertEqual(ts, ts2)
def CheckDateTimeSubSecondsFloatingPoint(self):
ts = sqlite.Timestamp(2004, 2, 14, 7, 15, 0, 510241)
self.cur.execute("insert into test(ts) values (?)", (ts,))
self.cur.execute("select ts from test")
ts2 = self.cur.fetchone()[0]
- self.failUnlessEqual(ts, ts2)
+ self.assertEqual(ts, ts2)
def suite():
sqlite_type_suite = unittest.makeSuite(SqliteTypeTests, "Check")
cur = self.con.cursor()
cur.execute("select returntext()")
val = cur.fetchone()[0]
- self.failUnlessEqual(type(val), str)
- self.failUnlessEqual(val, "foo")
+ self.assertEqual(type(val), str)
+ self.assertEqual(val, "foo")
def CheckFuncReturnUnicode(self):
cur = self.con.cursor()
cur.execute("select returnunicode()")
val = cur.fetchone()[0]
- self.failUnlessEqual(type(val), str)
- self.failUnlessEqual(val, "bar")
+ self.assertEqual(type(val), str)
+ self.assertEqual(val, "bar")
def CheckFuncReturnInt(self):
cur = self.con.cursor()
cur.execute("select returnint()")
val = cur.fetchone()[0]
- self.failUnlessEqual(type(val), int)
- self.failUnlessEqual(val, 42)
+ self.assertEqual(type(val), int)
+ self.assertEqual(val, 42)
def CheckFuncReturnFloat(self):
cur = self.con.cursor()
cur.execute("select returnfloat()")
val = cur.fetchone()[0]
- self.failUnlessEqual(type(val), float)
+ self.assertEqual(type(val), float)
if val < 3.139 or val > 3.141:
self.fail("wrong value")
cur = self.con.cursor()
cur.execute("select returnnull()")
val = cur.fetchone()[0]
- self.failUnlessEqual(type(val), type(None))
- self.failUnlessEqual(val, None)
+ self.assertEqual(type(val), type(None))
+ self.assertEqual(val, None)
def CheckFuncReturnBlob(self):
cur = self.con.cursor()
cur.execute("select returnblob()")
val = cur.fetchone()[0]
- self.failUnlessEqual(type(val), bytes)
- self.failUnlessEqual(val, b"blob")
+ self.assertEqual(type(val), bytes)
+ self.assertEqual(val, b"blob")
def CheckFuncException(self):
cur = self.con.cursor()
cur.fetchone()
self.fail("should have raised OperationalError")
except sqlite.OperationalError as e:
- self.failUnlessEqual(e.args[0], 'user-defined function raised exception')
+ self.assertEqual(e.args[0], 'user-defined function raised exception')
def CheckParamString(self):
cur = self.con.cursor()
cur.execute("select isstring(?)", ("foo",))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckParamInt(self):
cur = self.con.cursor()
cur.execute("select isint(?)", (42,))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckParamFloat(self):
cur = self.con.cursor()
cur.execute("select isfloat(?)", (3.14,))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckParamNone(self):
cur = self.con.cursor()
cur.execute("select isnone(?)", (None,))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckParamBlob(self):
cur = self.con.cursor()
cur.execute("select isblob(?)", (memoryview(b"blob"),))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
class AggregateTests(unittest.TestCase):
def setUp(self):
cur.execute("select nostep(t) from test")
self.fail("should have raised an AttributeError")
except AttributeError as e:
- self.failUnlessEqual(e.args[0], "'AggrNoStep' object has no attribute 'step'")
+ self.assertEqual(e.args[0], "'AggrNoStep' object has no attribute 'step'")
def CheckAggrNoFinalize(self):
cur = self.con.cursor()
val = cur.fetchone()[0]
self.fail("should have raised an OperationalError")
except sqlite.OperationalError as e:
- self.failUnlessEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error")
+ self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error")
def CheckAggrExceptionInInit(self):
cur = self.con.cursor()
val = cur.fetchone()[0]
self.fail("should have raised an OperationalError")
except sqlite.OperationalError as e:
- self.failUnlessEqual(e.args[0], "user-defined aggregate's '__init__' method raised error")
+ self.assertEqual(e.args[0], "user-defined aggregate's '__init__' method raised error")
def CheckAggrExceptionInStep(self):
cur = self.con.cursor()
val = cur.fetchone()[0]
self.fail("should have raised an OperationalError")
except sqlite.OperationalError as e:
- self.failUnlessEqual(e.args[0], "user-defined aggregate's 'step' method raised error")
+ self.assertEqual(e.args[0], "user-defined aggregate's 'step' method raised error")
def CheckAggrExceptionInFinalize(self):
cur = self.con.cursor()
val = cur.fetchone()[0]
self.fail("should have raised an OperationalError")
except sqlite.OperationalError as e:
- self.failUnlessEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error")
+ self.assertEqual(e.args[0], "user-defined aggregate's 'finalize' method raised error")
def CheckAggrCheckParamStr(self):
cur = self.con.cursor()
cur.execute("select checkType('str', ?)", ("foo",))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckAggrCheckParamInt(self):
cur = self.con.cursor()
cur.execute("select checkType('int', ?)", (42,))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckAggrCheckParamFloat(self):
cur = self.con.cursor()
cur.execute("select checkType('float', ?)", (3.14,))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckAggrCheckParamNone(self):
cur = self.con.cursor()
cur.execute("select checkType('None', ?)", (None,))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckAggrCheckParamBlob(self):
cur = self.con.cursor()
cur.execute("select checkType('blob', ?)", (memoryview(b"blob"),))
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 1)
+ self.assertEqual(val, 1)
def CheckAggrCheckAggrSum(self):
cur = self.con.cursor()
cur.executemany("insert into test(i) values (?)", [(10,), (20,), (30,)])
cur.execute("select mysum(i) from test")
val = cur.fetchone()[0]
- self.failUnlessEqual(val, 60)
+ self.assertEqual(val, 60)
def authorizer_cb(action, arg1, arg2, dbname, source):
if action != sqlite.SQLITE_SELECT:
error = IOError
else:
import select
+ _has_poll = hasattr(select, 'poll')
import errno
import fcntl
import pickle
except:
MAXFD = 256
+# When select or poll has indicated that the file is writable,
+# we can write up to _PIPE_BUF bytes without risk of blocking.
+# POSIX defines PIPE_BUF as >= 512.
+_PIPE_BUF = getattr(select, 'PIPE_BUF', 512)
+
_active = []
def _cleanup():
def _communicate(self, input):
- read_set = []
- write_set = []
- stdout = None # Return
- stderr = None # Return
-
if self.stdin:
# Flush stdio buffer. This might block, if the user has
# been writing to .stdin in an uncontrolled fashion.
self.stdin.flush()
- if input:
- write_set.append(self.stdin)
- else:
+ if not input:
self.stdin.close()
+
+ if _has_poll:
+ stdout, stderr = self._communicate_with_poll(input)
+ else:
+ stdout, stderr = self._communicate_with_select(input)
+
+ # All data exchanged. Translate lists into strings.
+ if stdout is not None:
+ stdout = b''.join(stdout)
+ if stderr is not None:
+ stderr = b''.join(stderr)
+
+ # Translate newlines, if requested.
+ # This also turns bytes into strings.
+ if self.universal_newlines:
+ if stdout is not None:
+ stdout = self._translate_newlines(stdout,
+ self.stdout.encoding)
+ if stderr is not None:
+ stderr = self._translate_newlines(stderr,
+ self.stderr.encoding)
+
+ self.wait()
+ return (stdout, stderr)
+
+
+ def _communicate_with_poll(self, input):
+ stdout = None # Return
+ stderr = None # Return
+ fd2file = {}
+ fd2output = {}
+
+ poller = select.poll()
+ def register_and_append(file_obj, eventmask):
+ poller.register(file_obj.fileno(), eventmask)
+ fd2file[file_obj.fileno()] = file_obj
+
+ def close_unregister_and_remove(fd):
+ poller.unregister(fd)
+ fd2file[fd].close()
+ fd2file.pop(fd)
+
+ if self.stdin and input:
+ register_and_append(self.stdin, select.POLLOUT)
+
+ select_POLLIN_POLLPRI = select.POLLIN | select.POLLPRI
+ if self.stdout:
+ register_and_append(self.stdout, select_POLLIN_POLLPRI)
+ fd2output[self.stdout.fileno()] = stdout = []
+ if self.stderr:
+ register_and_append(self.stderr, select_POLLIN_POLLPRI)
+ fd2output[self.stderr.fileno()] = stderr = []
+
+ input_offset = 0
+ while fd2file:
+ try:
+ ready = poller.poll()
+ except select.error as e:
+ if e.args[0] == errno.EINTR:
+ continue
+ raise
+
+ # XXX Rewrite these to use non-blocking I/O on the
+ # file objects; they are no longer using C stdio!
+
+ for fd, mode in ready:
+ if mode & select.POLLOUT:
+ chunk = input[input_offset : input_offset + _PIPE_BUF]
+ input_offset += os.write(fd, chunk)
+ if input_offset >= len(input):
+ close_unregister_and_remove(fd)
+ elif mode & select_POLLIN_POLLPRI:
+ data = os.read(fd, 4096)
+ if not data:
+ close_unregister_and_remove(fd)
+ fd2output[fd].append(data)
+ else:
+ # Ignore hang up or errors.
+ close_unregister_and_remove(fd)
+
+ return (stdout, stderr)
+
+
+ def _communicate_with_select(self, input):
+ read_set = []
+ write_set = []
+ stdout = None # Return
+ stderr = None # Return
+
+ if self.stdin and input:
+ write_set.append(self.stdin)
if self.stdout:
read_set.append(self.stdout)
stdout = []
# file objects; they are no longer using C stdio!
if self.stdin in wlist:
- # When select has indicated that the file is writable,
- # we can write up to PIPE_BUF bytes without risk
- # blocking. POSIX defines PIPE_BUF >= 512
- chunk = input[input_offset : input_offset + 512]
+ chunk = input[input_offset : input_offset + _PIPE_BUF]
bytes_written = os.write(self.stdin.fileno(), chunk)
input_offset += bytes_written
if input_offset >= len(input):
read_set.remove(self.stderr)
stderr.append(data)
- # All data exchanged. Translate lists into strings.
- if stdout is not None:
- stdout = b"".join(stdout)
- if stderr is not None:
- stderr = b"".join(stderr)
-
- # Translate newlines, if requested.
- # This also turns bytes into strings.
- if self.universal_newlines:
- if stdout is not None:
- stdout = self._translate_newlines(stdout,
- self.stdout.encoding)
- if stderr is not None:
- stderr = self._translate_newlines(stderr,
- self.stderr.encoding)
-
- self.wait()
return (stdout, stderr)
+
def send_signal(self, sig):
"""Send a signal to the process
"""
a,
array.array(self.typecode, self.example[::-1]+2*self.example)
)
+ a = array.array(self.typecode, self.example)
+ a += a
+ self.assertEqual(
+ a,
+ array.array(self.typecode, self.example + self.example)
+ )
b = array.array(self.badtypecode())
self.assertRaises(TypeError, a.__add__, b)
array.array(self.typecode, self.example+self.example[::-1])
)
+ a = array.array(self.typecode, self.example)
+ a.extend(a)
+ self.assertEqual(
+ a,
+ array.array(self.typecode, self.example+self.example)
+ )
+
b = array.array(self.badtypecode())
self.assertRaises(TypeError, a.extend, b)
ArraySubclassWithKwargs('b', newarg=1)
def test_create_from_bytes(self):
+ # XXX This test probably needs to be moved in a subclass or
+ # generalized to use self.typecode.
a = array.array('H', b"1234")
self.assertEqual(len(a) * a.itemsize, 4)
+ def test_memoryview_no_resize(self):
+ # Test for issue 4509.
+ a = array.array(self.typecode, self.example)
+ m = memoryview(a)
+ expected = m.tobytes()
+ self.assertRaises(BufferError, a.pop, 0)
+ self.assertEqual(m.tobytes(), expected)
+ self.assertRaises(BufferError, a.remove, a[0])
+ self.assertEqual(m.tobytes(), expected)
+ self.assertRaises(BufferError, a.__setitem__, slice(0, 0), a)
+ self.assertEqual(m.tobytes(), expected)
+ self.assertRaises(BufferError, a.__delitem__, slice(0, len(a)))
+ self.assertEqual(m.tobytes(), expected)
+ self.assertRaises(BufferError, a.__imul__, 2)
+ self.assertEqual(m.tobytes(), expected)
+ self.assertRaises(BufferError, a.__iadd__, a)
+ self.assertEqual(m.tobytes(), expected)
+ self.assertRaises(BufferError, a.extend, a)
+ self.assertEqual(m.tobytes(), expected)
+
class StringTest(BaseTest):
self.assertEqual(res1, RESOURCE_DATA)
res2 = pkgutil.get_data(pkg, 'sub/res.txt')
self.assertEqual(res2, RESOURCE_DATA)
+
+ names = []
+ for loader, name, ispkg in pkgutil.iter_modules([zip_file]):
+ names.append(name)
+ self.assertEqual(names, ['test_getdata_zipfile'])
+
del sys.path[0]
del sys.modules[pkg]
if dir is not None:
os.rmdir(dir)
+
+unit_tests = [ProcessTestCase, CommandTests]
+
+if subprocess._has_poll:
+ class ProcessTestCaseNoPoll(ProcessTestCase):
+ def setUp(self):
+ subprocess._has_poll = False
+ ProcessTestCase.setUp(self)
+
+ def tearDown(self):
+ subprocess._has_poll = True
+ ProcessTestCase.tearDown(self)
+
+ unit_tests.append(ProcessTestCaseNoPoll)
+
+
def test_main():
- support.run_unittest(ProcessTestCase, CommandTests)
+ support.run_unittest(*unit_tests)
support.reap_children()
if __name__ == "__main__":
- Issue #6323: The pdb debugger did not exit when running a script with a
syntax error.
+- Issue #3392: The subprocess communicate() method no longer fails in select()
+ when file descriptors are large; communicate() now uses poll() when possible.
+
- Issue #6369: Fix an RLE decompression bug in the binhex module.
- Issue #6344: Fixed a crash of mmap.read() when passed a negative argument.
+- Issue #4005: Fixed a crash of pydoc when there was a zip file present in
+ sys.path.
+
+Extension Modules
+-----------------
+
+- Issue #4509: array.array objects are no longer modified after an operation
+ failing due to the resize restriction in-place when the object has exported
+ buffers.
+
Build
-----
ihigh = Py_SIZE(a);
item = a->ob_item;
d = n - (ihigh-ilow);
+ /* Issue #4509: If the array has exported buffers and the slice
+ assignment would change the size of the array, fail early to make
+ sure we don't modify it. */
+ if (d != 0 && a->ob_exports > 0) {
+ PyErr_SetString(PyExc_BufferError,
+ "cannot resize an array that is exporting buffers");
+ return -1;
+ }
if (d < 0) { /* Delete -d items */
memmove(item + (ihigh+d)*a->ob_descr->itemsize,
item + ihigh*a->ob_descr->itemsize,
static int
array_do_extend(arrayobject *self, PyObject *bb)
{
- Py_ssize_t size, oldsize;
-
+ Py_ssize_t size, oldsize, bbsize;
+
if (!array_Check(bb))
return array_iter_extend(self, bb);
#define b ((arrayobject *)bb)
return -1;
}
oldsize = Py_SIZE(self);
+ /* Get the size of bb before resizing the array since bb could be self. */
+ bbsize = Py_SIZE(bb);
size = oldsize + Py_SIZE(b);
if (array_resize(self, size) == -1)
return -1;
memcpy(self->ob_item + oldsize * self->ob_descr->itemsize,
- b->ob_item, Py_SIZE(b) * b->ob_descr->itemsize);
+ b->ob_item, bbsize * b->ob_descr->itemsize);
return 0;
#undef b
projects / python / commitdiff