| Index: third_party/twisted_8_1/twisted/test/test_dirdbm.py
|
| diff --git a/third_party/twisted_8_1/twisted/test/test_dirdbm.py b/third_party/twisted_8_1/twisted/test/test_dirdbm.py
|
| deleted file mode 100644
|
| index 12268bd74253eeb1676def28a09983f4f476e9ff..0000000000000000000000000000000000000000
|
| --- a/third_party/twisted_8_1/twisted/test/test_dirdbm.py
|
| +++ /dev/null
|
| @@ -1,174 +0,0 @@
|
| -# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
|
| -# See LICENSE for details.
|
| -
|
| -
|
| -
|
| -"""
|
| -Test cases for dirdbm module.
|
| -"""
|
| -
|
| -from twisted.trial import unittest
|
| -from twisted.persisted import dirdbm
|
| -import os, shutil, glob
|
| -
|
| -class DirDbmTestCase(unittest.TestCase):
|
| -
|
| - def setUp(self):
|
| - self.path = self.mktemp()
|
| - self.dbm = dirdbm.open(self.path)
|
| - self.items = (('abc', 'foo'), ('/lalal', '\000\001'), ('\000\012', 'baz'))
|
| -
|
| - def tearDown(self):
|
| - shutil.rmtree(self.path)
|
| -
|
| - def testAll(self):
|
| - k = "//==".decode("base64")
|
| - self.dbm[k] = "a"
|
| - self.dbm[k] = "a"
|
| - self.assertEquals(self.dbm[k], "a")
|
| -
|
| - def testRebuildInteraction(self):
|
| - from twisted.persisted import dirdbm
|
| - from twisted.python import rebuild
|
| -
|
| - s = dirdbm.Shelf('dirdbm.rebuild.test')
|
| - s['key'] = 'value'
|
| - rebuild.rebuild(dirdbm)
|
| - # print s['key']
|
| -
|
| - def testDbm(self):
|
| - d = self.dbm
|
| -
|
| - # insert keys
|
| - keys = []
|
| - values = []
|
| - for k, v in self.items:
|
| - d[k] = v
|
| - keys.append(k)
|
| - values.append(v)
|
| - keys.sort()
|
| - values.sort()
|
| -
|
| - # check they exist
|
| - for k, v in self.items:
|
| - assert d.has_key(k), "has_key() failed"
|
| - assert d[k] == v, "database has wrong value"
|
| -
|
| - # check non existent key
|
| - try:
|
| - d["XXX"]
|
| - except KeyError:
|
| - pass
|
| - else:
|
| - assert 0, "didn't raise KeyError on non-existent key"
|
| -
|
| - # check keys(), values() and items()
|
| - dbkeys = list(d.keys())
|
| - dbvalues = list(d.values())
|
| - dbitems = list(d.items())
|
| - dbkeys.sort()
|
| - dbvalues.sort()
|
| - dbitems.sort()
|
| - items = list(self.items)
|
| - items.sort()
|
| - assert keys == dbkeys, ".keys() output didn't match: %s != %s" % (repr(keys), repr(dbkeys))
|
| - assert values == dbvalues, ".values() output didn't match: %s != %s" % (repr(values), repr(dbvalues))
|
| - assert items == dbitems, "items() didn't match: %s != %s" % (repr(items), repr(dbitems))
|
| -
|
| - copyPath = self.mktemp()
|
| - d2 = d.copyTo(copyPath)
|
| -
|
| - copykeys = list(d.keys())
|
| - copyvalues = list(d.values())
|
| - copyitems = list(d.items())
|
| - copykeys.sort()
|
| - copyvalues.sort()
|
| - copyitems.sort()
|
| -
|
| - assert dbkeys == copykeys, ".copyTo().keys() didn't match: %s != %s" % (repr(dbkeys), repr(copykeys))
|
| - assert dbvalues == copyvalues, ".copyTo().values() didn't match: %s != %s" % (repr(dbvalues), repr(copyvalues))
|
| - assert dbitems == copyitems, ".copyTo().items() didn't match: %s != %s" % (repr(dbkeys), repr(copyitems))
|
| -
|
| - d2.clear()
|
| - assert len(d2.keys()) == len(d2.values()) == len(d2.items()) == 0, ".clear() failed"
|
| - shutil.rmtree(copyPath)
|
| -
|
| - # delete items
|
| - for k, v in self.items:
|
| - del d[k]
|
| - assert not d.has_key(k), "has_key() even though we deleted it"
|
| - assert len(d.keys()) == 0, "database has keys"
|
| - assert len(d.values()) == 0, "database has values"
|
| - assert len(d.items()) == 0, "database has items"
|
| -
|
| -
|
| - def testModificationTime(self):
|
| - import time
|
| - # the mtime value for files comes from a different place than the
|
| - # gettimeofday() system call. On linux, gettimeofday() can be
|
| - # slightly ahead (due to clock drift which gettimeofday() takes into
|
| - # account but which open()/write()/close() do not), and if we are
|
| - # close to the edge of the next second, time.time() can give a value
|
| - # which is larger than the mtime which results from a subsequent
|
| - # write(). I consider this a kernel bug, but it is beyond the scope
|
| - # of this test. Thus we keep the range of acceptability to 3 seconds time.
|
| - # -warner
|
| - self.dbm["k"] = "v"
|
| - self.assert_(abs(time.time() - self.dbm.getModificationTime("k")) <= 3)
|
| -
|
| - def testRecovery(self):
|
| - """DirDBM: test recovery from directory after a faked crash"""
|
| - k = self.dbm._encode("key1")
|
| - f = open(os.path.join(self.path, k + ".rpl"), "wb")
|
| - f.write("value")
|
| - f.close()
|
| -
|
| - k2 = self.dbm._encode("key2")
|
| - f = open(os.path.join(self.path, k2), "wb")
|
| - f.write("correct")
|
| - f.close()
|
| - f = open(os.path.join(self.path, k2 + ".rpl"), "wb")
|
| - f.write("wrong")
|
| - f.close()
|
| -
|
| - f = open(os.path.join(self.path, "aa.new"), "wb")
|
| - f.write("deleted")
|
| - f.close()
|
| -
|
| - dbm = dirdbm.DirDBM(self.path)
|
| - assert dbm["key1"] == "value"
|
| - assert dbm["key2"] == "correct"
|
| - assert not glob.glob(os.path.join(self.path, "*.new"))
|
| - assert not glob.glob(os.path.join(self.path, "*.rpl"))
|
| -
|
| -
|
| - def test_nonStringKeys(self):
|
| - """
|
| - L{dirdbm.DirDBM} operations only support string keys: other types
|
| - should raise a C{AssertionError}. This really ought to be a
|
| - C{TypeError}, but it'll stay like this for backward compatibility.
|
| - """
|
| - self.assertRaises(AssertionError, self.dbm.__setitem__, 2, "3")
|
| - try:
|
| - self.assertRaises(AssertionError, self.dbm.__setitem__, "2", 3)
|
| - except unittest.FailTest:
|
| - # dirdbm.Shelf.__setitem__ supports non-string values
|
| - self.assertIsInstance(self.dbm, dirdbm.Shelf)
|
| - self.assertRaises(AssertionError, self.dbm.__getitem__, 2)
|
| - self.assertRaises(AssertionError, self.dbm.__delitem__, 2)
|
| - self.assertRaises(AssertionError, self.dbm.has_key, 2)
|
| - self.assertRaises(AssertionError, self.dbm.__contains__, 2)
|
| - self.assertRaises(AssertionError, self.dbm.getModificationTime, 2)
|
| -
|
| -
|
| -
|
| -class ShelfTestCase(DirDbmTestCase):
|
| -
|
| - def setUp(self):
|
| - self.path = self.mktemp()
|
| - self.dbm = dirdbm.Shelf(self.path)
|
| - self.items = (('abc', 'foo'), ('/lalal', '\000\001'), ('\000\012', 'baz'),
|
| - ('int', 12), ('float', 12.0), ('tuple', (None, 12)))
|
| -
|
| -
|
| -testCases = [DirDbmTestCase, ShelfTestCase]
|
|
|