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 foolscap.eventual import flushEventualQueue
   8 from twisted.application import service
   9 from allmydata.node import Node, formatTimeTahoeStyle
  10 from allmydata.util import fileutil
  11 import common_util as testutil
  12
  13 class LoggingMultiService(service.MultiService):
  14     def log(self, msg, **kw):
  15         pass
  16
  17 class TestNode(Node):
  18     CERTFILE='DEFAULT_CERTFILE_BLANK'
  19     PORTNUMFILE='DEFAULT_PORTNUMFILE_BLANK'
  20
  21 class TestCase(unittest.TestCase, testutil.SignalMixin):
  22     def setUp(self):
  23         self.parent = LoggingMultiService()
  24         self.parent.startService()
  25     def tearDown(self):
  26         log.msg("%s.tearDown" % self.__class__.__name__)
  27         d = defer.succeed(None)
  28         d.addCallback(lambda res: self.parent.stopService())
  29         d.addCallback(flushEventualQueue)
  30         return d
  31
  32     def test_location(self):
  33         basedir = "test_node/test_location"
  34         fileutil.make_dirs(basedir)
  35         f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
  36         f.write("[node]\n")
  37         f.write("tub.location = 1.2.3.4:5\n")
  38         f.close()
  39
  40         n = TestNode(basedir)
  41         n.setServiceParent(self.parent)
  42         d = n.when_tub_ready()
  43
  44         def _check_addresses(ignored_result):
  45             furl = n.tub.registerReference(n)
  46             self.failUnless("1.2.3.4:5" in furl, furl)
  47
  48         d.addCallback(_check_addresses)
  49         return d
  50
  51     def test_location2(self):
  52         basedir = "test_node/test_location2"
  53         fileutil.make_dirs(basedir)
  54         f = open(os.path.join(basedir, 'tahoe.cfg'), 'wt')
  55         f.write("[node]\n")
  56         f.write("tub.location = 1.2.3.4:5,example.org:8091\n")
  57         f.close()
  58
  59         n = TestNode(basedir)
  60         n.setServiceParent(self.parent)
  61         d = n.when_tub_ready()
  62
  63         def _check_addresses(ignored_result):
  64             furl = n.tub.registerReference(n)
  65             self.failUnless("1.2.3.4:5" in furl, furl)
  66             self.failUnless("example.org:8091" in furl, furl)
  67
  68         d.addCallback(_check_addresses)
  69         return d
  70
  71     def test_timestamp(self):
  72         # this modified logger doesn't seem to get used during the tests,
  73         # probably because we don't modify the LogObserver that trial
  74         # installs (only the one that twistd installs). So manually exercise
  75         # it a little bit.
  76         t = formatTimeTahoeStyle("ignored", time.time())
  77         self.failUnless("Z" in t)
  78         t2 = formatTimeTahoeStyle("ignored", int(time.time()))
  79         self.failUnless("Z" in t2)
  80
  81     def test_secrets_dir(self):
  82         basedir = "test_node/test_secrets_dir"
  83         fileutil.make_dirs(basedir)
  84         n = TestNode(basedir)
  85         self.failUnless(os.path.exists(os.path.join(basedir, "private")))
  86
  87     def test_secrets_dir_protected(self):
  88         if "win32" in sys.platform.lower() or "cygwin" in sys.platform.lower():
  89             # We don't know how to test that unprivileged users can't read this
  90             # thing.  (Also we don't know exactly how to set the permissions so
  91             # that unprivileged users can't read this thing.)
  92             raise unittest.SkipTest("We don't know how to set permissions on Windows.")
  93         basedir = "test_node/test_secrets_dir_protected"
  94         fileutil.make_dirs(basedir)
  95         n = TestNode(basedir)
  96         privdir = os.path.join(basedir, "private")
  97         st = os.stat(privdir)
  98         bits = stat.S_IMODE(st[stat.ST_MODE])
  99         self.failUnless(bits & 0001 == 0, bits)