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_log(self):
  54         basedir = "test_node/test_log"
  55         fileutil.make_dirs(basedir)
  56         n = TestNode(basedir)
  57         n.log("this is a message")
  58         n.log("with %d %s %s", args=(2, "interpolated", "parameters"))
  59         n.log("with bogus %d expansion", args=("not an integer",))
  60
  61     def test_logpublisher(self):
  62         basedir = "test_node/test_logpublisher"
  63         fileutil.make_dirs(basedir)
  64         observer = LogObserver()
  65         n = TestNode(basedir)
  66         n.setServiceParent(self.parent)
  67         d = n.when_tub_ready()
  68         def _ready(res):
  69             n.log("starting up")
  70             flogport = open(os.path.join(n.basedir,"logport.furl"), "r").read()
  71             return n.tub.getReference(flogport.strip())
  72         d.addCallback(_ready)
  73         def _got_logport(logport):
  74             d = logport.callRemote("get_versions")
  75             def _check(versions):
  76                 self.failUnlessEqual(versions["allmydata"],
  77                                      allmydata.__version__)
  78             d.addCallback(_check)
  79             d.addCallback(lambda res:
  80                           logport.callRemote("subscribe_to_all", observer))
  81             def _emit(subscription):
  82                 self._subscription = subscription
  83                 log.msg("message here")
  84             d.addCallback(_emit)
  85             d.addCallback(fireEventually)
  86             d.addCallback(fireEventually)
  87             def _check_observer(res):
  88                 msgs = observer.messages
  89                 self.failUnlessEqual(len(msgs), 1)
  90                 #print msgs
  91                 self.failUnlessEqual(msgs[0]["message"], ("message here",) )
  92             d.addCallback(_check_observer)
  93             def _done(res):
  94                 return logport.callRemote("unsubscribe", self._subscription)
  95             d.addCallback(_done)
  96             return d
  97         d.addCallback(_got_logport)
  98         return d
  99
 100     def test_log_gatherer(self):
 101         t = Tub()
 102         t.setServiceParent(self.parent)
 103         t.listenOn("tcp:0:interface=127.0.0.1")
 104         l = t.getListeners()[0]
 105         portnum = l.getPortnum()
 106         t.setLocation("127.0.0.1:%d" % portnum)
 107         gatherer = Gatherer()
 108         gatherer.d = defer.Deferred()
 109         gatherer_furl = t.registerReference(gatherer)
 110
 111         basedir = "test_node/test_log_gatherer"
 112         fileutil.make_dirs(basedir)
 113         f = open(os.path.join(basedir, "log_gatherer.furl"), "w")
 114         f.write(gatherer_furl + "\n")
 115         f.close()
 116
 117         n = TestNode(basedir)
 118         n.setServiceParent(self.parent)
 119         d = n.when_tub_ready()
 120         def _ready(res):
 121             n.log("starting up")
 122             # about now, the node will be contacting the Gatherer and
 123             # offering its logport.
 124             return gatherer.d
 125         d.addCallback(_ready)
 126         return d
 127
 128     def test_timestamp(self):
 129         # this modified logger doesn't seem to get used during the tests,
 130         # probably because we don't modify the LogObserver that trial
 131         # installs (only the one that twistd installs). So manually exercise
 132         # it a little bit.
 133         t = formatTimeTahoeStyle("ignored", time.time())
 134         self.failUnless("Z" in t)
 135         t2 = formatTimeTahoeStyle("ignored", int(time.time()))
 136         self.failUnless("Z" in t2)
 137
 138 class Gatherer(Referenceable):
 139     implements(logpublisher.RILogGatherer)
 140     def remote_logport(self, nodeid, logport):
 141         d = logport.callRemote("get_versions")
 142         d.addCallback(self.d.callback)
 143
 144 class LogObserver(Referenceable):
 145     implements(logpublisher.RILogObserver)
 146     def __init__(self):
 147         self.messages = []
 148     def remote_msg(self, d):
 149         self.messages.append(d)