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