← Back to branch summary

~vishvananda/nova/network-refactor

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_dirdbm.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/test_dirdbm.py
 
1
# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Test cases for dirdbm module.
 
6
"""
 
7
 
 
8
import os, shutil, glob
 
9
 
 
10
from twisted.trial import unittest
 
11
from twisted.persisted import dirdbm
 
12
 
 
13
 
 
14
 
 
15
class DirDbmTestCase(unittest.TestCase):
 
16
 
 
17
    def setUp(self):
 
18
        self.path = self.mktemp()
 
19
        self.dbm = dirdbm.open(self.path)
 
20
        self.items = (('abc', 'foo'), ('/lalal', '\000\001'), ('\000\012', 'baz'))
 
21
 
 
22
 
 
23
    def testAll(self):
 
24
        k = "//==".decode("base64")
 
25
        self.dbm[k] = "a"
 
26
        self.dbm[k] = "a"
 
27
        self.assertEquals(self.dbm[k], "a")
 
28
 
 
29
 
 
30
    def testRebuildInteraction(self):
 
31
        from twisted.persisted import dirdbm
 
32
        from twisted.python import rebuild
 
33
 
 
34
        s = dirdbm.Shelf('dirdbm.rebuild.test')
 
35
        s['key'] = 'value'
 
36
        rebuild.rebuild(dirdbm)
 
37
        # print s['key']
 
38
 
 
39
 
 
40
    def testDbm(self):
 
41
        d = self.dbm
 
42
 
 
43
        # insert keys
 
44
        keys = []
 
45
        values = []
 
46
        for k, v in self.items:
 
47
            d[k] = v
 
48
            keys.append(k)
 
49
            values.append(v)
 
50
        keys.sort()
 
51
        values.sort()
 
52
 
 
53
        # check they exist
 
54
        for k, v in self.items:
 
55
            assert d.has_key(k), "has_key() failed"
 
56
            assert d[k] == v, "database has wrong value"
 
57
 
 
58
        # check non existent key
 
59
        try:
 
60
            d["XXX"]
 
61
        except KeyError:
 
62
            pass
 
63
        else:
 
64
            assert 0, "didn't raise KeyError on non-existent key"
 
65
 
 
66
        # check keys(), values() and items()
 
67
        dbkeys = list(d.keys())
 
68
        dbvalues = list(d.values())
 
69
        dbitems = list(d.items())
 
70
        dbkeys.sort()
 
71
        dbvalues.sort()
 
72
        dbitems.sort()
 
73
        items = list(self.items)
 
74
        items.sort()
 
75
        assert keys == dbkeys, ".keys() output didn't match: %s != %s" % (repr(keys), repr(dbkeys))
 
76
        assert values == dbvalues, ".values() output didn't match: %s != %s" % (repr(values), repr(dbvalues))
 
77
        assert items == dbitems, "items() didn't match: %s != %s" % (repr(items), repr(dbitems))
 
78
 
 
79
        copyPath = self.mktemp()
 
80
        d2 = d.copyTo(copyPath)
 
81
 
 
82
        copykeys = list(d.keys())
 
83
        copyvalues = list(d.values())
 
84
        copyitems = list(d.items())
 
85
        copykeys.sort()
 
86
        copyvalues.sort()
 
87
        copyitems.sort()
 
88
 
 
89
        assert dbkeys == copykeys, ".copyTo().keys() didn't match: %s != %s" % (repr(dbkeys), repr(copykeys))
 
90
        assert dbvalues == copyvalues, ".copyTo().values() didn't match: %s != %s" % (repr(dbvalues), repr(copyvalues))
 
91
        assert dbitems == copyitems, ".copyTo().items() didn't match: %s != %s" % (repr(dbkeys), repr(copyitems))
 
92
 
 
93
        d2.clear()
 
94
        assert len(d2.keys()) == len(d2.values()) == len(d2.items()) == 0, ".clear() failed"
 
95
        shutil.rmtree(copyPath)
 
96
 
 
97
        # delete items
 
98
        for k, v in self.items:
 
99
            del d[k]
 
100
            assert not d.has_key(k), "has_key() even though we deleted it"
 
101
        assert len(d.keys()) == 0, "database has keys"
 
102
        assert len(d.values()) == 0, "database has values"
 
103
        assert len(d.items()) == 0, "database has items"
 
104
 
 
105
 
 
106
    def testModificationTime(self):
 
107
        import time
 
108
        # the mtime value for files comes from a different place than the
 
109
        # gettimeofday() system call. On linux, gettimeofday() can be
 
110
        # slightly ahead (due to clock drift which gettimeofday() takes into
 
111
        # account but which open()/write()/close() do not), and if we are
 
112
        # close to the edge of the next second, time.time() can give a value
 
113
        # which is larger than the mtime which results from a subsequent
 
114
        # write(). I consider this a kernel bug, but it is beyond the scope
 
115
        # of this test. Thus we keep the range of acceptability to 3 seconds time.
 
116
        # -warner
 
117
        self.dbm["k"] = "v"
 
118
        self.assert_(abs(time.time() - self.dbm.getModificationTime("k")) <= 3)
 
119
 
 
120
 
 
121
    def testRecovery(self):
 
122
        """DirDBM: test recovery from directory after a faked crash"""
 
123
        k = self.dbm._encode("key1")
 
124
        f = open(os.path.join(self.path, k + ".rpl"), "wb")
 
125
        f.write("value")
 
126
        f.close()
 
127
 
 
128
        k2 = self.dbm._encode("key2")
 
129
        f = open(os.path.join(self.path, k2), "wb")
 
130
        f.write("correct")
 
131
        f.close()
 
132
        f = open(os.path.join(self.path, k2 + ".rpl"), "wb")
 
133
        f.write("wrong")
 
134
        f.close()
 
135
 
 
136
        f = open(os.path.join(self.path, "aa.new"), "wb")
 
137
        f.write("deleted")
 
138
        f.close()
 
139
 
 
140
        dbm = dirdbm.DirDBM(self.path)
 
141
        assert dbm["key1"] == "value"
 
142
        assert dbm["key2"] == "correct"
 
143
        assert not glob.glob(os.path.join(self.path, "*.new"))
 
144
        assert not glob.glob(os.path.join(self.path, "*.rpl"))
 
145
 
 
146
 
 
147
    def test_nonStringKeys(self):
 
148
        """
 
149
        L{dirdbm.DirDBM} operations only support string keys: other types
 
150
        should raise a C{AssertionError}. This really ought to be a
 
151
        C{TypeError}, but it'll stay like this for backward compatibility.
 
152
        """
 
153
        self.assertRaises(AssertionError, self.dbm.__setitem__, 2, "3")
 
154
        try:
 
155
            self.assertRaises(AssertionError, self.dbm.__setitem__, "2", 3)
 
156
        except unittest.FailTest:
 
157
            # dirdbm.Shelf.__setitem__ supports non-string values
 
158
            self.assertIsInstance(self.dbm, dirdbm.Shelf)
 
159
        self.assertRaises(AssertionError, self.dbm.__getitem__, 2)
 
160
        self.assertRaises(AssertionError, self.dbm.__delitem__, 2)
 
161
        self.assertRaises(AssertionError, self.dbm.has_key, 2)
 
162
        self.assertRaises(AssertionError, self.dbm.__contains__, 2)
 
163
        self.assertRaises(AssertionError, self.dbm.getModificationTime, 2)
 
164
 
 
165
 
 
166
 
 
167
class ShelfTestCase(DirDbmTestCase):
 
168
 
 
169
    def setUp(self):
 
170
        self.path = self.mktemp()
 
171
        self.dbm = dirdbm.Shelf(self.path)
 
172
        self.items = (('abc', 'foo'), ('/lalal', '\000\001'), ('\000\012', 'baz'),
 
173
                      ('int', 12), ('float', 12.0), ('tuple', (None, 12)))
 
174
 
 
175
 
 
176
testCases = [DirDbmTestCase, ShelfTestCase]