source: trunk/src/allmydata/test/test_uri.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: 30.8 KB
Line 
1"""
2Tests for allmydata.uri.
3
4Ported to Python 3.
5"""
6
7import os
8from twisted.trial import unittest
9from allmydata import uri
10from allmydata.util import hashutil, base32
11from allmydata.interfaces import IURI, IFileURI, IDirnodeURI, IMutableFileURI, \
12    IVerifierURI, CapConstraintError
13import allmydata.test.common_util as testutil
14
15class Literal(testutil.ReallyEqualMixin, unittest.TestCase):
16    def _help_test(self, data):
17        u = uri.LiteralFileURI(data)
18        self.failUnless(IURI.providedBy(u))
19        self.failUnless(IFileURI.providedBy(u))
20        self.failIf(IDirnodeURI.providedBy(u))
21        self.failUnlessReallyEqual(u.data, data)
22        self.failUnlessReallyEqual(u.get_size(), len(data))
23        self.failUnless(u.is_readonly())
24        self.failIf(u.is_mutable())
25
26        u2 = uri.from_string(u.to_string())
27        self.failUnless(IURI.providedBy(u2))
28        self.failUnless(IFileURI.providedBy(u2))
29        self.failIf(IDirnodeURI.providedBy(u2))
30        self.failUnlessReallyEqual(u2.data, data)
31        self.failUnlessReallyEqual(u2.get_size(), len(data))
32        self.failUnless(u2.is_readonly())
33        self.failIf(u2.is_mutable())
34
35        u2i = uri.from_string(u.to_string(), deep_immutable=True)
36        self.failUnless(IFileURI.providedBy(u2i))
37        self.failIf(IDirnodeURI.providedBy(u2i))
38        self.failUnlessReallyEqual(u2i.data, data)
39        self.failUnlessReallyEqual(u2i.get_size(), len(data))
40        self.failUnless(u2i.is_readonly())
41        self.failIf(u2i.is_mutable())
42
43        u3 = u.get_readonly()
44        self.failUnlessIdentical(u, u3)
45        self.failUnlessReallyEqual(u.get_verify_cap(), None)
46
47    def test_empty(self):
48        data = b"" # This data is some *very* small data!
49        return self._help_test(data)
50
51    def test_pack(self):
52        data = b"This is some small data"
53        return self._help_test(data)
54
55    def test_nonascii(self):
56        data = b"This contains \x00 and URI:LIT: and \n, oh my."
57        return self._help_test(data)
58
59class Compare(testutil.ReallyEqualMixin, unittest.TestCase):
60    def test_compare(self):
61        lit1 = uri.LiteralFileURI(b"some data")
62        fileURI = b'URI:CHK:f5ahxa25t4qkktywz6teyfvcx4:opuioq7tj2y6idzfp6cazehtmgs5fdcebcz3cygrxyydvcozrmeq:3:10:345834'
63        chk1 = uri.CHKFileURI.init_from_string(fileURI)
64        chk2 = uri.CHKFileURI.init_from_string(fileURI)
65        unk = uri.UnknownURI(b"lafs://from_the_future")
66        self.failIfEqual(lit1, chk1)
67        self.failUnlessReallyEqual(chk1, chk2)
68        self.failIfEqual(chk1, "not actually a URI")
69        # these should be hashable too
70        s = set([lit1, chk1, chk2, unk])
71        self.failUnlessReallyEqual(len(s), 3) # since chk1==chk2
72
73    def test_is_uri(self):
74        lit1 = uri.LiteralFileURI(b"some data").to_string()
75        self.failUnless(uri.is_uri(lit1))
76        self.failIf(uri.is_uri(None))
77
78    def test_is_literal_file_uri(self):
79        lit1 = uri.LiteralFileURI(b"some data").to_string()
80        self.failUnless(uri.is_literal_file_uri(lit1))
81        self.failIf(uri.is_literal_file_uri(None))
82        self.failIf(uri.is_literal_file_uri("foo"))
83        self.failIf(uri.is_literal_file_uri("ro.foo"))
84        self.failIf(uri.is_literal_file_uri(b"URI:LITfoo"))
85        self.failUnless(uri.is_literal_file_uri("ro.URI:LIT:foo"))
86        self.failUnless(uri.is_literal_file_uri("imm.URI:LIT:foo"))
87
88    def test_has_uri_prefix(self):
89        self.failUnless(uri.has_uri_prefix(b"URI:foo"))
90        self.failUnless(uri.has_uri_prefix(b"ro.URI:foo"))
91        self.failUnless(uri.has_uri_prefix(b"imm.URI:foo"))
92        self.failUnless(uri.has_uri_prefix("URI:foo"))
93        self.failUnless(uri.has_uri_prefix("ro.URI:foo"))
94        self.failUnless(uri.has_uri_prefix("imm.URI:foo"))
95        self.failIf(uri.has_uri_prefix(None))
96        self.failIf(uri.has_uri_prefix("foo"))
97
98class CHKFile(testutil.ReallyEqualMixin, unittest.TestCase):
99    def test_pack(self):
100        key = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
101        storage_index = hashutil.storage_index_hash(key)
102        uri_extension_hash = hashutil.uri_extension_hash(b"stuff")
103        needed_shares = 25
104        total_shares = 100
105        size = 1234
106        u = uri.CHKFileURI(key=key,
107                           uri_extension_hash=uri_extension_hash,
108                           needed_shares=needed_shares,
109                           total_shares=total_shares,
110                           size=size)
111        self.failUnlessReallyEqual(u.get_storage_index(), storage_index)
112        self.failUnlessReallyEqual(u.key, key)
113        self.failUnlessReallyEqual(u.uri_extension_hash, uri_extension_hash)
114        self.failUnlessReallyEqual(u.needed_shares, needed_shares)
115        self.failUnlessReallyEqual(u.total_shares, total_shares)
116        self.failUnlessReallyEqual(u.size, size)
117        self.failUnless(u.is_readonly())
118        self.failIf(u.is_mutable())
119        self.failUnless(IURI.providedBy(u))
120        self.failUnless(IFileURI.providedBy(u))
121        self.failIf(IDirnodeURI.providedBy(u))
122        self.failUnlessReallyEqual(u.get_size(), 1234)
123
124        u_ro = u.get_readonly()
125        self.failUnlessIdentical(u, u_ro)
126
127        u2 = uri.from_string(u.to_string())
128        self.failUnlessReallyEqual(u2.get_storage_index(), storage_index)
129        self.failUnlessReallyEqual(u2.key, key)
130        self.failUnlessReallyEqual(u2.uri_extension_hash, uri_extension_hash)
131        self.failUnlessReallyEqual(u2.needed_shares, needed_shares)
132        self.failUnlessReallyEqual(u2.total_shares, total_shares)
133        self.failUnlessReallyEqual(u2.size, size)
134        self.failUnless(u2.is_readonly())
135        self.failIf(u2.is_mutable())
136        self.failUnless(IURI.providedBy(u2))
137        self.failUnless(IFileURI.providedBy(u2))
138        self.failIf(IDirnodeURI.providedBy(u2))
139        self.failUnlessReallyEqual(u2.get_size(), 1234)
140
141        u2i = uri.from_string(u.to_string(), deep_immutable=True)
142        self.failUnlessReallyEqual(u.to_string(), u2i.to_string())
143        u2ro = uri.from_string(uri.ALLEGED_READONLY_PREFIX + u.to_string())
144        self.failUnlessReallyEqual(u.to_string(), u2ro.to_string())
145        u2imm = uri.from_string(uri.ALLEGED_IMMUTABLE_PREFIX + u.to_string())
146        self.failUnlessReallyEqual(u.to_string(), u2imm.to_string())
147
148        v = u.get_verify_cap()
149        self.failUnless(isinstance(v.to_string(), bytes))
150        self.failUnless(v.is_readonly())
151        self.failIf(v.is_mutable())
152
153        v2 = uri.from_string(v.to_string())
154        self.failUnlessReallyEqual(v, v2)
155
156        v3 = uri.CHKFileVerifierURI(storage_index=b"\x00"*16,
157                                    uri_extension_hash=b"\x00"*32,
158                                    needed_shares=3,
159                                    total_shares=10,
160                                    size=1234)
161        self.failUnless(isinstance(v3.to_string(), bytes))
162        self.failUnless(v3.is_readonly())
163        self.failIf(v3.is_mutable())
164
165    def test_pack_badly(self):
166        key = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
167        storage_index = hashutil.storage_index_hash(key)
168        uri_extension_hash = hashutil.uri_extension_hash(b"stuff")
169        needed_shares = 25
170        total_shares = 100
171        size = 1234
172        self.failUnlessRaises(TypeError,
173                              uri.CHKFileURI,
174                              key=key,
175                              uri_extension_hash=uri_extension_hash,
176                              needed_shares=needed_shares,
177                              total_shares=total_shares,
178                              size=size,
179
180                              bogus_extra_argument="reject me",
181                              )
182        self.failUnlessRaises(TypeError,
183                              uri.CHKFileVerifierURI,
184                              bogus="bogus")
185        self.failUnlessRaises(TypeError,
186                              uri.CHKFileVerifierURI,
187                              storage_index=storage_index,
188                              uri_extension_hash=uri_extension_hash,
189                              needed_shares=3,
190                              total_shares=10,
191                              # leave size= missing
192                              )
193
194
195class Extension(testutil.ReallyEqualMixin, unittest.TestCase):
196    def test_pack(self):
197        data = {b"stuff": b"value",
198                b"size": 12,
199                b"needed_shares": 3,
200                b"big_hash": hashutil.tagged_hash(b"foo", b"bar"),
201                }
202        ext = uri.pack_extension(data)
203        d = uri.unpack_extension(ext)
204        self.failUnlessReallyEqual(d["stuff"], b"value")
205        self.failUnlessReallyEqual(d["size"], 12)
206        self.failUnlessReallyEqual(d["big_hash"], hashutil.tagged_hash(b"foo", b"bar"))
207
208        readable = uri.unpack_extension_readable(ext)
209        self.failUnlessReallyEqual(readable["needed_shares"], 3)
210        self.failUnlessReallyEqual(readable["stuff"], b"value")
211        self.failUnlessReallyEqual(readable["size"], 12)
212        self.failUnlessReallyEqual(readable["big_hash"],
213                             base32.b2a(hashutil.tagged_hash(b"foo", b"bar")))
214        self.failUnlessReallyEqual(readable["UEB_hash"],
215                             base32.b2a(hashutil.uri_extension_hash(ext)))
216
217class Unknown(testutil.ReallyEqualMixin, unittest.TestCase):
218    def test_from_future(self):
219        # any URI type that we don't recognize should be treated as unknown
220        future_uri = b"I am a URI from the future. Whatever you do, don't "
221        u = uri.from_string(future_uri)
222        self.failUnless(isinstance(u, uri.UnknownURI))
223        self.failUnlessReallyEqual(u.to_string(), future_uri)
224        self.failUnless(u.get_readonly() is None)
225        self.failUnless(u.get_error() is None)
226        future_uri_unicode = future_uri.decode("utf-8")
227        self.assertEqual(future_uri, uri.from_string(future_uri_unicode).to_string())
228
229        u2 = uri.UnknownURI(future_uri, error=CapConstraintError("..."))
230        self.failUnlessReallyEqual(u.to_string(), future_uri)
231        self.failUnless(u2.get_readonly() is None)
232        self.failUnless(isinstance(u2.get_error(), CapConstraintError))
233
234        # Future caps might have non-ASCII chars in them. (Or maybe not, who can tell about the future?)
235        future_uri = u"I am a cap from the \u263A future. Whatever you ".encode("utf-8")
236        u = uri.from_string(future_uri)
237        self.failUnless(isinstance(u, uri.UnknownURI))
238        self.failUnlessReallyEqual(u.to_string(), future_uri)
239        self.failUnless(u.get_readonly() is None)
240        self.failUnless(u.get_error() is None)
241
242        u2 = uri.UnknownURI(future_uri, error=CapConstraintError("..."))
243        self.failUnlessReallyEqual(u.to_string(), future_uri)
244        self.failUnless(u2.get_readonly() is None)
245        self.failUnless(isinstance(u2.get_error(), CapConstraintError))
246
247class Constraint(testutil.ReallyEqualMixin, unittest.TestCase):
248    def test_constraint(self):
249        bad = b"http://127.0.0.1:3456/uri/URI%3ADIR2%3Agh3l5rbvnv2333mrfvalmjfr4i%3Alz6l7u3z3b7g37s4zkdmfpx5ly4ib4m6thrpbusi6ys62qtc6mma/"
250        self.failUnlessRaises(uri.BadURIError, uri.DirectoryURI.init_from_string, bad)
251        fileURI = b'URI:CHK:gh3l5rbvnv2333mrfvalmjfr4i:lz6l7u3z3b7g37s4zkdmfpx5ly4ib4m6thrpbusi6ys62qtc6mma:3:10:345834'
252        uri.CHKFileURI.init_from_string(fileURI)
253
254class Mutable(testutil.ReallyEqualMixin, unittest.TestCase):
255    def setUp(self):
256        self.writekey = b"\x01" * 16
257        self.fingerprint = b"\x02" * 32
258        self.readkey = hashutil.ssk_readkey_hash(self.writekey)
259        self.storage_index = hashutil.ssk_storage_index_hash(self.readkey)
260
261    def test_pack(self):
262        u = uri.WriteableSSKFileURI(self.writekey, self.fingerprint)
263        self.failUnlessReallyEqual(u.writekey, self.writekey)
264        self.failUnlessReallyEqual(u.fingerprint, self.fingerprint)
265        self.failIf(u.is_readonly())
266        self.failUnless(u.is_mutable())
267        self.failUnless(IURI.providedBy(u))
268        self.failUnless(IMutableFileURI.providedBy(u))
269        self.failIf(IDirnodeURI.providedBy(u))
270        self.failUnless("WriteableSSKFileURI" in str(u))
271
272        u2 = uri.from_string(u.to_string())
273        self.failUnlessReallyEqual(u2.writekey, self.writekey)
274        self.failUnlessReallyEqual(u2.fingerprint, self.fingerprint)
275        self.failIf(u2.is_readonly())
276        self.failUnless(u2.is_mutable())
277        self.failUnless(IURI.providedBy(u2))
278        self.failUnless(IMutableFileURI.providedBy(u2))
279        self.failIf(IDirnodeURI.providedBy(u2))
280
281        u2i = uri.from_string(u.to_string(), deep_immutable=True)
282        self.failUnless(isinstance(u2i, uri.UnknownURI), u2i)
283        u2ro = uri.from_string(uri.ALLEGED_READONLY_PREFIX + u.to_string())
284        self.failUnless(isinstance(u2ro, uri.UnknownURI), u2ro)
285        u2imm = uri.from_string(uri.ALLEGED_IMMUTABLE_PREFIX + u.to_string())
286        self.failUnless(isinstance(u2imm, uri.UnknownURI), u2imm)
287
288        u3 = u2.get_readonly()
289        readkey = hashutil.ssk_readkey_hash(self.writekey)
290        self.failUnlessReallyEqual(u3.fingerprint, self.fingerprint)
291        self.failUnlessReallyEqual(u3.readkey, readkey)
292        self.failUnless(u3.is_readonly())
293        self.failUnless(u3.is_mutable())
294        self.failUnless(IURI.providedBy(u3))
295        self.failUnless(IMutableFileURI.providedBy(u3))
296        self.failIf(IDirnodeURI.providedBy(u3))
297
298        u3i = uri.from_string(u3.to_string(), deep_immutable=True)
299        self.failUnless(isinstance(u3i, uri.UnknownURI), u3i)
300        u3ro = uri.from_string(uri.ALLEGED_READONLY_PREFIX + u3.to_string())
301        self.failUnlessReallyEqual(u3.to_string(), u3ro.to_string())
302        u3imm = uri.from_string(uri.ALLEGED_IMMUTABLE_PREFIX + u3.to_string())
303        self.failUnless(isinstance(u3imm, uri.UnknownURI), u3imm)
304
305        u4 = uri.ReadonlySSKFileURI(readkey, self.fingerprint)
306        self.failUnlessReallyEqual(u4.fingerprint, self.fingerprint)
307        self.failUnlessReallyEqual(u4.readkey, readkey)
308        self.failUnless(u4.is_readonly())
309        self.failUnless(u4.is_mutable())
310        self.failUnless(IURI.providedBy(u4))
311        self.failUnless(IMutableFileURI.providedBy(u4))
312        self.failIf(IDirnodeURI.providedBy(u4))
313
314        u4i = uri.from_string(u4.to_string(), deep_immutable=True)
315        self.failUnless(isinstance(u4i, uri.UnknownURI), u4i)
316        u4ro = uri.from_string(uri.ALLEGED_READONLY_PREFIX + u4.to_string())
317        self.failUnlessReallyEqual(u4.to_string(), u4ro.to_string())
318        u4imm = uri.from_string(uri.ALLEGED_IMMUTABLE_PREFIX + u4.to_string())
319        self.failUnless(isinstance(u4imm, uri.UnknownURI), u4imm)
320
321        u4a = uri.from_string(u4.to_string())
322        self.failUnlessReallyEqual(u4a, u4)
323        self.failUnless("ReadonlySSKFileURI" in str(u4a))
324        self.failUnlessIdentical(u4a.get_readonly(), u4a)
325
326        u5 = u4.get_verify_cap()
327        self.failUnless(IVerifierURI.providedBy(u5))
328        self.failUnlessReallyEqual(u5.get_storage_index(), u.get_storage_index())
329        u7 = u.get_verify_cap()
330        self.failUnless(IVerifierURI.providedBy(u7))
331        self.failUnlessReallyEqual(u7.get_storage_index(), u.get_storage_index())
332
333    def test_writeable_mdmf_cap(self):
334        u1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
335        cap = u1.to_string()
336        u = uri.WriteableMDMFFileURI.init_from_string(cap)
337
338        self.failUnless(IMutableFileURI.providedBy(u))
339        self.failUnlessReallyEqual(u.fingerprint, self.fingerprint)
340        self.failUnlessReallyEqual(u.writekey, self.writekey)
341        self.failUnless(u.is_mutable())
342        self.failIf(u.is_readonly())
343        self.failUnlessEqual(cap, u.to_string())
344
345        # Now get a readonly cap from the writeable cap, and test that it
346        # degrades gracefully.
347        ru = u.get_readonly()
348        self.failUnlessReallyEqual(self.readkey, ru.readkey)
349        self.failUnlessReallyEqual(self.fingerprint, ru.fingerprint)
350        self.failUnless(ru.is_mutable())
351        self.failUnless(ru.is_readonly())
352
353        # Now get a verifier cap.
354        vu = ru.get_verify_cap()
355        self.failUnlessReallyEqual(self.storage_index, vu.storage_index)
356        self.failUnlessReallyEqual(self.fingerprint, vu.fingerprint)
357        self.failUnless(IVerifierURI.providedBy(vu))
358
359    def test_readonly_mdmf_cap(self):
360        u1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
361        cap = u1.to_string()
362        u2 = uri.ReadonlyMDMFFileURI.init_from_string(cap)
363
364        self.failUnlessReallyEqual(u2.fingerprint, self.fingerprint)
365        self.failUnlessReallyEqual(u2.readkey, self.readkey)
366        self.failUnless(u2.is_readonly())
367        self.failUnless(u2.is_mutable())
368
369        vu = u2.get_verify_cap()
370        self.failUnlessEqual(vu.storage_index, self.storage_index)
371        self.failUnlessEqual(vu.fingerprint, self.fingerprint)
372
373    def test_create_writeable_mdmf_cap_from_readcap(self):
374        # we shouldn't be able to create a writeable MDMF cap given only a
375        # readcap.
376        u1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
377        cap = u1.to_string()
378        self.failUnlessRaises(uri.BadURIError,
379                              uri.WriteableMDMFFileURI.init_from_string,
380                              cap)
381
382    def test_create_writeable_mdmf_cap_from_verifycap(self):
383        u1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
384        cap = u1.to_string()
385        self.failUnlessRaises(uri.BadURIError,
386                              uri.WriteableMDMFFileURI.init_from_string,
387                              cap)
388
389    def test_create_readonly_mdmf_cap_from_verifycap(self):
390        u1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
391        cap = u1.to_string()
392        self.failUnlessRaises(uri.BadURIError,
393                              uri.ReadonlyMDMFFileURI.init_from_string,
394                              cap)
395
396    def test_mdmf_verifier_cap(self):
397        u1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
398        self.failUnless(u1.is_readonly())
399        self.failIf(u1.is_mutable())
400        self.failUnlessReallyEqual(self.storage_index, u1.storage_index)
401        self.failUnlessReallyEqual(self.fingerprint, u1.fingerprint)
402
403        cap = u1.to_string()
404        u2 = uri.MDMFVerifierURI.init_from_string(cap)
405        self.failUnless(u2.is_readonly())
406        self.failIf(u2.is_mutable())
407        self.failUnlessReallyEqual(self.storage_index, u2.storage_index)
408        self.failUnlessReallyEqual(self.fingerprint, u2.fingerprint)
409
410        u3 = u2.get_readonly()
411        self.failUnlessReallyEqual(u3, u2)
412
413        u4 = u2.get_verify_cap()
414        self.failUnlessReallyEqual(u4, u2)
415
416    def test_mdmf_cap_ignore_extensions(self):
417        # MDMF caps can be arbitrarily extended after the fingerprint and
418        # key/storage index fields. tahoe-1.9 is supposed to ignore any
419        # extensions, and not add any itself.
420        u1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
421        cap = u1.to_string()
422
423        cap2 = cap+b":I COME FROM THE FUTURE"
424        u2 = uri.WriteableMDMFFileURI.init_from_string(cap2)
425        self.failUnlessReallyEqual(self.writekey, u2.writekey)
426        self.failUnlessReallyEqual(self.fingerprint, u2.fingerprint)
427        self.failIf(u2.is_readonly())
428        self.failUnless(u2.is_mutable())
429
430
431        cap3 = cap+b":" + os.urandom(40)
432        u3 = uri.WriteableMDMFFileURI.init_from_string(cap3)
433        self.failUnlessReallyEqual(self.writekey, u3.writekey)
434        self.failUnlessReallyEqual(self.fingerprint, u3.fingerprint)
435        self.failIf(u3.is_readonly())
436        self.failUnless(u3.is_mutable())
437
438        cap4 = u1.get_readonly().to_string()+b":ooh scary future stuff"
439        u4 = uri.from_string_mutable_filenode(cap4)
440        self.failUnlessReallyEqual(self.readkey, u4.readkey)
441        self.failUnlessReallyEqual(self.fingerprint, u4.fingerprint)
442        self.failUnless(u4.is_readonly())
443        self.failUnless(u4.is_mutable())
444
445        cap5 = u1.get_verify_cap().to_string()+b":spoilers!"
446        u5 = uri.from_string(cap5)
447        self.failUnlessReallyEqual(self.storage_index, u5.storage_index)
448        self.failUnlessReallyEqual(self.fingerprint, u5.fingerprint)
449        self.failUnless(u5.is_readonly())
450        self.failIf(u5.is_mutable())
451
452    def test_mdmf_from_string(self):
453        # Make sure that the from_string utility function works with
454        # MDMF caps.
455        u1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
456        cap = u1.to_string()
457        self.failUnless(uri.is_uri(cap))
458        u2 = uri.from_string(cap)
459        self.failUnlessReallyEqual(u1, u2)
460        u3 = uri.from_string_mutable_filenode(cap)
461        self.failUnlessEqual(u3, u1)
462
463        u1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
464        cap = u1.to_string()
465        self.failUnless(uri.is_uri(cap))
466        u2 = uri.from_string(cap)
467        self.failUnlessReallyEqual(u1, u2)
468        u3 = uri.from_string_mutable_filenode(cap)
469        self.failUnlessEqual(u3, u1)
470
471        u1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
472        cap = u1.to_string()
473        self.failUnless(uri.is_uri(cap))
474        u2 = uri.from_string(cap)
475        self.failUnlessReallyEqual(u1, u2)
476        u3 = uri.from_string_verifier(cap)
477        self.failUnlessEqual(u3, u1)
478
479
480class Dirnode(testutil.ReallyEqualMixin, unittest.TestCase):
481    def test_pack(self):
482        writekey = b"\x01" * 16
483        fingerprint = b"\x02" * 32
484
485        n = uri.WriteableSSKFileURI(writekey, fingerprint)
486        u1 = uri.DirectoryURI(n)
487        self.failIf(u1.is_readonly())
488        self.failUnless(u1.is_mutable())
489        self.failUnless(IURI.providedBy(u1))
490        self.failIf(IFileURI.providedBy(u1))
491        self.failUnless(IDirnodeURI.providedBy(u1))
492        self.failUnless("DirectoryURI" in str(u1))
493        u1_filenode = u1.get_filenode_cap()
494        self.failUnless(u1_filenode.is_mutable())
495        self.failIf(u1_filenode.is_readonly())
496
497        u2 = uri.from_string(u1.to_string())
498        self.failUnlessReallyEqual(u1.to_string(), u2.to_string())
499        self.failIf(u2.is_readonly())
500        self.failUnless(u2.is_mutable())
501        self.failUnless(IURI.providedBy(u2))
502        self.failIf(IFileURI.providedBy(u2))
503        self.failUnless(IDirnodeURI.providedBy(u2))
504
505        u2i = uri.from_string(u1.to_string(), deep_immutable=True)
506        self.failUnless(isinstance(u2i, uri.UnknownURI))
507
508        u3 = u2.get_readonly()
509        self.failUnless(u3.is_readonly())
510        self.failUnless(u3.is_mutable())
511        self.failUnless(IURI.providedBy(u3))
512        self.failIf(IFileURI.providedBy(u3))
513        self.failUnless(IDirnodeURI.providedBy(u3))
514
515        u3i = uri.from_string(u2.to_string(), deep_immutable=True)
516        self.failUnless(isinstance(u3i, uri.UnknownURI))
517
518        u3n = u3._filenode_uri
519        self.failUnless(u3n.is_readonly())
520        self.failUnless(u3n.is_mutable())
521        u3_filenode = u3.get_filenode_cap()
522        self.failUnless(u3_filenode.is_mutable())
523        self.failUnless(u3_filenode.is_readonly())
524
525        u3a = uri.from_string(u3.to_string())
526        self.failUnlessIdentical(u3a, u3a.get_readonly())
527
528        u4 = uri.ReadonlyDirectoryURI(u2._filenode_uri.get_readonly())
529        self.failUnlessReallyEqual(u4.to_string(), u3.to_string())
530        self.failUnless(u4.is_readonly())
531        self.failUnless(u4.is_mutable())
532        self.failUnless(IURI.providedBy(u4))
533        self.failIf(IFileURI.providedBy(u4))
534        self.failUnless(IDirnodeURI.providedBy(u4))
535
536        u4_verifier = u4.get_verify_cap()
537        u4_verifier_filenode = u4_verifier.get_filenode_cap()
538        self.failUnless(isinstance(u4_verifier_filenode, uri.SSKVerifierURI))
539
540        verifiers = [u1.get_verify_cap(), u2.get_verify_cap(),
541                     u3.get_verify_cap(), u4.get_verify_cap(),
542                     uri.DirectoryURIVerifier(n.get_verify_cap()),
543                     ]
544        for v in verifiers:
545            self.failUnless(IVerifierURI.providedBy(v))
546            self.failUnlessReallyEqual(v._filenode_uri,
547                                 u1.get_verify_cap()._filenode_uri)
548
549    def test_immutable(self):
550        readkey = b"\x01" * 16
551        uri_extension_hash = hashutil.uri_extension_hash(b"stuff")
552        needed_shares = 3
553        total_shares = 10
554        size = 1234
555
556        fnuri = uri.CHKFileURI(key=readkey,
557                               uri_extension_hash=uri_extension_hash,
558                               needed_shares=needed_shares,
559                               total_shares=total_shares,
560                               size=size)
561        fncap = fnuri.to_string()
562        self.failUnlessReallyEqual(fncap, b"URI:CHK:aeaqcaibaeaqcaibaeaqcaibae:nf3nimquen7aeqm36ekgxomalstenpkvsdmf6fplj7swdatbv5oa:3:10:1234")
563        u1 = uri.ImmutableDirectoryURI(fnuri)
564        self.failUnless(u1.is_readonly())
565        self.failIf(u1.is_mutable())
566        self.failUnless(IURI.providedBy(u1))
567        self.failIf(IFileURI.providedBy(u1))
568        self.failUnless(IDirnodeURI.providedBy(u1))
569        self.failUnless("DirectoryURI" in str(u1))
570        u1_filenode = u1.get_filenode_cap()
571        self.failIf(u1_filenode.is_mutable())
572        self.failUnless(u1_filenode.is_readonly())
573        self.failUnlessReallyEqual(u1_filenode.to_string(), fncap)
574        self.failUnless(str(u1))
575
576        u2 = uri.from_string(u1.to_string())
577        self.failUnlessReallyEqual(u1.to_string(), u2.to_string())
578        self.failUnless(u2.is_readonly())
579        self.failIf(u2.is_mutable())
580        self.failUnless(IURI.providedBy(u2))
581        self.failIf(IFileURI.providedBy(u2))
582        self.failUnless(IDirnodeURI.providedBy(u2))
583
584        u2i = uri.from_string(u1.to_string(), deep_immutable=True)
585        self.failUnlessReallyEqual(u1.to_string(), u2i.to_string())
586
587        u3 = u2.get_readonly()
588        self.failUnlessReallyEqual(u3.to_string(), u2.to_string())
589        self.failUnless(str(u3))
590
591        u3i = uri.from_string(u2.to_string(), deep_immutable=True)
592        self.failUnlessReallyEqual(u2.to_string(), u3i.to_string())
593
594        u2_verifier = u2.get_verify_cap()
595        self.failUnless(isinstance(u2_verifier,
596                                   uri.ImmutableDirectoryURIVerifier),
597                        u2_verifier)
598        self.failUnless(IVerifierURI.providedBy(u2_verifier))
599        u2vs = u2_verifier.to_string()
600        # URI:DIR2-CHK-Verifier:$key:$ueb:$k:$n:$size
601        self.failUnless(u2vs.startswith(b"URI:DIR2-CHK-Verifier:"), u2vs)
602        u2_verifier_fileuri = u2_verifier.get_filenode_cap()
603        self.failUnless(IVerifierURI.providedBy(u2_verifier_fileuri))
604        u2vfs = u2_verifier_fileuri.to_string()
605        # URI:CHK-Verifier:$key:$ueb:$k:$n:$size
606        self.failUnlessReallyEqual(u2vfs, fnuri.get_verify_cap().to_string())
607        self.failUnlessReallyEqual(u2vs[len(b"URI:DIR2-"):], u2vfs[len(b"URI:"):])
608        self.failUnless(str(u2_verifier))
609
610    def test_literal(self):
611        u0 = uri.LiteralFileURI(b"data")
612        u1 = uri.LiteralDirectoryURI(u0)
613        self.failUnless(str(u1))
614        self.failUnlessReallyEqual(u1.to_string(), b"URI:DIR2-LIT:mrqxiyi")
615        self.failUnless(u1.is_readonly())
616        self.failIf(u1.is_mutable())
617        self.failUnless(IURI.providedBy(u1))
618        self.failIf(IFileURI.providedBy(u1))
619        self.failUnless(IDirnodeURI.providedBy(u1))
620        self.failUnlessReallyEqual(u1.get_verify_cap(), None)
621        self.failUnlessReallyEqual(u1.get_storage_index(), None)
622        self.failUnlessReallyEqual(u1.abbrev_si(), b"<LIT>")
623
624    def test_mdmf(self):
625        writekey = b"\x01" * 16
626        fingerprint = b"\x02" * 32
627        uri1 = uri.WriteableMDMFFileURI(writekey, fingerprint)
628        d1 = uri.MDMFDirectoryURI(uri1)
629        self.failIf(d1.is_readonly())
630        self.failUnless(d1.is_mutable())
631        self.failUnless(IURI.providedBy(d1))
632        self.failUnless(IDirnodeURI.providedBy(d1))
633        d1_uri = d1.to_string()
634
635        d2 = uri.from_string(d1_uri)
636        self.failUnlessIsInstance(d2, uri.MDMFDirectoryURI)
637        self.failIf(d2.is_readonly())
638        self.failUnless(d2.is_mutable())
639        self.failUnless(IURI.providedBy(d2))
640        self.failUnless(IDirnodeURI.providedBy(d2))
641
642        # It doesn't make sense to ask for a deep immutable URI for a
643        # mutable directory, and we should get back a result to that
644        # effect.
645        d3 = uri.from_string(d2.to_string(), deep_immutable=True)
646        self.failUnlessIsInstance(d3, uri.UnknownURI)
647
648    def test_mdmf_attenuation(self):
649        writekey = b"\x01" * 16
650        fingerprint = b"\x02" * 32
651
652        uri1 = uri.WriteableMDMFFileURI(writekey, fingerprint)
653        d1 = uri.MDMFDirectoryURI(uri1)
654        self.failUnless(d1.is_mutable())
655        self.failIf(d1.is_readonly())
656        self.failUnless(IURI.providedBy(d1))
657        self.failUnless(IDirnodeURI.providedBy(d1))
658
659        d1_uri = d1.to_string()
660        d1_uri_from_fn = uri.MDMFDirectoryURI(d1.get_filenode_cap()).to_string()
661        self.failUnlessEqual(d1_uri_from_fn, d1_uri)
662
663        uri2 = uri.from_string(d1_uri)
664        self.failUnlessIsInstance(uri2, uri.MDMFDirectoryURI)
665        self.failUnless(IURI.providedBy(uri2))
666        self.failUnless(IDirnodeURI.providedBy(uri2))
667        self.failUnless(uri2.is_mutable())
668        self.failIf(uri2.is_readonly())
669
670        ro = uri2.get_readonly()
671        self.failUnlessIsInstance(ro, uri.ReadonlyMDMFDirectoryURI)
672        self.failUnless(ro.is_mutable())
673        self.failUnless(ro.is_readonly())
674        self.failUnless(IURI.providedBy(ro))
675        self.failUnless(IDirnodeURI.providedBy(ro))
676
677        ro_uri = ro.to_string()
678        n = uri.from_string(ro_uri, deep_immutable=True)
679        self.failUnlessIsInstance(n, uri.UnknownURI)
680
681        fn_cap = ro.get_filenode_cap()
682        fn_ro_cap = fn_cap.get_readonly()
683        d3 = uri.ReadonlyMDMFDirectoryURI(fn_ro_cap)
684        self.failUnlessEqual(ro.to_string(), d3.to_string())
685        self.failUnless(ro.is_mutable())
686        self.failUnless(ro.is_readonly())
687
688    def test_mdmf_verifier(self):
689        # I'm not sure what I want to write here yet.
690        writekey = b"\x01" * 16
691        fingerprint = b"\x02" * 32
692        uri1 = uri.WriteableMDMFFileURI(writekey, fingerprint)
693        d1 = uri.MDMFDirectoryURI(uri1)
694        v1 = d1.get_verify_cap()
695        self.failUnlessIsInstance(v1, uri.MDMFDirectoryURIVerifier)
696        self.failIf(v1.is_mutable())
697
698        d2 = uri.from_string(d1.to_string())
699        v2 = d2.get_verify_cap()
700        self.failUnlessIsInstance(v2, uri.MDMFDirectoryURIVerifier)
701        self.failIf(v2.is_mutable())
702        self.failUnlessEqual(v2.to_string(), v1.to_string())
703
704        # Now attenuate and make sure that works correctly.
705        r3 = d2.get_readonly()
706        v3 = r3.get_verify_cap()
707        self.failUnlessIsInstance(v3, uri.MDMFDirectoryURIVerifier)
708        self.failIf(v3.is_mutable())
709        self.failUnlessEqual(v3.to_string(), v1.to_string())
710        r4 = uri.from_string(r3.to_string())
711        v4 = r4.get_verify_cap()
712        self.failUnlessIsInstance(v4, uri.MDMFDirectoryURIVerifier)
713        self.failIf(v4.is_mutable())
714        self.failUnlessEqual(v4.to_string(), v3.to_string())
Note: See TracBrowser for help on using the repository browser.