src/allmydata/test/test_node.py

   1
   2 import os, time
   3 from zope.interface import implements
   4 from twisted.trial import unittest
   5 from twisted.internet import defer
   6 from twisted.python import log
   7
   8 from foolscap import Tub, Referenceable
   9 from foolscap.eventual import fireEventually, flushEventualQueue
  10 from twisted.application import service
  11 import allmydata
  12 from allmydata.node import Node, formatTimeTahoeStyle
  13 from allmydata.util import testutil, fileutil
  14 from allmydata import logpublisher
  15
  16 class LoggingMultiService(service.MultiService):
  17     def log(self, msg, **kw):
  18         pass
  19
  20 class TestNode(Node):
  21     CERTFILE='DEFAULT_CERTFILE_BLANK'
  22     PORTNUMFILE='DEFAULT_PORTNUMFILE_BLANK'
  23
  24 class TestCase(unittest.TestCase, testutil.SignalMixin):
  25     def setUp(self):
  26         self.parent = LoggingMultiService()
  27         self.parent.startService()
  28     def tearDown(self):
  29         log.msg("%s.tearDown" % self.__class__.__name__)
  30         d = defer.succeed(None)
  31         d.addCallback(lambda res: self.parent.stopService())
  32         d.addCallback(flushEventualQueue)
  33         return d
  34
  35     def test_advertised_ip_addresses(self):
  36         basedir = "test_node/test_advertised_ip_addresses"
  37         fileutil.make_dirs(basedir)
  38         f = open(os.path.join(basedir, 'advertised_ip_addresses'),'w')
  39         f.write('1.2.3.4:5')
  40         f.close()
  41
  42         n = TestNode(basedir)
  43         n.setServiceParent(self.parent)
  44         d = n.when_tub_ready()
  45
  46         def _check_addresses(ignored_result):
  47             furl = n.tub.registerReference(n)
  48             self.failUnless("1.2.3.4:5" in furl, furl)
  49
  50         d.addCallback(_check_addresses)
  51         return d
  52
  53     def test_timestamp(self):
  54         # this modified logger doesn't seem to get used during the tests,
  55         # probably because we don't modify the LogObserver that trial
  56         # installs (only the one that twistd installs). So manually exercise
  57         # it a little bit.
  58         t = formatTimeTahoeStyle("ignored", time.time())
  59         self.failUnless("Z" in t)
  60         t2 = formatTimeTahoeStyle("ignored", int(time.time()))
  61         self.failUnless("Z" in t2)
  62
  63 class Gatherer(Referenceable):
  64     implements(logpublisher.RILogGatherer)
  65     def remote_logport(self, nodeid, logport):
  66         d = logport.callRemote("get_versions")
  67         d.addCallback(self.d.callback)
  68
  69 class LogObserver(Referenceable):
  70     implements(logpublisher.RILogObserver)
  71     def __init__(self):
  72         self.messages = []
  73     def remote_msg(self, d):
  74         self.messages.append(d)