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 from foolscap.schema import Violation
   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     def test_empty(self):
  29         data = "" # This data is some *very* small data!
  30         return self._help_test(data)
  31
  32     def test_pack(self):
  33         data = "This is some small data"
  34         return self._help_test(data)
  35
  36     def test_nonascii(self):
  37         data = "This contains \x00 and URI:LIT: and \n, oh my."
  38         return self._help_test(data)
  39
  40 class CHKFile(unittest.TestCase):
  41     def test_pack(self):
  42         key = "\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f"
  43         storage_index = hashutil.storage_index_chk_hash(key)
  44         uri_extension_hash = hashutil.uri_extension_hash("stuff")
  45         needed_shares = 25
  46         total_shares = 100
  47         size = 1234
  48         u = uri.CHKFileURI(key=key,
  49                            uri_extension_hash=uri_extension_hash,
  50                            needed_shares=needed_shares,
  51                            total_shares=total_shares,
  52                            size=size)
  53         self.failUnlessEqual(u.storage_index, storage_index)
  54         self.failUnlessEqual(u.key, key)
  55         self.failUnlessEqual(u.uri_extension_hash, uri_extension_hash)
  56         self.failUnlessEqual(u.needed_shares, needed_shares)
  57         self.failUnlessEqual(u.total_shares, total_shares)
  58         self.failUnlessEqual(u.size, size)
  59         self.failUnless(u.is_readonly())
  60         self.failIf(u.is_mutable())
  61         self.failUnless(IURI.providedBy(u))
  62         self.failUnless(IFileURI.providedBy(u))
  63         self.failIf(IDirnodeURI.providedBy(u))
  64         self.failUnlessEqual(u.get_size(), 1234)
  65         self.failUnless(u.is_readonly())
  66         self.failIf(u.is_mutable())
  67
  68         u2 = uri.from_string(u.to_string())
  69         self.failUnlessEqual(u2.storage_index, storage_index)
  70         self.failUnlessEqual(u2.key, key)
  71         self.failUnlessEqual(u2.uri_extension_hash, uri_extension_hash)
  72         self.failUnlessEqual(u2.needed_shares, needed_shares)
  73         self.failUnlessEqual(u2.total_shares, total_shares)
  74         self.failUnlessEqual(u2.size, size)
  75         self.failUnless(u2.is_readonly())
  76         self.failIf(u2.is_mutable())
  77         self.failUnless(IURI.providedBy(u2))
  78         self.failUnless(IFileURI.providedBy(u2))
  79         self.failIf(IDirnodeURI.providedBy(u2))
  80         self.failUnlessEqual(u2.get_size(), 1234)
  81         self.failUnless(u2.is_readonly())
  82         self.failIf(u2.is_mutable())
  83
  84 class Extension(unittest.TestCase):
  85     def test_pack(self):
  86         data = {"stuff": "value",
  87                 "size": 12,
  88                 "needed_shares": 3,
  89                 "big_hash": hashutil.tagged_hash("foo", "bar"),
  90                 }
  91         ext = uri.pack_extension(data)
  92         d = uri.unpack_extension(ext)
  93         self.failUnlessEqual(d["stuff"], "value")
  94         self.failUnlessEqual(d["size"], 12)
  95         self.failUnlessEqual(d["big_hash"], hashutil.tagged_hash("foo", "bar"))
  96
  97         readable = uri.unpack_extension_readable(ext)
  98
  99 class Dirnode(unittest.TestCase):
 100     def test_pack(self):
 101         furl = "pb://stuff@morestuff:stuff/andstuff"
 102         writekey = "\x01" * 16
 103
 104         u = uri.DirnodeURI(furl, writekey)
 105         self.failUnlessEqual(u.furl, furl)
 106         self.failUnlessEqual(u.writekey, writekey)
 107         self.failIf(u.is_readonly())
 108         self.failUnless(u.is_mutable())
 109         self.failUnless(IURI.providedBy(u))
 110         self.failIf(IFileURI.providedBy(u))
 111         self.failUnless(IDirnodeURI.providedBy(u))
 112
 113         u2 = uri.from_string(u.to_string())
 114         self.failUnlessEqual(u2.furl, furl)
 115         self.failUnlessEqual(u2.writekey, writekey)
 116         self.failIf(u2.is_readonly())
 117         self.failUnless(u2.is_mutable())
 118         self.failUnless(IURI.providedBy(u2))
 119         self.failIf(IFileURI.providedBy(u2))
 120         self.failUnless(IDirnodeURI.providedBy(u2))
 121
 122         u3 = u2.get_readonly()
 123         readkey = hashutil.dir_read_key_hash(writekey)
 124         self.failUnlessEqual(u3.furl, furl)
 125         self.failUnlessEqual(u3.readkey, readkey)
 126         self.failUnless(u3.is_readonly())
 127         self.failUnless(u3.is_mutable())
 128         self.failUnless(IURI.providedBy(u3))
 129         self.failIf(IFileURI.providedBy(u3))
 130         self.failUnless(IDirnodeURI.providedBy(u3))
 131
 132         u4 = uri.ReadOnlyDirnodeURI(furl, readkey)
 133         self.failUnlessEqual(u4.furl, furl)
 134         self.failUnlessEqual(u4.readkey, readkey)
 135         self.failUnless(u4.is_readonly())
 136         self.failUnless(u4.is_mutable())
 137         self.failUnless(IURI.providedBy(u4))
 138         self.failIf(IFileURI.providedBy(u4))
 139         self.failUnless(IDirnodeURI.providedBy(u4))
 140
 141
 142 class Constraint(unittest.TestCase):
 143     def test_constraint(self):
 144        good = 'URI:DIR:pb://xextf3eap44o3wi27mf7ehiur6wvhzr6@207.7.153.180:56677,127.0.0.1:56677/vdrive:qj51rfpnukhjmo7cm9awe5ks5e'
 145        DirnodeURI.checkObject(good, False)
 146        bad = good + '==='
 147        self.failUnlessRaises(Violation, DirnodeURI.checkObject, bad, False)
 148        fileURI = 'URI:CHK:f3mf6az85wpcai8ma4qayfmxuc:nnw518w5hu3t5oohwtp7ah9n81z9rfg6c1ywk33ia3m64o67nsgo:3:10:345834'
 149        self.failUnlessRaises(Violation, DirnodeURI.checkObject, fileURI, False)
 150
 151