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, DirnodeURI, \
   6      IMutableFileURI, IVerifierURI
   7 from foolscap.schema import Violation
   8
   9 class Literal(unittest.TestCase):
  10     def _help_test(self, data):
  11         u = uri.LiteralFileURI(data)
  12         self.failUnless(IURI.providedBy(u))
  13         self.failUnless(IFileURI.providedBy(u))
  14         self.failIf(IDirnodeURI.providedBy(u))
  15         self.failUnlessEqual(u.data, data)
  16         self.failUnlessEqual(u.get_size(), len(data))
  17         self.failUnless(u.is_readonly())
  18         self.failIf(u.is_mutable())
  19
  20         u2 = uri.from_string(u.to_string())
  21         self.failUnless(IURI.providedBy(u2))
  22         self.failUnless(IFileURI.providedBy(u2))
  23         self.failIf(IDirnodeURI.providedBy(u2))
  24         self.failUnlessEqual(u2.data, data)
  25         self.failUnlessEqual(u2.get_size(), len(data))
  26         self.failUnless(u.is_readonly())
  27         self.failIf(u.is_mutable())
  28
  29     def test_empty(self):
  30         data = "" # This data is some *very* small data!
  31         return self._help_test(data)
  32
  33     def test_pack(self):
  34         data = "This is some small data"
  35         return self._help_test(data)
  36
  37     def test_nonascii(self):
  38         data = "This contains \x00 and URI:LIT: and \n, oh my."
  39         return self._help_test(data)
  40
  41 class Compare(unittest.TestCase):
  42     def test_compare(self):
  43         lit1 = uri.LiteralFileURI("some data")
  44         fileURI = 'URI:CHK:f3mf6az85wpcai8ma4qayfmxuc:nnw518w5hu3t5oohwtp7ah9n81z9rfg6c1ywk33ia3m64o67nsgo:3:10:345834'
  45         chk1 = uri.CHKFileURI().init_from_string(fileURI)
  46         chk2 = uri.CHKFileURI().init_from_string(fileURI)
  47         self.failIfEqual(lit1, chk1)
  48         self.failUnlessEqual(chk1, chk2)
  49         self.failIfEqual(chk1, "not actually a URI")
  50         # these should be hashable too
  51         s = set([lit1, chk1, chk2])
  52         self.failUnlessEqual(len(s), 2) # since chk1==chk2
  53
  54 class CHKFile(unittest.TestCase):
  55     def test_pack(self):
  56         key = "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
  57         storage_index = hashutil.storage_index_chk_hash(key)
  58         uri_extension_hash = hashutil.uri_extension_hash("stuff")
  59         needed_shares = 25
  60         total_shares = 100
  61         size = 1234
  62         u = uri.CHKFileURI(key=key,
  63                            uri_extension_hash=uri_extension_hash,
  64                            needed_shares=needed_shares,
  65                            total_shares=total_shares,
  66                            size=size)
  67         self.failUnlessEqual(u.storage_index, storage_index)
  68         self.failUnlessEqual(u.key, key)
  69         self.failUnlessEqual(u.uri_extension_hash, uri_extension_hash)
  70         self.failUnlessEqual(u.needed_shares, needed_shares)
  71         self.failUnlessEqual(u.total_shares, total_shares)
  72         self.failUnlessEqual(u.size, size)
  73         self.failUnless(u.is_readonly())
  74         self.failIf(u.is_mutable())
  75         self.failUnless(IURI.providedBy(u))
  76         self.failUnless(IFileURI.providedBy(u))
  77         self.failIf(IDirnodeURI.providedBy(u))
  78         self.failUnlessEqual(u.get_size(), 1234)
  79         self.failUnless(u.is_readonly())
  80         self.failIf(u.is_mutable())
  81
  82         u2 = uri.from_string(u.to_string())
  83         self.failUnlessEqual(u2.storage_index, storage_index)
  84         self.failUnlessEqual(u2.key, key)
  85         self.failUnlessEqual(u2.uri_extension_hash, uri_extension_hash)
  86         self.failUnlessEqual(u2.needed_shares, needed_shares)
  87         self.failUnlessEqual(u2.total_shares, total_shares)
  88         self.failUnlessEqual(u2.size, size)
  89         self.failUnless(u2.is_readonly())
  90         self.failIf(u2.is_mutable())
  91         self.failUnless(IURI.providedBy(u2))
  92         self.failUnless(IFileURI.providedBy(u2))
  93         self.failIf(IDirnodeURI.providedBy(u2))
  94         self.failUnlessEqual(u2.get_size(), 1234)
  95         self.failUnless(u2.is_readonly())
  96         self.failIf(u2.is_mutable())
  97
  98     def test_pack_badly(self):
  99         key = "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
 100         storage_index = hashutil.storage_index_chk_hash(key)
 101         uri_extension_hash = hashutil.uri_extension_hash("stuff")
 102         needed_shares = 25
 103         total_shares = 100
 104         size = 1234
 105         self.failUnlessRaises(TypeError,
 106                               uri.CHKFileURI,
 107                               key=key,
 108                               uri_extension_hash=uri_extension_hash,
 109                               needed_shares=needed_shares,
 110                               total_shares=total_shares,
 111                               size=size,
 112
 113                               bogus_extra_argument="reject me",
 114                               )
 115
 116 class Extension(unittest.TestCase):
 117     def test_pack(self):
 118         data = {"stuff": "value",
 119                 "size": 12,
 120                 "needed_shares": 3,
 121                 "big_hash": hashutil.tagged_hash("foo", "bar"),
 122                 }
 123         ext = uri.pack_extension(data)
 124         d = uri.unpack_extension(ext)
 125         self.failUnlessEqual(d["stuff"], "value")
 126         self.failUnlessEqual(d["size"], 12)
 127         self.failUnlessEqual(d["big_hash"], hashutil.tagged_hash("foo", "bar"))
 128
 129         readable = uri.unpack_extension_readable(ext)
 130
 131 class Invalid(unittest.TestCase):
 132     def test_create_invalid(self):
 133         not_uri = "I am not a URI"
 134         self.failUnlessRaises(TypeError, uri.from_string, not_uri)
 135
 136
 137 class Constraint(unittest.TestCase):
 138     def test_constraint(self):
 139        good = 'URI:DIR:pb://xextf3eap44o3wi27mf7ehiur6wvhzr6@207.7.153.180:56677,127.0.0.1:56677/vdrive:qj51rfpnukhjmo7cm9awe5ks5e'
 140        DirnodeURI.checkObject(good, False)
 141        bad = good + '==='
 142        self.failUnlessRaises(Violation, DirnodeURI.checkObject, bad, False)
 143        fileURI = 'URI:CHK:f3mf6az85wpcai8ma4qayfmxuc:nnw518w5hu3t5oohwtp7ah9n81z9rfg6c1ywk33ia3m64o67nsgo:3:10:345834'
 144        self.failUnlessRaises(Violation, DirnodeURI.checkObject, fileURI, False)
 145
 146
 147 class Mutable(unittest.TestCase):
 148     def test_pack(self):
 149         writekey = "\x01" * 16
 150         fingerprint = "\x02" * 32
 151
 152         u = uri.WriteableSSKFileURI(writekey, fingerprint)
 153         self.failUnlessEqual(u.writekey, writekey)
 154         self.failUnlessEqual(u.fingerprint, fingerprint)
 155         self.failIf(u.is_readonly())
 156         self.failUnless(u.is_mutable())
 157         self.failUnless(IURI.providedBy(u))
 158         self.failUnless(IMutableFileURI.providedBy(u))
 159         self.failIf(IDirnodeURI.providedBy(u))
 160
 161         u2 = uri.from_string(u.to_string())
 162         self.failUnlessEqual(u2.writekey, writekey)
 163         self.failUnlessEqual(u2.fingerprint, fingerprint)
 164         self.failIf(u2.is_readonly())
 165         self.failUnless(u2.is_mutable())
 166         self.failUnless(IURI.providedBy(u2))
 167         self.failUnless(IMutableFileURI.providedBy(u2))
 168         self.failIf(IDirnodeURI.providedBy(u2))
 169
 170         u3 = u2.get_readonly()
 171         readkey = hashutil.ssk_readkey_hash(writekey)
 172         self.failUnlessEqual(u3.fingerprint, fingerprint)
 173         self.failUnlessEqual(u3.readkey, readkey)
 174         self.failUnless(u3.is_readonly())
 175         self.failUnless(u3.is_mutable())
 176         self.failUnless(IURI.providedBy(u3))
 177         self.failUnless(IMutableFileURI.providedBy(u3))
 178         self.failIf(IDirnodeURI.providedBy(u3))
 179
 180         u4 = uri.ReadonlySSKFileURI(readkey, fingerprint)
 181         self.failUnlessEqual(u4.fingerprint, fingerprint)
 182         self.failUnlessEqual(u4.readkey, readkey)
 183         self.failUnless(u4.is_readonly())
 184         self.failUnless(u4.is_mutable())
 185         self.failUnless(IURI.providedBy(u4))
 186         self.failUnless(IMutableFileURI.providedBy(u4))
 187         self.failIf(IDirnodeURI.providedBy(u4))
 188
 189         u5 = u4.get_verifier()
 190         self.failUnless(IVerifierURI.providedBy(u5))
 191         self.failUnlessEqual(u5.storage_index, u.storage_index)
 192         u6 = IVerifierURI(u5.to_string())
 193         self.failUnless(IVerifierURI.providedBy(u6))
 194         self.failUnlessEqual(u6.storage_index, u.storage_index)
 195         u7 = u.get_verifier()
 196         self.failUnless(IVerifierURI.providedBy(u7))
 197         self.failUnlessEqual(u7.storage_index, u.storage_index)
 198
 199
 200 class NewDirnode(unittest.TestCase):
 201     def test_pack(self):
 202         writekey = "\x01" * 16
 203         fingerprint = "\x02" * 16
 204
 205         n = uri.WriteableSSKFileURI(writekey, fingerprint)
 206         u1 = uri.NewDirectoryURI(n)
 207         self.failIf(u1.is_readonly())
 208         self.failUnless(u1.is_mutable())
 209         self.failUnless(IURI.providedBy(u1))
 210         self.failIf(IFileURI.providedBy(u1))
 211         self.failUnless(IDirnodeURI.providedBy(u1))
 212
 213         u2 = uri.from_string(u1.to_string())
 214         self.failUnlessEqual(u1.to_string(), u2.to_string())
 215         self.failIf(u2.is_readonly())
 216         self.failUnless(u2.is_mutable())
 217         self.failUnless(IURI.providedBy(u2))
 218         self.failIf(IFileURI.providedBy(u2))
 219         self.failUnless(IDirnodeURI.providedBy(u2))
 220
 221         u3 = u2.get_readonly()
 222         self.failUnless(u3.is_readonly())
 223         self.failUnless(u3.is_mutable())
 224         self.failUnless(IURI.providedBy(u3))
 225         self.failIf(IFileURI.providedBy(u3))
 226         self.failUnless(IDirnodeURI.providedBy(u3))
 227         u3n = u3._filenode_uri
 228         self.failUnless(u3n.is_readonly())
 229         self.failUnless(u3n.is_mutable())
 230
 231         u4 = uri.ReadonlyNewDirectoryURI(u2._filenode_uri.get_readonly())
 232         self.failUnlessEqual(u4.to_string(), u3.to_string())
 233         self.failUnless(u4.is_readonly())
 234         self.failUnless(u4.is_mutable())
 235         self.failUnless(IURI.providedBy(u4))
 236         self.failIf(IFileURI.providedBy(u4))
 237         self.failUnless(IDirnodeURI.providedBy(u4))
 238
 239         verifiers = [u1.get_verifier(), u2.get_verifier(),
 240                      u3.get_verifier(), u4.get_verifier(),
 241                      IVerifierURI(u1.get_verifier().to_string()),
 242                      uri.NewDirectoryURIVerifier(n.get_verifier()),
 243                      uri.NewDirectoryURIVerifier(n.get_verifier().to_string()),
 244                      ]
 245         for v in verifiers:
 246             self.failUnless(IVerifierURI.providedBy(v))
 247             self.failUnlessEqual(v._filenode_uri,
 248                                  u1.get_verifier()._filenode_uri)
 249
 250