src/allmydata/test/test_system.py

   1 from base64 import b32encode
   2 import os, sys, time, simplejson
   3 from cStringIO import StringIO
   4 from twisted.trial import unittest
   5 from twisted.internet import defer
   6 from twisted.internet import threads # CLI tests use deferToThread
   7
   8 import allmydata
   9 from allmydata import uri
  10 from allmydata.storage.mutable import MutableShareFile
  11 from allmydata.storage.server import si_a2b
  12 from allmydata.immutable import offloaded, upload
  13 from allmydata.immutable.literal import LiteralFileNode
  14 from allmydata.immutable.filenode import ImmutableFileNode
  15 from allmydata.util import idlib, mathutil
  16 from allmydata.util import log, base32
  17 from allmydata.util.encodingutil import quote_output, unicode_to_argv, get_filesystem_encoding
  18 from allmydata.util.fileutil import abspath_expanduser_unicode
  19 from allmydata.util.consumer import MemoryConsumer, download_to_data
  20 from allmydata.scripts import runner
  21 from allmydata.interfaces import IDirectoryNode, IFileNode, \
  22      NoSuchChildError, NoSharesError
  23 from allmydata.monitor import Monitor
  24 from allmydata.mutable.common import NotWriteableError
  25 from allmydata.mutable import layout as mutable_layout
  26 from allmydata.mutable.publish import MutableData
  27 from foolscap.api import DeadReferenceError
  28 from twisted.python.failure import Failure
  29 from twisted.web.client import getPage
  30 from twisted.web.error import Error
  31
  32 from allmydata.test.common import SystemTestMixin
  33
  34 # TODO: move this to common or common_util
  35 from allmydata.test.test_runner import RunBinTahoeMixin
  36
  37 LARGE_DATA = """
  38 This is some data to publish to the remote grid.., which needs to be large
  39 enough to not fit inside a LIT uri.
  40 """
  41
  42 class CountingDataUploadable(upload.Data):
  43     bytes_read = 0
  44     interrupt_after = None
  45     interrupt_after_d = None
  46
  47     def read(self, length):
  48         self.bytes_read += length
  49         if self.interrupt_after is not None:
  50             if self.bytes_read > self.interrupt_after:
  51                 self.interrupt_after = None
  52                 self.interrupt_after_d.callback(self)
  53         return upload.Data.read(self, length)
  54
  55 class SystemTest(SystemTestMixin, RunBinTahoeMixin, unittest.TestCase):
  56     timeout = 3600 # It takes longer than 960 seconds on Zandr's ARM box.
  57
  58     def test_connections(self):
  59         self.basedir = "system/SystemTest/test_connections"
  60         d = self.set_up_nodes()
  61         self.extra_node = None
  62         d.addCallback(lambda res: self.add_extra_node(self.numclients))
  63         def _check(extra_node):
  64             self.extra_node = extra_node
  65             for c in self.clients:
  66                 all_peerids = c.get_storage_broker().get_all_serverids()
  67                 self.failUnlessEqual(len(all_peerids), self.numclients+1)
  68                 sb = c.storage_broker
  69                 permuted_peers = sb.get_servers_for_psi("a")
  70                 self.failUnlessEqual(len(permuted_peers), self.numclients+1)
  71
  72         d.addCallback(_check)
  73         def _shutdown_extra_node(res):
  74             if self.extra_node:
  75                 return self.extra_node.stopService()
  76             return res
  77         d.addBoth(_shutdown_extra_node)
  78         return d
  79     # test_connections is subsumed by test_upload_and_download, and takes
  80     # quite a while to run on a slow machine (because of all the TLS
  81     # connections that must be established). If we ever rework the introducer
  82     # code to such an extent that we're not sure if it works anymore, we can
  83     # reinstate this test until it does.
  84     del test_connections
  85
  86     def test_upload_and_download_random_key(self):
  87         self.basedir = "system/SystemTest/test_upload_and_download_random_key"
  88         return self._test_upload_and_download(convergence=None)
  89
  90     def test_upload_and_download_convergent(self):
  91         self.basedir = "system/SystemTest/test_upload_and_download_convergent"
  92         return self._test_upload_and_download(convergence="some convergence string")
  93
  94     def _test_upload_and_download(self, convergence):
  95         # we use 4000 bytes of data, which will result in about 400k written
  96         # to disk among all our simulated nodes
  97         DATA = "Some data to upload\n" * 200
  98         d = self.set_up_nodes()
  99         def _check_connections(res):
 100             for c in self.clients:
 101                 c.DEFAULT_ENCODING_PARAMETERS['happy'] = 5
 102                 all_peerids = c.get_storage_broker().get_all_serverids()
 103                 self.failUnlessEqual(len(all_peerids), self.numclients)
 104                 sb = c.storage_broker
 105                 permuted_peers = sb.get_servers_for_psi("a")
 106                 self.failUnlessEqual(len(permuted_peers), self.numclients)
 107         d.addCallback(_check_connections)
 108
 109         def _do_upload(res):
 110             log.msg("UPLOADING")
 111             u = self.clients[0].getServiceNamed("uploader")
 112             self.uploader = u
 113             # we crank the max segsize down to 1024b for the duration of this
 114             # test, so we can exercise multiple segments. It is important
 115             # that this is not a multiple of the segment size, so that the
 116             # tail segment is not the same length as the others. This actualy
 117             # gets rounded up to 1025 to be a multiple of the number of
 118             # required shares (since we use 25 out of 100 FEC).
 119             up = upload.Data(DATA, convergence=convergence)
 120             up.max_segment_size = 1024
 121             d1 = u.upload(up)
 122             return d1
 123         d.addCallback(_do_upload)
 124         def _upload_done(results):
 125             theuri = results.uri
 126             log.msg("upload finished: uri is %s" % (theuri,))
 127             self.uri = theuri
 128             assert isinstance(self.uri, str), self.uri
 129             self.cap = uri.from_string(self.uri)
 130             self.n = self.clients[1].create_node_from_uri(self.uri)
 131         d.addCallback(_upload_done)
 132
 133         def _upload_again(res):
 134             # Upload again. If using convergent encryption then this ought to be
 135             # short-circuited, however with the way we currently generate URIs
 136             # (i.e. because they include the roothash), we have to do all of the
 137             # encoding work, and only get to save on the upload part.
 138             log.msg("UPLOADING AGAIN")
 139             up = upload.Data(DATA, convergence=convergence)
 140             up.max_segment_size = 1024
 141             return self.uploader.upload(up)
 142         d.addCallback(_upload_again)
 143
 144         def _download_to_data(res):
 145             log.msg("DOWNLOADING")
 146             return download_to_data(self.n)
 147         d.addCallback(_download_to_data)
 148         def _download_to_data_done(data):
 149             log.msg("download finished")
 150             self.failUnlessEqual(data, DATA)
 151         d.addCallback(_download_to_data_done)
 152
 153         def _test_read(res):
 154             n = self.clients[1].create_node_from_uri(self.uri)
 155             d = download_to_data(n)
 156             def _read_done(data):
 157                 self.failUnlessEqual(data, DATA)
 158             d.addCallback(_read_done)
 159             d.addCallback(lambda ign:
 160                           n.read(MemoryConsumer(), offset=1, size=4))
 161             def _read_portion_done(mc):
 162                 self.failUnlessEqual("".join(mc.chunks), DATA[1:1+4])
 163             d.addCallback(_read_portion_done)
 164             d.addCallback(lambda ign:
 165                           n.read(MemoryConsumer(), offset=2, size=None))
 166             def _read_tail_done(mc):
 167                 self.failUnlessEqual("".join(mc.chunks), DATA[2:])
 168             d.addCallback(_read_tail_done)
 169             d.addCallback(lambda ign:
 170                           n.read(MemoryConsumer(), size=len(DATA)+1000))
 171             def _read_too_much(mc):
 172                 self.failUnlessEqual("".join(mc.chunks), DATA)
 173             d.addCallback(_read_too_much)
 174
 175             return d
 176         d.addCallback(_test_read)
 177
 178         def _test_bad_read(res):
 179             bad_u = uri.from_string_filenode(self.uri)
 180             bad_u.key = self.flip_bit(bad_u.key)
 181             bad_n = self.clients[1].create_node_from_uri(bad_u.to_string())
 182             # this should cause an error during download
 183
 184             d = self.shouldFail2(NoSharesError, "'download bad node'",
 185                                  None,
 186                                  bad_n.read, MemoryConsumer(), offset=2)
 187             return d
 188         d.addCallback(_test_bad_read)
 189
 190         def _download_nonexistent_uri(res):
 191             baduri = self.mangle_uri(self.uri)
 192             badnode = self.clients[1].create_node_from_uri(baduri)
 193             log.msg("about to download non-existent URI", level=log.UNUSUAL,
 194                     facility="tahoe.tests")
 195             d1 = download_to_data(badnode)
 196             def _baduri_should_fail(res):
 197                 log.msg("finished downloading non-existend URI",
 198                         level=log.UNUSUAL, facility="tahoe.tests")
 199                 self.failUnless(isinstance(res, Failure))
 200                 self.failUnless(res.check(NoSharesError),
 201                                 "expected NoSharesError, got %s" % res)
 202             d1.addBoth(_baduri_should_fail)
 203             return d1
 204         d.addCallback(_download_nonexistent_uri)
 205
 206         # add a new node, which doesn't accept shares, and only uses the
 207         # helper for upload.
 208         d.addCallback(lambda res: self.add_extra_node(self.numclients,
 209                                                       self.helper_furl,
 210                                                       add_to_sparent=True))
 211         def _added(extra_node):
 212             self.extra_node = extra_node
 213             self.extra_node.DEFAULT_ENCODING_PARAMETERS['happy'] = 5
 214         d.addCallback(_added)
 215
 216         HELPER_DATA = "Data that needs help to upload" * 1000
 217         def _upload_with_helper(res):
 218             u = upload.Data(HELPER_DATA, convergence=convergence)
 219             d = self.extra_node.upload(u)
 220             def _uploaded(results):
 221                 n = self.clients[1].create_node_from_uri(results.uri)
 222                 return download_to_data(n)
 223             d.addCallback(_uploaded)
 224             def _check(newdata):
 225                 self.failUnlessEqual(newdata, HELPER_DATA)
 226             d.addCallback(_check)
 227             return d
 228         d.addCallback(_upload_with_helper)
 229
 230         def _upload_duplicate_with_helper(res):
 231             u = upload.Data(HELPER_DATA, convergence=convergence)
 232             u.debug_stash_RemoteEncryptedUploadable = True
 233             d = self.extra_node.upload(u)
 234             def _uploaded(results):
 235                 n = self.clients[1].create_node_from_uri(results.uri)
 236                 return download_to_data(n)
 237             d.addCallback(_uploaded)
 238             def _check(newdata):
 239                 self.failUnlessEqual(newdata, HELPER_DATA)
 240                 self.failIf(hasattr(u, "debug_RemoteEncryptedUploadable"),
 241                             "uploadable started uploading, should have been avoided")
 242             d.addCallback(_check)
 243             return d
 244         if convergence is not None:
 245             d.addCallback(_upload_duplicate_with_helper)
 246
 247         def _upload_resumable(res):
 248             DATA = "Data that needs help to upload and gets interrupted" * 1000
 249             u1 = CountingDataUploadable(DATA, convergence=convergence)
 250             u2 = CountingDataUploadable(DATA, convergence=convergence)
 251
 252             # we interrupt the connection after about 5kB by shutting down
 253             # the helper, then restartingit.
 254             u1.interrupt_after = 5000
 255             u1.interrupt_after_d = defer.Deferred()
 256             u1.interrupt_after_d.addCallback(lambda res:
 257                                              self.bounce_client(0))
 258
 259             # sneak into the helper and reduce its chunk size, so that our
 260             # debug_interrupt will sever the connection on about the fifth
 261             # chunk fetched. This makes sure that we've started to write the
 262             # new shares before we abandon them, which exercises the
 263             # abort/delete-partial-share code. TODO: find a cleaner way to do
 264             # this. I know that this will affect later uses of the helper in
 265             # this same test run, but I'm not currently worried about it.
 266             offloaded.CHKCiphertextFetcher.CHUNK_SIZE = 1000
 267
 268             d = self.extra_node.upload(u1)
 269
 270             def _should_not_finish(res):
 271                 self.fail("interrupted upload should have failed, not finished"
 272                           " with result %s" % (res,))
 273             def _interrupted(f):
 274                 f.trap(DeadReferenceError)
 275
 276                 # make sure we actually interrupted it before finishing the
 277                 # file
 278                 self.failUnless(u1.bytes_read < len(DATA),
 279                                 "read %d out of %d total" % (u1.bytes_read,
 280                                                              len(DATA)))
 281
 282                 log.msg("waiting for reconnect", level=log.NOISY,
 283                         facility="tahoe.test.test_system")
 284                 # now, we need to give the nodes a chance to notice that this
 285                 # connection has gone away. When this happens, the storage
 286                 # servers will be told to abort their uploads, removing the
 287                 # partial shares. Unfortunately this involves TCP messages
 288                 # going through the loopback interface, and we can't easily
 289                 # predict how long that will take. If it were all local, we
 290                 # could use fireEventually() to stall. Since we don't have
 291                 # the right introduction hooks, the best we can do is use a
 292                 # fixed delay. TODO: this is fragile.
 293                 u1.interrupt_after_d.addCallback(self.stall, 2.0)
 294                 return u1.interrupt_after_d
 295             d.addCallbacks(_should_not_finish, _interrupted)
 296
 297             def _disconnected(res):
 298                 # check to make sure the storage servers aren't still hanging
 299                 # on to the partial share: their incoming/ directories should
 300                 # now be empty.
 301                 log.msg("disconnected", level=log.NOISY,
 302                         facility="tahoe.test.test_system")
 303                 for i in range(self.numclients):
 304                     incdir = os.path.join(self.getdir("client%d" % i),
 305                                           "storage", "shares", "incoming")
 306                     self.failIf(os.path.exists(incdir) and os.listdir(incdir))
 307             d.addCallback(_disconnected)
 308
 309             # then we need to give the reconnector a chance to
 310             # reestablish the connection to the helper.
 311             d.addCallback(lambda res:
 312                           log.msg("wait_for_connections", level=log.NOISY,
 313                                   facility="tahoe.test.test_system"))
 314             d.addCallback(lambda res: self.wait_for_connections())
 315
 316
 317             d.addCallback(lambda res:
 318                           log.msg("uploading again", level=log.NOISY,
 319                                   facility="tahoe.test.test_system"))
 320             d.addCallback(lambda res: self.extra_node.upload(u2))
 321
 322             def _uploaded(results):
 323                 cap = results.uri
 324                 log.msg("Second upload complete", level=log.NOISY,
 325                         facility="tahoe.test.test_system")
 326
 327                 # this is really bytes received rather than sent, but it's
 328                 # convenient and basically measures the same thing
 329                 bytes_sent = results.ciphertext_fetched
 330                 self.failUnless(isinstance(bytes_sent, (int, long)), bytes_sent)
 331
 332                 # We currently don't support resumption of upload if the data is
 333                 # encrypted with a random key.  (Because that would require us
 334                 # to store the key locally and re-use it on the next upload of
 335                 # this file, which isn't a bad thing to do, but we currently
 336                 # don't do it.)
 337                 if convergence is not None:
 338                     # Make sure we did not have to read the whole file the
 339                     # second time around .
 340                     self.failUnless(bytes_sent < len(DATA),
 341                                     "resumption didn't save us any work:"
 342                                     " read %r bytes out of %r total" %
 343                                     (bytes_sent, len(DATA)))
 344                 else:
 345                     # Make sure we did have to read the whole file the second
 346                     # time around -- because the one that we partially uploaded
 347                     # earlier was encrypted with a different random key.
 348                     self.failIf(bytes_sent < len(DATA),
 349                                 "resumption saved us some work even though we were using random keys:"
 350                                 " read %r bytes out of %r total" %
 351                                 (bytes_sent, len(DATA)))
 352                 n = self.clients[1].create_node_from_uri(cap)
 353                 return download_to_data(n)
 354             d.addCallback(_uploaded)
 355
 356             def _check(newdata):
 357                 self.failUnlessEqual(newdata, DATA)
 358                 # If using convergent encryption, then also check that the
 359                 # helper has removed the temp file from its directories.
 360                 if convergence is not None:
 361                     basedir = os.path.join(self.getdir("client0"), "helper")
 362                     files = os.listdir(os.path.join(basedir, "CHK_encoding"))
 363                     self.failUnlessEqual(files, [])
 364                     files = os.listdir(os.path.join(basedir, "CHK_incoming"))
 365                     self.failUnlessEqual(files, [])
 366             d.addCallback(_check)
 367             return d
 368         d.addCallback(_upload_resumable)
 369
 370         def _grab_stats(ignored):
 371             # the StatsProvider doesn't normally publish a FURL:
 372             # instead it passes a live reference to the StatsGatherer
 373             # (if and when it connects). To exercise the remote stats
 374             # interface, we manually publish client0's StatsProvider
 375             # and use client1 to query it.
 376             sp = self.clients[0].stats_provider
 377             sp_furl = self.clients[0].tub.registerReference(sp)
 378             d = self.clients[1].tub.getReference(sp_furl)
 379             d.addCallback(lambda sp_rref: sp_rref.callRemote("get_stats"))
 380             def _got_stats(stats):
 381                 #print "STATS"
 382                 #from pprint import pprint
 383                 #pprint(stats)
 384                 s = stats["stats"]
 385                 self.failUnlessEqual(s["storage_server.accepting_immutable_shares"], 1)
 386                 c = stats["counters"]
 387                 self.failUnless("storage_server.allocate" in c)
 388             d.addCallback(_got_stats)
 389             return d
 390         d.addCallback(_grab_stats)
 391
 392         return d
 393
 394     def _find_all_shares(self, basedir):
 395         shares = []
 396         for (dirpath, dirnames, filenames) in os.walk(basedir):
 397             if "storage" not in dirpath:
 398                 continue
 399             if not filenames:
 400                 continue
 401             pieces = dirpath.split(os.sep)
 402             if (len(pieces) >= 5
 403                 and pieces[-4] == "storage"
 404                 and pieces[-3] == "shares"):
 405                 # we're sitting in .../storage/shares/$START/$SINDEX , and there
 406                 # are sharefiles here
 407                 assert pieces[-5].startswith("client")
 408                 client_num = int(pieces[-5][-1])
 409                 storage_index_s = pieces[-1]
 410                 storage_index = si_a2b(storage_index_s)
 411                 for sharename in filenames:
 412                     shnum = int(sharename)
 413                     filename = os.path.join(dirpath, sharename)
 414                     data = (client_num, storage_index, filename, shnum)
 415                     shares.append(data)
 416         if not shares:
 417             self.fail("unable to find any share files in %s" % basedir)
 418         return shares
 419
 420     def _corrupt_mutable_share(self, filename, which):
 421         msf = MutableShareFile(filename)
 422         datav = msf.readv([ (0, 1000000) ])
 423         final_share = datav[0]
 424         assert len(final_share) < 1000000 # ought to be truncated
 425         pieces = mutable_layout.unpack_share(final_share)
 426         (seqnum, root_hash, IV, k, N, segsize, datalen,
 427          verification_key, signature, share_hash_chain, block_hash_tree,
 428          share_data, enc_privkey) = pieces
 429
 430         if which == "seqnum":
 431             seqnum = seqnum + 15
 432         elif which == "R":
 433             root_hash = self.flip_bit(root_hash)
 434         elif which == "IV":
 435             IV = self.flip_bit(IV)
 436         elif which == "segsize":
 437             segsize = segsize + 15
 438         elif which == "pubkey":
 439             verification_key = self.flip_bit(verification_key)
 440         elif which == "signature":
 441             signature = self.flip_bit(signature)
 442         elif which == "share_hash_chain":
 443             nodenum = share_hash_chain.keys()[0]
 444             share_hash_chain[nodenum] = self.flip_bit(share_hash_chain[nodenum])
 445         elif which == "block_hash_tree":
 446             block_hash_tree[-1] = self.flip_bit(block_hash_tree[-1])
 447         elif which == "share_data":
 448             share_data = self.flip_bit(share_data)
 449         elif which == "encprivkey":
 450             enc_privkey = self.flip_bit(enc_privkey)
 451
 452         prefix = mutable_layout.pack_prefix(seqnum, root_hash, IV, k, N,
 453                                             segsize, datalen)
 454         final_share = mutable_layout.pack_share(prefix,
 455                                                 verification_key,
 456                                                 signature,
 457                                                 share_hash_chain,
 458                                                 block_hash_tree,
 459                                                 share_data,
 460                                                 enc_privkey)
 461         msf.writev( [(0, final_share)], None)
 462
 463
 464     def test_mutable(self):
 465         self.basedir = "system/SystemTest/test_mutable"
 466         DATA = "initial contents go here."  # 25 bytes % 3 != 0
 467         DATA_uploadable = MutableData(DATA)
 468         NEWDATA = "new contents yay"
 469         NEWDATA_uploadable = MutableData(NEWDATA)
 470         NEWERDATA = "this is getting old"
 471         NEWERDATA_uploadable = MutableData(NEWERDATA)
 472
 473         d = self.set_up_nodes(use_key_generator=True)
 474
 475         def _create_mutable(res):
 476             c = self.clients[0]
 477             log.msg("starting create_mutable_file")
 478             d1 = c.create_mutable_file(DATA_uploadable)
 479             def _done(res):
 480                 log.msg("DONE: %s" % (res,))
 481                 self._mutable_node_1 = res
 482             d1.addCallback(_done)
 483             return d1
 484         d.addCallback(_create_mutable)
 485
 486         def _test_debug(res):
 487             # find a share. It is important to run this while there is only
 488             # one slot in the grid.
 489             shares = self._find_all_shares(self.basedir)
 490             (client_num, storage_index, filename, shnum) = shares[0]
 491             log.msg("test_system.SystemTest.test_mutable._test_debug using %s"
 492                     % filename)
 493             log.msg(" for clients[%d]" % client_num)
 494
 495             out,err = StringIO(), StringIO()
 496             rc = runner.runner(["debug", "dump-share", "--offsets",
 497                                 filename],
 498                                stdout=out, stderr=err)
 499             output = out.getvalue()
 500             self.failUnlessEqual(rc, 0)
 501             try:
 502                 self.failUnless("Mutable slot found:\n" in output)
 503                 self.failUnless("share_type: SDMF\n" in output)
 504                 peerid = idlib.nodeid_b2a(self.clients[client_num].nodeid)
 505                 self.failUnless(" WE for nodeid: %s\n" % peerid in output)
 506                 self.failUnless(" num_extra_leases: 0\n" in output)
 507                 self.failUnless("  secrets are for nodeid: %s\n" % peerid
 508                                 in output)
 509                 self.failUnless(" SDMF contents:\n" in output)
 510                 self.failUnless("  seqnum: 1\n" in output)
 511                 self.failUnless("  required_shares: 3\n" in output)
 512                 self.failUnless("  total_shares: 10\n" in output)
 513                 self.failUnless("  segsize: 27\n" in output, (output, filename))
 514                 self.failUnless("  datalen: 25\n" in output)
 515                 # the exact share_hash_chain nodes depends upon the sharenum,
 516                 # and is more of a hassle to compute than I want to deal with
 517                 # now
 518                 self.failUnless("  share_hash_chain: " in output)
 519                 self.failUnless("  block_hash_tree: 1 nodes\n" in output)
 520                 expected = ("  verify-cap: URI:SSK-Verifier:%s:" %
 521                             base32.b2a(storage_index))
 522                 self.failUnless(expected in output)
 523             except unittest.FailTest:
 524                 print
 525                 print "dump-share output was:"
 526                 print output
 527                 raise
 528         d.addCallback(_test_debug)
 529
 530         # test retrieval
 531
 532         # first, let's see if we can use the existing node to retrieve the
 533         # contents. This allows it to use the cached pubkey and maybe the
 534         # latest-known sharemap.
 535
 536         d.addCallback(lambda res: self._mutable_node_1.download_best_version())
 537         def _check_download_1(res):
 538             self.failUnlessEqual(res, DATA)
 539             # now we see if we can retrieve the data from a new node,
 540             # constructed using the URI of the original one. We do this test
 541             # on the same client that uploaded the data.
 542             uri = self._mutable_node_1.get_uri()
 543             log.msg("starting retrieve1")
 544             newnode = self.clients[0].create_node_from_uri(uri)
 545             newnode_2 = self.clients[0].create_node_from_uri(uri)
 546             self.failUnlessIdentical(newnode, newnode_2)
 547             return newnode.download_best_version()
 548         d.addCallback(_check_download_1)
 549
 550         def _check_download_2(res):
 551             self.failUnlessEqual(res, DATA)
 552             # same thing, but with a different client
 553             uri = self._mutable_node_1.get_uri()
 554             newnode = self.clients[1].create_node_from_uri(uri)
 555             log.msg("starting retrieve2")
 556             d1 = newnode.download_best_version()
 557             d1.addCallback(lambda res: (res, newnode))
 558             return d1
 559         d.addCallback(_check_download_2)
 560
 561         def _check_download_3((res, newnode)):
 562             self.failUnlessEqual(res, DATA)
 563             # replace the data
 564             log.msg("starting replace1")
 565             d1 = newnode.overwrite(NEWDATA_uploadable)
 566             d1.addCallback(lambda res: newnode.download_best_version())
 567             return d1
 568         d.addCallback(_check_download_3)
 569
 570         def _check_download_4(res):
 571             self.failUnlessEqual(res, NEWDATA)
 572             # now create an even newer node and replace the data on it. This
 573             # new node has never been used for download before.
 574             uri = self._mutable_node_1.get_uri()
 575             newnode1 = self.clients[2].create_node_from_uri(uri)
 576             newnode2 = self.clients[3].create_node_from_uri(uri)
 577             self._newnode3 = self.clients[3].create_node_from_uri(uri)
 578             log.msg("starting replace2")
 579             d1 = newnode1.overwrite(NEWERDATA_uploadable)
 580             d1.addCallback(lambda res: newnode2.download_best_version())
 581             return d1
 582         d.addCallback(_check_download_4)
 583
 584         def _check_download_5(res):
 585             log.msg("finished replace2")
 586             self.failUnlessEqual(res, NEWERDATA)
 587         d.addCallback(_check_download_5)
 588
 589         def _corrupt_shares(res):
 590             # run around and flip bits in all but k of the shares, to test
 591             # the hash checks
 592             shares = self._find_all_shares(self.basedir)
 593             ## sort by share number
 594             #shares.sort( lambda a,b: cmp(a[3], b[3]) )
 595             where = dict([ (shnum, filename)
 596                            for (client_num, storage_index, filename, shnum)
 597                            in shares ])
 598             assert len(where) == 10 # this test is designed for 3-of-10
 599             for shnum, filename in where.items():
 600                 # shares 7,8,9 are left alone. read will check
 601                 # (share_hash_chain, block_hash_tree, share_data). New
 602                 # seqnum+R pairs will trigger a check of (seqnum, R, IV,
 603                 # segsize, signature).
 604                 if shnum == 0:
 605                     # read: this will trigger "pubkey doesn't match
 606                     # fingerprint".
 607                     self._corrupt_mutable_share(filename, "pubkey")
 608                     self._corrupt_mutable_share(filename, "encprivkey")
 609                 elif shnum == 1:
 610                     # triggers "signature is invalid"
 611                     self._corrupt_mutable_share(filename, "seqnum")
 612                 elif shnum == 2:
 613                     # triggers "signature is invalid"
 614                     self._corrupt_mutable_share(filename, "R")
 615                 elif shnum == 3:
 616                     # triggers "signature is invalid"
 617                     self._corrupt_mutable_share(filename, "segsize")
 618                 elif shnum == 4:
 619                     self._corrupt_mutable_share(filename, "share_hash_chain")
 620                 elif shnum == 5:
 621                     self._corrupt_mutable_share(filename, "block_hash_tree")
 622                 elif shnum == 6:
 623                     self._corrupt_mutable_share(filename, "share_data")
 624                 # other things to correct: IV, signature
 625                 # 7,8,9 are left alone
 626
 627                 # note that initial_query_count=5 means that we'll hit the
 628                 # first 5 servers in effectively random order (based upon
 629                 # response time), so we won't necessarily ever get a "pubkey
 630                 # doesn't match fingerprint" error (if we hit shnum>=1 before
 631                 # shnum=0, we pull the pubkey from there). To get repeatable
 632                 # specific failures, we need to set initial_query_count=1,
 633                 # but of course that will change the sequencing behavior of
 634                 # the retrieval process. TODO: find a reasonable way to make
 635                 # this a parameter, probably when we expand this test to test
 636                 # for one failure mode at a time.
 637
 638                 # when we retrieve this, we should get three signature
 639                 # failures (where we've mangled seqnum, R, and segsize). The
 640                 # pubkey mangling
 641         d.addCallback(_corrupt_shares)
 642
 643         d.addCallback(lambda res: self._newnode3.download_best_version())
 644         d.addCallback(_check_download_5)
 645
 646         def _check_empty_file(res):
 647             # make sure we can create empty files, this usually screws up the
 648             # segsize math
 649             d1 = self.clients[2].create_mutable_file(MutableData(""))
 650             d1.addCallback(lambda newnode: newnode.download_best_version())
 651             d1.addCallback(lambda res: self.failUnlessEqual("", res))
 652             return d1
 653         d.addCallback(_check_empty_file)
 654
 655         d.addCallback(lambda res: self.clients[0].create_dirnode())
 656         def _created_dirnode(dnode):
 657             log.msg("_created_dirnode(%s)" % (dnode,))
 658             d1 = dnode.list()
 659             d1.addCallback(lambda children: self.failUnlessEqual(children, {}))
 660             d1.addCallback(lambda res: dnode.has_child(u"edgar"))
 661             d1.addCallback(lambda answer: self.failUnlessEqual(answer, False))
 662             d1.addCallback(lambda res: dnode.set_node(u"see recursive", dnode))
 663             d1.addCallback(lambda res: dnode.has_child(u"see recursive"))
 664             d1.addCallback(lambda answer: self.failUnlessEqual(answer, True))
 665             d1.addCallback(lambda res: dnode.build_manifest().when_done())
 666             d1.addCallback(lambda res:
 667                            self.failUnlessEqual(len(res["manifest"]), 1))
 668             return d1
 669         d.addCallback(_created_dirnode)
 670
 671         def wait_for_c3_kg_conn():
 672             return self.clients[3]._key_generator is not None
 673         d.addCallback(lambda junk: self.poll(wait_for_c3_kg_conn))
 674
 675         def check_kg_poolsize(junk, size_delta):
 676             self.failUnlessEqual(len(self.key_generator_svc.key_generator.keypool),
 677                                  self.key_generator_svc.key_generator.pool_size + size_delta)
 678
 679         d.addCallback(check_kg_poolsize, 0)
 680         d.addCallback(lambda junk:
 681             self.clients[3].create_mutable_file(MutableData('hello, world')))
 682         d.addCallback(check_kg_poolsize, -1)
 683         d.addCallback(lambda junk: self.clients[3].create_dirnode())
 684         d.addCallback(check_kg_poolsize, -2)
 685         # use_helper induces use of clients[3], which is the using-key_gen client
 686         d.addCallback(lambda junk:
 687                       self.POST("uri?t=mkdir&name=george", use_helper=True))
 688         d.addCallback(check_kg_poolsize, -3)
 689
 690         return d
 691
 692     def flip_bit(self, good):
 693         return good[:-1] + chr(ord(good[-1]) ^ 0x01)
 694
 695     def mangle_uri(self, gooduri):
 696         # change the key, which changes the storage index, which means we'll
 697         # be asking about the wrong file, so nobody will have any shares
 698         u = uri.from_string(gooduri)
 699         u2 = uri.CHKFileURI(key=self.flip_bit(u.key),
 700                             uri_extension_hash=u.uri_extension_hash,
 701                             needed_shares=u.needed_shares,
 702                             total_shares=u.total_shares,
 703                             size=u.size)
 704         return u2.to_string()
 705
 706     # TODO: add a test which mangles the uri_extension_hash instead, and
 707     # should fail due to not being able to get a valid uri_extension block.
 708     # Also a test which sneakily mangles the uri_extension block to change
 709     # some of the validation data, so it will fail in the post-download phase
 710     # when the file's crypttext integrity check fails. Do the same thing for
 711     # the key, which should cause the download to fail the post-download
 712     # plaintext_hash check.
 713
 714     def test_filesystem(self):
 715         self.basedir = "system/SystemTest/test_filesystem"
 716         self.data = LARGE_DATA
 717         d = self.set_up_nodes(use_stats_gatherer=True)
 718         def _new_happy_semantics(ign):
 719             for c in self.clients:
 720                 c.DEFAULT_ENCODING_PARAMETERS['happy'] = 1
 721         d.addCallback(_new_happy_semantics)
 722         d.addCallback(self._test_introweb)
 723         d.addCallback(self.log, "starting publish")
 724         d.addCallback(self._do_publish1)
 725         d.addCallback(self._test_runner)
 726         d.addCallback(self._do_publish2)
 727         # at this point, we have the following filesystem (where "R" denotes
 728         # self._root_directory_uri):
 729         # R
 730         # R/subdir1
 731         # R/subdir1/mydata567
 732         # R/subdir1/subdir2/
 733         # R/subdir1/subdir2/mydata992
 734
 735         d.addCallback(lambda res: self.bounce_client(0))
 736         d.addCallback(self.log, "bounced client0")
 737
 738         d.addCallback(self._check_publish1)
 739         d.addCallback(self.log, "did _check_publish1")
 740         d.addCallback(self._check_publish2)
 741         d.addCallback(self.log, "did _check_publish2")
 742         d.addCallback(self._do_publish_private)
 743         d.addCallback(self.log, "did _do_publish_private")
 744         # now we also have (where "P" denotes a new dir):
 745         #  P/personal/sekrit data
 746         #  P/s2-rw -> /subdir1/subdir2/
 747         #  P/s2-ro -> /subdir1/subdir2/ (read-only)
 748         d.addCallback(self._check_publish_private)
 749         d.addCallback(self.log, "did _check_publish_private")
 750         d.addCallback(self._test_web)
 751         d.addCallback(self._test_control)
 752         d.addCallback(self._test_cli)
 753         # P now has four top-level children:
 754         # P/personal/sekrit data
 755         # P/s2-ro/
 756         # P/s2-rw/
 757         # P/test_put/  (empty)
 758         d.addCallback(self._test_checker)
 759         return d
 760
 761     def _test_introweb(self, res):
 762         d = getPage(self.introweb_url, method="GET", followRedirect=True)
 763         def _check(res):
 764             try:
 765                 self.failUnless("%s: %s" % (allmydata.__appname__, allmydata.__version__) in res)
 766                 verstr = str(allmydata.__version__)
 767
 768                 # The Python "rational version numbering" convention
 769                 # disallows "-r$REV" but allows ".post$REV"
 770                 # instead. Eventually we'll probably move to
 771                 # that. When we do, this test won't go red:
 772                 ix = verstr.rfind('-r')
 773                 if ix != -1:
 774                     altverstr = verstr[:ix] + '.post' + verstr[ix+2:]
 775                 else:
 776                     ix = verstr.rfind('.post')
 777                     if ix != -1:
 778                         altverstr = verstr[:ix] + '-r' + verstr[ix+5:]
 779                     else:
 780                         altverstr = verstr
 781
 782                 appverstr = "%s: %s" % (allmydata.__appname__, verstr)
 783                 newappverstr = "%s: %s" % (allmydata.__appname__, altverstr)
 784
 785                 self.failUnless((appverstr in res) or (newappverstr in res), (appverstr, newappverstr, res))
 786                 self.failUnless("Announcement Summary: storage: 5, stub_client: 5" in res)
 787                 self.failUnless("Subscription Summary: storage: 5" in res)
 788             except unittest.FailTest:
 789                 print
 790                 print "GET %s output was:" % self.introweb_url
 791                 print res
 792                 raise
 793         d.addCallback(_check)
 794         d.addCallback(lambda res:
 795                       getPage(self.introweb_url + "?t=json",
 796                               method="GET", followRedirect=True))
 797         def _check_json(res):
 798             data = simplejson.loads(res)
 799             try:
 800                 self.failUnlessEqual(data["subscription_summary"],
 801                                      {"storage": 5})
 802                 self.failUnlessEqual(data["announcement_summary"],
 803                                      {"storage": 5, "stub_client": 5})
 804                 self.failUnlessEqual(data["announcement_distinct_hosts"],
 805                                      {"storage": 1, "stub_client": 1})
 806             except unittest.FailTest:
 807                 print
 808                 print "GET %s?t=json output was:" % self.introweb_url
 809                 print res
 810                 raise
 811         d.addCallback(_check_json)
 812         return d
 813
 814     def _do_publish1(self, res):
 815         ut = upload.Data(self.data, convergence=None)
 816         c0 = self.clients[0]
 817         d = c0.create_dirnode()
 818         def _made_root(new_dirnode):
 819             self._root_directory_uri = new_dirnode.get_uri()
 820             return c0.create_node_from_uri(self._root_directory_uri)
 821         d.addCallback(_made_root)
 822         d.addCallback(lambda root: root.create_subdirectory(u"subdir1"))
 823         def _made_subdir1(subdir1_node):
 824             self._subdir1_node = subdir1_node
 825             d1 = subdir1_node.add_file(u"mydata567", ut)
 826             d1.addCallback(self.log, "publish finished")
 827             def _stash_uri(filenode):
 828                 self.uri = filenode.get_uri()
 829                 assert isinstance(self.uri, str), (self.uri, filenode)
 830             d1.addCallback(_stash_uri)
 831             return d1
 832         d.addCallback(_made_subdir1)
 833         return d
 834
 835     def _do_publish2(self, res):
 836         ut = upload.Data(self.data, convergence=None)
 837         d = self._subdir1_node.create_subdirectory(u"subdir2")
 838         d.addCallback(lambda subdir2: subdir2.add_file(u"mydata992", ut))
 839         return d
 840
 841     def log(self, res, *args, **kwargs):
 842         # print "MSG: %s  RES: %s" % (msg, args)
 843         log.msg(*args, **kwargs)
 844         return res
 845
 846     def _do_publish_private(self, res):
 847         self.smalldata = "sssh, very secret stuff"
 848         ut = upload.Data(self.smalldata, convergence=None)
 849         d = self.clients[0].create_dirnode()
 850         d.addCallback(self.log, "GOT private directory")
 851         def _got_new_dir(privnode):
 852             rootnode = self.clients[0].create_node_from_uri(self._root_directory_uri)
 853             d1 = privnode.create_subdirectory(u"personal")
 854             d1.addCallback(self.log, "made P/personal")
 855             d1.addCallback(lambda node: node.add_file(u"sekrit data", ut))
 856             d1.addCallback(self.log, "made P/personal/sekrit data")
 857             d1.addCallback(lambda res: rootnode.get_child_at_path([u"subdir1", u"subdir2"]))
 858             def _got_s2(s2node):
 859                 d2 = privnode.set_uri(u"s2-rw", s2node.get_uri(),
 860                                       s2node.get_readonly_uri())
 861                 d2.addCallback(lambda node:
 862                                privnode.set_uri(u"s2-ro",
 863                                                 s2node.get_readonly_uri(),
 864                                                 s2node.get_readonly_uri()))
 865                 return d2
 866             d1.addCallback(_got_s2)
 867             d1.addCallback(lambda res: privnode)
 868             return d1
 869         d.addCallback(_got_new_dir)
 870         return d
 871
 872     def _check_publish1(self, res):
 873         # this one uses the iterative API
 874         c1 = self.clients[1]
 875         d = defer.succeed(c1.create_node_from_uri(self._root_directory_uri))
 876         d.addCallback(self.log, "check_publish1 got /")
 877         d.addCallback(lambda root: root.get(u"subdir1"))
 878         d.addCallback(lambda subdir1: subdir1.get(u"mydata567"))
 879         d.addCallback(lambda filenode: download_to_data(filenode))
 880         d.addCallback(self.log, "get finished")
 881         def _get_done(data):
 882             self.failUnlessEqual(data, self.data)
 883         d.addCallback(_get_done)
 884         return d
 885
 886     def _check_publish2(self, res):
 887         # this one uses the path-based API
 888         rootnode = self.clients[1].create_node_from_uri(self._root_directory_uri)
 889         d = rootnode.get_child_at_path(u"subdir1")
 890         d.addCallback(lambda dirnode:
 891                       self.failUnless(IDirectoryNode.providedBy(dirnode)))
 892         d.addCallback(lambda res: rootnode.get_child_at_path(u"subdir1/mydata567"))
 893         d.addCallback(lambda filenode: download_to_data(filenode))
 894         d.addCallback(lambda data: self.failUnlessEqual(data, self.data))
 895
 896         d.addCallback(lambda res: rootnode.get_child_at_path(u"subdir1/mydata567"))
 897         def _got_filenode(filenode):
 898             fnode = self.clients[1].create_node_from_uri(filenode.get_uri())
 899             assert fnode == filenode
 900         d.addCallback(_got_filenode)
 901         return d
 902
 903     def _check_publish_private(self, resnode):
 904         # this one uses the path-based API
 905         self._private_node = resnode
 906
 907         d = self._private_node.get_child_at_path(u"personal")
 908         def _got_personal(personal):
 909             self._personal_node = personal
 910             return personal
 911         d.addCallback(_got_personal)
 912
 913         d.addCallback(lambda dirnode:
 914                       self.failUnless(IDirectoryNode.providedBy(dirnode), dirnode))
 915         def get_path(path):
 916             return self._private_node.get_child_at_path(path)
 917
 918         d.addCallback(lambda res: get_path(u"personal/sekrit data"))
 919         d.addCallback(lambda filenode: download_to_data(filenode))
 920         d.addCallback(lambda data: self.failUnlessEqual(data, self.smalldata))
 921         d.addCallback(lambda res: get_path(u"s2-rw"))
 922         d.addCallback(lambda dirnode: self.failUnless(dirnode.is_mutable()))
 923         d.addCallback(lambda res: get_path(u"s2-ro"))
 924         def _got_s2ro(dirnode):
 925             self.failUnless(dirnode.is_mutable(), dirnode)
 926             self.failUnless(dirnode.is_readonly(), dirnode)
 927             d1 = defer.succeed(None)
 928             d1.addCallback(lambda res: dirnode.list())
 929             d1.addCallback(self.log, "dirnode.list")
 930
 931             d1.addCallback(lambda res: self.shouldFail2(NotWriteableError, "mkdir(nope)", None, dirnode.create_subdirectory, u"nope"))
 932
 933             d1.addCallback(self.log, "doing add_file(ro)")
 934             ut = upload.Data("I will disappear, unrecorded and unobserved. The tragedy of my demise is made more poignant by its silence, but this beauty is not for you to ever know.", convergence="99i-p1x4-xd4-18yc-ywt-87uu-msu-zo -- completely and totally unguessable string (unless you read this)")
 935             d1.addCallback(lambda res: self.shouldFail2(NotWriteableError, "add_file(nope)", None, dirnode.add_file, u"hope", ut))
 936
 937             d1.addCallback(self.log, "doing get(ro)")
 938             d1.addCallback(lambda res: dirnode.get(u"mydata992"))
 939             d1.addCallback(lambda filenode:
 940                            self.failUnless(IFileNode.providedBy(filenode)))
 941
 942             d1.addCallback(self.log, "doing delete(ro)")
 943             d1.addCallback(lambda res: self.shouldFail2(NotWriteableError, "delete(nope)", None, dirnode.delete, u"mydata992"))
 944
 945             d1.addCallback(lambda res: self.shouldFail2(NotWriteableError, "set_uri(nope)", None, dirnode.set_uri, u"hopeless", self.uri, self.uri))
 946
 947             d1.addCallback(lambda res: self.shouldFail2(NoSuchChildError, "get(missing)", "missing", dirnode.get, u"missing"))
 948
 949             personal = self._personal_node
 950             d1.addCallback(lambda res: self.shouldFail2(NotWriteableError, "mv from readonly", None, dirnode.move_child_to, u"mydata992", personal, u"nope"))
 951
 952             d1.addCallback(self.log, "doing move_child_to(ro)2")
 953             d1.addCallback(lambda res: self.shouldFail2(NotWriteableError, "mv to readonly", None, personal.move_child_to, u"sekrit data", dirnode, u"nope"))
 954
 955             d1.addCallback(self.log, "finished with _got_s2ro")
 956             return d1
 957         d.addCallback(_got_s2ro)
 958         def _got_home(dummy):
 959             home = self._private_node
 960             personal = self._personal_node
 961             d1 = defer.succeed(None)
 962             d1.addCallback(self.log, "mv 'P/personal/sekrit data' to P/sekrit")
 963             d1.addCallback(lambda res:
 964                            personal.move_child_to(u"sekrit data",home,u"sekrit"))
 965
 966             d1.addCallback(self.log, "mv P/sekrit 'P/sekrit data'")
 967             d1.addCallback(lambda res:
 968                            home.move_child_to(u"sekrit", home, u"sekrit data"))
 969
 970             d1.addCallback(self.log, "mv 'P/sekret data' P/personal/")
 971             d1.addCallback(lambda res:
 972                            home.move_child_to(u"sekrit data", personal))
 973
 974             d1.addCallback(lambda res: home.build_manifest().when_done())
 975             d1.addCallback(self.log, "manifest")
 976             #  five items:
 977             # P/
 978             # P/personal/
 979             # P/personal/sekrit data
 980             # P/s2-rw  (same as P/s2-ro)
 981             # P/s2-rw/mydata992 (same as P/s2-rw/mydata992)
 982             d1.addCallback(lambda res:
 983                            self.failUnlessEqual(len(res["manifest"]), 5))
 984             d1.addCallback(lambda res: home.start_deep_stats().when_done())
 985             def _check_stats(stats):
 986                 expected = {"count-immutable-files": 1,
 987                             "count-mutable-files": 0,
 988                             "count-literal-files": 1,
 989                             "count-files": 2,
 990                             "count-directories": 3,
 991                             "size-immutable-files": 112,
 992                             "size-literal-files": 23,
 993                             #"size-directories": 616, # varies
 994                             #"largest-directory": 616,
 995                             "largest-directory-children": 3,
 996                             "largest-immutable-file": 112,
 997                             }
 998                 for k,v in expected.iteritems():
 999                     self.failUnlessEqual(stats[k], v,
1000                                          "stats[%s] was %s, not %s" %
1001                                          (k, stats[k], v))
1002                 self.failUnless(stats["size-directories"] > 1300,
1003                                 stats["size-directories"])
1004                 self.failUnless(stats["largest-directory"] > 800,
1005                                 stats["largest-directory"])
1006                 self.failUnlessEqual(stats["size-files-histogram"],
1007                                      [ (11, 31, 1), (101, 316, 1) ])
1008             d1.addCallback(_check_stats)
1009             return d1
1010         d.addCallback(_got_home)
1011         return d
1012
1013     def shouldFail(self, res, expected_failure, which, substring=None):
1014         if isinstance(res, Failure):
1015             res.trap(expected_failure)
1016             if substring:
1017                 self.failUnless(substring in str(res),
1018                                 "substring '%s' not in '%s'"
1019                                 % (substring, str(res)))
1020         else:
1021             self.fail("%s was supposed to raise %s, not get '%s'" %
1022                       (which, expected_failure, res))
1023
1024     def shouldFail2(self, expected_failure, which, substring, callable, *args, **kwargs):
1025         assert substring is None or isinstance(substring, str)
1026         d = defer.maybeDeferred(callable, *args, **kwargs)
1027         def done(res):
1028             if isinstance(res, Failure):
1029                 res.trap(expected_failure)
1030                 if substring:
1031                     self.failUnless(substring in str(res),
1032                                     "substring '%s' not in '%s'"
1033                                     % (substring, str(res)))
1034             else:
1035                 self.fail("%s was supposed to raise %s, not get '%s'" %
1036                           (which, expected_failure, res))
1037         d.addBoth(done)
1038         return d
1039
1040     def PUT(self, urlpath, data):
1041         url = self.webish_url + urlpath
1042         return getPage(url, method="PUT", postdata=data)
1043
1044     def GET(self, urlpath, followRedirect=False):
1045         url = self.webish_url + urlpath
1046         return getPage(url, method="GET", followRedirect=followRedirect)
1047
1048     def POST(self, urlpath, followRedirect=False, use_helper=False, **fields):
1049         sepbase = "boogabooga"
1050         sep = "--" + sepbase
1051         form = []
1052         form.append(sep)
1053         form.append('Content-Disposition: form-data; name="_charset"')
1054         form.append('')
1055         form.append('UTF-8')
1056         form.append(sep)
1057         for name, value in fields.iteritems():
1058             if isinstance(value, tuple):
1059                 filename, value = value
1060                 form.append('Content-Disposition: form-data; name="%s"; '
1061                             'filename="%s"' % (name, filename.encode("utf-8")))
1062             else:
1063                 form.append('Content-Disposition: form-data; name="%s"' % name)
1064             form.append('')
1065             form.append(str(value))
1066             form.append(sep)
1067         form[-1] += "--"
1068         body = ""
1069         headers = {}
1070         if fields:
1071             body = "\r\n".join(form) + "\r\n"
1072             headers["content-type"] = "multipart/form-data; boundary=%s" % sepbase
1073         return self.POST2(urlpath, body, headers, followRedirect, use_helper)
1074
1075     def POST2(self, urlpath, body="", headers={}, followRedirect=False,
1076               use_helper=False):
1077         if use_helper:
1078             url = self.helper_webish_url + urlpath
1079         else:
1080             url = self.webish_url + urlpath
1081         return getPage(url, method="POST", postdata=body, headers=headers,
1082                        followRedirect=followRedirect)
1083
1084     def _test_web(self, res):
1085         base = self.webish_url
1086         public = "uri/" + self._root_directory_uri
1087         d = getPage(base)
1088         def _got_welcome(page):
1089             # XXX This test is oversensitive to formatting
1090             expected = "Connected to <span>%d</span>\n     of <span>%d</span> known storage servers:" % (self.numclients, self.numclients)
1091             self.failUnless(expected in page,
1092                             "I didn't see the right 'connected storage servers'"
1093                             " message in: %s" % page
1094                             )
1095             expected = "<th>My nodeid:</th> <td class=\"nodeid mine data-chars\">%s</td>" % (b32encode(self.clients[0].nodeid).lower(),)
1096             self.failUnless(expected in page,
1097                             "I didn't see the right 'My nodeid' message "
1098                             "in: %s" % page)
1099             self.failUnless("Helper: 0 active uploads" in page)
1100         d.addCallback(_got_welcome)
1101         d.addCallback(self.log, "done with _got_welcome")
1102
1103         # get the welcome page from the node that uses the helper too
1104         d.addCallback(lambda res: getPage(self.helper_webish_url))
1105         def _got_welcome_helper(page):
1106             self.failUnless("Connected to helper?: <span>yes</span>" in page,
1107                             page)
1108             self.failUnless("Not running helper" in page)
1109         d.addCallback(_got_welcome_helper)
1110
1111         d.addCallback(lambda res: getPage(base + public))
1112         d.addCallback(lambda res: getPage(base + public + "/subdir1"))
1113         def _got_subdir1(page):
1114             # there ought to be an href for our file
1115             self.failUnlessIn('<td align="right">%d</td>' % len(self.data), page)
1116             self.failUnless(">mydata567</a>" in page)
1117         d.addCallback(_got_subdir1)
1118         d.addCallback(self.log, "done with _got_subdir1")
1119         d.addCallback(lambda res:
1120                       getPage(base + public + "/subdir1/mydata567"))
1121         def _got_data(page):
1122             self.failUnlessEqual(page, self.data)
1123         d.addCallback(_got_data)
1124
1125         # download from a URI embedded in a URL
1126         d.addCallback(self.log, "_get_from_uri")
1127         def _get_from_uri(res):
1128             return getPage(base + "uri/%s?filename=%s"
1129                            % (self.uri, "mydata567"))
1130         d.addCallback(_get_from_uri)
1131         def _got_from_uri(page):
1132             self.failUnlessEqual(page, self.data)
1133         d.addCallback(_got_from_uri)
1134
1135         # download from a URI embedded in a URL, second form
1136         d.addCallback(self.log, "_get_from_uri2")
1137         def _get_from_uri2(res):
1138             return getPage(base + "uri?uri=%s" % (self.uri,))
1139         d.addCallback(_get_from_uri2)
1140         d.addCallback(_got_from_uri)
1141
1142         # download from a bogus URI, make sure we get a reasonable error
1143         d.addCallback(self.log, "_get_from_bogus_uri", level=log.UNUSUAL)
1144         def _get_from_bogus_uri(res):
1145             d1 = getPage(base + "uri/%s?filename=%s"
1146                          % (self.mangle_uri(self.uri), "mydata567"))
1147             d1.addBoth(self.shouldFail, Error, "downloading bogus URI",
1148                        "410")
1149             return d1
1150         d.addCallback(_get_from_bogus_uri)
1151         d.addCallback(self.log, "_got_from_bogus_uri", level=log.UNUSUAL)
1152
1153         # upload a file with PUT
1154         d.addCallback(self.log, "about to try PUT")
1155         d.addCallback(lambda res: self.PUT(public + "/subdir3/new.txt",
1156                                            "new.txt contents"))
1157         d.addCallback(lambda res: self.GET(public + "/subdir3/new.txt"))
1158         d.addCallback(self.failUnlessEqual, "new.txt contents")
1159         # and again with something large enough to use multiple segments,
1160         # and hopefully trigger pauseProducing too
1161         def _new_happy_semantics(ign):
1162             for c in self.clients:
1163                 # these get reset somewhere? Whatever.
1164                 c.DEFAULT_ENCODING_PARAMETERS['happy'] = 1
1165         d.addCallback(_new_happy_semantics)
1166         d.addCallback(lambda res: self.PUT(public + "/subdir3/big.txt",
1167                                            "big" * 500000)) # 1.5MB
1168         d.addCallback(lambda res: self.GET(public + "/subdir3/big.txt"))
1169         d.addCallback(lambda res: self.failUnlessEqual(len(res), 1500000))
1170
1171         # can we replace files in place?
1172         d.addCallback(lambda res: self.PUT(public + "/subdir3/new.txt",
1173                                            "NEWER contents"))
1174         d.addCallback(lambda res: self.GET(public + "/subdir3/new.txt"))
1175         d.addCallback(self.failUnlessEqual, "NEWER contents")
1176
1177         # test unlinked POST
1178         d.addCallback(lambda res: self.POST("uri", t="upload",
1179                                             file=("new.txt", "data" * 10000)))
1180         # and again using the helper, which exercises different upload-status
1181         # display code
1182         d.addCallback(lambda res: self.POST("uri", use_helper=True, t="upload",
1183                                             file=("foo.txt", "data2" * 10000)))
1184
1185         # check that the status page exists
1186         d.addCallback(lambda res: self.GET("status", followRedirect=True))
1187         def _got_status(res):
1188             # find an interesting upload and download to look at. LIT files
1189             # are not interesting.
1190             h = self.clients[0].get_history()
1191             for ds in h.list_all_download_statuses():
1192                 if ds.get_size() > 200:
1193                     self._down_status = ds.get_counter()
1194             for us in h.list_all_upload_statuses():
1195                 if us.get_size() > 200:
1196                     self._up_status = us.get_counter()
1197             rs = list(h.list_all_retrieve_statuses())[0]
1198             self._retrieve_status = rs.get_counter()
1199             ps = list(h.list_all_publish_statuses())[0]
1200             self._publish_status = ps.get_counter()
1201             us = list(h.list_all_mapupdate_statuses())[0]
1202             self._update_status = us.get_counter()
1203
1204             # and that there are some upload- and download- status pages
1205             return self.GET("status/up-%d" % self._up_status)
1206         d.addCallback(_got_status)
1207         def _got_up(res):
1208             return self.GET("status/down-%d" % self._down_status)
1209         d.addCallback(_got_up)
1210         def _got_down(res):
1211             return self.GET("status/mapupdate-%d" % self._update_status)
1212         d.addCallback(_got_down)
1213         def _got_update(res):
1214             return self.GET("status/publish-%d" % self._publish_status)
1215         d.addCallback(_got_update)
1216         def _got_publish(res):
1217             return self.GET("status/retrieve-%d" % self._retrieve_status)
1218         d.addCallback(_got_publish)
1219
1220         # check that the helper status page exists
1221         d.addCallback(lambda res:
1222                       self.GET("helper_status", followRedirect=True))
1223         def _got_helper_status(res):
1224             self.failUnless("Bytes Fetched:" in res)
1225             # touch a couple of files in the helper's working directory to
1226             # exercise more code paths
1227             workdir = os.path.join(self.getdir("client0"), "helper")
1228             incfile = os.path.join(workdir, "CHK_incoming", "spurious")
1229             f = open(incfile, "wb")
1230             f.write("small file")
1231             f.close()
1232             then = time.time() - 86400*3
1233             now = time.time()
1234             os.utime(incfile, (now, then))
1235             encfile = os.path.join(workdir, "CHK_encoding", "spurious")
1236             f = open(encfile, "wb")
1237             f.write("less small file")
1238             f.close()
1239             os.utime(encfile, (now, then))
1240         d.addCallback(_got_helper_status)
1241         # and that the json form exists
1242         d.addCallback(lambda res:
1243                       self.GET("helper_status?t=json", followRedirect=True))
1244         def _got_helper_status_json(res):
1245             data = simplejson.loads(res)
1246             self.failUnlessEqual(data["chk_upload_helper.upload_need_upload"],
1247                                  1)
1248             self.failUnlessEqual(data["chk_upload_helper.incoming_count"], 1)
1249             self.failUnlessEqual(data["chk_upload_helper.incoming_size"], 10)
1250             self.failUnlessEqual(data["chk_upload_helper.incoming_size_old"],
1251                                  10)
1252             self.failUnlessEqual(data["chk_upload_helper.encoding_count"], 1)
1253             self.failUnlessEqual(data["chk_upload_helper.encoding_size"], 15)
1254             self.failUnlessEqual(data["chk_upload_helper.encoding_size_old"],
1255                                  15)
1256         d.addCallback(_got_helper_status_json)
1257
1258         # and check that client[3] (which uses a helper but does not run one
1259         # itself) doesn't explode when you ask for its status
1260         d.addCallback(lambda res: getPage(self.helper_webish_url + "status/"))
1261         def _got_non_helper_status(res):
1262             self.failUnless("Upload and Download Status" in res)
1263         d.addCallback(_got_non_helper_status)
1264
1265         # or for helper status with t=json
1266         d.addCallback(lambda res:
1267                       getPage(self.helper_webish_url + "helper_status?t=json"))
1268         def _got_non_helper_status_json(res):
1269             data = simplejson.loads(res)
1270             self.failUnlessEqual(data, {})
1271         d.addCallback(_got_non_helper_status_json)
1272
1273         # see if the statistics page exists
1274         d.addCallback(lambda res: self.GET("statistics"))
1275         def _got_stats(res):
1276             self.failUnless("Node Statistics" in res)
1277             self.failUnless("  'downloader.files_downloaded': 5," in res, res)
1278         d.addCallback(_got_stats)
1279         d.addCallback(lambda res: self.GET("statistics?t=json"))
1280         def _got_stats_json(res):
1281             data = simplejson.loads(res)
1282             self.failUnlessEqual(data["counters"]["uploader.files_uploaded"], 5)
1283             self.failUnlessEqual(data["stats"]["chk_upload_helper.upload_need_upload"], 1)
1284         d.addCallback(_got_stats_json)
1285
1286         # TODO: mangle the second segment of a file, to test errors that
1287         # occur after we've already sent some good data, which uses a
1288         # different error path.
1289
1290         # TODO: download a URI with a form
1291         # TODO: create a directory by using a form
1292         # TODO: upload by using a form on the directory page
1293         #    url = base + "somedir/subdir1/freeform_post!!upload"
1294         # TODO: delete a file by using a button on the directory page
1295
1296         return d
1297
1298     def _test_runner(self, res):
1299         # exercise some of the diagnostic tools in runner.py
1300
1301         # find a share
1302         for (dirpath, dirnames, filenames) in os.walk(unicode(self.basedir)):
1303             if "storage" not in dirpath:
1304                 continue
1305             if not filenames:
1306                 continue
1307             pieces = dirpath.split(os.sep)
1308             if (len(pieces) >= 4
1309                 and pieces[-4] == "storage"
1310                 and pieces[-3] == "shares"):
1311                 # we're sitting in .../storage/shares/$START/$SINDEX , and there
1312                 # are sharefiles here
1313                 filename = os.path.join(dirpath, filenames[0])
1314                 # peek at the magic to see if it is a chk share
1315                 magic = open(filename, "rb").read(4)
1316                 if magic == '\x00\x00\x00\x01':
1317                     break
1318         else:
1319             self.fail("unable to find any uri_extension files in %r"
1320                       % self.basedir)
1321         log.msg("test_system.SystemTest._test_runner using %r" % filename)
1322
1323         out,err = StringIO(), StringIO()
1324         rc = runner.runner(["debug", "dump-share", "--offsets",
1325                             unicode_to_argv(filename)],
1326                            stdout=out, stderr=err)
1327         output = out.getvalue()
1328         self.failUnlessEqual(rc, 0)
1329
1330         # we only upload a single file, so we can assert some things about
1331         # its size and shares.
1332         self.failUnlessIn("share filename: %s" % quote_output(abspath_expanduser_unicode(filename)), output)
1333         self.failUnlessIn("size: %d\n" % len(self.data), output)
1334         self.failUnlessIn("num_segments: 1\n", output)
1335         # segment_size is always a multiple of needed_shares
1336         self.failUnlessIn("segment_size: %d\n" % mathutil.next_multiple(len(self.data), 3), output)
1337         self.failUnlessIn("total_shares: 10\n", output)
1338         # keys which are supposed to be present
1339         for key in ("size", "num_segments", "segment_size",
1340                     "needed_shares", "total_shares",
1341                     "codec_name", "codec_params", "tail_codec_params",
1342                     #"plaintext_hash", "plaintext_root_hash",
1343                     "crypttext_hash", "crypttext_root_hash",
1344                     "share_root_hash", "UEB_hash"):
1345             self.failUnlessIn("%s: " % key, output)
1346         self.failUnlessIn("  verify-cap: URI:CHK-Verifier:", output)
1347
1348         # now use its storage index to find the other shares using the
1349         # 'find-shares' tool
1350         sharedir, shnum = os.path.split(filename)
1351         storagedir, storage_index_s = os.path.split(sharedir)
1352         storage_index_s = str(storage_index_s)
1353         out,err = StringIO(), StringIO()
1354         nodedirs = [self.getdir("client%d" % i) for i in range(self.numclients)]
1355         cmd = ["debug", "find-shares", storage_index_s] + nodedirs
1356         rc = runner.runner(cmd, stdout=out, stderr=err)
1357         self.failUnlessEqual(rc, 0)
1358         out.seek(0)
1359         sharefiles = [sfn.strip() for sfn in out.readlines()]
1360         self.failUnlessEqual(len(sharefiles), 10)
1361
1362         # also exercise the 'catalog-shares' tool
1363         out,err = StringIO(), StringIO()
1364         nodedirs = [self.getdir("client%d" % i) for i in range(self.numclients)]
1365         cmd = ["debug", "catalog-shares"] + nodedirs
1366         rc = runner.runner(cmd, stdout=out, stderr=err)
1367         self.failUnlessEqual(rc, 0)
1368         out.seek(0)
1369         descriptions = [sfn.strip() for sfn in out.readlines()]
1370         self.failUnlessEqual(len(descriptions), 30)
1371         matching = [line
1372                     for line in descriptions
1373                     if line.startswith("CHK %s " % storage_index_s)]
1374         self.failUnlessEqual(len(matching), 10)
1375
1376     def _test_control(self, res):
1377         # exercise the remote-control-the-client foolscap interfaces in
1378         # allmydata.control (mostly used for performance tests)
1379         c0 = self.clients[0]
1380         control_furl_file = os.path.join(c0.basedir, "private", "control.furl")
1381         control_furl = open(control_furl_file, "r").read().strip()
1382         # it doesn't really matter which Tub we use to connect to the client,
1383         # so let's just use our IntroducerNode's
1384         d = self.introducer.tub.getReference(control_furl)
1385         d.addCallback(self._test_control2, control_furl_file)
1386         return d
1387     def _test_control2(self, rref, filename):
1388         d = rref.callRemote("upload_from_file_to_uri",
1389                             filename.encode(get_filesystem_encoding()), convergence=None)
1390         downfile = os.path.join(self.basedir, "control.downfile").encode(get_filesystem_encoding())
1391         d.addCallback(lambda uri:
1392                       rref.callRemote("download_from_uri_to_file",
1393                                       uri, downfile))
1394         def _check(res):
1395             self.failUnlessEqual(res, downfile)
1396             data = open(downfile, "r").read()
1397             expected_data = open(filename, "r").read()
1398             self.failUnlessEqual(data, expected_data)
1399         d.addCallback(_check)
1400         d.addCallback(lambda res: rref.callRemote("speed_test", 1, 200, False))
1401         if sys.platform in ("linux2", "linux3"):
1402             d.addCallback(lambda res: rref.callRemote("get_memory_usage"))
1403         d.addCallback(lambda res: rref.callRemote("measure_peer_response_time"))
1404         return d
1405
1406     def _test_cli(self, res):
1407         # run various CLI commands (in a thread, since they use blocking
1408         # network calls)
1409
1410         private_uri = self._private_node.get_uri()
1411         client0_basedir = self.getdir("client0")
1412
1413         nodeargs = [
1414             "--node-directory", client0_basedir,
1415             ]
1416
1417         d = defer.succeed(None)
1418
1419         # for compatibility with earlier versions, private/root_dir.cap is
1420         # supposed to be treated as an alias named "tahoe:". Start by making
1421         # sure that works, before we add other aliases.
1422
1423         root_file = os.path.join(client0_basedir, "private", "root_dir.cap")
1424         f = open(root_file, "w")
1425         f.write(private_uri)
1426         f.close()
1427
1428         def run(ignored, verb, *args, **kwargs):
1429             stdin = kwargs.get("stdin", "")
1430             newargs = [verb] + nodeargs + list(args)
1431             return self._run_cli(newargs, stdin=stdin)
1432
1433         def _check_ls((out,err), expected_children, unexpected_children=[]):
1434             self.failUnlessEqual(err, "")
1435             for s in expected_children:
1436                 self.failUnless(s in out, (s,out))
1437             for s in unexpected_children:
1438                 self.failIf(s in out, (s,out))
1439
1440         def _check_ls_root((out,err)):
1441             self.failUnless("personal" in out)
1442             self.failUnless("s2-ro" in out)
1443             self.failUnless("s2-rw" in out)
1444             self.failUnlessEqual(err, "")
1445
1446         # this should reference private_uri
1447         d.addCallback(run, "ls")
1448         d.addCallback(_check_ls, ["personal", "s2-ro", "s2-rw"])
1449
1450         d.addCallback(run, "list-aliases")
1451         def _check_aliases_1((out,err)):
1452             self.failUnlessEqual(err, "")
1453             self.failUnlessEqual(out.strip(" \n"), "tahoe: %s" % private_uri)
1454         d.addCallback(_check_aliases_1)
1455
1456         # now that that's out of the way, remove root_dir.cap and work with
1457         # new files
1458         d.addCallback(lambda res: os.unlink(root_file))
1459         d.addCallback(run, "list-aliases")
1460         def _check_aliases_2((out,err)):
1461             self.failUnlessEqual(err, "")
1462             self.failUnlessEqual(out, "")
1463         d.addCallback(_check_aliases_2)
1464
1465         d.addCallback(run, "mkdir")
1466         def _got_dir( (out,err) ):
1467             self.failUnless(uri.from_string_dirnode(out.strip()))
1468             return out.strip()
1469         d.addCallback(_got_dir)
1470         d.addCallback(lambda newcap: run(None, "add-alias", "tahoe", newcap))
1471
1472         d.addCallback(run, "list-aliases")
1473         def _check_aliases_3((out,err)):
1474             self.failUnlessEqual(err, "")
1475             self.failUnless("tahoe: " in out)
1476         d.addCallback(_check_aliases_3)
1477
1478         def _check_empty_dir((out,err)):
1479             self.failUnlessEqual(out, "")
1480             self.failUnlessEqual(err, "")
1481         d.addCallback(run, "ls")
1482         d.addCallback(_check_empty_dir)
1483
1484         def _check_missing_dir((out,err)):
1485             # TODO: check that rc==2
1486             self.failUnlessEqual(out, "")
1487             self.failUnlessEqual(err, "No such file or directory\n")
1488         d.addCallback(run, "ls", "bogus")
1489         d.addCallback(_check_missing_dir)
1490
1491         files = []
1492         datas = []
1493         for i in range(10):
1494             fn = os.path.join(self.basedir, "file%d" % i)
1495             files.append(fn)
1496             data = "data to be uploaded: file%d\n" % i
1497             datas.append(data)
1498             open(fn,"wb").write(data)
1499
1500         def _check_stdout_against((out,err), filenum=None, data=None):
1501             self.failUnlessEqual(err, "")
1502             if filenum is not None:
1503                 self.failUnlessEqual(out, datas[filenum])
1504             if data is not None:
1505                 self.failUnlessEqual(out, data)
1506
1507         # test all both forms of put: from a file, and from stdin
1508         #  tahoe put bar FOO
1509         d.addCallback(run, "put", files[0], "tahoe-file0")
1510         def _put_out((out,err)):
1511             self.failUnless("URI:LIT:" in out, out)
1512             self.failUnless("201 Created" in err, err)
1513             uri0 = out.strip()
1514             return run(None, "get", uri0)
1515         d.addCallback(_put_out)
1516         d.addCallback(lambda (out,err): self.failUnlessEqual(out, datas[0]))
1517
1518         d.addCallback(run, "put", files[1], "subdir/tahoe-file1")
1519         #  tahoe put bar tahoe:FOO
1520         d.addCallback(run, "put", files[2], "tahoe:file2")
1521         d.addCallback(run, "put", "--mutable", files[3], "tahoe:file3")
1522         def _check_put_mutable((out,err)):
1523             self._mutable_file3_uri = out.strip()
1524         d.addCallback(_check_put_mutable)
1525         d.addCallback(run, "get", "tahoe:file3")
1526         d.addCallback(_check_stdout_against, 3)
1527
1528         #  tahoe put FOO
1529         STDIN_DATA = "This is the file to upload from stdin."
1530         d.addCallback(run, "put", "-", "tahoe-file-stdin", stdin=STDIN_DATA)
1531         #  tahoe put tahoe:FOO
1532         d.addCallback(run, "put", "-", "tahoe:from-stdin",
1533                       stdin="Other file from stdin.")
1534
1535         d.addCallback(run, "ls")
1536         d.addCallback(_check_ls, ["tahoe-file0", "file2", "file3", "subdir",
1537                                   "tahoe-file-stdin", "from-stdin"])
1538         d.addCallback(run, "ls", "subdir")
1539         d.addCallback(_check_ls, ["tahoe-file1"])
1540
1541         # tahoe mkdir FOO
1542         d.addCallback(run, "mkdir", "subdir2")
1543         d.addCallback(run, "ls")
1544         # TODO: extract the URI, set an alias with it
1545         d.addCallback(_check_ls, ["subdir2"])
1546
1547         # tahoe get: (to stdin and to a file)
1548         d.addCallback(run, "get", "tahoe-file0")
1549         d.addCallback(_check_stdout_against, 0)
1550         d.addCallback(run, "get", "tahoe:subdir/tahoe-file1")
1551         d.addCallback(_check_stdout_against, 1)
1552         outfile0 = os.path.join(self.basedir, "outfile0")
1553         d.addCallback(run, "get", "file2", outfile0)
1554         def _check_outfile0((out,err)):
1555             data = open(outfile0,"rb").read()
1556             self.failUnlessEqual(data, "data to be uploaded: file2\n")
1557         d.addCallback(_check_outfile0)
1558         outfile1 = os.path.join(self.basedir, "outfile0")
1559         d.addCallback(run, "get", "tahoe:subdir/tahoe-file1", outfile1)
1560         def _check_outfile1((out,err)):
1561             data = open(outfile1,"rb").read()
1562             self.failUnlessEqual(data, "data to be uploaded: file1\n")
1563         d.addCallback(_check_outfile1)
1564
1565         d.addCallback(run, "rm", "tahoe-file0")
1566         d.addCallback(run, "rm", "tahoe:file2")
1567         d.addCallback(run, "ls")
1568         d.addCallback(_check_ls, [], ["tahoe-file0", "file2"])
1569
1570         d.addCallback(run, "ls", "-l")
1571         def _check_ls_l((out,err)):
1572             lines = out.split("\n")
1573             for l in lines:
1574                 if "tahoe-file-stdin" in l:
1575                     self.failUnless(l.startswith("-r-- "), l)
1576                     self.failUnless(" %d " % len(STDIN_DATA) in l)
1577                 if "file3" in l:
1578                     self.failUnless(l.startswith("-rw- "), l) # mutable
1579         d.addCallback(_check_ls_l)
1580
1581         d.addCallback(run, "ls", "--uri")
1582         def _check_ls_uri((out,err)):
1583             lines = out.split("\n")
1584             for l in lines:
1585                 if "file3" in l:
1586                     self.failUnless(self._mutable_file3_uri in l)
1587         d.addCallback(_check_ls_uri)
1588
1589         d.addCallback(run, "ls", "--readonly-uri")
1590         def _check_ls_rouri((out,err)):
1591             lines = out.split("\n")
1592             for l in lines:
1593                 if "file3" in l:
1594                     rw_uri = self._mutable_file3_uri
1595                     u = uri.from_string_mutable_filenode(rw_uri)
1596                     ro_uri = u.get_readonly().to_string()
1597                     self.failUnless(ro_uri in l)
1598         d.addCallback(_check_ls_rouri)
1599
1600
1601         d.addCallback(run, "mv", "tahoe-file-stdin", "tahoe-moved")
1602         d.addCallback(run, "ls")
1603         d.addCallback(_check_ls, ["tahoe-moved"], ["tahoe-file-stdin"])
1604
1605         d.addCallback(run, "ln", "tahoe-moved", "newlink")
1606         d.addCallback(run, "ls")
1607         d.addCallback(_check_ls, ["tahoe-moved", "newlink"])
1608
1609         d.addCallback(run, "cp", "tahoe:file3", "tahoe:file3-copy")
1610         d.addCallback(run, "ls")
1611         d.addCallback(_check_ls, ["file3", "file3-copy"])
1612         d.addCallback(run, "get", "tahoe:file3-copy")
1613         d.addCallback(_check_stdout_against, 3)
1614
1615         # copy from disk into tahoe
1616         d.addCallback(run, "cp", files[4], "tahoe:file4")
1617         d.addCallback(run, "ls")
1618         d.addCallback(_check_ls, ["file3", "file3-copy", "file4"])
1619         d.addCallback(run, "get", "tahoe:file4")
1620         d.addCallback(_check_stdout_against, 4)
1621
1622         # copy from tahoe into disk
1623         target_filename = os.path.join(self.basedir, "file-out")
1624         d.addCallback(run, "cp", "tahoe:file4", target_filename)
1625         def _check_cp_out((out,err)):
1626             self.failUnless(os.path.exists(target_filename))
1627             got = open(target_filename,"rb").read()
1628             self.failUnlessEqual(got, datas[4])
1629         d.addCallback(_check_cp_out)
1630
1631         # copy from disk to disk (silly case)
1632         target2_filename = os.path.join(self.basedir, "file-out-copy")
1633         d.addCallback(run, "cp", target_filename, target2_filename)
1634         def _check_cp_out2((out,err)):
1635             self.failUnless(os.path.exists(target2_filename))
1636             got = open(target2_filename,"rb").read()
1637             self.failUnlessEqual(got, datas[4])
1638         d.addCallback(_check_cp_out2)
1639
1640         # copy from tahoe into disk, overwriting an existing file
1641         d.addCallback(run, "cp", "tahoe:file3", target_filename)
1642         def _check_cp_out3((out,err)):
1643             self.failUnless(os.path.exists(target_filename))
1644             got = open(target_filename,"rb").read()
1645             self.failUnlessEqual(got, datas[3])
1646         d.addCallback(_check_cp_out3)
1647
1648         # copy from disk into tahoe, overwriting an existing immutable file
1649         d.addCallback(run, "cp", files[5], "tahoe:file4")
1650         d.addCallback(run, "ls")
1651         d.addCallback(_check_ls, ["file3", "file3-copy", "file4"])
1652         d.addCallback(run, "get", "tahoe:file4")
1653         d.addCallback(_check_stdout_against, 5)
1654
1655         # copy from disk into tahoe, overwriting an existing mutable file
1656         d.addCallback(run, "cp", files[5], "tahoe:file3")
1657         d.addCallback(run, "ls")
1658         d.addCallback(_check_ls, ["file3", "file3-copy", "file4"])
1659         d.addCallback(run, "get", "tahoe:file3")
1660         d.addCallback(_check_stdout_against, 5)
1661
1662         # recursive copy: setup
1663         dn = os.path.join(self.basedir, "dir1")
1664         os.makedirs(dn)
1665         open(os.path.join(dn, "rfile1"), "wb").write("rfile1")
1666         open(os.path.join(dn, "rfile2"), "wb").write("rfile2")
1667         open(os.path.join(dn, "rfile3"), "wb").write("rfile3")
1668         sdn2 = os.path.join(dn, "subdir2")
1669         os.makedirs(sdn2)
1670         open(os.path.join(sdn2, "rfile4"), "wb").write("rfile4")
1671         open(os.path.join(sdn2, "rfile5"), "wb").write("rfile5")
1672
1673         # from disk into tahoe
1674         d.addCallback(run, "cp", "-r", dn, "tahoe:dir1")
1675         d.addCallback(run, "ls")
1676         d.addCallback(_check_ls, ["dir1"])
1677         d.addCallback(run, "ls", "dir1")
1678         d.addCallback(_check_ls, ["rfile1", "rfile2", "rfile3", "subdir2"],
1679                       ["rfile4", "rfile5"])
1680         d.addCallback(run, "ls", "tahoe:dir1/subdir2")
1681         d.addCallback(_check_ls, ["rfile4", "rfile5"],
1682                       ["rfile1", "rfile2", "rfile3"])
1683         d.addCallback(run, "get", "dir1/subdir2/rfile4")
1684         d.addCallback(_check_stdout_against, data="rfile4")
1685
1686         # and back out again
1687         dn_copy = os.path.join(self.basedir, "dir1-copy")
1688         d.addCallback(run, "cp", "--verbose", "-r", "tahoe:dir1", dn_copy)
1689         def _check_cp_r_out((out,err)):
1690             def _cmp(name):
1691                 old = open(os.path.join(dn, name), "rb").read()
1692                 newfn = os.path.join(dn_copy, name)
1693                 self.failUnless(os.path.exists(newfn))
1694                 new = open(newfn, "rb").read()
1695                 self.failUnlessEqual(old, new)
1696             _cmp("rfile1")
1697             _cmp("rfile2")
1698             _cmp("rfile3")
1699             _cmp(os.path.join("subdir2", "rfile4"))
1700             _cmp(os.path.join("subdir2", "rfile5"))
1701         d.addCallback(_check_cp_r_out)
1702
1703         # and copy it a second time, which ought to overwrite the same files
1704         d.addCallback(run, "cp", "-r", "tahoe:dir1", dn_copy)
1705
1706         # and again, only writing filecaps
1707         dn_copy2 = os.path.join(self.basedir, "dir1-copy-capsonly")
1708         d.addCallback(run, "cp", "-r", "--caps-only", "tahoe:dir1", dn_copy2)
1709         def _check_capsonly((out,err)):
1710             # these should all be LITs
1711             x = open(os.path.join(dn_copy2, "subdir2", "rfile4")).read()
1712             y = uri.from_string_filenode(x)
1713             self.failUnlessEqual(y.data, "rfile4")
1714         d.addCallback(_check_capsonly)
1715
1716         # and tahoe-to-tahoe
1717         d.addCallback(run, "cp", "-r", "tahoe:dir1", "tahoe:dir1-copy")
1718         d.addCallback(run, "ls")
1719         d.addCallback(_check_ls, ["dir1", "dir1-copy"])
1720         d.addCallback(run, "ls", "dir1-copy")
1721         d.addCallback(_check_ls, ["rfile1", "rfile2", "rfile3", "subdir2"],
1722                       ["rfile4", "rfile5"])
1723         d.addCallback(run, "ls", "tahoe:dir1-copy/subdir2")
1724         d.addCallback(_check_ls, ["rfile4", "rfile5"],
1725                       ["rfile1", "rfile2", "rfile3"])
1726         d.addCallback(run, "get", "dir1-copy/subdir2/rfile4")
1727         d.addCallback(_check_stdout_against, data="rfile4")
1728
1729         # and copy it a second time, which ought to overwrite the same files
1730         d.addCallback(run, "cp", "-r", "tahoe:dir1", "tahoe:dir1-copy")
1731
1732         # tahoe_ls doesn't currently handle the error correctly: it tries to
1733         # JSON-parse a traceback.
1734 ##         def _ls_missing(res):
1735 ##             argv = ["ls"] + nodeargs + ["bogus"]
1736 ##             return self._run_cli(argv)
1737 ##         d.addCallback(_ls_missing)
1738 ##         def _check_ls_missing((out,err)):
1739 ##             print "OUT", out
1740 ##             print "ERR", err
1741 ##             self.failUnlessEqual(err, "")
1742 ##         d.addCallback(_check_ls_missing)
1743
1744         return d
1745
1746     def test_filesystem_with_cli_in_subprocess(self):
1747         # We do this in a separate test so that test_filesystem doesn't skip if we can't run bin/tahoe.
1748
1749         self.basedir = "system/SystemTest/test_filesystem_with_cli_in_subprocess"
1750         d = self.set_up_nodes()
1751         def _new_happy_semantics(ign):
1752             for c in self.clients:
1753                 c.DEFAULT_ENCODING_PARAMETERS['happy'] = 1
1754         d.addCallback(_new_happy_semantics)
1755
1756         def _run_in_subprocess(ignored, verb, *args, **kwargs):
1757             stdin = kwargs.get("stdin")
1758             env = kwargs.get("env")
1759             newargs = [verb, "--node-directory", self.getdir("client0")] + list(args)
1760             return self.run_bintahoe(newargs, stdin=stdin, env=env)
1761
1762         def _check_succeeded(res, check_stderr=True):
1763             out, err, rc_or_sig = res
1764             self.failUnlessEqual(rc_or_sig, 0, str(res))
1765             if check_stderr:
1766                 self.failUnlessEqual(err, "")
1767
1768         d.addCallback(_run_in_subprocess, "create-alias", "newalias")
1769         d.addCallback(_check_succeeded)
1770
1771         STDIN_DATA = "This is the file to upload from stdin."
1772         d.addCallback(_run_in_subprocess, "put", "-", "newalias:tahoe-file", stdin=STDIN_DATA)
1773         d.addCallback(_check_succeeded, check_stderr=False)
1774
1775         def _mv_with_http_proxy(ign):
1776             env = os.environ
1777             env['http_proxy'] = env['HTTP_PROXY'] = "http://127.0.0.0:12345"  # invalid address
1778             return _run_in_subprocess(None, "mv", "newalias:tahoe-file", "newalias:tahoe-moved", env=env)
1779         d.addCallback(_mv_with_http_proxy)
1780         d.addCallback(_check_succeeded)
1781
1782         d.addCallback(_run_in_subprocess, "ls", "newalias:")
1783         def _check_ls(res):
1784             out, err, rc_or_sig = res
1785             self.failUnlessEqual(rc_or_sig, 0, str(res))
1786             self.failUnlessEqual(err, "", str(res))
1787             self.failUnlessIn("tahoe-moved", out)
1788             self.failIfIn("tahoe-file", out)
1789         d.addCallback(_check_ls)
1790         return d
1791
1792     def test_debug_trial(self):
1793         def _check_for_line(lines, result, test):
1794             for l in lines:
1795                 if result in l and test in l:
1796                     return
1797             self.fail("output (prefixed with '##') does not have a line containing both %r and %r:\n## %s"
1798                       % (result, test, "\n## ".join(lines)))
1799
1800         def _check_for_outcome(lines, out, outcome):
1801             self.failUnlessIn(outcome, out, "output (prefixed with '##') does not contain %r:\n## %s"
1802                                             % (outcome, "\n## ".join(lines)))
1803
1804         d = self.run_bintahoe(['debug', 'trial', '--reporter=verbose',
1805                                'allmydata.test.trialtest'])
1806         def _check_failure( (out, err, rc) ):
1807             self.failUnlessEqual(rc, 1)
1808             lines = out.split('\n')
1809             _check_for_line(lines, "[SKIPPED]", "test_skip")
1810             _check_for_line(lines, "[TODO]",    "test_todo")
1811             _check_for_line(lines, "[FAIL]",    "test_fail")
1812             _check_for_line(lines, "[ERROR]",   "test_deferred_error")
1813             _check_for_line(lines, "[ERROR]",   "test_error")
1814             _check_for_outcome(lines, out, "FAILED")
1815         d.addCallback(_check_failure)
1816
1817         # the --quiet argument regression-tests a problem in finding which arguments to pass to trial
1818         d.addCallback(lambda ign: self.run_bintahoe(['--quiet', 'debug', 'trial', '--reporter=verbose',
1819                                                      'allmydata.test.trialtest.Success']))
1820         def _check_success( (out, err, rc) ):
1821             self.failUnlessEqual(rc, 0)
1822             lines = out.split('\n')
1823             _check_for_line(lines, "[SKIPPED]", "test_skip")
1824             _check_for_line(lines, "[TODO]",    "test_todo")
1825             _check_for_outcome(lines, out, "PASSED")
1826         d.addCallback(_check_success)
1827         return d
1828
1829     def _run_cli(self, argv, stdin=""):
1830         #print "CLI:", argv
1831         stdout, stderr = StringIO(), StringIO()
1832         d = threads.deferToThread(runner.runner, argv, run_by_human=False,
1833                                   stdin=StringIO(stdin),
1834                                   stdout=stdout, stderr=stderr)
1835         def _done(res):
1836             return stdout.getvalue(), stderr.getvalue()
1837         d.addCallback(_done)
1838         return d
1839
1840     def _test_checker(self, res):
1841         ut = upload.Data("too big to be literal" * 200, convergence=None)
1842         d = self._personal_node.add_file(u"big file", ut)
1843
1844         d.addCallback(lambda res: self._personal_node.check(Monitor()))
1845         def _check_dirnode_results(r):
1846             self.failUnless(r.is_healthy())
1847         d.addCallback(_check_dirnode_results)
1848         d.addCallback(lambda res: self._personal_node.check(Monitor(), verify=True))
1849         d.addCallback(_check_dirnode_results)
1850
1851         d.addCallback(lambda res: self._personal_node.get(u"big file"))
1852         def _got_chk_filenode(n):
1853             self.failUnless(isinstance(n, ImmutableFileNode))
1854             d = n.check(Monitor())
1855             def _check_filenode_results(r):
1856                 self.failUnless(r.is_healthy())
1857             d.addCallback(_check_filenode_results)
1858             d.addCallback(lambda res: n.check(Monitor(), verify=True))
1859             d.addCallback(_check_filenode_results)
1860             return d
1861         d.addCallback(_got_chk_filenode)
1862
1863         d.addCallback(lambda res: self._personal_node.get(u"sekrit data"))
1864         def _got_lit_filenode(n):
1865             self.failUnless(isinstance(n, LiteralFileNode))
1866             d = n.check(Monitor())
1867             def _check_lit_filenode_results(r):
1868                 self.failUnlessEqual(r, None)
1869             d.addCallback(_check_lit_filenode_results)
1870             d.addCallback(lambda res: n.check(Monitor(), verify=True))
1871             d.addCallback(_check_lit_filenode_results)
1872             return d
1873         d.addCallback(_got_lit_filenode)
1874         return d