import os, stat, sys, time
+
from twisted.trial import unittest
from twisted.internet import defer
from twisted.python import log
-from foolscap.eventual import flushEventualQueue
+from foolscap.api import flushEventualQueue
+import foolscap.logging.log
+
from twisted.application import service
-from allmydata.node import Node, formatTimeTahoeStyle
-from allmydata.util import testutil, 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):
CERTFILE='DEFAULT_CERTFILE_BLANK'
PORTNUMFILE='DEFAULT_PORTNUMFILE_BLANK'
-class TestCase(unittest.TestCase, testutil.SignalMixin):
+class TestCase(testutil.SignalMixin, unittest.TestCase):
def setUp(self):
+ testutil.SignalMixin.setUp(self)
self.parent = LoggingMultiService()
self.parent.startService()
def tearDown(self):
log.msg("%s.tearDown" % self.__class__.__name__)
+ testutil.SignalMixin.tearDown(self)
d = defer.succeed(None)
d.addCallback(lambda res: self.parent.stopService())
d.addCallback(flushEventualQueue)
return d
- def test_advertised_ip_addresses(self):
- basedir = "test_node/test_advertised_ip_addresses"
+ 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, 'advertised_ip_addresses'),'w')
- f.write('1.2.3.4:5')
+ f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
+ f.write("[node]\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_advertised_ip_addresses2(self):
- basedir = "test_node/test_advertised_ip_addresses2"
+ 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):
+ 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(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()
- # this lets the 'port' file get written
- d.addCallback(lambda res: n.disownServiceParent())
- def _new_node(res):
- f = open(os.path.join(basedir, 'advertised_ip_addresses'),'w')
- f.write('1.2.3.4\n')
- f.write("6.7.8.9\n")
- f.close()
- n2 = self.node = TestNode(basedir)
- n2.setServiceParent(self.parent)
- return n2.when_tub_ready()
- d.addCallback(_new_node)
+ d.addCallback(lambda ign: self.failUnlessEqual(n.get_config("node", "nickname").decode('utf-8'),
+ u"\u2621"))
+ return d
- def _check_addresses(ignored_result):
- portfile = os.path.join(basedir, self.node.PORTNUMFILE)
- port = int(open(portfile, "r").read().strip())
- furl = self.node.tub.registerReference(n)
- self.failUnless(("1.2.3.4:%d" % port) in furl, furl)
- self.failUnless(("6.7.8.9:%d" % port) 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,
basedir = "test_node/test_secrets_dir"
fileutil.make_dirs(basedir)
n = TestNode(basedir)
+ self.failUnless(isinstance(n, TestNode))
self.failUnless(os.path.exists(os.path.join(basedir, "private")))
def test_secrets_dir_protected(self):
basedir = "test_node/test_secrets_dir_protected"
fileutil.make_dirs(basedir)
n = TestNode(basedir)
+ self.failUnless(isinstance(n, TestNode))
privdir = os.path.join(basedir, "private")
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)