unknown_immcap = u"imm.lafs://immutable_from_the_future_imm_\u263A".encode('utf-8')
FAVICON_MARKUP = '<link href="/icon.png" rel="shortcut icon" />'
-
+DIR_HTML_TAG = '<html lang="en">'
class FakeStatsProvider:
def get_stats(self):
return []
class FakeDisplayableServer(StubServer):
- def __init__(self, serverid, nickname):
+ def __init__(self, serverid, nickname, connected,
+ last_connect_time, last_lost_time, last_rx_time):
StubServer.__init__(self, serverid)
self.announcement = {"my-version": "allmydata-tahoe-fake",
"service-name": "storage",
"nickname": nickname}
+ self.connected = connected
+ self.last_lost_time = last_lost_time
+ self.last_rx_time = last_rx_time
+ self.last_connect_time = last_connect_time
def is_connected(self):
- return True
+ return self.connected
def get_permutation_seed(self):
return ""
def get_remote_host(self):
return ""
def get_last_loss_time(self):
- return None
- def get_announcement_time(self):
- return None
+ return self.last_lost_time
+ def get_last_received_data_time(self):
+ return self.last_rx_time
+ def get_last_connect_time(self):
+ return self.last_connect_time
def get_announcement(self):
return self.announcement
def get_nickname(self):
return self.announcement["nickname"]
+ def get_available_space(self):
+ return 123456
class FakeBucketCounter(object):
def get_state(self):
self.storage_broker = StorageFarmBroker(None, permute_peers=True)
# fake knowledge of another server
self.storage_broker.test_add_server("other_nodeid",
- FakeDisplayableServer("other_nodeid", u"other_nickname \u263B"))
+ FakeDisplayableServer(
+ serverid="other_nodeid", nickname=u"other_nickname \u263B", connected = True,
+ last_connect_time = 10, last_lost_time = 20, last_rx_time = 30))
+ self.storage_broker.test_add_server("disconnected_nodeid",
+ FakeDisplayableServer(
+ serverid="other_nodeid", nickname=u"disconnected_nickname \u263B", connected = False,
+ last_connect_time = 15, last_lost_time = 25, last_rx_time = 35))
self.introducer_client = None
self.history = FakeHistory()
self.uploader = FakeUploader()
self.s.startService()
self.staticdir = self.mktemp()
self.clock = Clock()
+ self.fakeTime = 86460 # 1d 0h 1m 0s
self.ws = webish.WebishServer(self.s, "0", staticdir=self.staticdir,
- clock=self.clock)
+ clock=self.clock, now_fn=lambda:self.fakeTime)
self.ws.setServiceParent(self.s)
self.webish_port = self.ws.getPortnum()
self.webish_url = self.ws.getURL()
res_u = res.decode('utf-8')
self.failUnlessIn(u'<td>fake_nickname \u263A</td>', res_u)
self.failUnlessIn(u'<div class="nickname">other_nickname \u263B</div>', res_u)
+ self.failUnlessIn(u'Connected to <span>1</span>\n of <span>2</span> known storage servers', res_u)
+ self.failUnlessIn(u'<div class="status-indicator"><img src="img/connected-yes.png" alt="Connected" /></div>\n <a class="timestamp" title="1970-01-01 00:00:10">1d\u00A00h\u00A00m\u00A050s</a>', res_u)
+ self.failUnlessIn(u'<div class="status-indicator"><img src="img/connected-no.png" alt="Disconnected" /></div>\n <a class="timestamp" title="1970-01-01 00:00:25">1d\u00A00h\u00A00m\u00A035s</a>', res_u)
+ self.failUnlessIn(u'<td class="service-last-received-data"><a class="timestamp" title="1970-01-01 00:00:30">1d\u00A00h\u00A00m\u00A030s</a></td>', res_u)
+ self.failUnlessIn(u'<td class="service-last-received-data"><a class="timestamp" title="1970-01-01 00:00:35">1d\u00A00h\u00A00m\u00A025s</a></td>', res_u)
self.failUnlessIn(u'\u00A9 <a href="https://tahoe-lafs.org/">Tahoe-LAFS Software Foundation', res_u)
+ self.failUnlessIn('<td><h3>Available</h3></td>', res)
+ self.failUnlessIn('123.5kB', res)
self.s.basedir = 'web/test_welcome'
fileutil.make_dirs("web/test_welcome")
html = res.replace('\n', ' ')
self.failUnlessIn('<div class="furl">pb://someIntroducer/[censored]</div>', html)
self.failIfIn('pb://someIntroducer/secret', html)
- self.failUnless(re.search('<div class="status-indicator connected-no"></div>[ ]*<div>Introducer not connected</div>', html), res)
+ self.failUnless(re.search('<img src="img/connected-no.png" alt="Disconnected" />', html), res)
d.addCallback(_check_introducer_not_connected_unguessable)
# introducer connected, unguessable furl
html = res.replace('\n', ' ')
self.failUnlessIn('<div class="furl">pb://someIntroducer/[censored]</div>', html)
self.failIfIn('pb://someIntroducer/secret', html)
- self.failUnless(re.search('<div class="status-indicator connected-yes"></div>[ ]*<div>Introducer</div>', html), res)
+ self.failUnless(re.search('<img src="img/connected-yes.png" alt="Connected" />', html), res)
d.addCallback(_check_introducer_connected_unguessable)
# introducer connected, guessable furl
def _check_introducer_connected_guessable(res):
html = res.replace('\n', ' ')
self.failUnlessIn('<div class="furl">pb://someIntroducer/introducer</div>', html)
- self.failUnless(re.search('<div class="status-indicator connected-yes"></div>[ ]*<div>Introducer</div>', html), res)
+ self.failUnless(re.search('<img src="img/connected-yes.png" alt="Connected" />', html), res)
d.addCallback(_check_introducer_connected_guessable)
return d
d.addCallback(_set_no_helper)
def _check_no_helper(res):
html = res.replace('\n', ' ')
- self.failUnless(re.search('<div class="status-indicator connected-not-configured"></div>[ ]*<div>Helper</div>', html), res)
+ self.failUnless(re.search('<img src="img/connected-not-configured.png" alt="Not Configured" />', html), res)
d.addCallback(_check_no_helper)
# enable helper, not connected
html = res.replace('\n', ' ')
self.failUnlessIn('<div class="furl">pb://someHelper/[censored]</div>', html)
self.failIfIn('pb://someHelper/secret', html)
- self.failUnless(re.search('<div class="status-indicator connected-no"></div>[ ]*<div>Helper not connected</div>', html), res)
+ self.failUnless(re.search('<img src="img/connected-no.png" alt="Disconnected" />', html), res)
d.addCallback(_check_helper_not_connected)
# enable helper, connected
html = res.replace('\n', ' ')
self.failUnlessIn('<div class="furl">pb://someHelper/[censored]</div>', html)
self.failIfIn('pb://someHelper/secret', html)
- self.failUnless(re.search('<div class="status-indicator connected-yes"></div>[ ]*<div>Helper</div>', html), res)
+ self.failUnless(re.search('<img src="img/connected-yes.png" alt="Connected" />', html), res)
d.addCallback(_check_helper_connected)
return d
def test_GET_DIRECTORY_html(self):
d = self.GET(self.public_url + "/foo", followRedirect=True)
def _check(html):
- self.failUnlessIn('<div class="toolbar-item"><a href="../../..">Return to Welcome page</a></div>', html)
+ self.failUnlessIn('<li class="toolbar-item"><a href="../../..">Return to Welcome page</a></li>', html)
self._check_upload_and_mkdir_forms(html)
self.failUnlessIn("quux", html)
d.addCallback(_check)
d.addCallback(lambda res: self.GET(self.public_url + "/foo/empty/"))
def _check4(res):
self.failUnlessIn("directory is empty", res)
- MKDIR_BUTTON_RE=re.compile('<input type="hidden" name="t" value="mkdir" />.*<legend class="freeform-form-label">Create a new directory in this directory</legend>.*<input type="submit" value="Create" />', re.I)
+ MKDIR_BUTTON_RE=re.compile('<input type="hidden" name="t" value="mkdir" />.*<legend class="freeform-form-label">Create a new directory in this directory</legend>.*<input type="submit" class="btn" value="Create" />', re.I)
self.failUnless(MKDIR_BUTTON_RE.search(res), res)
d.addCallback(_check4)
def test_welcome(self):
basedir = "web.IntroducerWeb.test_welcome"
os.mkdir(basedir)
- fileutil.write(os.path.join(basedir, "tahoe.cfg"), "[node]\nweb.port = tcp:0\n")
+ cfg = "\n".join(["[node]",
+ "tub.location = 127.0.0.1:1",
+ "web.port = tcp:0",
+ ]) + "\n"
+ fileutil.write(os.path.join(basedir, "tahoe.cfg"), cfg)
self.node = IntroducerNode(basedir)
self.ws = self.node.getServiceNamed("webish")
self.basedir = "web/Grid/exceptions"
self.set_up_grid(num_clients=1, num_servers=2)
c0 = self.g.clients[0]
- c0.DEFAULT_ENCODING_PARAMETERS['happy'] = 2
+ c0.encoding_params['happy'] = 2
self.fileurls = {}
DATA = "data" * 100
d = c0.create_dirnode()
d.addCallback(lambda ignored: self.GET(self.fileurls["dir-0share"]))
def _check_0shares_dir_html(body):
- self.failUnlessIn("<html>", body)
+ self.failUnlessIn(DIR_HTML_TAG, body)
# we should see the regular page, but without the child table or
# the dirops forms
body = " ".join(body.strip().split())
# and some-shares like we did for immutable files (since there
# are different sorts of advice to offer in each case). For now,
# they present the same way.
- self.failUnlessIn("<html>", body)
+ self.failUnlessIn(DIR_HTML_TAG, body)
body = " ".join(body.strip().split())
self.failUnlessIn('href="?t=info">More info on this directory',
body)
d.addCallback(_stash_dir)
d.addCallback(lambda ign: self.GET(self.dir_url, followRedirect=True))
def _check_dir_html(body):
- self.failUnlessIn("<html>", body)
+ self.failUnlessIn(DIR_HTML_TAG, body)
self.failUnlessIn("blacklisted.txt</a>", body)
d.addCallback(_check_dir_html)
d.addCallback(lambda ign: self.GET(self.url))
# We should still be able to list the parent directory, in HTML...
d.addCallback(lambda ign: self.GET(self.dir_url, followRedirect=True))
def _check_dir_html2(body):
- self.failUnlessIn("<html>", body)
+ self.failUnlessIn(DIR_HTML_TAG, body)
self.failUnlessIn("blacklisted.txt</strike>", body)
d.addCallback(_check_dir_html2)
self.child_url = "uri/"+dn.get_readonly_uri()+"/child"
d.addCallback(_get_dircap)
d.addCallback(lambda ign: self.GET(self.dir_url_base, followRedirect=True))
- d.addCallback(lambda body: self.failUnlessIn("<html>", body))
+ d.addCallback(lambda body: self.failUnlessIn(DIR_HTML_TAG, body))
d.addCallback(lambda ign: self.GET(self.dir_url_json1))
d.addCallback(lambda res: simplejson.loads(res)) # just check it decodes
d.addCallback(lambda ign: self.GET(self.dir_url_json2))
projects / tahoe-lafs / tahoe-lafs.git / blobdiff