-from base64 import b32encode
import os, re, sys, time, simplejson
from cStringIO import StringIO
d = getPage(base)
def _got_welcome(page):
html = page.replace('\n', ' ')
- connected_re = "Connected to <span>%d</span>[ ]*of <span>%d</span> known storage servers" % (self.numclients, self.numclients)
+ 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))
- nodeid_re = "<th>Node ID:</th>[ ]*<td>%s</td>" % (re.escape(b32encode(self.clients[0].nodeid).lower()),)
+ # 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(lambda res: getPage(self.helper_webish_url))
def _got_welcome_helper(page):
html = page.replace('\n', ' ')
- self.failUnless(re.search('<div class="status-indicator connected-yes"></div>[ ]*<div>Helper</div>', html), page)
+ 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)
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):