src/allmydata/test/common.py

   1
   2 import os
   3 from zope.interface import implements
   4 from twisted.internet import defer
   5 from twisted.python import failure
   6 from twisted.application import service
   7 from allmydata import uri, dirnode
   8 from allmydata.interfaces import IURI, IMutableFileNode, IFileNode
   9 from allmydata.encode import NotEnoughSharesError
  10 from allmydata.util import log
  11
  12 class FakeCHKFileNode:
  13     """I provide IFileNode, but all of my data is stored in a class-level
  14     dictionary."""
  15     implements(IFileNode)
  16     all_contents = {}
  17
  18     def __init__(self, u, client):
  19         self.client = client
  20         self.my_uri = u.to_string()
  21
  22     def get_uri(self):
  23         return self.my_uri
  24     def get_readonly_uri(self):
  25         return self.my_uri
  26     def get_verifier(self):
  27         return IURI(self.my_uri).get_verifier()
  28     def check(self):
  29         return defer.succeed(None)
  30     def is_mutable(self):
  31         return False
  32     def is_readonly(self):
  33         return True
  34
  35     def download(self, target):
  36         if self.my_uri not in self.all_contents:
  37             f = failure.Failure(NotEnoughSharesError())
  38             target.fail(f)
  39             return defer.fail(f)
  40         data = self.all_contents[self.my_uri]
  41         target.open(len(data))
  42         target.write(data)
  43         target.close()
  44         return defer.maybeDeferred(target.finish)
  45     def download_to_data(self):
  46         if self.my_uri not in self.all_contents:
  47             return defer.fail(NotEnoughSharesError())
  48         data = self.all_contents[self.my_uri]
  49         return defer.succeed(data)
  50     def get_size(self):
  51         data = self.all_contents[self.my_uri]
  52         return len(data)
  53
  54 def make_chk_file_uri(size):
  55     return uri.CHKFileURI(key=os.urandom(16),
  56                           uri_extension_hash=os.urandom(32),
  57                           needed_shares=3,
  58                           total_shares=10,
  59                           size=size)
  60
  61 def create_chk_filenode(client, contents):
  62     u = make_chk_file_uri(len(contents))
  63     n = FakeCHKFileNode(u, client)
  64     FakeCHKFileNode.all_contents[u.to_string()] = contents
  65     return n
  66
  67
  68 class FakeMutableFileNode:
  69     """I provide IMutableFileNode, but all of my data is stored in a
  70     class-level dictionary."""
  71
  72     implements(IMutableFileNode)
  73     all_contents = {}
  74     def __init__(self, client):
  75         self.client = client
  76         self.my_uri = make_mutable_file_uri()
  77         self.storage_index = self.my_uri.storage_index
  78     def create(self, initial_contents, key_generator=None):
  79         self.all_contents[self.storage_index] = initial_contents
  80         return defer.succeed(self)
  81     def init_from_uri(self, myuri):
  82         self.my_uri = IURI(myuri)
  83         self.storage_index = self.my_uri.storage_index
  84         return self
  85     def get_uri(self):
  86         return self.my_uri.to_string()
  87     def get_readonly_uri(self):
  88         return self.my_uri.get_readonly().to_string()
  89     def is_readonly(self):
  90         return self.my_uri.is_readonly()
  91     def is_mutable(self):
  92         return self.my_uri.is_mutable()
  93     def download_to_data(self):
  94         return defer.succeed(self.all_contents[self.storage_index])
  95     def get_writekey(self):
  96         return "\x00"*16
  97     def get_size(self):
  98         return "?" # TODO: see mutable.MutableFileNode.get_size
  99
 100     def update(self, new_contents):
 101         assert not self.is_readonly()
 102         self.all_contents[self.storage_index] = new_contents
 103         return defer.succeed(None)
 104
 105     def overwrite(self, new_contents):
 106         return self.update(new_contents)
 107
 108
 109 def make_mutable_file_uri():
 110     return uri.WriteableSSKFileURI(writekey=os.urandom(16),
 111                                    fingerprint=os.urandom(32))
 112 def make_verifier_uri():
 113     return uri.SSKVerifierURI(storage_index=os.urandom(16),
 114                               fingerprint=os.urandom(32))
 115
 116 class FakeDirectoryNode(dirnode.NewDirectoryNode):
 117     """This offers IDirectoryNode, but uses a FakeMutableFileNode for the
 118     backing store, so it doesn't go to the grid. The child data is still
 119     encrypted and serialized, so this isn't useful for tests that want to
 120     look inside the dirnodes and check their contents.
 121     """
 122     filenode_class = FakeMutableFileNode
 123
 124 class LoggingServiceParent(service.MultiService):
 125     def log(self, *args, **kwargs):
 126         return log.msg(*args, **kwargs)