source: trunk/src/allmydata/test/test_filenode.py

Last change on this file was 1cfe843d, checked in by Alexandre Detiste <alexandre.detiste@…>, at 2024-02-22T23:40:25Z

more python2 removal
  • Property mode set to 100644
File size: 8.2 KB
Line 
1"""
2Ported to Python 3.
3"""
4
5from twisted.trial import unittest
6from allmydata import uri, client
7from allmydata.monitor import Monitor
8from allmydata.immutable.literal import LiteralFileNode
9from allmydata.immutable.filenode import ImmutableFileNode
10from allmydata.mutable.filenode import MutableFileNode
11from allmydata.util import hashutil
12from allmydata.util.consumer import download_to_data
13
14class NotANode(object):
15    pass
16
17class FakeClient(object):
18    # just enough to let the node acquire a downloader (which it won't use),
19    # and to get default encoding parameters
20    def getServiceNamed(self, name):
21        return None
22    def get_encoding_parameters(self):
23        return {"k": 3, "n": 10}
24    def get_storage_broker(self):
25        return None
26    def get_history(self):
27        return None
28    _secret_holder = client.SecretHolder(b"lease secret", b"convergence secret")
29
30class Node(unittest.TestCase):
31    def test_chk_filenode(self):
32        u = uri.CHKFileURI(key=b"\x00"*16,
33                           uri_extension_hash=b"\x00"*32,
34                           needed_shares=3,
35                           total_shares=10,
36                           size=1000)
37        fn1 = ImmutableFileNode(u, None, None, None, None)
38        fn2 = ImmutableFileNode(u, None, None, None, None)
39        self.failUnlessEqual(fn1, fn2)
40        self.failIfEqual(fn1, "I am not a filenode")
41        self.failIfEqual(fn1, NotANode())
42        self.failUnlessEqual(fn1.get_uri(), u.to_string())
43        self.failUnlessEqual(fn1.get_cap(), u)
44        self.failUnlessEqual(fn1.get_readcap(), u)
45        self.failUnless(fn1.is_readonly())
46        self.failIf(fn1.is_mutable())
47        self.failIf(fn1.is_unknown())
48        self.failUnless(fn1.is_allowed_in_immutable_directory())
49        self.failUnlessEqual(fn1.get_write_uri(), None)
50        self.failUnlessEqual(fn1.get_readonly_uri(), u.to_string())
51        self.failUnlessEqual(fn1.get_size(), 1000)
52        self.failUnlessEqual(fn1.get_storage_index(), u.get_storage_index())
53        fn1.raise_error()
54        fn2.raise_error()
55        d = {}
56        d[fn1] = 1 # exercise __hash__
57        v = fn1.get_verify_cap()
58        self.failUnless(isinstance(v, uri.CHKFileVerifierURI))
59        self.failUnlessEqual(fn1.get_repair_cap(), v)
60        self.failUnless(v.is_readonly())
61        self.failIf(v.is_mutable())
62
63
64    def test_literal_filenode(self):
65        DATA = b"I am a short file."
66        u = uri.LiteralFileURI(data=DATA)
67        fn1 = LiteralFileNode(u)
68        fn2 = LiteralFileNode(u)
69        self.failUnlessEqual(fn1, fn2)
70        self.failIfEqual(fn1, "I am not a filenode")
71        self.failIfEqual(fn1, NotANode())
72        self.failUnlessEqual(fn1.get_uri(), u.to_string())
73        self.failUnlessEqual(fn1.get_cap(), u)
74        self.failUnlessEqual(fn1.get_readcap(), u)
75        self.failUnless(fn1.is_readonly())
76        self.failIf(fn1.is_mutable())
77        self.failIf(fn1.is_unknown())
78        self.failUnless(fn1.is_allowed_in_immutable_directory())
79        self.failUnlessEqual(fn1.get_write_uri(), None)
80        self.failUnlessEqual(fn1.get_readonly_uri(), u.to_string())
81        self.failUnlessEqual(fn1.get_size(), len(DATA))
82        self.failUnlessEqual(fn1.get_storage_index(), None)
83        fn1.raise_error()
84        fn2.raise_error()
85        d = {}
86        d[fn1] = 1 # exercise __hash__
87
88        v = fn1.get_verify_cap()
89        self.failUnlessEqual(v, None)
90        self.failUnlessEqual(fn1.get_repair_cap(), None)
91
92        d = download_to_data(fn1)
93        def _check(res):
94            self.failUnlessEqual(res, DATA)
95        d.addCallback(_check)
96
97        d.addCallback(lambda res: download_to_data(fn1, 1, 5))
98        def _check_segment(res):
99            self.failUnlessEqual(res, DATA[1:1+5])
100        d.addCallback(_check_segment)
101        d.addCallback(lambda ignored: fn1.get_best_readable_version())
102        d.addCallback(lambda fn2: self.failUnlessEqual(fn1, fn2))
103        d.addCallback(lambda ignored:
104            fn1.get_size_of_best_version())
105        d.addCallback(lambda size:
106            self.failUnlessEqual(size, len(DATA)))
107        d.addCallback(lambda ignored:
108            fn1.download_to_data())
109        d.addCallback(lambda data:
110            self.failUnlessEqual(data, DATA))
111        d.addCallback(lambda ignored:
112            fn1.download_best_version())
113        d.addCallback(lambda data:
114            self.failUnlessEqual(data, DATA))
115
116        return d
117
118    def test_mutable_filenode(self):
119        client = FakeClient()
120        wk = b"\x00"*16
121        rk = hashutil.ssk_readkey_hash(wk)
122        si = hashutil.ssk_storage_index_hash(rk)
123
124        u = uri.WriteableSSKFileURI(b"\x00"*16, b"\x00"*32)
125        n = MutableFileNode(None, None, client.get_encoding_parameters(),
126                            None).init_from_cap(u)
127
128        self.failUnlessEqual(n.get_writekey(), wk)
129        self.failUnlessEqual(n.get_readkey(), rk)
130        self.failUnlessEqual(n.get_storage_index(), si)
131        # these items are populated on first read (or create), so until that
132        # happens they'll be None
133        self.failUnlessEqual(n.get_privkey(), None)
134        self.failUnlessEqual(n.get_encprivkey(), None)
135        self.failUnlessEqual(n.get_pubkey(), None)
136
137        self.failUnlessEqual(n.get_uri(), u.to_string())
138        self.failUnlessEqual(n.get_write_uri(), u.to_string())
139        self.failUnlessEqual(n.get_readonly_uri(), u.get_readonly().to_string())
140        self.failUnlessEqual(n.get_cap(), u)
141        self.failUnlessEqual(n.get_readcap(), u.get_readonly())
142        self.failUnless(n.is_mutable())
143        self.failIf(n.is_readonly())
144        self.failIf(n.is_unknown())
145        self.failIf(n.is_allowed_in_immutable_directory())
146        n.raise_error()
147
148        n2 = MutableFileNode(None, None, client.get_encoding_parameters(),
149                             None).init_from_cap(u)
150        self.failUnlessEqual(n, n2)
151        self.failIfEqual(n, "not even the right type")
152        self.failIfEqual(n, u) # not the right class
153        n.raise_error()
154        d = {n: "can these be used as dictionary keys?"}
155        d[n2] = "replace the old one"
156        self.failUnlessEqual(len(d), 1)
157
158        nro = n.get_readonly()
159        self.failUnless(isinstance(nro, MutableFileNode))
160
161        self.failUnlessEqual(nro.get_readonly(), nro)
162        self.failUnlessEqual(nro.get_cap(), u.get_readonly())
163        self.failUnlessEqual(nro.get_readcap(), u.get_readonly())
164        self.failUnless(nro.is_mutable())
165        self.failUnless(nro.is_readonly())
166        self.failIf(nro.is_unknown())
167        self.failIf(nro.is_allowed_in_immutable_directory())
168        nro_u = nro.get_uri()
169        self.failUnlessEqual(nro_u, nro.get_readonly_uri())
170        self.failUnlessEqual(nro_u, u.get_readonly().to_string())
171        self.failUnlessEqual(nro.get_write_uri(), None)
172        self.failUnlessEqual(nro.get_repair_cap(), None) # RSAmut needs writecap
173        nro.raise_error()
174
175        v = n.get_verify_cap()
176        self.failUnless(isinstance(v, uri.SSKVerifierURI))
177        self.failUnlessEqual(n.get_repair_cap(), n._uri) # TODO: n.get_uri()
178
179    def test_mutable_filenode_equality(self):
180        client = FakeClient()
181        u = uri.WriteableSSKFileURI(b"\x00"*16, b"\x00"*32)
182        n = MutableFileNode(None, None, client.get_encoding_parameters(),
183                            None).init_from_cap(u)
184        u2 = uri.WriteableSSKFileURI(b"\x01"*16, b"\x01"*32)
185        n2 = MutableFileNode(None, None, client.get_encoding_parameters(),
186                             None).init_from_cap(u2)
187        n2b = MutableFileNode(None, None, client.get_encoding_parameters(),
188                              None).init_from_cap(u2)
189        self.assertTrue(n2 == n2b)
190        self.assertFalse(n2 != n2b)
191        self.assertTrue(n2 != n)
192        self.assertTrue(n != n2)
193        self.assertFalse(n == n2)
194        self.assertTrue(n != 3)
195        self.assertFalse(n == 3)
196
197
198class LiteralChecker(unittest.TestCase):
199    def test_literal_filenode(self):
200        DATA = b"I am a short file."
201        u = uri.LiteralFileURI(data=DATA)
202        fn1 = LiteralFileNode(u)
203
204        d = fn1.check(Monitor())
205        def _check_checker_results(cr):
206            self.failUnlessEqual(cr, None)
207        d.addCallback(_check_checker_results)
208
209        d.addCallback(lambda res: fn1.check(Monitor(), verify=True))
210        d.addCallback(_check_checker_results)
211
212        return d
Note: See TracBrowser for help on using the repository browser.