import os, stat, sys, time
+
from twisted.trial import unittest
from twisted.internet import defer
from twisted.python import log
from foolscap.api import flushEventualQueue
+import foolscap.logging.log
+
from twisted.application import service
-from allmydata.node import Node, formatTimeTahoeStyle
-from allmydata.util import fileutil
+from allmydata.node import Node, formatTimeTahoeStyle, MissingConfigEntry
+from allmydata.util import fileutil, iputil
+from allmydata.util.namespace import Namespace
import allmydata.test.common_util as testutil
+
class LoggingMultiService(service.MultiService):
def log(self, msg, **kw):
pass
d.addCallback(flushEventualQueue)
return d
- def test_location(self):
- basedir = "test_node/test_location"
+ def _test_location(self, basedir, expected_addresses, tub_port=None, tub_location=None, local_addresses=None):
fileutil.make_dirs(basedir)
f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
f.write("[node]\n")
- f.write("tub.location = 1.2.3.4:5\n")
+ if tub_port:
+ f.write("tub.port = %d\n" % (tub_port,))
+ if tub_location is not None:
+ f.write("tub.location = %s\n" % (tub_location,))
f.close()
+ if local_addresses:
+ self.patch(iputil, 'get_local_addresses_async', lambda target=None: defer.succeed(local_addresses))
+
n = TestNode(basedir)
n.setServiceParent(self.parent)
d = n.when_tub_ready()
def _check_addresses(ignored_result):
furl = n.tub.registerReference(n)
- self.failUnless("1.2.3.4:5" in furl, furl)
+ for address in expected_addresses:
+ self.failUnlessIn(address, furl)
d.addCallback(_check_addresses)
return d
+ def test_location1(self):
+ return self._test_location(basedir="test_node/test_location1",
+ expected_addresses=["192.0.2.0:1234"],
+ tub_location="192.0.2.0:1234")
+
def test_location2(self):
- basedir = "test_node/test_location2"
+ return self._test_location(basedir="test_node/test_location2",
+ expected_addresses=["192.0.2.0:1234", "example.org:8091"],
+ tub_location="192.0.2.0:1234,example.org:8091")
+
+ def test_location_not_set(self):
+ """Checks the autogenerated furl when tub.location is not set."""
+ return self._test_location(basedir="test_node/test_location3",
+ expected_addresses=["127.0.0.1:1234", "192.0.2.0:1234"],
+ tub_port=1234,
+ local_addresses=["127.0.0.1", "192.0.2.0"])
+
+ def test_location_auto_and_explicit(self):
+ """Checks the autogenerated furl when tub.location contains 'AUTO'."""
+ return self._test_location(basedir="test_node/test_location4",
+ expected_addresses=["127.0.0.1:1234", "192.0.2.0:1234", "example.com:4321"],
+ tub_port=1234,
+ tub_location="AUTO,example.com:4321",
+ local_addresses=["127.0.0.1", "192.0.2.0", "example.com:4321"])
+
+ def test_tahoe_cfg_utf8(self):
+ basedir = "test_node/test_tahoe_cfg_utf8"
fileutil.make_dirs(basedir)
f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
- f.write("[node]\n")
- f.write("tub.location = 1.2.3.4:5,example.org:8091\n")
+ f.write(u"\uFEFF[node]\n".encode('utf-8'))
+ f.write(u"nickname = \u2621\n".encode('utf-8'))
f.close()
n = TestNode(basedir)
n.setServiceParent(self.parent)
d = n.when_tub_ready()
+ d.addCallback(lambda ign: self.failUnlessEqual(n.get_config("node", "nickname").decode('utf-8'),
+ u"\u2621"))
+ return d
- def _check_addresses(ignored_result):
- furl = n.tub.registerReference(n)
- self.failUnless("1.2.3.4:5" in furl, furl)
- self.failUnless("example.org:8091" in furl, furl)
+ def test_tahoe_cfg_hash_in_name(self):
+ basedir = "test_node/test_cfg_hash_in_name"
+ nickname = "Hash#Bang!" # a clever nickname containing a hash
+ fileutil.make_dirs(basedir)
+ f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
+ f.write("[node]\n")
+ f.write("nickname = %s\n" % (nickname,))
+ f.close()
+ n = TestNode(basedir)
+ self.failUnless(n.nickname == nickname)
- d.addCallback(_check_addresses)
- return d
+ def test_private_config(self):
+ basedir = "test_node/test_private_config"
+ privdir = os.path.join(basedir, "private")
+ fileutil.make_dirs(privdir)
+ f = open(os.path.join(privdir, 'already'), 'wt')
+ f.write("secret")
+ f.close()
+
+ n = TestNode(basedir)
+ self.failUnlessEqual(n.get_private_config("already"), "secret")
+ self.failUnlessEqual(n.get_private_config("not", "default"), "default")
+ self.failUnlessRaises(MissingConfigEntry, n.get_private_config, "not")
+ value = n.get_or_create_private_config("new", "start")
+ self.failUnlessEqual(value, "start")
+ self.failUnlessEqual(n.get_private_config("new"), "start")
+ counter = []
+ def make_newer():
+ counter.append("called")
+ return "newer"
+ value = n.get_or_create_private_config("newer", make_newer)
+ self.failUnlessEqual(len(counter), 1)
+ self.failUnlessEqual(value, "newer")
+ self.failUnlessEqual(n.get_private_config("newer"), "newer")
+
+ value = n.get_or_create_private_config("newer", make_newer)
+ self.failUnlessEqual(len(counter), 1) # don't call unless necessary
+ self.failUnlessEqual(value, "newer")
def test_timestamp(self):
# this modified logger doesn't seem to get used during the tests,
st = os.stat(privdir)
bits = stat.S_IMODE(st[stat.ST_MODE])
self.failUnless(bits & 0001 == 0, bits)
+
+ def test_logdir_is_str(self):
+ basedir = "test_node/test_logdir_is_str"
+ fileutil.make_dirs(basedir)
+
+ ns = Namespace()
+ ns.called = False
+ def call_setLogDir(logdir):
+ ns.called = True
+ self.failUnless(isinstance(logdir, str), logdir)
+ self.patch(foolscap.logging.log, 'setLogDir', call_setLogDir)
+
+ TestNode(basedir)
+ self.failUnless(ns.called)
projects / tahoe-lafs / tahoe-lafs.git / blobdiff