src/allmydata/test/no_network.py

   1
   2 # This contains a test harness that creates a full Tahoe grid in a single
   3 # process (actually in a single MultiService) which does not use the network.
   4 # It does not use an Introducer, and there are no foolscap Tubs. Each storage
   5 # server puts real shares on disk, but is accessed through loopback
   6 # RemoteReferences instead of over serialized SSL. It is not as complete as
   7 # the common.SystemTestMixin framework (which does use the network), but
   8 # should be considerably faster: on my laptop, it takes 50-80ms to start up,
   9 # whereas SystemTestMixin takes close to 2s.
  10
  11 # This should be useful for tests which want to examine and/or manipulate the
  12 # uploaded shares, checker/verifier/repairer tests, etc. The clients have no
  13 # Tubs, so it is not useful for tests that involve a Helper, a KeyGenerator,
  14 # or the control.furl .
  15
  16 import os.path
  17 import sha
  18 from twisted.application import service
  19 from foolscap import Referenceable
  20 from foolscap.eventual import fireEventually
  21 from base64 import b32encode
  22 from allmydata.client import Client
  23 from allmydata.storage import StorageServer
  24 from allmydata.util import fileutil, idlib, hashutil, rrefutil
  25 from allmydata.introducer.client import RemoteServiceConnector
  26
  27 class IntentionalError(Exception):
  28     pass
  29
  30 class Marker:
  31     pass
  32
  33 class LocalWrapper:
  34     def __init__(self, original):
  35         self.original = original
  36         self.broken = False
  37         self.post_call_notifier = None
  38         self.disconnectors = {}
  39
  40     def callRemoteOnly(self, methname, *args, **kwargs):
  41         d = self.callRemote(methname, *args, **kwargs)
  42         return None
  43
  44     def callRemote(self, methname, *args, **kwargs):
  45         # this is ideally a Membrane, but that's too hard. We do a shallow
  46         # wrapping of inbound arguments, and per-methodname wrapping of
  47         # selected return values.
  48         def wrap(a):
  49             if isinstance(a, Referenceable):
  50                 return LocalWrapper(a)
  51             else:
  52                 return a
  53         args = tuple([wrap(a) for a in args])
  54         kwargs = dict([(k,wrap(kwargs[k])) for k in kwargs])
  55         def _call():
  56             if self.broken:
  57                 raise IntentionalError("I was asked to break")
  58             meth = getattr(self.original, "remote_" + methname)
  59             return meth(*args, **kwargs)
  60         d = fireEventually()
  61         d.addCallback(lambda res: _call())
  62         def _return_membrane(res):
  63             # rather than complete the difficult task of building a
  64             # fully-general Membrane (which would locate all Referenceable
  65             # objects that cross the simulated wire and replace them with
  66             # wrappers), we special-case certain methods that we happen to
  67             # know will return Referenceables.
  68             if methname == "allocate_buckets":
  69                 (alreadygot, allocated) = res
  70                 for shnum in allocated:
  71                     allocated[shnum] = LocalWrapper(allocated[shnum])
  72             if methname == "get_buckets":
  73                 for shnum in res:
  74                     res[shnum] = LocalWrapper(res[shnum])
  75             return res
  76         d.addCallback(_return_membrane)
  77         if self.post_call_notifier:
  78             d.addCallback(self.post_call_notifier, methname)
  79         return d
  80
  81     def notifyOnDisconnect(self, f, *args, **kwargs):
  82         m = Marker()
  83         self.disconnectors[m] = (f, args, kwargs)
  84         return m
  85     def dontNotifyOnDisconnect(self, marker):
  86         del self.disconnectors[marker]
  87
  88 def wrap(original, service_name):
  89     # The code in immutable.checker insists upon asserting the truth of
  90     # isinstance(rref, rrefutil.WrappedRemoteReference). Much of the
  91     # upload/download code uses rref.version (which normally comes from
  92     # rrefutil.VersionedRemoteReference). To avoid using a network, we want a
  93     # LocalWrapper here. Try to satisfy all these constraints at the same
  94     # time.
  95     local = LocalWrapper(original)
  96     wrapped = rrefutil.WrappedRemoteReference(local)
  97     try:
  98         version = original.remote_get_version()
  99     except AttributeError:
 100         version = RemoteServiceConnector.VERSION_DEFAULTS[service_name]
 101     wrapped.version = version
 102     return wrapped
 103
 104 class NoNetworkClient(Client):
 105
 106     def create_tub(self):
 107         pass
 108     def init_introducer_client(self):
 109         pass
 110     def setup_logging(self):
 111         pass
 112     def startService(self):
 113         service.MultiService.startService(self)
 114     def stopService(self):
 115         service.MultiService.stopService(self)
 116     def when_tub_ready(self):
 117         raise RuntimeError("NoNetworkClient has no Tub")
 118     def init_control(self):
 119         pass
 120     def init_helper(self):
 121         pass
 122     def init_key_gen(self):
 123         pass
 124     def init_storage(self):
 125         pass
 126     def init_stub_client(self):
 127         pass
 128
 129     def get_servers(self, service_name):
 130         return self._servers
 131
 132     def get_permuted_peers(self, service_name, key):
 133         return sorted(self._servers, key=lambda x: sha.new(key+x[0]).digest())
 134
 135
 136 class NoNetworkGrid(service.MultiService):
 137     def __init__(self, basedir, num_clients=1, num_servers=10,
 138                  client_config_hooks={}):
 139         service.MultiService.__init__(self)
 140         self.basedir = basedir
 141         fileutil.make_dirs(basedir)
 142
 143         self.servers_by_number = {}
 144         self.servers_by_id = {}
 145         self.clients = []
 146
 147         for i in range(num_servers):
 148             serverid = hashutil.tagged_hash("serverid", str(i))[:20]
 149             serverdir = os.path.join(basedir, "servers",
 150                                      idlib.shortnodeid_b2a(serverid))
 151             fileutil.make_dirs(serverdir)
 152             ss = StorageServer(serverdir)
 153             self.add_server(i, serverid, ss)
 154
 155         for i in range(num_clients):
 156             clientid = hashutil.tagged_hash("clientid", str(i))[:20]
 157             clientdir = os.path.join(basedir, "clients",
 158                                      idlib.shortnodeid_b2a(clientid))
 159             fileutil.make_dirs(clientdir)
 160             f = open(os.path.join(clientdir, "tahoe.cfg"), "w")
 161             f.write("[node]\n")
 162             f.write("nickname = client-%d\n" % i)
 163             f.write("web.port = tcp:0:interface=127.0.0.1\n")
 164             f.write("[storage]\n")
 165             f.write("enabled = false\n")
 166             f.close()
 167             c = None
 168             if i in client_config_hooks:
 169                 # this hook can either modify tahoe.cfg, or return an
 170                 # entirely new Client instance
 171                 c = client_config_hooks[i](clientdir)
 172             if not c:
 173                 c = NoNetworkClient(clientdir)
 174             c.nodeid = clientid
 175             c.short_nodeid = b32encode(clientid).lower()[:8]
 176             c._servers = self.all_servers # can be updated later
 177             c.setServiceParent(self)
 178             self.clients.append(c)
 179
 180     def add_server(self, i, serverid, ss):
 181         # TODO: ss.setServiceParent(self), but first remove the goofy
 182         # self.parent.nodeid from Storage.startService . At the moment,
 183         # Storage doesn't really need to be startService'd, but it will in
 184         # the future.
 185         ss.setNodeID(serverid)
 186         self.servers_by_number[i] = ss
 187         self.servers_by_id[serverid] = wrap(ss, "storage")
 188         self.all_servers = frozenset(self.servers_by_id.items())
 189         for c in self.clients:
 190             c._servers = self.all_servers
 191
 192 class GridTestMixin:
 193     def setUp(self):
 194         self.s = service.MultiService()
 195         self.s.startService()
 196
 197     def tearDown(self):
 198         return self.s.stopService()
 199
 200     def set_up_grid(self, client_config_hooks={}):
 201         # self.basedir must be set
 202         self.g = NoNetworkGrid(self.basedir,
 203                                client_config_hooks=client_config_hooks)
 204         self.g.setServiceParent(self.s)
 205
 206     def get_clientdir(self, i=0):
 207         return self.g.clients[i].basedir
 208
 209     def get_serverdir(self, i):
 210         return self.g.servers_by_number[i].storedir
 211
 212     def iterate_servers(self):
 213         for i in sorted(self.g.servers_by_number.keys()):
 214             ss = self.g.servers_by_number[i]
 215             yield (i, ss, ss.storedir)