src/allmydata/test/test_node.py

   1
   2 import 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 testutil
  11
  12 class LoggingMultiService(service.MultiService):
  13     def log(self, msg):
  14         pass
  15
  16 class TestNode(Node):
  17     CERTFILE='DEFAULT_CERTFILE_BLANK'
  18     PORTNUMFILE='DEFAULT_PORTNUMFILE_BLANK'
  19
  20 class TestCase(unittest.TestCase, testutil.SignalMixin):
  21     def setUp(self):
  22         self.parent = LoggingMultiService()
  23         self.parent.startService()
  24     def tearDown(self):
  25         log.msg("%s.tearDown" % self.__class__.__name__)
  26         d = defer.succeed(None)
  27         d.addCallback(lambda res: self.parent.stopService())
  28         d.addCallback(flushEventualQueue)
  29         return d
  30
  31     def test_advertised_ip_addresses(self):
  32         open('advertised_ip_addresses','w').write('1.2.3.4:5')
  33
  34         n = TestNode()
  35         n.setServiceParent(self.parent)
  36         d = n.when_tub_ready()
  37
  38         def _check_addresses(ignored_result):
  39             self.failUnless("1.2.3.4:5" in n.tub.registerReference(n), n.tub.registerReference(n))
  40
  41         d.addCallback(_check_addresses)
  42         return d
  43
  44     def test_log(self):
  45         n = TestNode()
  46         n.log("this is a message")
  47         n.log("with %d %s %s", args=(2, "interpolated", "parameters"))
  48         n.log("with bogus %d expansion", args=("not an integer",))
  49
  50     def test_timestamp(self):
  51         # this modified logger doesn't seem to get used during the tests,
  52         # probably because we don't modify the LogObserver that trial
  53         # installs (only the one that twistd installs). So manually exercise
  54         # it a little bit.
  55         t = formatTimeTahoeStyle("ignored", time.time())
  56         self.failUnless("Z" in t)
  57         t2 = formatTimeTahoeStyle("ignored", int(time.time()))
  58         self.failUnless("Z" in t2)
  59