-from base64 import b32encode
-import os, sys, time, simplejson
+
+import os, re, sys, time, simplejson
from cStringIO import StringIO
+
from twisted.trial import unittest
from twisted.internet import defer
from twisted.internet import threads # CLI tests use deferToThread
from allmydata.immutable.filenode import ImmutableFileNode
from allmydata.util import idlib, mathutil
from allmydata.util import log, base32
+from allmydata.util.verlib import NormalizedVersion
from allmydata.util.encodingutil import quote_output, unicode_to_argv, get_filesystem_encoding
from allmydata.util.fileutil import abspath_expanduser_unicode
from allmydata.util.consumer import MemoryConsumer, download_to_data
from allmydata.mutable.common import NotWriteableError
from allmydata.mutable import layout as mutable_layout
from allmydata.mutable.publish import MutableData
+
+import foolscap
from foolscap.api import DeadReferenceError, fireEventually
from twisted.python.failure import Failure
from twisted.web.client import getPage
public = "uri/" + self._root_directory_uri
d = getPage(base)
def _got_welcome(page):
- # XXX This test is oversensitive to formatting
- expected = "Connected to <span>%d</span>\n of <span>%d</span> known storage servers:" % (self.numclients, self.numclients)
- self.failUnless(expected in page,
- "I didn't see the right 'connected storage servers'"
- " message in: %s" % page
- )
- expected = "<th>My nodeid:</th> <td class=\"nodeid mine data-chars\">%s</td>" % (b32encode(self.clients[0].nodeid).lower(),)
- self.failUnless(expected in page,
- "I didn't see the right 'My nodeid' message "
- "in: %s" % page)
+ html = page.replace('\n', ' ')
+ connected_re = r'Connected to <span>%d</span>\s*of <span>%d</span> known storage servers' % (self.numclients, self.numclients)
+ self.failUnless(re.search(connected_re, html),
+ "I didn't see the right '%s' message in:\n%s" % (connected_re, page))
+ # nodeids/tubids don't have any regexp-special characters
+ nodeid_re = r'<th>Node ID:</th>\s*<td title="TubID: %s">%s</td>' % (
+ self.clients[0].get_long_tubid(), self.clients[0].get_long_nodeid())
+ self.failUnless(re.search(nodeid_re, html),
+ "I didn't see the right '%s' message in:\n%s" % (nodeid_re, page))
self.failUnless("Helper: 0 active uploads" in page)
d.addCallback(_got_welcome)
d.addCallback(self.log, "done with _got_welcome")
# get the welcome page from the node that uses the helper too
d.addCallback(lambda res: getPage(self.helper_webish_url))
def _got_welcome_helper(page):
- self.failUnless("Connected to helper?: <span>yes</span>" in page,
- page)
- self.failUnless("Not running helper" in page)
+ html = page.replace('\n', ' ')
+ self.failUnless(re.search(r'<div class="status-indicator connected-yes"></div>\s*<div>Helper</div>', html), page)
+ self.failUnlessIn("Not running helper", page)
d.addCallback(_got_welcome_helper)
d.addCallback(lambda res: getPage(base + public))
# itself) doesn't explode when you ask for its status
d.addCallback(lambda res: getPage(self.helper_webish_url + "status/"))
def _got_non_helper_status(res):
- self.failUnless("Upload and Download Status" in res)
+ self.failUnlessIn("Recent and Active Operations", res)
d.addCallback(_got_non_helper_status)
# or for helper status with t=json
# see if the statistics page exists
d.addCallback(lambda res: self.GET("statistics"))
def _got_stats(res):
- self.failUnless("Node Statistics" in res)
- self.failUnless(" 'downloader.files_downloaded': 5," in res, res)
+ self.failUnlessIn("Operational Statistics", res)
+ self.failUnlessIn(" 'downloader.files_downloaded': 5,", res)
d.addCallback(_got_stats)
d.addCallback(lambda res: self.GET("statistics?t=json"))
def _got_stats_json(res):
def run(ignored, verb, *args, **kwargs):
stdin = kwargs.get("stdin", "")
- newargs = [verb] + nodeargs + list(args)
+ newargs = nodeargs + [verb] + list(args)
return self._run_cli(newargs, stdin=stdin)
def _check_ls((out,err), expected_children, unexpected_children=[]):
# tahoe_ls doesn't currently handle the error correctly: it tries to
# JSON-parse a traceback.
## def _ls_missing(res):
-## argv = ["ls"] + nodeargs + ["bogus"]
+## argv = nodeargs + ["ls", "bogus"]
## return self._run_cli(argv)
## d.addCallback(_ls_missing)
## def _check_ls_missing((out,err)):
def _run_in_subprocess(ignored, verb, *args, **kwargs):
stdin = kwargs.get("stdin")
env = kwargs.get("env")
- newargs = [verb, "--node-directory", self.getdir("client0")] + list(args)
+ newargs = ["--node-directory", self.getdir("client0"), verb] + list(args)
return self.run_bintahoe(newargs, stdin=stdin, env=env)
def _check_succeeded(res, check_stderr=True):
d.addCallback(_got_lit_filenode)
return d
+
class Connections(SystemTestMixin, unittest.TestCase):
def test_rref(self):
+ if NormalizedVersion(foolscap.__version__) < NormalizedVersion('0.6.4'):
+ raise unittest.SkipTest("skipped due to http://foolscap.lothar.com/trac/ticket/196 "
+ "(which does not affect normal usage of Tahoe-LAFS)")
+
self.basedir = "system/Connections/rref"
d = self.set_up_nodes(2)
def _start(ign):
self.c0 = self.clients[0]
- for s in self.c0.storage_broker.get_connected_servers():
- if "pub-"+s.get_longname() != self.c0.node_key_s:
- break
- self.s1 = s # s1 is the server, not c0
- self.s1_rref = s.get_rref()
+ nonclients = [s for s in self.c0.storage_broker.get_connected_servers()
+ if s.get_serverid() != self.c0.nodeid]
+ self.failUnlessEqual(len(nonclients), 1)
+
+ self.s1 = nonclients[0] # s1 is the server, not c0
+ self.s1_rref = self.s1.get_rref()
self.failIfEqual(self.s1_rref, None)
self.failUnless(self.s1.is_connected())
d.addCallback(_start)
projects / tahoe-lafs / tahoe-lafs.git / blobdiff