src/allmydata/test/test_system.py

   1
   2 import os
   3 from twisted.trial import unittest
   4 from twisted.internet import defer, reactor
   5 from twisted.application import service
   6 from allmydata import client, queen
   7 from allmydata.util import idlib
   8 from foolscap.eventual import flushEventualQueue
   9 from twisted.python import log
  10 from twisted.web.client import getPage
  11
  12 class SystemTest(unittest.TestCase):
  13     # it takes a little while for a disconnected loopback TCP connection to
  14     # be noticed by the other side. This is not directly visible to us, so we
  15     # have to wait for time to pass rather than just waiting on a deferred.
  16     # This is unfortunate, both because it arbitrarily slows down the test
  17     # process, and because it is hard to predict what the minimum time
  18     # necessary would be (on a slow or heavily loaded system, 100ms might not
  19     # be enough).
  20     DISCONNECT_DELAY = 0.1
  21
  22     def setUp(self):
  23         self.sparent = service.MultiService()
  24         self.sparent.startService()
  25     def tearDown(self):
  26         log.msg("shutting down SystemTest services")
  27         d = self.sparent.stopService()
  28         d.addCallback(lambda res: flushEventualQueue())
  29         def _done(res):
  30             d1 = defer.Deferred()
  31             reactor.callLater(self.DISCONNECT_DELAY, d1.callback, None)
  32             return d1
  33         d.addCallback(_done)
  34         return d
  35
  36     def add_service(self, s):
  37         s.setServiceParent(self.sparent)
  38         return s
  39
  40     def set_up_nodes(self, NUMCLIENTS=5):
  41         self.numclients = NUMCLIENTS
  42         if not os.path.isdir("queen"):
  43             os.mkdir("queen")
  44         self.queen = self.add_service(queen.Queen(basedir="queen"))
  45         d = self.queen.when_tub_ready()
  46         d.addCallback(self._set_up_nodes_2)
  47         return d
  48
  49     def _set_up_nodes_2(self, res):
  50         q = self.queen
  51         self.queen_furl = q.urls["introducer"]
  52         self.vdrive_furl = q.urls["vdrive"]
  53         self.clients = []
  54         for i in range(self.numclients):
  55             basedir = "client%d" % i
  56             if not os.path.isdir(basedir):
  57                 os.mkdir(basedir)
  58             if i == 0:
  59                 open(os.path.join(basedir, "webport"), "w").write("tcp:0:interface=127.0.0.1")
  60             open(os.path.join(basedir, "introducer.furl"), "w").write(self.queen_furl)
  61             open(os.path.join(basedir, "vdrive.furl"), "w").write(self.vdrive_furl)
  62             c = self.add_service(client.Client(basedir=basedir))
  63             self.clients.append(c)
  64         log.msg("STARTING")
  65         d = self.wait_for_connections()
  66         def _connected(res):
  67             # now find out where the web port was
  68             l = self.clients[0].getServiceNamed("webish").listener
  69             port = l._port.getHost().port
  70             self.webish_url = "http://localhost:%d/" % port
  71         d.addCallback(_connected)
  72         return d
  73
  74     def add_extra_node(self, client_num):
  75         # this node is *not* parented to our self.sparent, so we can shut it
  76         # down separately from the rest, to exercise the connection-lost code
  77         basedir = "client%d" % client_num
  78         if not os.path.isdir(basedir):
  79             os.mkdir(basedir)
  80         open(os.path.join(basedir, "introducer.furl"), "w").write(self.queen_furl)
  81         open(os.path.join(basedir, "vdrive.furl"), "w").write(self.vdrive_furl)
  82
  83         c = client.Client(basedir=basedir)
  84         self.clients.append(c)
  85         self.numclients += 1
  86         c.startService()
  87         d = self.wait_for_connections()
  88         d.addCallback(lambda res: c)
  89         return d
  90
  91     def wait_for_connections(self, ignored=None):
  92         for c in self.clients:
  93             if (not c.introducer_client or
  94                 len(list(c.get_all_peerids())) != self.numclients):
  95                 d = defer.Deferred()
  96                 d.addCallback(self.wait_for_connections)
  97                 reactor.callLater(0.05, d.callback, None)
  98                 return d
  99         return defer.succeed(None)
 100
 101     def test_connections(self):
 102         d = self.set_up_nodes()
 103         self.extra_node = None
 104         d.addCallback(lambda res: self.add_extra_node(5))
 105         def _check(extra_node):
 106             self.extra_node = extra_node
 107             for c in self.clients:
 108                 self.failUnlessEqual(len(list(c.get_all_peerids())), 6)
 109         d.addCallback(_check)
 110         def _shutdown_extra_node(res):
 111             if self.extra_node:
 112                 d1 = self.extra_node.stopService()
 113                 d2 = defer.Deferred()
 114                 reactor.callLater(self.DISCONNECT_DELAY, d2.callback, res)
 115                 d1.addCallback(lambda ignored: d2)
 116                 return d1
 117             return res
 118         d.addBoth(_shutdown_extra_node)
 119         return d
 120
 121     def test_upload_and_download(self):
 122         DATA = "Some data to upload\n"
 123         d = self.set_up_nodes()
 124         def _do_upload(res):
 125             log.msg("UPLOADING")
 126             u = self.clients[0].getServiceNamed("uploader")
 127             d1 = u.upload_data(DATA)
 128             return d1
 129         d.addCallback(_do_upload)
 130         def _upload_done(uri):
 131             log.msg("upload finished: uri is %s" % (uri,))
 132             dl = self.clients[1].getServiceNamed("downloader")
 133             d1 = dl.download_to_data(uri)
 134             return d1
 135         d.addCallback(_upload_done)
 136         def _download_done(data):
 137             log.msg("download finished")
 138             self.failUnlessEqual(data, DATA)
 139         d.addCallback(_download_done)
 140         return d
 141
 142     def test_vdrive(self):
 143         self.data = DATA = "Some data to publish to the virtual drive\n"
 144         d = self.set_up_nodes()
 145         def _do_publish(res):
 146             log.msg("PUBLISHING")
 147             v0 = self.clients[0].getServiceNamed("vdrive")
 148             d1 = v0.make_directory("/", "subdir1")
 149             d1.addCallback(lambda subdir1:
 150                            v0.put_file_by_data(subdir1, "mydata567", DATA))
 151             return d1
 152         d.addCallback(_do_publish)
 153         def _publish_done(res):
 154             log.msg("publish finished")
 155             v1 = self.clients[1].getServiceNamed("vdrive")
 156             d1 = v1.get_file_to_data("/subdir1/mydata567")
 157             return d1
 158         d.addCallback(_publish_done)
 159         def _get_done(data):
 160             log.msg("get finished")
 161             self.failUnlessEqual(data, DATA)
 162         d.addCallback(_get_done)
 163         d.addCallback(self._test_web)
 164         return d
 165
 166     def _test_web(self, res):
 167         base = self.webish_url
 168         d = getPage(base)
 169         def _got_welcome(page):
 170             expected = "Connected Peers: <span>%d</span>" % (self.numclients)
 171             self.failUnless(expected in page,
 172                             "I didn't see the right 'connected peers' message "
 173                             "in: %s" % page
 174                             )
 175             expected = "My nodeid: <span>%s</span>" % idlib.b2a(self.clients[0].nodeid)
 176             self.failUnless(expected in page,
 177                             "I didn't see the right 'My nodeid' message "
 178                             "in: %s" % page)
 179         d.addCallback(_got_welcome)
 180         d.addCallback(lambda res: getPage(base + "vdrive/subdir1"))
 181         def _got_subdir1(page):
 182             # there ought to be an href for our file
 183             self.failUnless(">mydata567</a>" in page)
 184         d.addCallback(_got_subdir1)
 185         d.addCallback(lambda res: getPage(base + "vdrive/subdir1/mydata567"))
 186         def _got_data(page):
 187             self.failUnlessEqual(page, self.data)
 188         d.addCallback(_got_data)
 189         return d
 190