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