src/allmydata/test/test_filenode.py

   1
   2 from twisted.trial import unittest
   3 from allmydata import uri
   4 from allmydata.monitor import Monitor
   5 from allmydata.immutable import filenode, download
   6 from allmydata.mutable.node import MutableFileNode
   7 from allmydata.util import hashutil
   8 from allmydata.test.common import download_to_data
   9
  10 class NotANode:
  11     pass
  12
  13 class Node(unittest.TestCase):
  14     def test_chk_filenode(self):
  15         u = uri.CHKFileURI(key="\x00"*16,
  16                            uri_extension_hash="\x00"*32,
  17                            needed_shares=3,
  18                            total_shares=10,
  19                            size=1000)
  20         c = None
  21         fn1 = filenode.FileNode(u, c, "cachefile")
  22         fn2 = filenode.FileNode(u.to_string(), c, "cachefile")
  23         self.failUnlessEqual(fn1, fn2)
  24         self.failIfEqual(fn1, "I am not a filenode")
  25         self.failIfEqual(fn1, NotANode())
  26         self.failUnlessEqual(fn1.get_uri(), u.to_string())
  27         self.failUnlessEqual(fn1.is_readonly(), True)
  28         self.failUnlessEqual(fn1.is_mutable(), False)
  29         self.failUnlessEqual(fn1.get_readonly_uri(), u.to_string())
  30         self.failUnlessEqual(fn1.get_size(), 1000)
  31         self.failUnlessEqual(fn1.get_storage_index(), u.storage_index)
  32         d = {}
  33         d[fn1] = 1 # exercise __hash__
  34         v = fn1.get_verifier()
  35         self.failUnless(isinstance(v, uri.CHKFileVerifierURI))
  36
  37
  38     def test_literal_filenode(self):
  39         DATA = "I am a short file."
  40         u = uri.LiteralFileURI(data=DATA)
  41         c = None
  42         fn1 = filenode.LiteralFileNode(u, c)
  43         fn2 = filenode.LiteralFileNode(u.to_string(), c)
  44         self.failUnlessEqual(fn1, fn2)
  45         self.failIfEqual(fn1, "I am not a filenode")
  46         self.failIfEqual(fn1, NotANode())
  47         self.failUnlessEqual(fn1.get_uri(), u.to_string())
  48         self.failUnlessEqual(fn1.is_readonly(), True)
  49         self.failUnlessEqual(fn1.is_mutable(), False)
  50         self.failUnlessEqual(fn1.get_readonly_uri(), u.to_string())
  51         self.failUnlessEqual(fn1.get_size(), len(DATA))
  52         self.failUnlessEqual(fn1.get_storage_index(), None)
  53         d = {}
  54         d[fn1] = 1 # exercise __hash__
  55
  56         v = fn1.get_verifier()
  57         self.failUnlessEqual(v, None)
  58
  59         d = fn1.download(download.Data())
  60         def _check(res):
  61             self.failUnlessEqual(res, DATA)
  62         d.addCallback(_check)
  63
  64         d.addCallback(lambda res: fn1.download_to_data())
  65         d.addCallback(_check)
  66
  67         d.addCallback(lambda res: download_to_data(fn1))
  68         d.addCallback(_check)
  69
  70         d.addCallback(lambda res: download_to_data(fn1, 1, 5))
  71         def _check_segment(res):
  72             self.failUnlessEqual(res, DATA[1:1+5])
  73         d.addCallback(_check_segment)
  74
  75         return d
  76
  77     def test_mutable_filenode(self):
  78         client = None
  79         wk = "\x00"*16
  80         fp = "\x00"*32
  81         rk = hashutil.ssk_readkey_hash(wk)
  82         si = hashutil.ssk_storage_index_hash(rk)
  83
  84         u = uri.WriteableSSKFileURI("\x00"*16, "\x00"*32)
  85         n = MutableFileNode(client).init_from_uri(u)
  86
  87         self.failUnlessEqual(n.get_writekey(), wk)
  88         self.failUnlessEqual(n.get_readkey(), rk)
  89         self.failUnlessEqual(n.get_storage_index(), si)
  90         # these itmes are populated on first read (or create), so until that
  91         # happens they'll be None
  92         self.failUnlessEqual(n.get_privkey(), None)
  93         self.failUnlessEqual(n.get_encprivkey(), None)
  94         self.failUnlessEqual(n.get_pubkey(), None)
  95
  96         self.failUnlessEqual(n.get_uri(), u.to_string())
  97         self.failUnlessEqual(n.get_readonly_uri(), u.get_readonly().to_string())
  98         self.failUnlessEqual(n.is_mutable(), True)
  99         self.failUnlessEqual(n.is_readonly(), False)
 100
 101         n2 = MutableFileNode(client).init_from_uri(u)
 102         self.failUnlessEqual(n, n2)
 103         self.failIfEqual(n, "not even the right type")
 104         self.failIfEqual(n, u) # not the right class
 105         d = {n: "can these be used as dictionary keys?"}
 106         d[n2] = "replace the old one"
 107         self.failUnlessEqual(len(d), 1)
 108
 109         nro = n.get_readonly()
 110         self.failUnless(isinstance(nro, MutableFileNode))
 111
 112         self.failUnlessEqual(nro.get_readonly(), nro)
 113         nro_u = nro.get_uri()
 114         self.failUnlessEqual(nro_u, nro.get_readonly_uri())
 115         self.failUnlessEqual(nro_u, u.get_readonly().to_string())
 116         self.failUnlessEqual(nro.is_mutable(), True)
 117         self.failUnlessEqual(nro.is_readonly(), True)
 118
 119         v = n.get_verifier()
 120         self.failUnless(isinstance(v, uri.SSKVerifierURI))
 121
 122 class LiteralChecker(unittest.TestCase):
 123     def test_literal_filenode(self):
 124         DATA = "I am a short file."
 125         u = uri.LiteralFileURI(data=DATA)
 126         fn1 = filenode.LiteralFileNode(u, None)
 127
 128         d = fn1.check(Monitor())
 129         def _check_checker_results(cr):
 130             self.failUnlessEqual(cr, None)
 131         d.addCallback(_check_checker_results)
 132
 133         d.addCallback(lambda res: fn1.check(Monitor(), verify=True))
 134         d.addCallback(_check_checker_results)
 135
 136         return d