src/allmydata/test/test_uri.py

   1
   2 from twisted.trial import unittest
   3 from allmydata import uri
   4 from allmydata.util import hashutil
   5 from allmydata.interfaces import IURI, IFileURI, IDirnodeURI, IMutableFileURI, \
   6     IVerifierURI
   7
   8 class Literal(unittest.TestCase):
   9     def _help_test(self, data):
  10         u = uri.LiteralFileURI(data)
  11         self.failUnless(IURI.providedBy(u))
  12         self.failUnless(IFileURI.providedBy(u))
  13         self.failIf(IDirnodeURI.providedBy(u))
  14         self.failUnlessEqual(u.data, data)
  15         self.failUnlessEqual(u.get_size(), len(data))
  16         self.failUnless(u.is_readonly())
  17         self.failIf(u.is_mutable())
  18
  19         u2 = uri.from_string(u.to_string())
  20         self.failUnless(IURI.providedBy(u2))
  21         self.failUnless(IFileURI.providedBy(u2))
  22         self.failIf(IDirnodeURI.providedBy(u2))
  23         self.failUnlessEqual(u2.data, data)
  24         self.failUnlessEqual(u2.get_size(), len(data))
  25         self.failUnless(u.is_readonly())
  26         self.failIf(u.is_mutable())
  27
  28         u3 = u.get_readonly()
  29         self.failUnlessIdentical(u, u3)
  30         self.failUnlessEqual(u.get_verifier(), None)
  31
  32     def test_empty(self):
  33         data = "" # This data is some *very* small data!
  34         return self._help_test(data)
  35
  36     def test_pack(self):
  37         data = "This is some small data"
  38         return self._help_test(data)
  39
  40     def test_nonascii(self):
  41         data = "This contains \x00 and URI:LIT: and \n, oh my."
  42         return self._help_test(data)
  43
  44 class Compare(unittest.TestCase):
  45     def test_compare(self):
  46         lit1 = uri.LiteralFileURI("some data")
  47         fileURI = 'URI:CHK:f3mf6az85wpcai8ma4qayfmxuc:nnw518w5hu3t5oohwtp7ah9n81z9rfg6c1ywk33ia3m64o67nsgo:3:10:345834'
  48         chk1 = uri.CHKFileURI().init_from_string(fileURI)
  49         chk2 = uri.CHKFileURI().init_from_string(fileURI)
  50         self.failIfEqual(lit1, chk1)
  51         self.failUnlessEqual(chk1, chk2)
  52         self.failIfEqual(chk1, "not actually a URI")
  53         # these should be hashable too
  54         s = set([lit1, chk1, chk2])
  55         self.failUnlessEqual(len(s), 2) # since chk1==chk2
  56
  57 class CHKFile(unittest.TestCase):
  58     def test_pack(self):
  59         key = "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
  60         storage_index = hashutil.storage_index_chk_hash(key)
  61         uri_extension_hash = hashutil.uri_extension_hash("stuff")
  62         needed_shares = 25
  63         total_shares = 100
  64         size = 1234
  65         u = uri.CHKFileURI(key=key,
  66                            uri_extension_hash=uri_extension_hash,
  67                            needed_shares=needed_shares,
  68                            total_shares=total_shares,
  69                            size=size)
  70         self.failUnlessEqual(u.storage_index, storage_index)
  71         self.failUnlessEqual(u.key, key)
  72         self.failUnlessEqual(u.uri_extension_hash, uri_extension_hash)
  73         self.failUnlessEqual(u.needed_shares, needed_shares)
  74         self.failUnlessEqual(u.total_shares, total_shares)
  75         self.failUnlessEqual(u.size, size)
  76         self.failUnless(u.is_readonly())
  77         self.failIf(u.is_mutable())
  78         self.failUnless(IURI.providedBy(u))
  79         self.failUnless(IFileURI.providedBy(u))
  80         self.failIf(IDirnodeURI.providedBy(u))
  81         self.failUnlessEqual(u.get_size(), 1234)
  82         self.failUnless(u.is_readonly())
  83         self.failIf(u.is_mutable())
  84         u_ro = u.get_readonly()
  85         self.failUnlessIdentical(u, u_ro)
  86         u1a = IFileURI(u.to_string())
  87         self.failUnlessEqual(u1a, u)
  88
  89         u2 = uri.from_string(u.to_string())
  90         self.failUnlessEqual(u2.storage_index, storage_index)
  91         self.failUnlessEqual(u2.key, key)
  92         self.failUnlessEqual(u2.uri_extension_hash, uri_extension_hash)
  93         self.failUnlessEqual(u2.needed_shares, needed_shares)
  94         self.failUnlessEqual(u2.total_shares, total_shares)
  95         self.failUnlessEqual(u2.size, size)
  96         self.failUnless(u2.is_readonly())
  97         self.failIf(u2.is_mutable())
  98         self.failUnless(IURI.providedBy(u2))
  99         self.failUnless(IFileURI.providedBy(u2))
 100         self.failIf(IDirnodeURI.providedBy(u2))
 101         self.failUnlessEqual(u2.get_size(), 1234)
 102         self.failUnless(u2.is_readonly())
 103         self.failIf(u2.is_mutable())
 104
 105         v = u.get_verifier()
 106         self.failUnless(isinstance(v.to_string(), str))
 107         v2 = uri.from_string(v.to_string())
 108         self.failUnlessEqual(v, v2)
 109
 110         v3 = uri.CHKFileVerifierURI(storage_index="\x00"*16,
 111                                     uri_extension_hash="\x00"*32,
 112                                     needed_shares=3,
 113                                     total_shares=10,
 114                                     size=1234)
 115         self.failUnless(isinstance(v3.to_string(), str))
 116
 117     def test_pack_badly(self):
 118         key = "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
 119         storage_index = hashutil.storage_index_chk_hash(key)
 120         uri_extension_hash = hashutil.uri_extension_hash("stuff")
 121         needed_shares = 25
 122         total_shares = 100
 123         size = 1234
 124         self.failUnlessRaises(TypeError,
 125                               uri.CHKFileURI,
 126                               key=key,
 127                               uri_extension_hash=uri_extension_hash,
 128                               needed_shares=needed_shares,
 129                               total_shares=total_shares,
 130                               size=size,
 131
 132                               bogus_extra_argument="reject me",
 133                               )
 134         self.failUnlessRaises(TypeError,
 135                               uri.CHKFileVerifierURI,
 136                               bogus="bogus")
 137         self.failUnlessRaises(TypeError,
 138                               uri.CHKFileVerifierURI,
 139                               storage_index=storage_index,
 140                               uri_extension_hash=uri_extension_hash,
 141                               needed_shares=3,
 142                               total_shares=10,
 143                               # leave size= missing
 144                               )
 145
 146
 147 class Extension(unittest.TestCase):
 148     def test_pack(self):
 149         data = {"stuff": "value",
 150                 "size": 12,
 151                 "needed_shares": 3,
 152                 "big_hash": hashutil.tagged_hash("foo", "bar"),
 153                 }
 154         ext = uri.pack_extension(data)
 155         d = uri.unpack_extension(ext)
 156         self.failUnlessEqual(d["stuff"], "value")
 157         self.failUnlessEqual(d["size"], 12)
 158         self.failUnlessEqual(d["big_hash"], hashutil.tagged_hash("foo", "bar"))
 159
 160         readable = uri.unpack_extension_readable(ext)
 161
 162 class Invalid(unittest.TestCase):
 163     def test_create_invalid(self):
 164         not_uri = "I am not a URI"
 165         self.failUnlessRaises(TypeError, uri.from_string, not_uri)
 166
 167
 168 class Constraint(unittest.TestCase):
 169     def test_constraint(self):
 170        good="http://127.0.0.1:8123/uri/URI%3ADIR2%3Aqo8ayna47cpw3rx3kho3mu7q4h%3Abk9qbgx76gh6eyj5ps8p6buz8fffw1ofc37e9w9d6ncsfpuz7icy/"
 171        self.failUnless(uri.DirnodeURI_RE.search(good))
 172        bad = good + '==='
 173        self.failIf(uri.DirnodeURI_RE.search(bad))
 174        fileURI = 'URI:CHK:f3mf6az85wpcai8ma4qayfmxuc:nnw518w5hu3t5oohwtp7ah9n81z9rfg6c1ywk33ia3m64o67nsgo:3:10:345834'
 175        self.failIf(uri.DirnodeURI_RE.search(fileURI))
 176
 177
 178 class Mutable(unittest.TestCase):
 179     def test_pack(self):
 180         writekey = "\x01" * 16
 181         fingerprint = "\x02" * 32
 182
 183         u = uri.WriteableSSKFileURI(writekey, fingerprint)
 184         self.failUnlessEqual(u.writekey, writekey)
 185         self.failUnlessEqual(u.fingerprint, fingerprint)
 186         self.failIf(u.is_readonly())
 187         self.failUnless(u.is_mutable())
 188         self.failUnless(IURI.providedBy(u))
 189         self.failUnless(IMutableFileURI.providedBy(u))
 190         self.failIf(IDirnodeURI.providedBy(u))
 191         self.failUnless("WriteableSSKFileURI" in str(u))
 192         u1a = IMutableFileURI(u.to_string())
 193         self.failUnlessEqual(u1a, u)
 194
 195         u2 = uri.from_string(u.to_string())
 196         self.failUnlessEqual(u2.writekey, writekey)
 197         self.failUnlessEqual(u2.fingerprint, fingerprint)
 198         self.failIf(u2.is_readonly())
 199         self.failUnless(u2.is_mutable())
 200         self.failUnless(IURI.providedBy(u2))
 201         self.failUnless(IMutableFileURI.providedBy(u2))
 202         self.failIf(IDirnodeURI.providedBy(u2))
 203
 204         u3 = u2.get_readonly()
 205         readkey = hashutil.ssk_readkey_hash(writekey)
 206         self.failUnlessEqual(u3.fingerprint, fingerprint)
 207         self.failUnlessEqual(u3.readkey, readkey)
 208         self.failUnless(u3.is_readonly())
 209         self.failUnless(u3.is_mutable())
 210         self.failUnless(IURI.providedBy(u3))
 211         self.failUnless(IMutableFileURI.providedBy(u3))
 212         self.failIf(IDirnodeURI.providedBy(u3))
 213
 214         u4 = uri.ReadonlySSKFileURI(readkey, fingerprint)
 215         self.failUnlessEqual(u4.fingerprint, fingerprint)
 216         self.failUnlessEqual(u4.readkey, readkey)
 217         self.failUnless(u4.is_readonly())
 218         self.failUnless(u4.is_mutable())
 219         self.failUnless(IURI.providedBy(u4))
 220         self.failUnless(IMutableFileURI.providedBy(u4))
 221         self.failIf(IDirnodeURI.providedBy(u4))
 222
 223         u4a = uri.from_string(u4.to_string())
 224         self.failUnlessEqual(u4a, u4)
 225         self.failUnless("ReadonlySSKFileURI" in str(u4a))
 226         self.failUnlessIdentical(u4a.get_readonly(), u4a)
 227
 228         u5 = u4.get_verifier()
 229         self.failUnless(IVerifierURI.providedBy(u5))
 230         self.failUnlessEqual(u5.storage_index, u.storage_index)
 231         u6 = IVerifierURI(u5.to_string())
 232         self.failUnless(IVerifierURI.providedBy(u6))
 233         self.failUnlessEqual(u6.storage_index, u.storage_index)
 234         u7 = u.get_verifier()
 235         self.failUnless(IVerifierURI.providedBy(u7))
 236         self.failUnlessEqual(u7.storage_index, u.storage_index)
 237
 238
 239 class NewDirnode(unittest.TestCase):
 240     def test_pack(self):
 241         writekey = "\x01" * 16
 242         fingerprint = "\x02" * 16
 243
 244         n = uri.WriteableSSKFileURI(writekey, fingerprint)
 245         u1 = uri.NewDirectoryURI(n)
 246         self.failIf(u1.is_readonly())
 247         self.failUnless(u1.is_mutable())
 248         self.failUnless(IURI.providedBy(u1))
 249         self.failIf(IFileURI.providedBy(u1))
 250         self.failUnless(IDirnodeURI.providedBy(u1))
 251         self.failUnless("NewDirectoryURI" in str(u1))
 252         u1_filenode = u1.get_filenode_uri()
 253         self.failUnless(u1_filenode.is_mutable())
 254         self.failIf(u1_filenode.is_readonly())
 255         u1a = IDirnodeURI(u1.to_string())
 256         self.failUnlessEqual(u1a, u1)
 257
 258         u2 = uri.from_string(u1.to_string())
 259         self.failUnlessEqual(u1.to_string(), u2.to_string())
 260         self.failIf(u2.is_readonly())
 261         self.failUnless(u2.is_mutable())
 262         self.failUnless(IURI.providedBy(u2))
 263         self.failIf(IFileURI.providedBy(u2))
 264         self.failUnless(IDirnodeURI.providedBy(u2))
 265
 266         u3 = u2.get_readonly()
 267         self.failUnless(u3.is_readonly())
 268         self.failUnless(u3.is_mutable())
 269         self.failUnless(IURI.providedBy(u3))
 270         self.failIf(IFileURI.providedBy(u3))
 271         self.failUnless(IDirnodeURI.providedBy(u3))
 272         u3n = u3._filenode_uri
 273         self.failUnless(u3n.is_readonly())
 274         self.failUnless(u3n.is_mutable())
 275         u3_filenode = u3.get_filenode_uri()
 276         self.failUnless(u3_filenode.is_mutable())
 277         self.failUnless(u3_filenode.is_readonly())
 278
 279         u3a = uri.from_string(u3.to_string())
 280         self.failUnlessIdentical(u3a, u3a.get_readonly())
 281
 282         u4 = uri.ReadonlyNewDirectoryURI(u2._filenode_uri.get_readonly())
 283         self.failUnlessEqual(u4.to_string(), u3.to_string())
 284         self.failUnless(u4.is_readonly())
 285         self.failUnless(u4.is_mutable())
 286         self.failUnless(IURI.providedBy(u4))
 287         self.failIf(IFileURI.providedBy(u4))
 288         self.failUnless(IDirnodeURI.providedBy(u4))
 289
 290         u4_verifier = u4.get_verifier()
 291         u4_verifier_filenode = u4_verifier.get_filenode_uri()
 292         self.failUnless(isinstance(u4_verifier_filenode, uri.SSKVerifierURI))
 293
 294         verifiers = [u1.get_verifier(), u2.get_verifier(),
 295                      u3.get_verifier(), u4.get_verifier(),
 296                      IVerifierURI(u1.get_verifier().to_string()),
 297                      uri.NewDirectoryURIVerifier(n.get_verifier()),
 298                      uri.NewDirectoryURIVerifier(n.get_verifier().to_string()),
 299                      ]
 300         for v in verifiers:
 301             self.failUnless(IVerifierURI.providedBy(v))
 302             self.failUnlessEqual(v._filenode_uri,
 303                                  u1.get_verifier()._filenode_uri)
 304
 305     def test_is_string_newdirnode_rw(self):
 306         writekey = "\x01" * 16
 307         fingerprint = "\x02" * 32
 308
 309         n = uri.WriteableSSKFileURI(writekey, fingerprint)
 310         u1 = uri.NewDirectoryURI(n)
 311
 312         self.failUnless(uri.is_string_newdirnode_rw(u1.to_string()), u1.to_string())
 313
 314         self.failIf(uri.is_string_newdirnode_rw("bogus"))
 315         self.failIf(uri.is_string_newdirnode_rw("URI:DIR2:bogus"))
 316