src/allmydata/test/test_node.py

   1
   2 import os, stat, sys, time
   3 from twisted.trial import unittest
   4 from twisted.internet import defer
   5 from twisted.python import log
   6
   7 from mock import patch
   8
   9 from foolscap.api import flushEventualQueue
  10 from twisted.application import service
  11 from allmydata.node import Node, formatTimeTahoeStyle
  12 from allmydata.util import fileutil
  13 import allmydata.test.common_util as testutil
  14
  15 class LoggingMultiService(service.MultiService):
  16     def log(self, msg, **kw):
  17         pass
  18
  19 class TestNode(Node):
  20     CERTFILE='DEFAULT_CERTFILE_BLANK'
  21     PORTNUMFILE='DEFAULT_PORTNUMFILE_BLANK'
  22
  23 class TestCase(testutil.SignalMixin, unittest.TestCase):
  24     def setUp(self):
  25         testutil.SignalMixin.setUp(self)
  26         self.parent = LoggingMultiService()
  27         self.parent.startService()
  28     def tearDown(self):
  29         log.msg("%s.tearDown" % self.__class__.__name__)
  30         testutil.SignalMixin.tearDown(self)
  31         d = defer.succeed(None)
  32         d.addCallback(lambda res: self.parent.stopService())
  33         d.addCallback(flushEventualQueue)
  34         return d
  35
  36     def test_location(self):
  37         basedir = "test_node/test_location"
  38         fileutil.make_dirs(basedir)
  39         f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
  40         f.write("[node]\n")
  41         f.write("tub.location = 1.2.3.4:5\n")
  42         f.close()
  43
  44         n = TestNode(basedir)
  45         n.setServiceParent(self.parent)
  46         d = n.when_tub_ready()
  47
  48         def _check_addresses(ignored_result):
  49             furl = n.tub.registerReference(n)
  50             self.failUnless("1.2.3.4:5" in furl, furl)
  51
  52         d.addCallback(_check_addresses)
  53         return d
  54
  55     def test_location2(self):
  56         basedir = "test_node/test_location2"
  57         fileutil.make_dirs(basedir)
  58         f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
  59         f.write("[node]\n")
  60         f.write("tub.location = 1.2.3.4:5,example.org:8091\n")
  61         f.close()
  62
  63         n = TestNode(basedir)
  64         n.setServiceParent(self.parent)
  65         d = n.when_tub_ready()
  66
  67         def _check_addresses(ignored_result):
  68             furl = n.tub.registerReference(n)
  69             self.failUnless("1.2.3.4:5" in furl, furl)
  70             self.failUnless("example.org:8091" in furl, furl)
  71
  72         d.addCallback(_check_addresses)
  73         return d
  74
  75     def test_tahoe_cfg_utf8(self):
  76         basedir = "test_node/test_tahoe_cfg_utf8"
  77         fileutil.make_dirs(basedir)
  78         f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
  79         f.write(u"\uFEFF[node]\n".encode('utf-8'))
  80         f.write(u"nickname = \u2621\n".encode('utf-8'))
  81         f.close()
  82
  83         n = TestNode(basedir)
  84         n.setServiceParent(self.parent)
  85         d = n.when_tub_ready()
  86         d.addCallback(lambda ign: self.failUnlessEqual(n.get_config("node", "nickname").decode('utf-8'),
  87                                                        u"\u2621"))
  88         return d
  89
  90     def test_timestamp(self):
  91         # this modified logger doesn't seem to get used during the tests,
  92         # probably because we don't modify the LogObserver that trial
  93         # installs (only the one that twistd installs). So manually exercise
  94         # it a little bit.
  95         t = formatTimeTahoeStyle("ignored", time.time())
  96         self.failUnless("Z" in t)
  97         t2 = formatTimeTahoeStyle("ignored", int(time.time()))
  98         self.failUnless("Z" in t2)
  99
 100     def test_secrets_dir(self):
 101         basedir = "test_node/test_secrets_dir"
 102         fileutil.make_dirs(basedir)
 103         n = TestNode(basedir)
 104         self.failUnless(isinstance(n, TestNode))
 105         self.failUnless(os.path.exists(os.path.join(basedir, "private")))
 106
 107     def test_secrets_dir_protected(self):
 108         if "win32" in sys.platform.lower() or "cygwin" in sys.platform.lower():
 109             # We don't know how to test that unprivileged users can't read this
 110             # thing.  (Also we don't know exactly how to set the permissions so
 111             # that unprivileged users can't read this thing.)
 112             raise unittest.SkipTest("We don't know how to set permissions on Windows.")
 113         basedir = "test_node/test_secrets_dir_protected"
 114         fileutil.make_dirs(basedir)
 115         n = TestNode(basedir)
 116         self.failUnless(isinstance(n, TestNode))
 117         privdir = os.path.join(basedir, "private")
 118         st = os.stat(privdir)
 119         bits = stat.S_IMODE(st[stat.ST_MODE])
 120         self.failUnless(bits & 0001 == 0, bits)
 121
 122     @patch("foolscap.logging.log.setLogDir")
 123     def test_logdir_is_str(self, mock_setLogDir):
 124         basedir = "test_node/test_logdir_is_str"
 125         fileutil.make_dirs(basedir)
 126
 127         def call_setLogDir(logdir):
 128             self.failUnless(isinstance(logdir, str), logdir)
 129         mock_setLogDir.side_effect = call_setLogDir
 130
 131         TestNode(basedir)
 132         self.failUnless(mock_setLogDir.called)