def keys(self, txn=None):
- if txn != None:
+ if txn is not None:
return self.db.keys(txn)
else:
return self.db.keys()
def items(self, txn=None):
- if txn != None:
+ if txn is not None:
items = self.db.items(txn)
else:
items = self.db.items()
return newitems
def values(self, txn=None):
- if txn != None:
+ if txn is not None:
values = self.db.values(txn)
else:
values = self.db.values()
def set_range(self, search) :
v = self._dbcursor.set_range(bytes(search, "iso8859-1"))
- if v != None :
+ if v is not None :
v = (v[0].decode("iso8859-1"),
v[1].decode("iso8859-1"))
return v
def __next__(self) :
v = getattr(self._dbcursor, "next")()
- if v != None :
+ if v is not None :
v = (v[0].decode("iso8859-1"),
v[1].decode("iso8859-1"))
return v
def put(self, key, value, flags=0, txn=None) :
key = bytes(key, "iso8859-1")
- if value != None :
+ if value is not None :
value = bytes(value, "iso8859-1")
return self._db.put(key, value, flags=flags, txn=txn)
def get(self, key, txn=None, flags=0) :
key = bytes(key, "iso8859-1")
v = self._db.get(key, txn=txn, flags=flags)
- if v != None :
+ if v is not None :
v = v.decode("iso8859-1")
return v
# error
dataitem = None
dataitem = mappings[column](dataitem)
- if dataitem != None:
+ if dataitem is not None:
self.db.put(
_data_key(table, column, rowid),
dataitem, txn=txn)
return getattr(self._dbcursor, v)
def _fix(self, v) :
- if v == None : return None
+ if v is None : return None
key, value = v
if isinstance(key, bytes) :
key = key.decode(charset)
def pget(self, key=None, data=None, flags=0) :
# Incorrect because key can be a bare number,
# but enough to pass testsuite
- if isinstance(key, int) and (data==None) and (flags==0) :
+ if isinstance(key, int) and (data is None) and (flags == 0) :
flags = key
key = None
if isinstance(key, str) :
if isinstance(data, str) :
data = bytes(data, charset)
v=self._dbcursor.pget(key=key, data=data, flags=flags)
- if v != None :
+ if v is not None :
v1, v2, v3 = v
if isinstance(v1, bytes) :
v1 = v1.decode(charset)
def join_item(self) :
v = self._dbcursor.join_item()
- if v != None :
+ if v is not None :
v = v.decode(charset)
return v
args =(k, d, f)
v = self._dbcursor.get(*args, **kwargs)
- if v != None :
+ if v is not None :
k, v = v
if isinstance(k, bytes) :
k = k.decode(charset)
if isinstance(k, str) :
k = bytes(k, charset)
v = self._db[k]
- if v != None :
+ if v is not None :
v = v.decode(charset)
return v
else :
v=self._db.get(key, txn=txn, flags=flags,
dlen=dlen, doff=doff)
- if (v != None) and isinstance(v, bytes) :
+ if (v is not None) and isinstance(v, bytes) :
v = v.decode(charset)
return v
if isinstance(key, str) :
key = bytes(key, charset)
v=self._db.pget(key, txn=txn)
- if v != None :
+ if v is not None :
v1, v2 = v
if isinstance(v1, bytes) :
v1 = v1.decode(charset)
if isinstance(value, str) :
value = bytes(value, charset)
v=self._db.get_both(key, value, txn=txn, flags=flags)
- if v != None :
+ if v is not None :
v = v.decode(charset)
return v
else:
if set_raises_error:
self.fail("expected exception")
- if n != None:
+ if n is not None:
self.fail("expected None: %r" % (n,))
rec = c.get_both('0404', self.makeData('0404'))
else:
if get_raises_error:
self.fail("expected exception")
- if n != None:
+ if n is not None:
self.fail("expected None: %r" % (n,))
if self.d.get_type() == db.DB_BTREE:
self.startTest ()
self.createDB (my_compare)
- try:
- self.db.set_bt_compare (my_compare)
- self.assert_(0, "this set should fail")
-
- except RuntimeError, msg:
- pass
+ self.assertRaises (RuntimeError, self.db.set_bt_compare, my_compare)
def test_suite ():
res = unittest.TestSuite ()
self.tdb.Insert(tabname, {'Type': 'Unknown', 'Access': '0'})
def set_type(type):
- if type == None:
+ if type is None:
return 'MP3'
return type
for x in letters:
recno = d.append(x * 60)
- self.assertEqual(type(recno), type(0))
- self.assert_(recno >= 1)
+ self.assertIsInstance(recno, int)
+ self.assertGreaterEqual(recno, 1)
if verbose:
print recno,
if verbose:
print data
- self.assertEqual(type(data), type(""))
+ self.assertIsInstance(data, str)
self.assertEqual(data, d.get(recno))
try:
keys = d.keys()
if verbose:
print keys
- self.assertEqual(type(keys), type([]))
- self.assertEqual(type(keys[0]), type(123))
+ self.assertIsInstance(keys, list)
+ self.assertIsInstance(keys[0], int)
self.assertEqual(len(keys), len(d))
items = d.items()
if verbose:
pprint(items)
- self.assertEqual(type(items), type([]))
- self.assertEqual(type(items[0]), type(()))
+ self.assertIsInstance(items, list)
+ self.assertIsInstance(items[0], tuple)
self.assertEqual(len(items[0]), 2)
- self.assertEqual(type(items[0][0]), type(123))
- self.assertEqual(type(items[0][1]), type(""))
+ self.assertIsInstance(items[0][0], int)
+ self.assertIsInstance(items[0][1], str)
self.assertEqual(len(items), len(d))
- self.assert_(d.has_key(25))
+ self.assertTrue(d.has_key(25))
del d[25]
self.assertFalse(d.has_key(25))
if get_returns_none:
self.fail("unexpected DBKeyEmptyError exception")
else:
- self.assertEqual(val[0], db.DB_KEYEMPTY)
+ self.assertEqual(val.args[0], db.DB_KEYEMPTY)
if verbose: print val
else:
if not get_returns_none:
import time
timeout=time.time()+10
v=None
- while (time.time()<timeout) and (v==None) :
+ while (time.time()<timeout) and (v is None) :
txn=self.dbenvClient.txn_begin()
v=self.dbClient.get("ABC", txn=txn)
txn.commit()
- if v==None :
+ if v is None :
time.sleep(0.02)
self.assertTrue(time.time()<timeout)
self.assertEquals("123", v)
self.dbMaster.delete("ABC", txn=txn)
txn.commit()
timeout=time.time()+10
- while (time.time()<timeout) and (v!=None) :
+ while (time.time()<timeout) and (v is not None) :
txn=self.dbenvClient.txn_begin()
v=self.dbClient.get("ABC", txn=txn)
txn.commit()
- if v==None :
+ if v is None :
time.sleep(0.02)
self.assertTrue(time.time()<timeout)
self.assertEquals(None, v)
def thread_do(env, q, envid, election_status, must_be_master) :
while True :
v=q.get()
- if v == None : return
+ if v is None : return
env.rep_process_message(v[0], v[1], envid)
self.thread_do = thread_do
import time
timeout=time.time()+10
v=None
- while (time.time()<timeout) and (v==None) :
+ while (time.time()<timeout) and (v is None) :
txn=self.dbenvClient.txn_begin()
v=self.dbClient.get("ABC", txn=txn)
txn.commit()
- if v==None :
+ if v is None :
time.sleep(0.02)
self.assertTrue(time.time()<timeout)
self.assertEquals("123", v)
self.dbMaster.delete("ABC", txn=txn)
txn.commit()
timeout=time.time()+10
- while (time.time()<timeout) and (v!=None) :
+ while (time.time()<timeout) and (v is not None) :
txn=self.dbenvClient.txn_begin()
v=self.dbClient.get("ABC", txn=txn)
txn.commit()
- if v==None :
+ if v is None :
time.sleep(0.02)
self.assertTrue(time.time()<timeout)
self.assertEquals(None, v)
def thread_do(env, q, envid, election_status, must_be_master) :
while True :
v=q.get()
- if v == None : return
+ if v is None : return
r = env.rep_process_message(v[0],v[1],envid)
if must_be_master and self.confirmed_master :
self.dbenvMaster.rep_start(flags = db.DB_REP_MASTER)
readers.pop().start()
except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val:
if verbose:
- print "%s: Aborting transaction (%s)" % (name, val[1])
+ print "%s: Aborting transaction (%s)" % (name, val.args[1])
txn.abort()
if verbose:
finished = True
except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val:
if verbose:
- print "%s: Aborting transaction (%s)" % (name, val[1])
+ print "%s: Aborting transaction (%s)" % (name, val.args[1])
c.close()
txn.abort()
import tempfile
import time
import unittest
-from test.test_support import (requires, verbose, run_unittest, unlink, rmtree,
- import_module)
+from test.test_support import requires, run_unittest, import_module
# Skip test if _bsddb module was not built.
import_module('_bsddb')